fsfw
/
fsfw-example-tmtc-common
11
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

moved some helpers into framework

This commit is contained in:
Robin Müller 2022-05-24 14:50:37 +02:00
parent cdbd40363d
commit 3cca54f66f
No known key found for this signature in database
GPG Key ID: 11D4952C8CCEF814
1 changed files with 47 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

92
pus_tc/pus_11_tc_sched.py
View File

@ -16,6 +16,7 @@ from tmtccmd.tc.pus_11_tc_sched import (
generate_enable_tc_sched_cmd, generate_enable_tc_sched_cmd,
generate_disable_tc_sched_cmd, generate_disable_tc_sched_cmd,
generate_reset_tc_sched_cmd, generate_reset_tc_sched_cmd,
pack_time_tagged_tc_app_data,
) )
from tmtccmd.tc.definitions import TcQueueT from tmtccmd.tc.definitions import TcQueueT
@ -25,16 +26,18 @@ class OpCodes:
DISABLE = ["1", "disable"] DISABLE = ["1", "disable"]
RESET = ["2", "reset"] RESET = ["2", "reset"]
INSERT = ["12", "test-insert"] TEST_INSERT = ["12", "test-insert"]
DELETE = ["13", "del-test"] TEST_DELETE = ["13", "test-del"]
TEST_RESET = ["14", "test-clear"]
class Info: class Info:
ENABLE = "Enable TC scheduling" ENABLE = "Enable TC scheduling"
DISABLE = "Disable TC scheduling" DISABLE = "Disable TC scheduling"
RESET = "Reset TC scheduling" RESET = "Reset TC scheduling"
INSERT = "Test TC scheduling insertion" TEST_INSERT = "Test TC scheduling insertion"
DELETE = "Test TC scheduling deletion" TEST_DELETE = "Test TC scheduling deletion"
TEST_RESET = "Test TC scheduling reset command"
def add_tc_sched_cmds(cmd_dict: ServiceOpCodeDictT): def add_tc_sched_cmds(cmd_dict: ServiceOpCodeDictT):
@ -54,17 +57,22 @@ def add_tc_sched_cmds(cmd_dict: ServiceOpCodeDictT):
add_op_code_entry(op_code_dict=op_code, keys=OpCodes.RESET, info=Info.RESET) add_op_code_entry(op_code_dict=op_code, keys=OpCodes.RESET, info=Info.RESET)
add_op_code_entry( add_op_code_entry(
op_code_dict=op_code, op_code_dict=op_code,
keys=OpCodes.INSERT, keys=OpCodes.TEST_INSERT,
info=Info.INSERT, info=Info.TEST_INSERT,
options=generate_op_code_options(custom_timeout=0.2), options=generate_op_code_options(custom_timeout=0.2),
) )
add_op_code_entry( add_op_code_entry(
op_code_dict=op_code, op_code_dict=op_code,
keys=OpCodes.DELETE, keys=OpCodes.TEST_DELETE,
info=Info.DELETE, info=Info.TEST_DELETE,
options=generate_op_code_options(custom_timeout=0.2),
)
add_op_code_entry(
op_code_dict=op_code,
keys=OpCodes.TEST_RESET,
info=Info.TEST_RESET,
options=generate_op_code_options(custom_timeout=0.2), options=generate_op_code_options(custom_timeout=0.2),
) )
add_service_op_code_entry( add_service_op_code_entry(
srv_op_code_dict=cmd_dict, srv_op_code_dict=cmd_dict,
name="11", name="11",
@ -73,39 +81,41 @@ def add_tc_sched_cmds(cmd_dict: ServiceOpCodeDictT):
) )
def pack_service_11_commands(op_code: str, tc_queue: TcQueueT): def __generic_pack_three_time_tagged_cmds(tc_queue: TcQueueT):
if op_code in OpCodes.INSERT: tc_queue.appendleft((QueueCommands.PRINT, "Testing Time-Tagged Command insertion"))
tc_queue.appendleft( tc_queue.appendleft(generate_enable_tc_sched_cmd(ssc=0).pack_command_tuple())
(QueueCommands.PRINT, "Testing Time-Tagged Command insertion") current_time = int(round(time.time()))
)
tc_queue.appendleft(generate_enable_tc_sched_cmd(ssc=0).pack_command_tuple())
current_time = int(round(time.time()))
# these TC[17,1] (ping commands) shall be inserted # these TC[17,1] (ping commands) shall be inserted
ping_tcs = [ ping_tcs = [
pack_service_17_ping_command(1701), pack_service_17_ping_command(1701),
pack_service_17_ping_command(1702), pack_service_17_ping_command(1702),
pack_service_17_ping_command(1703), pack_service_17_ping_command(1703),
] ]
for idx, tc in enumerate(ping_tcs): for idx, tc in enumerate(ping_tcs):
release_time = current_time + (idx + 2) * 5 release_time = current_time + (idx + 2) * 5
time_tagged_tc = PusTelecommand( time_tagged_tc = PusTelecommand(
service=11, service=11,
subservice=Subservices.TC_INSERT, subservice=Subservices.TC_INSERT,
app_data=pack_insert_tc_app_data(struct.pack("!I", release_time), tc), app_data=pack_time_tagged_tc_app_data(struct.pack("!I", release_time), tc),
) )
tc_queue.appendleft(time_tagged_tc.pack_command_tuple()) tc_queue.appendleft(time_tagged_tc.pack_command_tuple())
tc_queue.appendleft((QueueCommands.WAIT, 25)) tc_queue.appendleft((QueueCommands.WAIT, 25))
if op_code in OpCodes.DELETE:
def pack_service_11_commands(op_code: str, tc_queue: TcQueueT):
if op_code in OpCodes.TEST_INSERT:
tc_queue.appendleft( tc_queue.appendleft(
(QueueCommands.PRINT, "Testing Time-Tagged Command deletion") (QueueCommands.PRINT, "Testing Time-Tagged Command deletion")
) )
__generic_pack_three_time_tagged_cmds(tc_queue=tc_queue)
if op_code in OpCodes.TEST_DELETE:
current_time = int(round(time.time())) current_time = int(round(time.time()))
tc_to_schedule = pack_service_17_ping_command(1703) tc_to_schedule = pack_service_17_ping_command(1703)
time_tagged_tc = PusTelecommand( time_tagged_tc = PusTelecommand(
service=11, service=11,
subservice=Subservices.TC_INSERT, subservice=Subservices.TC_INSERT,
app_data=pack_insert_tc_app_data( app_data=pack_time_tagged_tc_app_data(
struct.pack("!I", current_time + 20), tc_to_schedule struct.pack("!I", current_time + 20), tc_to_schedule
), ),
) )
@ -126,6 +136,10 @@ def pack_service_11_commands(op_code: str, tc_queue: TcQueueT):
if op_code in OpCodes.RESET: if op_code in OpCodes.RESET:
tc_queue.appendleft((QueueCommands.PRINT, "Reset TC scheduler")) tc_queue.appendleft((QueueCommands.PRINT, "Reset TC scheduler"))
tc_queue.appendleft(generate_reset_tc_sched_cmd(ssc=0).pack_command_tuple()) tc_queue.appendleft(generate_reset_tc_sched_cmd(ssc=0).pack_command_tuple())
if op_code in OpCodes.TEST_RESET:
tc_queue.appendleft((QueueCommands.PRINT, "Testing Reset command"))
__generic_pack_three_time_tagged_cmds(tc_queue=tc_queue)
tc_queue.appendleft(generate_reset_tc_sched_cmd(ssc=0).pack_command_tuple())
# a TC[11,5] for 3rd inserted ping TC # a TC[11,5] for 3rd inserted ping TC
# TODO: This should be an independent test # TODO: This should be an independent test
# a TC[11,6] for some other previously inserted TCs # a TC[11,6] for some other previously inserted TCs
@ -149,18 +163,6 @@ def pack_service_11_commands(op_code: str, tc_queue: TcQueueT):
# tc_queue.appendleft((service_11_8_tc.pack_command_tuple())) # tc_queue.appendleft((service_11_8_tc.pack_command_tuple()))
# this function packs another TC into an insert activity TC[11,4]
# parameter: release_time: Absolute time when TC shall be released/run
# parameter tc_to_insert: The TC which shall be inserted
def pack_insert_tc_app_data(release_time: bytes, tc_to_insert: PusTelecommand) -> bytes:
app_data = bytearray()
# pack the release time
app_data.extend(release_time)
# followed by the tc
app_data.extend(tc_to_insert.pack())
return app_data
def pack_delete_corresponding_tc_app_data(tc: PusTelecommand) -> bytes: def pack_delete_corresponding_tc_app_data(tc: PusTelecommand) -> bytes:
return TcSchedReqId.build_from_tc(tc).pack() return TcSchedReqId.build_from_tc(tc).pack()