jensandClaude Sonnet 4.6
53ed665f44
- network_backend.py -> backend/network.py - a_backend.py -> backend/a_backend.py - update all imports accordingly Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
135 lines
3.4 KiB
Python
135 lines
3.4 KiB
Python
import time
|
|
import signal
|
|
import argparse
|
|
|
|
from backend.network import NetworkBackend
|
|
from msg.frame_logger import FrameLogger
|
|
from transceiver import Transceiver
|
|
|
|
from nmea.packetizer import NmeaPacketizer
|
|
from nmea.messages import Gll, Gsa, Gsv, Gga, Rmc
|
|
|
|
from ubx.ubx import frame_create
|
|
from ubx.msg_types import *
|
|
from ubx.packetizer import UbxPacketizer
|
|
from ubx.messages import Sig, Sat, RawX, UbxMsg
|
|
from ubx.listener import UbxListener
|
|
|
|
|
|
def handler(signum, frame, be: NetworkBackend, loggers: list):
|
|
sig_name = signal.Signals(signum).name
|
|
print(f'Signal handler called with signal {sig_name} ({signum})')
|
|
be.stop()
|
|
be.disconnect()
|
|
for lg in loggers:
|
|
lg.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument('--log-dir', default='./log', help='Directory for frame log files (default: ./log)')
|
|
args = ap.parse_args()
|
|
|
|
server_list = [
|
|
{'name': " neo-f9p", 'host': "192.168.22.93", 'port': 8721},
|
|
{'name': "zed-x20p", 'host': "192.168.22.93", 'port': 8731}
|
|
]
|
|
|
|
transceiver = {}
|
|
loggers = []
|
|
|
|
# TCPIP multi source receiver
|
|
mc = NetworkBackend(server_list)
|
|
|
|
# Set the signal handler
|
|
signal.signal(signal.SIGINT, lambda signum, frame: handler(signum, frame, mc, loggers))
|
|
|
|
for grm in [" neo-f9p", "zed-x20p"]:
|
|
nmea_rx_list = [
|
|
Gll(grm, 'GN'),
|
|
Gsa(grm, 'GN'),
|
|
Gsv(grm, 'GP'),
|
|
Gga(grm, 'GP'),
|
|
Rmc(grm, 'GN'),
|
|
Gga(grm, 'GN'),
|
|
]
|
|
ubx_rx_list = [
|
|
UbxListener(Sig()),
|
|
UbxListener(Sat()),
|
|
UbxListener(RawX()),
|
|
|
|
UbxListener(UbxMsg(UBX_CLASS_ACK, UBX_ID_ACK_ACK)),
|
|
UbxListener(UbxMsg(UBX_CLASS_ACK, UBX_ID_ACK_NACK)),
|
|
]
|
|
|
|
# Topology
|
|
# Receiver-0.0 \
|
|
# Receiver-0.1 <- Packetizer-0 <- Transceiver-0 \
|
|
# / \
|
|
# Send-0->/ \
|
|
# + Backend
|
|
# Send-1->\ /
|
|
# \ /
|
|
# Receiver-1.0 <- Packetizer-1 <- Transceiver-1 /
|
|
# Receiver-1.1 /
|
|
|
|
# Create transceiver
|
|
transceiver[grm] = Transceiver(grm)
|
|
mc.register_xcvr(transceiver[grm])
|
|
|
|
# Create and register frame logger
|
|
lg = FrameLogger(grm, args.log_dir)
|
|
loggers.append(lg)
|
|
transceiver[grm].register_listener(lg)
|
|
|
|
# Create NMEA-Packetizer
|
|
nmea_packetizer = NmeaPacketizer()
|
|
ubx_packetizer = UbxPacketizer()
|
|
|
|
# Register NMEA-Packetizer to transceiver
|
|
transceiver[grm].register_listener(nmea_packetizer)
|
|
transceiver[grm].register_listener(ubx_packetizer)
|
|
|
|
# Register Receiver to NMEA-Packetizer
|
|
for rx in nmea_rx_list:
|
|
nmea_packetizer.register_listener(rx)
|
|
|
|
# Register Receiver to UBX-Packetizer
|
|
for rx in ubx_rx_list:
|
|
ubx_packetizer.register_listener(rx)
|
|
|
|
|
|
# Connect all sources
|
|
mc.connect()
|
|
|
|
# Start thread
|
|
mc.start()
|
|
|
|
while True:
|
|
try:
|
|
for grm in [" neo-f9p", "zed-x20p"]:
|
|
# Poll UBX-NAV-CLOCK (0x01 0x22)
|
|
frame = frame_create(UBX_CLASS_NAV, UBX_ID_CLOCK)
|
|
transceiver[grm].send(frame)
|
|
# Poll UBX-NAV-SIG (0x01 0x43)
|
|
frame = frame_create(UBX_CLASS_NAV, UBX_ID_NAV_SIG)
|
|
transceiver[grm].send(frame)
|
|
# Poll UBX-NAV-SAT (0x01 0x35)
|
|
frame = frame_create(UBX_CLASS_NAV, UBX_ID_NAV_SAT)
|
|
transceiver[grm].send(frame)
|
|
# Poll UBX-RXM-RAWX (0x02 0x15)
|
|
frame = frame_create(UBX_CLASS_RXM, UBX_ID_RXM_RAW)
|
|
transceiver[grm].send(frame)
|
|
except ValueError:
|
|
break
|
|
|
|
time.sleep(1)
|
|
|
|
# Stop thread
|
|
mc.stop()
|
|
|
|
# code never reached
|
|
mc.disconnect()
|
|
|
|
for lg in loggers:
|
|
lg.close() |