Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

extcap: make the code compatible with Python3.7+ #79

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

extcap: make the code compatible with Python3.7+ #79

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 19 additions & 13 deletions nrf802154_sniffer/nrf802154_sniffer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,9 @@
import sys
import os

if sys.version_info < (3, 7):
raise RuntimeError("This script requires Python 3.7 or higher.")

is_standalone = __name__ == "__main__"


Expand All @@ -59,6 +62,7 @@
from dataclasses import dataclass
from threading import Thread
from enum import IntEnum
from typing import List


@dataclass
Expand Down Expand Up @@ -108,7 +112,7 @@ def __init__(self, connection_open_timeout=None):
self.control_out = None
self.fifo = None
self.dlt = DLT.DLT_IEEE802_15_4_TAP
self.processes: list[Process] = []
self.processes: List[Process] = []
self.windows_mode = is_standalone and os.name == "nt"
self.first_local_timestamp = None
self.first_sniffer_timestamp = None
Expand Down Expand Up @@ -411,18 +415,20 @@ def _start(
fifo.write(self.pcap_header())
fifo.flush()

while packet := self.queue.get():
match packet:
case SnifferPacket(content, timestamp, lqi, rssi):
pcap = self.pcap_packet(
content, self.dlt, self.channel, rssi, lqi, self.correct_time(timestamp)
)
fifo.write(pcap)
case ExitEvent(reason):
if reason:
sys.stderr.write(reason)
self._stop()
break
while True:
packet = self.queue.get()
if packet is None:
break
if isinstance(packet, SnifferPacket):
pcap = self.pcap_packet(
content, self.dlt, self.channel, rssi, lqi, self.correct_time(timestamp)
)
fifo.write(pcap)
elif isinstance(packet, ExitEvent):
if packet.reason:
sys.stderr.write(packet.reason)
self._stop()
break
except BrokenPipeError:
self._stop()

Expand Down