FreeDATA/modem/frame_dispatcher.py

275 lines
10 KiB
Python
Raw Normal View History

"""
FRAME DISPATCHER - We are dispatching the received frames to the needed functions
2023-11-23 09:14:11 +00:00
"""
2023-11-23 09:14:11 +00:00
import threading
2023-11-28 19:00:39 +00:00
import structlog
2023-11-23 09:14:11 +00:00
from modem_frametypes import FRAME_TYPE as FR_TYPE
import event_manager
from data_frame_factory import DataFrameFactory
#from deprecated_data_handler_broadcasts import BROADCAST
#from deprecated_data_handler_data_broadcasts import DATABROADCAST
#from deprecated_data_handler_ping import PING
2023-11-23 09:14:11 +00:00
from deprecated_protocol_arq_session_iss import ISS
from deprecated_protocol_arq_session_irs import IRS
from deprecated_protocol_arq_session import ARQ
from deprecated_protocol_arq_connection import SESSION
2023-11-23 09:14:11 +00:00
from frame_handler import FrameHandler
2023-11-29 23:15:16 +00:00
from frame_handler_ping import PingFrameHandler
from frame_handler_cq import CQFrameHandler
from frame_handler_arq_session import ARQFrameHandler
2023-11-23 09:14:11 +00:00
class DISPATCHER():
FRAME_HANDLER = {
FR_TYPE.ARQ_SESSION_OPEN_ACK_N.value: {"class": ARQFrameHandler, "name": "ARQ OPEN ACK (Narrow)"},
FR_TYPE.ARQ_SESSION_OPEN_ACK_W.value: {"class": ARQFrameHandler, "name": "ARQ OPEN ACK (Wide)"},
FR_TYPE.ARQ_SESSION_OPEN_N.value: {"class": ARQFrameHandler, "name": "ARQ Data Channel Open (Narrow)"},
FR_TYPE.ARQ_SESSION_OPEN_W.value: {"class": ARQFrameHandler, "name": "ARQ Data Channel Open (Wide)"},
FR_TYPE.ARQ_CONNECTION_CLOSE.value: {"class": ARQFrameHandler, "name": "ARQ CLOSE SESSION"},
FR_TYPE.ARQ_CONNECTION_HB.value: {"class": ARQFrameHandler, "name": "ARQ HEARTBEAT"},
FR_TYPE.ARQ_CONNECTION_OPEN.value: {"class": ARQFrameHandler, "name": "ARQ OPEN SESSION"},
2023-12-05 17:50:39 +00:00
FR_TYPE.ARQ_STOP.value: {"class": ARQFrameHandler, "name": "ARQ STOP TX"},
FR_TYPE.BEACON.value: {"class": FrameHandler, "name": "BEACON"},
FR_TYPE.BURST_ACK.value: {"class": FrameHandler, "name": "BURST ACK"},
FR_TYPE.BURST_NACK.value: {"class": FrameHandler, "name": "BURST NACK"},
FR_TYPE.CQ.value: {"class": CQFrameHandler, "name": "CQ"},
FR_TYPE.FR_ACK.value: {"class": FrameHandler, "name": "FRAME ACK"},
FR_TYPE.FR_NACK.value: {"class": FrameHandler, "name": "FRAME NACK"},
FR_TYPE.FR_REPEAT.value: {"class": FrameHandler, "name": "REPEAT REQUEST"},
FR_TYPE.PING_ACK.value: {"class": FrameHandler, "name": "PING ACK"},
2023-11-29 23:15:16 +00:00
FR_TYPE.PING.value: {"class": PingFrameHandler, "name": "PING"},
FR_TYPE.QRV.value: {"class": FrameHandler, "name": "QRV"},
FR_TYPE.IS_WRITING.value: {"class": FrameHandler, "name": "IS_WRITING"},
FR_TYPE.FEC.value: {"class": FrameHandler, "name": "FEC"},
FR_TYPE.FEC_WAKEUP.value: {"class": FrameHandler, "name": "FEC WAKEUP"},
}
def __init__(self, config, event_queue, states, data_q_rx, modem_tx_q):
2023-11-28 19:00:39 +00:00
self.log = structlog.get_logger("frame_dispatcher")
self.log.info("loading frame dispatcher.....\n")
2023-11-23 09:14:11 +00:00
self.config = config
self.event_queue = event_queue
self.states = states
self._initialize_handlers(config, event_queue, states)
#self._initialize_dispatchers()
2023-11-29 16:35:23 +00:00
self.data_queue_received = data_q_rx
self.modem_transmit_queue = modem_tx_q
2023-11-23 09:14:11 +00:00
self.arq_sessions = []
2023-11-23 09:14:11 +00:00
def _initialize_handlers(self, config, event_queue, states):
"""Initializes various data handlers."""
self.frame_factory = DataFrameFactory(config)
#self.broadcasts = BROADCAST(config, event_queue, states)
#self.data_broadcasts = DATABROADCAST(config, event_queue, states)
#self.ping = PING(config, event_queue, states)
2023-11-23 09:14:11 +00:00
self.arq = ARQ(config, event_queue, states)
self.arq_irs = IRS(config, event_queue, states)
self.arq_iss = ISS(config, event_queue, states)
self.arq_session = SESSION(config, event_queue, states)
self.event_manager = event_manager.EventManager([event_queue])
2023-11-26 13:51:58 +00:00
def start(self):
2023-11-23 09:14:11 +00:00
"""Starts worker threads for transmit and receive operations."""
threading.Thread(target=self.worker_receive, name="Receive Worker", daemon=True).start()
def worker_receive(self) -> None:
"""Queue received data for processing"""
while True:
data = self.data_queue_received.get()
self.new_process_data(
data['payload'],
data['freedv'],
data['bytes_per_frame'],
data['snr'],
data['frequency_offset'],
2023-11-23 09:14:11 +00:00
)
def new_process_data(self, bytes_out, freedv, bytes_per_frame: int, snr, offset) -> None:
# get frame as dictionary
deconstructed_frame = self.frame_factory.deconstruct(bytes_out)
frametype = deconstructed_frame["frame_type_int"]
if frametype not in self.FRAME_HANDLER:
self.log.warning(
"[Modem] ARQ - other frame type", frametype=FR_TYPE(frametype).name)
return
# instantiate handler
handler_class = self.FRAME_HANDLER[frametype]['class']
handler = handler_class(self.FRAME_HANDLER[frametype]['name'],
2023-11-29 23:15:16 +00:00
self.config,
self.states,
self.event_manager,
2023-12-05 17:50:39 +00:00
self.modem_transmit_queue)
2023-12-01 15:39:52 +00:00
handler.handle(deconstructed_frame, snr, offset, freedv, bytes_per_frame)
def get_id_from_frame(self, data):
if data[:1] in [FR_TYPE.ARQ_SESSION_OPEN_N, FR_TYPE.ARQ_SESSION_OPEN_W]:
return data[13:14]
return None
def initialize_arq_instance(self):
self.arq = ARQ(self.config, self.event_queue, self.states)
self.arq_irs = IRS(self.config, self.event_queue, self.states)
self.arq_iss = ISS(self.config, self.event_queue, self.states)
self.arq_session = SESSION(self.config, self.event_queue, self.states)
return {
'arq': self.arq,
'arq_irs': self.arq_irs,
'arq_iss': self.arq_iss,
'arq_session': self.arq_session
}
def initialize_arq_transmission_irs(self, data):
if id := self.get_id_from_frame(data):
instance = self.initialize_arq_instance()
self.states.register_arq_instance_by_id(id, instance)
instance['arq_irs'].arq_received_data_channel_opener()
def old_process_data(self, bytes_out, freedv, bytes_per_frame: int, snr) -> None:
2023-11-23 09:14:11 +00:00
"""
Process incoming data and decide what to do with the frame.
Args:
bytes_out:
freedv:
bytes_per_frame:
snr:
Returns:
"""
# get frame as dictionary
deconstructed_frame = self.frame_factory.deconstruct(bytes_out)
frametype = deconstructed_frame["frame_type_int"]
if frametype not in self.rx_dispatcher:
self.log.warning(
"[Modem] ARQ - other frame type", frametype=FR_TYPE(frametype).name)
return
# Process frametypes requiring a different set of arguments.
if FR_TYPE.BURST_51.value >= frametype >= FR_TYPE.BURST_01.value:
self.arq_irs.arq_data_received(
deconstructed_frame, bytes_per_frame, snr, freedv
)
# if we received the last frame of a burst or the last remaining rpt frame, do a modem unsync
# if self.arq_rx_burst_buffer.count(None) <= 1 or (frame+1) == n_frames_per_burst:
# self.log.debug(f"[Modem] LAST FRAME OF BURST --> UNSYNC {frame+1}/{n_frames_per_burst}")
# self.c_lib.freedv_set_sync(freedv, 0)
return
# TESTFRAMES
if frametype == FR_TYPE.TEST_FRAME.value:
self.log.debug("[Modem] TESTFRAME RECEIVED", frame=deconstructed_frame)
return
# Process frames "known" by rx_dispatcher
# self.log.debug(f"[Modem] {self.rx_dispatcher[frametype][1]} RECEIVED....")
self.rx_dispatcher[frametype][0](deconstructed_frame, snr)
def _initialize_dispatchers(self):
"""Initializes dispatchers for handling different frame types."""
# Dictionary of functions and log messages used in process_data
# instead of a long series of if-elif-else statements.
self.rx_dispatcher = {
FR_TYPE.ARQ_SESSION_OPEN_ACK_N.value: (
self.arq_iss.arq_received_channel_is_open,
"ARQ OPEN ACK (Narrow)",
),
FR_TYPE.ARQ_SESSION_OPEN_ACK_W.value: (
self.arq_iss.arq_received_channel_is_open,
"ARQ OPEN ACK (Wide)",
),
FR_TYPE.ARQ_SESSION_OPEN_N.value: (
self.initialize_arq_transmission_irs,
"ARQ Data Channel Open (Narrow)",
),
FR_TYPE.ARQ_SESSION_OPEN_W.value: (
self.initialize_arq_transmission_irs,
"ARQ Data Channel Open (Wide)",
),
FR_TYPE.ARQ_CONNECTION_CLOSE.value: (
self.arq_session.received_session_close,
"ARQ CLOSE SESSION",
),
FR_TYPE.ARQ_CONNECTION_HB.value: (
self.arq_session.received_session_heartbeat,
"ARQ HEARTBEAT",
),
FR_TYPE.ARQ_CONNECTION_OPEN.value: (
self.arq_session.received_session_opener,
"ARQ OPEN SESSION",
),
FR_TYPE.ARQ_STOP.value: (self.arq.received_stop_transmission, "ARQ STOP TX"),
FR_TYPE.BEACON.value: (self.broadcasts.received_beacon, "BEACON"),
FR_TYPE.BURST_ACK.value: (self.arq_iss.burst_ack_nack_received, "BURST ACK"),
FR_TYPE.BURST_NACK.value: (self.arq_iss.burst_ack_nack_received, "BURST NACK"),
FR_TYPE.CQ.value: (self.broadcasts.received_cq, "CQ"),
FR_TYPE.FR_ACK.value: (self.arq_iss.frame_ack_received, "FRAME ACK"),
FR_TYPE.FR_NACK.value: (self.arq_iss.frame_nack_received, "FRAME NACK"),
FR_TYPE.FR_REPEAT.value: (self.arq_iss.burst_rpt_received, "REPEAT REQUEST"),
FR_TYPE.PING_ACK.value: (self.ping.received_ping_ack, "PING ACK"),
FR_TYPE.PING.value: (self.ping.received_ping, "PING"),
FR_TYPE.QRV.value: (self.broadcasts.received_qrv, "QRV"),
FR_TYPE.IS_WRITING.value: (self.broadcasts.received_is_writing, "IS_WRITING"),
FR_TYPE.FEC.value: (self.data_broadcasts.received_fec, "FEC"),
FR_TYPE.FEC_WAKEUP.value: (self.data_broadcasts.received_fec_wakeup, "FEC WAKEUP"),
}