mirror of https://github.com/DJ2LS/FreeDATA.git
450 lines
17 KiB
Python
450 lines
17 KiB
Python
from freedata_server import helpers
|
|
import structlog
|
|
import time
|
|
from freedata_server.codec2 import FREEDV_MODE
|
|
from freedata_server.message_system_db_manager import DatabaseManager
|
|
from freedata_server.message_system_db_station import DatabaseManagerStations
|
|
from freedata_server.message_system_db_messages import DatabaseManagerMessages
|
|
|
|
from freedata_server import maidenhead
|
|
|
|
TESTMODE = False
|
|
|
|
|
|
class FrameHandler:
|
|
"""Base class for handling received frames.
|
|
|
|
This class provides common functionality for processing received frames,
|
|
including checking if the frame is addressed to the current station,
|
|
adding activity to the activity list, managing heard stations, emitting
|
|
events, and transmitting responses. Subclasses implement the
|
|
`follow_protocol` method to handle specific frame types and protocols.
|
|
"""
|
|
|
|
def __init__(self, ctx, name: str) -> None:
|
|
"""Initializes a new FrameHandler instance.
|
|
|
|
Args:
|
|
name (str): The name of the frame handler.
|
|
config (dict): The configuration dictionary.
|
|
states (StateManager): The state manager object.
|
|
event_manager (EventManager): The event manager object.
|
|
modem: The modem object.
|
|
"""
|
|
|
|
self.ctx = ctx
|
|
|
|
self.name = name
|
|
self.logger = structlog.get_logger("Frame Handler")
|
|
|
|
self.details = {
|
|
"frame": None,
|
|
"snr": 0,
|
|
"frequency_offset": 0,
|
|
"freedv_inst": None,
|
|
"bytes_per_frame": 0,
|
|
}
|
|
|
|
def is_frame_for_me(self):
|
|
"""Checks if the received frame is addressed to this station.
|
|
|
|
This method checks if the received frame is intended for this
|
|
station by verifying the destination callsign CRC and SSID against
|
|
the station's configured callsign and SSID list. It also checks for
|
|
session IDs in the case of ARQ and P2P frames.
|
|
|
|
Returns:
|
|
bool: True if the frame is for this station, False otherwise.
|
|
"""
|
|
call_with_ssid = (
|
|
self.ctx.config_manager.config["STATION"]["mycall"]
|
|
+ "-"
|
|
+ str(self.ctx.config_manager.config["STATION"]["myssid"])
|
|
)
|
|
ft = self.details["frame"]["frame_type"]
|
|
valid = False
|
|
|
|
# Check for callsign checksum
|
|
if ft in ["ARQ_SESSION_OPEN", "ARQ_SESSION_OPEN_ACK", "PING", "PING_ACK"]:
|
|
valid, mycallsign = helpers.check_callsign(
|
|
call_with_ssid,
|
|
self.details["frame"]["destination_crc"],
|
|
self.ctx.config_manager.config["STATION"]["ssid_list"],
|
|
)
|
|
|
|
# Check for session id on IRS side
|
|
elif ft in ["ARQ_SESSION_INFO", "ARQ_BURST_FRAME", "ARQ_STOP"]:
|
|
session_id = self.details["frame"]["session_id"]
|
|
if session_id in self.ctx.state_manager.arq_irs_sessions:
|
|
valid = True
|
|
|
|
# Check for session id on ISS side
|
|
elif ft in ["ARQ_SESSION_INFO_ACK", "ARQ_BURST_ACK", "ARQ_STOP_ACK"]:
|
|
session_id = self.details["frame"]["session_id"]
|
|
if session_id in self.ctx.state_manager.arq_iss_sessions:
|
|
valid = True
|
|
|
|
# check for p2p connection
|
|
elif ft in ["P2P_CONNECTION_CONNECT"]:
|
|
# Need to make sure this does not affect any other features in FreeDATA.
|
|
# This will allow the client to respond to any call sent in the "MYCALL" command
|
|
|
|
self.details["frame"]["destination_crc"] = helpers.get_crc_24(
|
|
self.details["frame"]["destination"]
|
|
)
|
|
if (
|
|
self.ctx.socket_interface_manager
|
|
and self.ctx.socket_interface_manager.socket_interface_callsigns
|
|
):
|
|
print("checking callsings....")
|
|
print(self.ctx.socket_interface_manager.socket_interface_callsigns)
|
|
for callsign in self.ctx.socket_interface_manager.socket_interface_callsigns:
|
|
print("check:", callsign)
|
|
valid, mycallsign = helpers.check_callsign(
|
|
callsign,
|
|
self.details["frame"]["destination_crc"].hex(),
|
|
self.ctx.config_manager.config["STATION"]["ssid_list"],
|
|
)
|
|
if valid is True:
|
|
break
|
|
else:
|
|
print("no socket interface manager")
|
|
valid, mycallsign = helpers.check_callsign(
|
|
call_with_ssid,
|
|
self.details["frame"]["destination_crc"].hex(),
|
|
self.ctx.config_manager.config["STATION"]["ssid_list"],
|
|
)
|
|
print("check done .... ")
|
|
|
|
# check for p2p connection
|
|
elif ft in [
|
|
"P2P_CONNECTION_CONNECT_ACK",
|
|
"P2P_CONNECTION_PAYLOAD",
|
|
"P2P_CONNECTION_PAYLOAD_ACK",
|
|
"P2P_CONNECTION_HEARTBEAT",
|
|
"P2P_CONNECTION_HEARTBEAT_ACK",
|
|
"P2P_CONNECTION_DISCONNECT",
|
|
"P2P_CONNECTION_DISCONNECT_ACK",
|
|
]:
|
|
session_id = self.details["frame"]["session_id"]
|
|
if session_id in self.ctx.state_manager.p2p_connection_sessions:
|
|
valid = True
|
|
|
|
else:
|
|
valid = False
|
|
|
|
if not valid:
|
|
self.logger.info(f"[Frame handler] {ft} received but not for us.")
|
|
return valid
|
|
|
|
def should_respond(self):
|
|
"""Checks if the frame handler should respond to the received frame.
|
|
|
|
This method simply calls is_frame_for_me() to determine if a
|
|
response is necessary. It can be overridden by subclasses to
|
|
implement more complex response logic.
|
|
|
|
Returns:
|
|
bool: True if the handler should respond, False otherwise.
|
|
"""
|
|
return self.is_frame_for_me()
|
|
|
|
def is_origin_on_blacklist(self):
|
|
"""Checks if the origin callsign is on the blacklist.
|
|
|
|
This method checks if the origin callsign of the received frame is
|
|
present in the callsign blacklist defined in the configuration.
|
|
It handles callsigns with SSIDs by removing the suffix and performs
|
|
a case-insensitive comparison.
|
|
|
|
Returns:
|
|
bool: True if the origin callsign is blacklisted, False otherwise.
|
|
"""
|
|
origin_callsign = self.details["frame"]["origin"]
|
|
|
|
for blacklist_callsign in self.ctx.config_manager.config["STATION"]["callsign_blacklist"]:
|
|
if (
|
|
helpers.get_crc_24(origin_callsign).hex()
|
|
== helpers.get_crc_24(blacklist_callsign).hex()
|
|
):
|
|
return True
|
|
|
|
if origin_callsign == blacklist_callsign or origin_callsign.startswith(
|
|
blacklist_callsign
|
|
):
|
|
return True
|
|
return False
|
|
|
|
def add_to_activity_list(self):
|
|
"""Adds the received frame to the activity list.
|
|
|
|
This method extracts relevant information from the received frame,
|
|
such as origin, destination, gridsquare, SNR, frequency offset,
|
|
activity type, session ID, and away-from-key status, and adds it
|
|
as a new activity to the state manager's activity list.
|
|
"""
|
|
frame = self.details["frame"]
|
|
|
|
activity = {
|
|
"direction": "received",
|
|
"snr": self.details["snr"],
|
|
"frequency_offset": self.details["frequency_offset"],
|
|
"activity_type": frame["frame_type"],
|
|
}
|
|
if "origin" in frame:
|
|
activity["origin"] = frame["origin"]
|
|
|
|
if "destination" in frame:
|
|
activity["destination"] = frame["destination"]
|
|
|
|
if "gridsquare" in frame:
|
|
activity["gridsquare"] = frame["gridsquare"]
|
|
|
|
if "session_id" in frame:
|
|
activity["session_id"] = frame["session_id"]
|
|
|
|
if "flag" in frame:
|
|
if "AWAY_FROM_KEY" in frame["flag"]:
|
|
activity["away_from_key"] = frame["flag"]["AWAY_FROM_KEY"]
|
|
|
|
self.ctx.state_manager.add_activity(activity)
|
|
|
|
def add_to_heard_stations(self):
|
|
"""Adds the received frame's origin station to the heard stations list.
|
|
|
|
This method extracts information from the received frame, including
|
|
callsign, gridsquare, signal strength, frequency offset, and
|
|
away-from-key status, and adds it to the heard stations list in the
|
|
state manager. It also calculates the distance between the current
|
|
station and the received station if gridsquares are available.
|
|
"""
|
|
frame = self.details["frame"]
|
|
|
|
if "origin" not in frame:
|
|
return
|
|
|
|
dxgrid = frame.get("gridsquare", "------")
|
|
|
|
# Initialize distance values
|
|
distance_km = None
|
|
distance_miles = None
|
|
if dxgrid != "------":
|
|
distance_dict = maidenhead.distance_between_locators(
|
|
self.ctx.config_manager.config["STATION"]["mygrid"], dxgrid
|
|
)
|
|
distance_km = distance_dict["kilometers"]
|
|
distance_miles = distance_dict["miles"]
|
|
|
|
away_from_key = False
|
|
if "flag" in self.details["frame"]:
|
|
if "AWAY_FROM_KEY" in self.details["frame"]["flag"]:
|
|
away_from_key = self.details["frame"]["flag"]["AWAY_FROM_KEY"]
|
|
|
|
helpers.add_to_heard_stations(
|
|
frame["origin"],
|
|
dxgrid,
|
|
self.name,
|
|
self.details["snr"],
|
|
self.details["frequency_offset"],
|
|
self.ctx.state_manager.radio_frequency,
|
|
self.ctx.state_manager.heard_stations,
|
|
distance_km=distance_km, # Pass the kilometer distance
|
|
distance_miles=distance_miles, # Pass the miles distance
|
|
away_from_key=away_from_key,
|
|
)
|
|
|
|
def make_event(self):
|
|
"""Creates a frame received event dictionary.
|
|
|
|
This method constructs a dictionary containing information about the
|
|
received frame, including timestamps, callsigns, gridsquares, signal
|
|
strength, and distance. This dictionary is used for emitting events
|
|
related to frame reception.
|
|
|
|
Returns:
|
|
dict: A dictionary containing the event data.
|
|
"""
|
|
|
|
event = {
|
|
"type": "frame-handler",
|
|
"received": self.details["frame"]["frame_type"],
|
|
"timestamp": int(time.time()),
|
|
"mycallsign": self.ctx.config_manager.config["STATION"]["mycall"],
|
|
"myssid": self.ctx.config_manager.config["STATION"]["myssid"],
|
|
"snr": str(self.details["snr"]),
|
|
}
|
|
if "origin" in self.details["frame"]:
|
|
event["dxcallsign"] = self.details["frame"]["origin"]
|
|
|
|
if "gridsquare" in self.details["frame"]:
|
|
event["gridsquare"] = self.details["frame"]["gridsquare"]
|
|
if event["gridsquare"] != "------":
|
|
distance = maidenhead.distance_between_locators(
|
|
self.ctx.config_manager.config["STATION"]["mygrid"],
|
|
self.details["frame"]["gridsquare"],
|
|
)
|
|
event["distance_kilometers"] = distance["kilometers"]
|
|
event["distance_miles"] = distance["miles"]
|
|
else:
|
|
event["distance_kilometers"] = 0
|
|
event["distance_miles"] = 0
|
|
|
|
if "flag" in self.details["frame"] and "AWAY_FROM_KEY" in self.details["frame"]["flag"]:
|
|
event["away_from_key"] = self.details["frame"]["flag"]["AWAY_FROM_KEY"]
|
|
|
|
return event
|
|
|
|
def emit_event(self):
|
|
"""Emits a frame received event.
|
|
|
|
This method creates an event dictionary containing information about
|
|
the received frame, such as the frame type, timestamp, callsigns,
|
|
gridsquare, SNR, distance, and away-from-key status. It then
|
|
broadcasts this event through the event manager.
|
|
"""
|
|
event_data = self.make_event()
|
|
print(event_data)
|
|
self.ctx.event_manager.broadcast(event_data)
|
|
|
|
def get_tx_mode(self):
|
|
"""Returns the transmission mode for acknowledgements.
|
|
|
|
This method returns the FreeDV mode used for transmitting
|
|
acknowledgement frames. Currently, it always returns the signalling
|
|
mode.
|
|
|
|
Returns:
|
|
FREEDV_MODE: The FreeDV mode for transmissions.
|
|
"""
|
|
return FREEDV_MODE.signalling
|
|
|
|
def transmit(self, frame):
|
|
"""Transmits a frame using the modem.
|
|
|
|
This method transmits the given frame using the modem. In test mode,
|
|
it broadcasts the frame through the event manager instead of using
|
|
the modem.
|
|
|
|
Args:
|
|
frame: The frame to transmit.
|
|
"""
|
|
if not TESTMODE:
|
|
self.ctx.rf_modem.transmit(self.get_tx_mode(), 1, 0, frame)
|
|
else:
|
|
self.ctx.event_manager.broadcast(frame)
|
|
|
|
def follow_protocol(self):
|
|
"""Handles protocol-specific actions for the received frame.
|
|
|
|
This method is intended to be overridden by subclasses to implement
|
|
specific protocol handling logic for different frame types. The base
|
|
implementation does nothing.
|
|
"""
|
|
pass
|
|
|
|
def log(self):
|
|
"""Logs the frame type being handled."""
|
|
self.logger.info(f"[Frame Handler] Handling frame {self.details['frame']['frame_type']}")
|
|
|
|
def handle(self, frame, snr, frequency_offset, freedv_inst, bytes_per_frame):
|
|
"""Handles a received frame.
|
|
|
|
This method processes the received frame, updates internal state,
|
|
performs blacklist checks, adds the frame to activity lists and heard
|
|
stations, emits an event, and calls the follow_protocol method for
|
|
subclass-specific handling.
|
|
|
|
Args:
|
|
frame (dict): The received frame data.
|
|
snr (float): The signal-to-noise ratio of the received frame.
|
|
frequency_offset (float): The frequency offset of the received frame.
|
|
freedv_inst: The FreeDV instance.
|
|
bytes_per_frame (int): The number of bytes per frame.
|
|
|
|
Returns:
|
|
bool: True if the frame was processed successfully, False if it was blocked due to blacklisting.
|
|
"""
|
|
|
|
self.details["frame"] = frame
|
|
self.details["snr"] = snr
|
|
self.details["frequency_offset"] = frequency_offset
|
|
self.details["freedv_inst"] = freedv_inst
|
|
self.details["bytes_per_frame"] = bytes_per_frame
|
|
|
|
print(self.details)
|
|
|
|
if "origin" not in self.details["frame"] and "session_id" in self.details["frame"]:
|
|
dxcall = self.ctx.state_manager.get_dxcall_by_session_id(
|
|
self.details["frame"]["session_id"]
|
|
)
|
|
if dxcall:
|
|
self.details["frame"]["origin"] = dxcall
|
|
|
|
# look in database for a full callsign if only crc is present
|
|
if "origin" not in self.details["frame"] and "origin_crc" in self.details["frame"]:
|
|
self.details["frame"]["origin"] = DatabaseManager(self.ctx).get_callsign_by_checksum(
|
|
frame["origin_crc"]
|
|
)
|
|
|
|
if (
|
|
"location" in self.details["frame"]
|
|
and "gridsquare" in self.details["frame"]["location"]
|
|
):
|
|
DatabaseManagerStations(self.ctx).update_station_location(
|
|
self.details["frame"]["origin"], frame["gridsquare"]
|
|
)
|
|
|
|
if "origin" in self.details["frame"]:
|
|
# try to find station info in database
|
|
try:
|
|
station = DatabaseManagerStations(self.ctx).get_station(
|
|
self.details["frame"]["origin"]
|
|
)
|
|
if station and station["location"] and "gridsquare" in station["location"]:
|
|
dxgrid = station["location"]["gridsquare"]
|
|
else:
|
|
dxgrid = "------"
|
|
|
|
# overwrite gridsquare only if not provided by frame
|
|
if "gridsquare" not in self.details["frame"]:
|
|
self.details["frame"]["gridsquare"] = dxgrid
|
|
|
|
except Exception as e:
|
|
self.logger.info(
|
|
f"[Frame Handler] Error getting gridsquare from callsign info: {e}"
|
|
)
|
|
|
|
# check if callsign is blacklisted
|
|
if self.ctx.config_manager.config["STATION"]["enable_callsign_blacklist"]:
|
|
if self.is_origin_on_blacklist():
|
|
self.logger.info(
|
|
f"[Frame Handler] Callsign blocked: {self.details['frame']['origin']}"
|
|
)
|
|
return False
|
|
|
|
self.log()
|
|
self.add_to_heard_stations()
|
|
self.add_to_activity_list()
|
|
self.emit_event()
|
|
self.follow_protocol()
|
|
return True
|
|
|
|
def check_for_queued_message(self):
|
|
"""Checks for queued messages to send.
|
|
|
|
This method checks if auto-repeat is enabled in the configuration
|
|
and if the received signal strength is above a certain threshold.
|
|
If both conditions are met, it sets any messages addressed to the
|
|
originating station to 'queued' status in the message database.
|
|
"""
|
|
|
|
# only check for queued messages, if we have enabled this and if we have a minimum snr received
|
|
if (
|
|
self.ctx.config_manager.config["MESSAGES"]["enable_auto_repeat"]
|
|
and self.details["snr"] >= -2
|
|
):
|
|
# set message to queued if beacon received
|
|
DatabaseManagerMessages(self.ctx).set_message_to_queued_for_callsign(
|
|
self.details["frame"]["origin"]
|
|
)
|