Crash_Override/FreeDATA
1
0
Fork 0
mirror of https://github.com/DJ2LS/FreeDATA synced 2024-05-14 10:04:33 +02:00
Code Issues Packages Projects Releases Wiki Activity

Small improvements

Browse source
This commit is contained in:
Pedro 2023-12-20 12:26:28 +01:00
parent c81698a698
commit 2faf398e4b
4 changed files with 15 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
modem/arq_session.py
View file

@ -87,4 +87,4 @@ class ARQSession():
getattr(self, action_name)(frame)
return
self.log(f"Ignoring unknow transition from state {self.state} with frame {frame['frame_type']}")
self.log(f"Ignoring unknow transition from state {self.state.name} with frame {frame['frame_type']}")

3
modem/arq_session_irs.py
View file

@ -37,6 +37,9 @@ class ARQSessionIRS(arq_session.ARQSession):
IRS_State.ENDED: {
FRAME_TYPE.ARQ_BURST_FRAME.value: 'receive_data',
},
IRS_State.FAILED: {
FRAME_TYPE.ARQ_BURST_FRAME.value: 'receive_data',
},
}
def __init__(self, config: dict, modem, dxcall: str, session_id: int):

2
modem/frame_dispatcher.py
View file

@ -88,7 +88,7 @@ class DISPATCHER():
# instantiate handler
handler_class = self.FRAME_HANDLER[frametype]['class']
handler = handler_class(self.FRAME_HANDLER[frametype]['name'],
handler: FrameHandler = handler_class(self.FRAME_HANDLER[frametype]['name'],
self.config,
self.states,
self.event_manager,

12
tests/test_arq_session.py
View file

@ -15,6 +15,7 @@ import random
import structlog
import numpy as np
from event_manager import EventManager
from data_frame_factory import DataFrameFactory
class TestModem:
def __init__(self, event_q):
@ -23,7 +24,11 @@ class TestModem:
self.event_manager = EventManager([event_q])
def transmit(self, mode, repeats: int, repeat_delay: int, frames: bytearray) -> bool:
self.data_queue_received.put(frames)
transmission = {
'mode': mode,
'bytes': frames,
}
self.data_queue_received.put(transmission)
class TestARQSession(unittest.TestCase):
@ -32,6 +37,7 @@ class TestARQSession(unittest.TestCase):
config_manager = CONFIG('modem/config.ini.example')
cls.config = config_manager.read()
cls.logger = structlog.get_logger("TESTS")
cls.frame_factory = DataFrameFactory(cls.config)
# ISS
cls.iss_state_manager = StateManager(queue.Queue())
@ -60,10 +66,12 @@ class TestARQSession(unittest.TestCase):
while self.channels_running:
# Transfer data between both parties
try:
frame_bytes = modem_transmit_queue.get(timeout=1)
transmission = modem_transmit_queue.get(timeout=1)
if random.randint(0, 100) < self.loss_probability:
self.logger.info(f"[{threading.current_thread().name}] Frame lost...")
continue
frame_bytes = transmission['bytes']
frame_dispatcher.new_process_data(frame_bytes, None, len(frame_bytes), 0, 0)
except queue.Empty:
continue