dependencies update, migrate from kamene to scapy

hytera_homebrew_bridge/dmrlib/transmission.py

@@ -3,13 +3,13 @@
 from typing import List, Optional, Union
 
 from kaitaistruct import KaitaiStruct
-from kamene.layers.inet import IP
 from okdmr.dmrlib.etsi.fec.trellis import Trellis34
 from okdmr.kaitai.etsi.dmr_csbk import DmrCsbk
 from okdmr.kaitai.etsi.dmr_data import DmrData
 from okdmr.kaitai.etsi.dmr_data_header import DmrDataHeader
 from okdmr.kaitai.etsi.dmr_ip_udp import DmrIpUdp
 from okdmr.kaitai.etsi.full_link_control import FullLinkControl
+from scapy.layers.inet import IP
 
 from hytera_homebrew_bridge.dmrlib.burst_info import BurstInfo
 from hytera_homebrew_bridge.dmrlib.data_type import DataType

hytera_homebrew_bridge/tests/pcap_test.py

@@ -1,14 +1,19 @@
 #!/usr/bin/env python3
+import os
+import sys
 from argparse import ArgumentParser, Namespace
 
 from kaitaistruct import KaitaiStruct
-from kamene.all import *
-from kamene.layers.inet import UDP, IP
-from kamene.layers.inet6 import IPv6
+from okdmr.kaitai.hytera.ip_site_connect_heartbeat import IpSiteConnectHeartbeat
 from okdmr.kaitai.hytera.real_time_transport_protocol import RealTimeTransportProtocol
+from scapy.layers.inet import UDP, IP
+from scapy.layers.inet6 import IPv6
+from scapy.layers.l2 import Ether
+from scapy.utils import PcapReader
 
 from hytera_homebrew_bridge.dmrlib.packet_utils import try_parse_packet
 from hytera_homebrew_bridge.lib.packet_format import common_log_format
+from hytera_homebrew_bridge.tests.prettyprint import prettyprint
 
 try:
     import hytera_homebrew_bridge
@@ -62,14 +67,17 @@ def debug_udp_packet(
     known_packet: KaitaiStruct = try_parse_packet(udpdata=udp.load)
     if _args.no_rttp and isinstance(known_packet, RealTimeTransportProtocol):
         return
-    print(udp.load.hex())
-    if not known_packet and not hide_unknown:
-        print("Unknown UDP packet")
-        packet.display()
+    if isinstance(known_packet, IpSiteConnectHeartbeat):
+        return
+    if not known_packet:
+        if not hide_unknown:
+            print("Unknown UDP packet")
+            packet.display()
     else:
-        ip: IP = packet.getlayer(IP.name)
+        print(udp.load.hex())
+        ip: IP = packet.getlayer(IP)
         if not ip:
-            ip: IPv6 = packet.getlayer(IPv6.name)
+            ip: IPv6 = packet.getlayer(IPv6)
         logline = common_log_format(
             proto=known_packet.__class__.__name__,
             from_ip_port=(ip.src, udp.sport),
@@ -85,11 +93,12 @@ def debug_udp_packet(
 def read_pcap(filepath: str, _args: Namespace, hide_unknown: bool = False):
     with PcapReader(filepath) as pcap_reader:
         for pkt in pcap_reader:
-            if isinstance(pkt, Ether) and pkt.haslayer(UDP.name):
+            if isinstance(pkt, Ether) and pkt.haslayer(UDP):
                 debug_udp_packet(packet=pkt, _args=_args, hide_unknown=hide_unknown)
+            else:
+                print("not Ether / ", isinstance(pkt, Ether), pkt.haslayer(UDP))
 
 
 if __name__ == "__main__":
     args = arguments().parse_args(sys.argv[1:])
-    print(args)
     read_pcap(args.file, hide_unknown=args.hide_unknown, _args=args)

hytera_homebrew_bridge/tests/transmission_test.py

@@ -6,11 +6,11 @@
 from typing import Optional
 
 from kaitaistruct import KaitaiStruct
-from kamene.layers.inet import UDP
-from kamene.layers.l2 import Ether
-from kamene.utils import PcapReader
 from okdmr.kaitai.homebrew.mmdvm2020 import Mmdvm2020
 from okdmr.kaitai.hytera.ip_site_connect_protocol import IpSiteConnectProtocol
+from scapy.layers.inet import UDP
+from scapy.layers.l2 import Ether
+from scapy.utils import PcapReader
 
 from hytera_homebrew_bridge.dmrlib.packet_utils import try_parse_packet
 from hytera_homebrew_bridge.dmrlib.transmission_watcher import TransmissionWatcher
@@ -108,7 +108,7 @@ def feed_from_file(filepath: str, _watcher: TransmissionWatcher, _args: Namespac
 def feed_from_pcapng(filepath: str, _watcher: TransmissionWatcher, _args: Namespace):
     with PcapReader(filepath) as pcap_reader:
         for block in pcap_reader:
-            if isinstance(block, Ether) and block.haslayer(UDP.name):
+            if isinstance(block, Ether) and block.haslayer(UDP):
                 udp = block.getlayer(UDP)
                 if not udp:
                     continue

requirements.txt

@@ -2,14 +2,14 @@ configparser>=5.0.1
 puresnmp>=1.10.2
 asyncio>=3.4.3
 dmr-utils3>=0.1.29
-setuptools==60.7.1
+setuptools>=60.9.1
 bitarray>=2.3.5
-dmr-kaitai==0.8
-ok-dmrlib==0.2
+dmr-kaitai>=0.8
+ok-dmrlib>=0.2
 
 # developers / contributors
-kamene>=0.32
+scapy>=2.4.5
 uvloop>=0.15.2
 pytest
 pytest-cov
-pytest-asyncio==0.17.2
+pytest-asyncio>=0.18.1