mirror of https://github.com/DJ2LS/FreeDATA.git
first basic work on NORM implementation
DJ2LS
parent
14f3d65fbd
commit
5433a001b4
|
|
@ -0,0 +1,47 @@
|
|||
import queue
|
||||
from command import TxCommand
|
||||
import api_validations
|
||||
import base64
|
||||
from queue import Queue
|
||||
import numpy as np
|
||||
import threading
|
||||
from norm.norm_transmission_iss import NormTransmissionISS
|
||||
|
||||
class Norm(TxCommand):
|
||||
def set_params_from_api(self, apiParams):
|
||||
self.origin = apiParams['origin']
|
||||
if not api_validations.validate_freedata_callsign(self.origin):
|
||||
self.origin = f"{self.origin}-0"
|
||||
|
||||
self.domain = apiParams['domain']
|
||||
if not api_validations.validate_freedata_callsign(self.domain):
|
||||
self.domain = f"{self.domain}-0"
|
||||
|
||||
self.data = base64.b64decode(apiParams['data'])
|
||||
|
||||
if 'priority' not in apiParams:
|
||||
self.priority = 1
|
||||
else:
|
||||
self.priority = apiParams['priority']
|
||||
|
||||
self.msgtype = apiParams['type']
|
||||
self.gridsquare = apiParams['gridsquare']
|
||||
|
||||
|
||||
def run(self):
|
||||
try:
|
||||
self.emit_event()
|
||||
self.logger.info(self.log_message())
|
||||
|
||||
# wait some random time and wait if we have an ongoing codec2 transmission
|
||||
# on our channel. This should prevent some packet collision
|
||||
random_delay = np.random.randint(0, 6)
|
||||
threading.Event().wait(random_delay)
|
||||
self.ctx.state_manager.channel_busy_condition_codec2.wait(0.5)
|
||||
|
||||
NormTransmissionISS(self.ctx, self.origin, self.domain, self.gridsquare, self.data, self.priority, self.msgtype).prepare_and_transmit()
|
||||
|
||||
except Exception as e:
|
||||
self.log(f"Error starting NORM transmission: {e}", isWarning=True)
|
||||
|
||||
return False
|
||||
|
|
@ -8,6 +8,7 @@ class DataFrameFactory:
|
|||
LENGTH_SIG0_FRAME = 14
|
||||
LENGTH_SIG1_FRAME = 14
|
||||
LENGTH_ACK_FRAME = 3
|
||||
LENGTH_NORM_FRAME = 126
|
||||
|
||||
"""
|
||||
helpers.set_flag(byte, 'DATA-ACK-NACK', True, FLAG_POSITIONS)
|
||||
|
|
@ -28,6 +29,10 @@ class DataFrameFactory:
|
|||
'ANNOUNCE_ARQ': 1, # Bit-position for announcing an ARQ session
|
||||
}
|
||||
|
||||
NORM_FLAGS = {
|
||||
'LAST_DATA': 0, # Bit-position for indicating the LAST DATA state
|
||||
}
|
||||
|
||||
def __init__(self, ctx):
|
||||
self.ctx = ctx
|
||||
|
||||
|
|
@ -41,6 +46,7 @@ class DataFrameFactory:
|
|||
self._load_ping_templates()
|
||||
self._load_arq_templates()
|
||||
self._load_p2p_connection_templates()
|
||||
self._load_norm_templates()
|
||||
|
||||
def _load_broadcast_templates(self):
|
||||
# cq frame
|
||||
|
|
@ -224,7 +230,50 @@ class DataFrameFactory:
|
|||
"session_id": 1,
|
||||
}
|
||||
|
||||
def _load_norm_templates(self):
|
||||
# data frame
|
||||
self.template_list[FR_TYPE.NORM_DATA.value] = {
|
||||
"frame_length": self.LENGTH_NORM_FRAME,
|
||||
"origin": 6,
|
||||
"domain": 6,
|
||||
"gridsquare": 4,
|
||||
"flag": 1,
|
||||
"timestamp": 4,
|
||||
"burst_info": 1,
|
||||
"payload_size": 1,
|
||||
"payload_data": 30
|
||||
}
|
||||
|
||||
# repair frame
|
||||
# FIXME
|
||||
self.template_list[FR_TYPE.NORM_REPAIR.value] = {
|
||||
"frame_length": self.LENGTH_NORM_FRAME,
|
||||
"origin": 6,
|
||||
"domain": 6,
|
||||
"flag": 1,
|
||||
"timestamp": 4,
|
||||
"burst_info": 1,
|
||||
"payload_size": 1,
|
||||
"payload_data": 34
|
||||
}
|
||||
|
||||
# nack frame
|
||||
# FIXME
|
||||
self.template_list[FR_TYPE.NORM_NACK.value] = {
|
||||
"frame_length": self.LENGTH_NORM_FRAME,
|
||||
"origin": 6,
|
||||
"domain": 4,
|
||||
"flag": 2
|
||||
}
|
||||
|
||||
# cmd frame
|
||||
# FIXME
|
||||
self.template_list[FR_TYPE.NORM_CMD.value] = {
|
||||
"frame_length": self.LENGTH_NORM_FRAME,
|
||||
"origin": 6,
|
||||
"domain": 4,
|
||||
"flag": 1
|
||||
}
|
||||
|
||||
def construct(self, frametype, content, frame_length = LENGTH_SIG1_FRAME):
|
||||
frame_template = self.template_list[frametype.value]
|
||||
|
|
@ -250,14 +299,13 @@ class DataFrameFactory:
|
|||
#print(item_length)
|
||||
#print(content)
|
||||
if buffer_position + item_length > frame_length:
|
||||
raise OverflowError("Frame data overflow!")
|
||||
raise OverflowError(f"Frame data overflow! {buffer_position + item_length} of max {frame_length}")
|
||||
frame[buffer_position: buffer_position + item_length] = content[key]
|
||||
buffer_position += item_length
|
||||
|
||||
return frame
|
||||
|
||||
def deconstruct(self, frame, mode_name=None):
|
||||
|
||||
buffer_position = 1
|
||||
# Handle the case where the frame type is not recognized
|
||||
#raise ValueError(f"Unknown frame type: {frametype}")
|
||||
|
|
@ -266,6 +314,9 @@ class DataFrameFactory:
|
|||
frame_template = self.template_list.get(frametype)
|
||||
frame = bytes([frametype]) + frame
|
||||
else:
|
||||
print("------------------------")
|
||||
print(frame)
|
||||
print(type(frame))
|
||||
# Extract frametype and get the corresponding template
|
||||
frametype = int.from_bytes(frame[:1], "big")
|
||||
frame_template = self.template_list.get(frametype)
|
||||
|
|
@ -284,9 +335,12 @@ class DataFrameFactory:
|
|||
data = frame[buffer_position: buffer_position + item_length]
|
||||
|
||||
# Process the data based on the key
|
||||
if key in ["origin", "destination"]:
|
||||
if key in ["origin", "destination", "domain"]:
|
||||
extracted_data[key] = helpers.bytes_to_callsign(data).decode()
|
||||
|
||||
elif key in ["payload_data"]:
|
||||
extracted_data[key] = data
|
||||
|
||||
elif key in ["origin_crc", "destination_crc", "total_crc"]:
|
||||
extracted_data[key] = data.hex()
|
||||
|
||||
|
|
@ -295,9 +349,9 @@ class DataFrameFactory:
|
|||
|
||||
elif key in ["session_id", "speed_level",
|
||||
"frames_per_burst", "version",
|
||||
"offset", "total_length", "state", "type", "maximum_bandwidth", "protocol_version"]:
|
||||
"offset", "total_length", "state", "type", "maximum_bandwidth", "protocol_version", "burst_info", "timestamp", "payload_size"]:
|
||||
extracted_data[key] = int.from_bytes(data, 'big')
|
||||
|
||||
print(key, data)
|
||||
elif key in ["snr"]:
|
||||
extracted_data[key] = helpers.snr_from_bytes(data)
|
||||
|
||||
|
|
@ -327,6 +381,14 @@ class DataFrameFactory:
|
|||
# get_flag returns True or False based on the bit value at the flag's position
|
||||
extracted_data[key][flag] = helpers.get_flag(data, flag, flag_dict)
|
||||
|
||||
if frametype in [FR_TYPE.NORM_DATA.value, FR_TYPE.NORM_NACK.value, FR_TYPE.NORM_REPAIR.value, FR_TYPE.NORM_CMD.value]:
|
||||
extracted_data[key] = data
|
||||
# flag_dict = self.NORM_FLAGS
|
||||
# for flag in flag_dict:
|
||||
# # Update extracted_data with the status of each flag
|
||||
# # get_flag returns True or False based on the bit value at the flag's position
|
||||
# extracted_data[key][flag] = helpers.get_flag(data, flag, flag_dict)
|
||||
|
||||
else:
|
||||
extracted_data[key] = data
|
||||
|
||||
|
|
@ -604,3 +666,29 @@ class DataFrameFactory:
|
|||
"session_id": session_id.to_bytes(1, 'big'),
|
||||
}
|
||||
return self.construct(FR_TYPE.P2P_CONNECTION_DISCONNECT_ACK, payload)
|
||||
|
||||
def build_norm_data(self, origin, domain, gridsquare, timestamp, burst_info, payload_size, payload_data, flag):
|
||||
|
||||
payload = {
|
||||
"origin": helpers.callsign_to_bytes(origin),
|
||||
"domain": helpers.callsign_to_bytes(domain),
|
||||
"gridsquare": helpers.encode_grid(gridsquare),
|
||||
"flag": flag.to_bytes(1, 'big'),
|
||||
"timestamp": timestamp.to_bytes(4, 'big'),
|
||||
"burst_info": burst_info.to_bytes(1, 'big'),
|
||||
"payload_size": payload_size.to_bytes(1, 'big'),
|
||||
"payload_data": payload_data,
|
||||
}
|
||||
return self.construct(FR_TYPE.NORM_DATA, payload)
|
||||
|
||||
|
||||
|
||||
|
||||
def build_norm_nack(self):
|
||||
pass
|
||||
|
||||
def build_norm_repair(self, origin, domain, timestamp, burst_info, payload_size, payload, flag=None):
|
||||
pass
|
||||
|
||||
def build_norm_cmd(self):
|
||||
pass
|
||||
|
|
@ -15,6 +15,7 @@ from frame_handler_cq import CQFrameHandler
|
|||
from frame_handler_arq_session import ARQFrameHandler
|
||||
from frame_handler_p2p_connection import P2PConnectionFrameHandler
|
||||
from frame_handler_beacon import BeaconFrameHandler
|
||||
from frame_handler_norm import NORMFrameHandler
|
||||
|
||||
|
||||
|
||||
|
|
@ -54,6 +55,9 @@ class DISPATCHER:
|
|||
FR_TYPE.PING_ACK.value: {"class": FrameHandler, "name": "PING ACK"},
|
||||
FR_TYPE.PING.value: {"class": PingFrameHandler, "name": "PING"},
|
||||
FR_TYPE.QRV.value: {"class": FrameHandler, "name": "QRV"},
|
||||
FR_TYPE.NORM_DATA.value: {"class": NORMFrameHandler, "name": "NORM DATA"},
|
||||
|
||||
|
||||
#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"},
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ from codec2 import FREEDV_MODE
|
|||
from message_system_db_manager import DatabaseManager
|
||||
from message_system_db_station import DatabaseManagerStations
|
||||
from message_system_db_messages import DatabaseManagerMessages
|
||||
|
||||
from collections.abc import Iterable
|
||||
import maidenhead
|
||||
|
||||
TESTMODE = False
|
||||
|
|
@ -81,6 +81,13 @@ class FrameHandler():
|
|||
if session_id in self.ctx.state_manager.arq_iss_sessions:
|
||||
valid = True
|
||||
|
||||
# check for NORM data
|
||||
elif ft in ['NORM_DATA']:
|
||||
# TODO
|
||||
# maybe we can add a list of domains, we are listening to in state manager?
|
||||
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.
|
||||
|
|
@ -184,7 +191,7 @@ class FrameHandler():
|
|||
if "session_id" in frame:
|
||||
activity["session_id"] = frame["session_id"]
|
||||
|
||||
if "flag" in frame:
|
||||
if "flag" in frame and isinstance(frame["flag"], (list, dict, Iterable)):
|
||||
if "AWAY_FROM_KEY" in frame["flag"]:
|
||||
activity["away_from_key"] = frame["flag"]["AWAY_FROM_KEY"]
|
||||
|
||||
|
|
@ -216,7 +223,7 @@ class FrameHandler():
|
|||
distance_miles = distance_dict['miles']
|
||||
|
||||
away_from_key = False
|
||||
if "flag" in self.details['frame']:
|
||||
if "flag" in self.details['frame'] and isinstance(frame["flag"], (list, dict, Iterable)):
|
||||
if "AWAY_FROM_KEY" in self.details['frame']["flag"]:
|
||||
away_from_key = self.details['frame']["flag"]["AWAY_FROM_KEY"]
|
||||
|
||||
|
|
@ -265,7 +272,9 @@ class FrameHandler():
|
|||
event['distance_kilometers'] = 0
|
||||
event['distance_miles'] = 0
|
||||
|
||||
if "flag" in self.details['frame'] and "AWAY_FROM_KEY" in self.details['frame']["flag"]:
|
||||
|
||||
|
||||
if "flag" in self.details and isinstance(self.details["flag"], (list, dict, Iterable)) and "AWAY_FROM_KEY" in self.details['frame']["flag"]:
|
||||
event['away_from_key'] = self.details['frame']["flag"]["AWAY_FROM_KEY"]
|
||||
|
||||
return event
|
||||
|
|
@ -279,7 +288,6 @@ class FrameHandler():
|
|||
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):
|
||||
|
|
@ -347,8 +355,6 @@ class FrameHandler():
|
|||
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:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,22 @@
|
|||
import threading
|
||||
|
||||
import frame_handler_ping
|
||||
import helpers
|
||||
import data_frame_factory
|
||||
import frame_handler
|
||||
from message_system_db_messages import DatabaseManagerMessages
|
||||
import numpy as np
|
||||
|
||||
from norm.norm_transmission_irs import NormTransmissionIRS
|
||||
|
||||
|
||||
class NORMFrameHandler(frame_handler.FrameHandler):
|
||||
|
||||
def follow_protocol(self):
|
||||
#self.logger.debug(f"[NORM] handling burst:{self.details}")
|
||||
|
||||
#origin = self.details["frame"]["origin"]
|
||||
#print(origin)
|
||||
|
||||
|
||||
NormTransmissionIRS(self.details["frame"])
|
||||
|
|
@ -25,6 +25,10 @@ class FRAME_TYPE(Enum):
|
|||
#MESH_BROADCAST = 100
|
||||
#MESH_SIGNALLING_PING = 101
|
||||
#MESH_SIGNALLING_PING_ACK = 102
|
||||
NORM_DATA = 100
|
||||
NORM_NACK = 101
|
||||
NORM_REPAIR = 102
|
||||
NORM_CMD = 103
|
||||
CQ = 200
|
||||
QRV = 201
|
||||
PING = 210
|
||||
|
|
|
|||
|
|
@ -0,0 +1,123 @@
|
|||
# base class for norm transmission
|
||||
|
||||
|
||||
import structlog
|
||||
import time
|
||||
import data_frame_factory
|
||||
from enum import IntEnum
|
||||
|
||||
class NORMMsgType(IntEnum):
|
||||
UNDEFINED = 0
|
||||
MESSAGE = 1 # Generic text/data message
|
||||
POSITION = 2 # GPS or grid locator info
|
||||
SITREP = 3 # Situation report
|
||||
PING = 4 # Ping or keepalive
|
||||
ACK = 5 # Acknowledgement
|
||||
COMMAND = 6 # Control or remote command
|
||||
STATUS = 7 # System or device status
|
||||
ALERT = 8 # High-priority broadcast
|
||||
|
||||
|
||||
class NORMMsgPriority(IntEnum):
|
||||
LOW = 0
|
||||
NORMAL = 1
|
||||
HIGH = 2
|
||||
CRITICAL = 3
|
||||
ALERT = 4
|
||||
EMERGENCY = 5
|
||||
|
||||
class NormTransmission:
|
||||
def __init__(self, ctx, origin, domain):
|
||||
self.logger = structlog.get_logger(type(self).__name__)
|
||||
self.ctx = ctx
|
||||
self.origin = origin
|
||||
self.domain = domain
|
||||
|
||||
self.frame_factory = data_frame_factory.DataFrameFactory(self.ctx)
|
||||
|
||||
def log(self, message, isWarning=False):
|
||||
"""Logs a message with session context.
|
||||
|
||||
Logs a message, including the class name, session ID, and current state,
|
||||
using the appropriate log level (warning or info).
|
||||
|
||||
Args:
|
||||
message: The message to be logged.
|
||||
isWarning: A boolean indicating whether the message should be logged as a warning.
|
||||
"""
|
||||
msg = f"[{type(self).__name__}][origin={self.origin},domain={self.domain}][state={self.state.name}]: {message}"
|
||||
logger = self.logger.warn if isWarning else self.logger.info
|
||||
logger(msg)
|
||||
|
||||
def set_state(self, state):
|
||||
|
||||
self.last_state_change_timestamp = time.time()
|
||||
if self.state == state:
|
||||
self.log(f"{type(self).__name__} state {self.state.name} unchanged.")
|
||||
else:
|
||||
self.log(f"{type(self).__name__} state change from {self.state.name} to {state.name} at {self.last_state_change_timestamp}")
|
||||
self.state = state
|
||||
|
||||
def on_frame_received(self, frame):
|
||||
"""Handles received frames based on the current session state.
|
||||
|
||||
This method processes incoming frames, triggering state transitions and
|
||||
data handling based on the frame type and current session state.
|
||||
It logs received frame types and ignores unknown state transitions.
|
||||
|
||||
Args:
|
||||
frame: The received frame.
|
||||
"""
|
||||
self.event_frame_received.set()
|
||||
self.log(f"Received {frame['frame_type']}")
|
||||
frame_type = frame['frame_type_int']
|
||||
if self.state in self.STATE_TRANSITION and frame_type in self.STATE_TRANSITION[self.state]:
|
||||
action_name = self.STATE_TRANSITION[self.state][frame_type]
|
||||
received_data, type_byte = getattr(self, action_name)(frame)
|
||||
|
||||
if isinstance(received_data, bytearray) and isinstance(type_byte, int):
|
||||
self.arq_data_type_handler.dispatch(type_byte, received_data,
|
||||
self.update_histograms(len(received_data), len(received_data)))
|
||||
return
|
||||
|
||||
self.log(f"Ignoring unknown transition from state {self.state.name} with frame {frame['frame_type']}")
|
||||
|
||||
def encode_flags(self, msg_type, priority, is_last):
|
||||
"""
|
||||
Encodes message type, priority and 'last burst' flag into a single byte.
|
||||
|
||||
Bit layout:
|
||||
Bit 7 → is_last (1 = letzter Burst)
|
||||
Bits 6–3 → msg_type (0–15)
|
||||
Bits 2–0 → priority (0–7)
|
||||
"""
|
||||
if isinstance(msg_type, IntEnum): # e.g., MsgType
|
||||
msg_type = int(msg_type)
|
||||
|
||||
assert 0 <= msg_type <= 15, "msg_type must be 0–15"
|
||||
assert 0 <= priority <= 7, "priority must be 0–7"
|
||||
|
||||
return ((1 if is_last else 0) << 7) | ((msg_type & 0x0F) << 3) | (priority & 0x07)
|
||||
|
||||
|
||||
def decode_flags(self, flags):
|
||||
"""
|
||||
Decodes a flags byte into (is_last, msg_type, priority).
|
||||
|
||||
Bit layout:
|
||||
Bit 7 → is_last
|
||||
Bits 6–3 → msg_type (0–15)
|
||||
Bits 2–0 → priority (0–7)
|
||||
"""
|
||||
is_last = bool((flags >> 7) & 0x01)
|
||||
msg_type = (flags >> 3) & 0x0F
|
||||
priority = flags & 0x07
|
||||
return is_last, msg_type, priority
|
||||
|
||||
def encode_burst_info(self, burst_number, total_bursts):
|
||||
return ((burst_number & 0x0F) << 4) | (total_bursts & 0x0F)
|
||||
|
||||
def decode_burst_info(self, burst_info):
|
||||
burst_number = (burst_info >> 4) & 0x0F
|
||||
burst_total = burst_info & 0x0F
|
||||
return burst_number, burst_total
|
||||
|
|
@ -0,0 +1,42 @@
|
|||
# file for handling received data
|
||||
from norm.norm_transmission import NormTransmission
|
||||
|
||||
class NormTransmissionIRS(NormTransmission):
|
||||
MAX_PAYLOAD_SIZE = 96
|
||||
|
||||
def __init__(self, frame):
|
||||
print("burst:", frame)
|
||||
|
||||
is_last, msg_type, priority = self.decode_flags(frame["flag"])
|
||||
burst_number, total_bursts = self.decode_burst_info(frame["burst_info"])
|
||||
payload_size = frame["payload_size"]
|
||||
payload_data = frame["payload_data"]
|
||||
self.origin = frame["origin"]
|
||||
self.domain = frame["domain"]
|
||||
self.gridsquare = frame["gridsquare"]
|
||||
|
||||
# FIXME
|
||||
#if payload_size > len(frame["payload_data"]) and total_bursts > 1:
|
||||
# payload_data = frame["payload_data"]
|
||||
#else:
|
||||
# payload_data = frame["payload_data"][:self.MAX_PAYLOAD_SIZE * burst_number]
|
||||
|
||||
|
||||
|
||||
|
||||
print("####################################")
|
||||
|
||||
print("payload_size:", payload_size)
|
||||
print("payload_data:", payload_data)
|
||||
|
||||
print("origin", self.origin)
|
||||
print("domain", self.domain)
|
||||
print("gridsquare", self.gridsquare)
|
||||
print("is_last", is_last)
|
||||
print("msg_type", msg_type)
|
||||
print("priority", priority)
|
||||
print("burst_number", burst_number)
|
||||
print("total_bursts", total_bursts)
|
||||
|
||||
# TODO:
|
||||
# add data to database or update it
|
||||
|
|
@ -0,0 +1,83 @@
|
|||
# file for handling transmitting data
|
||||
from norm.norm_transmission import NormTransmission
|
||||
from norm.norm_transmission import NORMMsgType, NORMMsgPriority
|
||||
from enum import Enum
|
||||
import time
|
||||
from codec2 import FREEDV_MODE
|
||||
|
||||
class NORM_ISS_State(Enum):
|
||||
NEW = 0
|
||||
TRANSMITTING = 1
|
||||
ENDED = 2
|
||||
FAILED = 3
|
||||
ABORTING = 4
|
||||
ABORTED = 5
|
||||
|
||||
class NormTransmissionISS(NormTransmission):
|
||||
MAX_PAYLOAD_SIZE = 96
|
||||
|
||||
def __init__(self, ctx, origin, domain, gridsquare, data, priority=NORMMsgPriority.NORMAL, message_type=NORMMsgType.UNDEFINED):
|
||||
|
||||
super().__init__(ctx, origin, domain)
|
||||
self.ctx = ctx
|
||||
self.origin = origin
|
||||
self.domain = domain
|
||||
self.gridsquare = gridsquare
|
||||
self.data = data
|
||||
self.priority = priority
|
||||
self.message_type = message_type
|
||||
self.payload_size = len(data)
|
||||
|
||||
self.timestamp = int(time.time())
|
||||
|
||||
self.state = NORM_ISS_State.NEW
|
||||
|
||||
self.log("Initialized")
|
||||
|
||||
def prepare_and_transmit(self):
|
||||
bursts = self.create_bursts()
|
||||
self.transmit_bursts(bursts)
|
||||
|
||||
def create_bursts(self):
|
||||
self.message_type = NORMMsgType.MESSAGE
|
||||
self.message_priority = NORMMsgPriority.NORMAL
|
||||
|
||||
|
||||
full_data = self.data
|
||||
|
||||
total_bursts = (len(full_data) + self.MAX_PAYLOAD_SIZE - 1) // self.MAX_PAYLOAD_SIZE
|
||||
bursts = []
|
||||
|
||||
for burst_number in range(1, total_bursts + 1):
|
||||
offset = (burst_number-1) * self.MAX_PAYLOAD_SIZE
|
||||
payload = full_data[offset: offset + self.MAX_PAYLOAD_SIZE]
|
||||
|
||||
burst_info = self.encode_burst_info(burst_number, total_bursts)
|
||||
|
||||
# set flag for last burst
|
||||
is_last = (burst_number == total_bursts - 1)
|
||||
flags = self.encode_flags(
|
||||
msg_type=self.message_type,
|
||||
priority=self.message_priority,
|
||||
is_last=is_last
|
||||
)
|
||||
|
||||
burst_frame = self.frame_factory.build_norm_data(
|
||||
origin=self.origin,
|
||||
domain=self.domain,
|
||||
gridsquare=self.gridsquare,
|
||||
timestamp=self.timestamp,
|
||||
burst_info=burst_info,
|
||||
payload_size=len(payload),
|
||||
payload_data=payload,
|
||||
flag=flags
|
||||
)
|
||||
print(burst_frame)
|
||||
bursts.append(burst_frame)
|
||||
|
||||
return bursts
|
||||
|
||||
def transmit_bursts(self, bursts):
|
||||
|
||||
for burst in bursts:
|
||||
self.ctx.rf_modem.transmit(FREEDV_MODE.datac4, 1, 100, burst)
|
||||
|
|
@ -0,0 +1 @@
|
|||
# file for "healing requested transmissions...
|
||||
|
|
@ -0,0 +1,174 @@
|
|||
import sys
|
||||
import time
|
||||
import unittest
|
||||
import unittest.mock
|
||||
import queue
|
||||
import threading
|
||||
import random
|
||||
import structlog
|
||||
import base64
|
||||
import numpy as np
|
||||
|
||||
sys.path.append('freedata_server')
|
||||
|
||||
from config import CONFIG
|
||||
from context import AppContext
|
||||
from event_manager import EventManager
|
||||
from state_manager import StateManager
|
||||
from data_frame_factory import DataFrameFactory
|
||||
from frame_dispatcher import DISPATCHER
|
||||
import codec2
|
||||
import command_norm
|
||||
|
||||
|
||||
class TestModem:
|
||||
def __init__(self, event_q, state_q):
|
||||
self.data_queue_received = queue.Queue()
|
||||
self.demodulator = unittest.mock.Mock()
|
||||
self.event_manager = EventManager([event_q])
|
||||
self.logger = structlog.get_logger('Modem')
|
||||
self.states = StateManager(state_q)
|
||||
|
||||
def getFrameTransmissionTime(self, mode):
|
||||
samples = 0
|
||||
c2instance = codec2.open_instance(mode.value)
|
||||
samples += codec2.api.freedv_get_n_tx_preamble_modem_samples(c2instance)
|
||||
samples += codec2.api.freedv_get_n_tx_modem_samples(c2instance)
|
||||
samples += codec2.api.freedv_get_n_tx_postamble_modem_samples(c2instance)
|
||||
time = samples / 8000
|
||||
return time
|
||||
|
||||
def transmit(self, mode, repeats: int, repeat_delay: int, frames: bytearray) -> bool:
|
||||
tx_time = self.getFrameTransmissionTime(mode) + 0.1
|
||||
self.logger.info(f"TX {tx_time} seconds...")
|
||||
threading.Event().wait(tx_time)
|
||||
|
||||
transmission = {
|
||||
'mode': mode,
|
||||
'bytes': frames,
|
||||
}
|
||||
self.data_queue_received.put(transmission)
|
||||
|
||||
|
||||
class TestMessageProtocol(unittest.TestCase):
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.logger = structlog.get_logger("TESTS")
|
||||
|
||||
# ISS
|
||||
|
||||
cls.ctx_ISS = AppContext('freedata_server/config.ini.example')
|
||||
cls.ctx_ISS.TESTMODE = True
|
||||
cls.ctx_ISS.startup()
|
||||
|
||||
|
||||
# IRS
|
||||
cls.ctx_IRS = AppContext('freedata_server/config.ini.example')
|
||||
cls.ctx_IRS.TESTMODE = True
|
||||
cls.ctx_IRS.startup()
|
||||
|
||||
# simulate a busy condition
|
||||
cls.ctx_IRS.state_manager.channel_busy_slot = [True, False, False, False, False]
|
||||
# Frame loss probability in %
|
||||
cls.loss_probability = 0
|
||||
|
||||
|
||||
cls.channels_running = False
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.ctx_IRS.shutdown()
|
||||
cls.ctx_ISS.shutdown()
|
||||
|
||||
|
||||
def channelWorker(self, ctx_a, ctx_b):
|
||||
while self.channels_running:
|
||||
try:
|
||||
# Station A gets the data from its transmit queue
|
||||
transmission = ctx_a.TESTMODE_TRANSMIT_QUEUE.get(timeout=1)
|
||||
print(f"Station A sending: {transmission[1]}", len(transmission[1]), transmission[0])
|
||||
|
||||
transmission[1] += bytes(2) # 2bytes crc simulation
|
||||
|
||||
if random.randint(0, 100) < self.loss_probability:
|
||||
self.logger.info(f"[{threading.current_thread().name}] Frame lost...")
|
||||
continue
|
||||
|
||||
# Forward data from Station A to Station B's receive queue
|
||||
if ctx_b:
|
||||
for burst in transmission:
|
||||
ctx_b.TESTMODE_RECEIVE_QUEUE.put(burst)
|
||||
self.logger.info(f"Data forwarded to Station B")
|
||||
|
||||
frame_bytes = transmission[1]
|
||||
if len(frame_bytes) == 5:
|
||||
mode_name = "SIGNALLING_ACK"
|
||||
else:
|
||||
mode_name = transmission[0]
|
||||
|
||||
snr = 15
|
||||
ctx_b.service_manager.frame_dispatcher.process_data(
|
||||
frame_bytes, None, len(frame_bytes), snr, 0, mode_name=mode_name
|
||||
)
|
||||
|
||||
except queue.Empty:
|
||||
continue
|
||||
|
||||
self.logger.info(f"[{threading.current_thread().name}] Channel closed.")
|
||||
|
||||
def waitForSession(self, event_queue, outbound=False):
|
||||
key = 'arq-transfer-outbound' if outbound else 'arq-transfer-inbound'
|
||||
while self.channels_running:
|
||||
try:
|
||||
ev = event_queue.get(timeout=2)
|
||||
if key in ev and ('success' in ev[key] or 'ABORTED' in ev[key]):
|
||||
self.logger.info(f"[{threading.current_thread().name}] {key} session ended.")
|
||||
break
|
||||
except queue.Empty:
|
||||
continue
|
||||
|
||||
def establishChannels(self):
|
||||
self.channels_running = True
|
||||
self.channelA = threading.Thread(target=self.channelWorker,args=[self.ctx_ISS, self.ctx_IRS],name = "channelA")
|
||||
self.channelA.start()
|
||||
|
||||
self.channelB = threading.Thread(target=self.channelWorker,args=[self.ctx_IRS, self.ctx_ISS],name = "channelB")
|
||||
self.channelB.start()
|
||||
|
||||
def waitAndCloseChannels(self):
|
||||
self.waitForSession(self.ctx_ISS.modem_events, True)
|
||||
self.channels_running = False
|
||||
self.waitForSession(self.ctx_IRS.modem_events, False)
|
||||
self.channels_running = False
|
||||
|
||||
def testNormBroadcast(self):
|
||||
self.loss_probability = 0 # no loss
|
||||
self.establishChannels()
|
||||
|
||||
params = {
|
||||
'origin': "AA1AAA-1",
|
||||
'domain': "BB1BBB-1",
|
||||
'gridsquare': "JN48ea",
|
||||
'type': 'MESSAGE',
|
||||
'priority': '1',
|
||||
'data': str(base64.b64encode(b"hello world!"), 'utf-8')
|
||||
}
|
||||
try:
|
||||
command = command_norm.Norm(self.ctx_ISS, params)
|
||||
command.run()
|
||||
except Exception as e:
|
||||
print(e)
|
||||
|
||||
|
||||
#del cmd
|
||||
#print(self.ctx_ISS.TESTMODE_EVENTS.empty())
|
||||
|
||||
while not self.ctx_ISS.TESTMODE_EVENTS.empty():
|
||||
event = self.ctx_ISS.TESTMODE_EVENTS.get()
|
||||
success = event.get('arq-transfer-outbound', {}).get('success', None)
|
||||
if success is not None:
|
||||
self.assertTrue(success, f"Test failed because of wrong success: {success}")
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Loading…
Reference in New Issue