From 3d1a4044fe7be4289558e848ff01407c317d6cca Mon Sep 17 00:00:00 2001 From: Robin Mueller Date: Thu, 28 Apr 2022 19:04:17 +0200 Subject: [PATCH] run black afmt --- pus_tc/cmd_definitions.py | 30 ++++++++--- pus_tc/devs/ploc_mpsoc.py | 6 ++- pus_tc/devs/ploc_supervisor.py | 82 +++++++++++++++--------------- pus_tc/devs/syrlinks_hk_handler.py | 14 ++--- pus_tc/tc_packer_hook.py | 2 +- 5 files changed, 78 insertions(+), 56 deletions(-) diff --git a/pus_tc/cmd_definitions.py b/pus_tc/cmd_definitions.py index 37e4a8e..887622c 100644 --- a/pus_tc/cmd_definitions.py +++ b/pus_tc/cmd_definitions.py @@ -627,18 +627,36 @@ def add_ploc_supv_cmds(cmd_dict: ServiceOpCodeDictT): ), "41": ("PLOC Supervisor: CAN loopback test", {OpCodeDictKeys.TIMEOUT: 2.0}), "42": ("PLOC Supervisor: Perform update", {OpCodeDictKeys.TIMEOUT: 2.0}), - "43": ("PLOC Supervisor: Terminate supervisor process", {OpCodeDictKeys.TIMEOUT: 2.0}), + "43": ( + "PLOC Supervisor: Terminate supervisor process", + {OpCodeDictKeys.TIMEOUT: 2.0}, + ), "44": ("PLOC Supervisor: Start MPSoC quiet", {OpCodeDictKeys.TIMEOUT: 2.0}), "45": ("PLOC Supervisor: Set shutdown timeout", {OpCodeDictKeys.TIMEOUT: 2.0}), "46": ("PLOC Supervisor: Factory flash", {OpCodeDictKeys.TIMEOUT: 2.0}), "47": ("PLOC Supervisor: Enable auto TM", {OpCodeDictKeys.TIMEOUT: 2.0}), "48": ("PLOC Supervisor: Disable auto TM", {OpCodeDictKeys.TIMEOUT: 2.0}), - "49": ("PLOC Supervisor: Enable ADC monitor task", {OpCodeDictKeys.TIMEOUT: 2.0}), - "50": ("PLOC Supervisor: Disable ADC monitor task", {OpCodeDictKeys.TIMEOUT: 2.0}), - "51": ("PLOC Supervisor: Logging request event buffers", {OpCodeDictKeys.TIMEOUT: 2.0}), - "52": ("PLOC Supervisor: Logging clear counters", {OpCodeDictKeys.TIMEOUT: 2.0}), + "49": ( + "PLOC Supervisor: Enable ADC monitor task", + {OpCodeDictKeys.TIMEOUT: 2.0}, + ), + "50": ( + "PLOC Supervisor: Disable ADC monitor task", + {OpCodeDictKeys.TIMEOUT: 2.0}, + ), + "51": ( + "PLOC Supervisor: Logging request event buffers", + {OpCodeDictKeys.TIMEOUT: 2.0}, + ), + "52": ( + "PLOC Supervisor: Logging clear counters", + {OpCodeDictKeys.TIMEOUT: 2.0}, + ), "53": ("PLOC Supervisor: Logging set topic", {OpCodeDictKeys.TIMEOUT: 2.0}), - "54": ("PLOC Supervisor: Logging request counters", {OpCodeDictKeys.TIMEOUT: 2.0}), + "54": ( + "PLOC Supervisor: Logging request counters", + {OpCodeDictKeys.TIMEOUT: 2.0}, + ), "55": ("PLOC Supervisor: Request ADC Report", {OpCodeDictKeys.TIMEOUT: 2.0}), } service_ploc_supv_tuple = ("PLOC Supervisor", op_code_dict_srv_ploc_supv) diff --git a/pus_tc/devs/ploc_mpsoc.py b/pus_tc/devs/ploc_mpsoc.py index 9a39acb..ab9e95c 100644 --- a/pus_tc/devs/ploc_mpsoc.py +++ b/pus_tc/devs/ploc_mpsoc.py @@ -172,7 +172,11 @@ def pack_ploc_mpsoc_commands( elif op_code == "16": tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: Tc cam command send")) cam_cmd = input("Specify cam command string: ") - command = object_id + struct.pack("!I", CommandIds.TC_CAM_CMD_SEND) + bytearray(cam_cmd, 'utf-8') + command = ( + object_id + + struct.pack("!I", CommandIds.TC_CAM_CMD_SEND) + + bytearray(cam_cmd, "utf-8") + ) command = PusTelecommand(service=8, subservice=128, ssc=32, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) diff --git a/pus_tc/devs/ploc_supervisor.py b/pus_tc/devs/ploc_supervisor.py index 6e8effb..7509959 100644 --- a/pus_tc/devs/ploc_supervisor.py +++ b/pus_tc/devs/ploc_supervisor.py @@ -384,69 +384,65 @@ def pack_ploc_supv_commands( command = PusTelecommand(service=8, subservice=128, ssc=57, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "43": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Terminate supervisor process")) - command = object_id + struct.pack( - "!I", SupvActionIds.TERMINATE_SUPV_HELPER + tc_queue.appendleft( + (QueueCommands.PRINT, "PLOC Supervisor: Terminate supervisor process") ) + command = object_id + struct.pack("!I", SupvActionIds.TERMINATE_SUPV_HELPER) command = PusTelecommand(service=8, subservice=128, ssc=58, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "44": tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Start MPSoC quiet")) - command = object_id + struct.pack( - "!I", SupvActionIds.START_MPSOC_QUIET - ) + command = object_id + struct.pack("!I", SupvActionIds.START_MPSOC_QUIET) command = PusTelecommand(service=8, subservice=128, ssc=59, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "45": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set shutdown timeout")) + tc_queue.appendleft( + (QueueCommands.PRINT, "PLOC Supervisor: Set shutdown timeout") + ) command = pack_set_shutdown_timeout_command(object_id) command = PusTelecommand(service=8, subservice=128, ssc=60, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "46": tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Factory flash")) - command = object_id + struct.pack( - "!I", SupvActionIds.FACTORY_FLASH - ) + command = object_id + struct.pack("!I", SupvActionIds.FACTORY_FLASH) command = PusTelecommand(service=8, subservice=128, ssc=61, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "47": tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Enable auto TM")) - command = object_id + struct.pack( - "!I", SupvActionIds.ENABLE_AUTO_TM - ) + command = object_id + struct.pack("!I", SupvActionIds.ENABLE_AUTO_TM) command = PusTelecommand(service=8, subservice=128, ssc=62, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "48": tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Disable auto TM")) - command = object_id + struct.pack( - "!I", SupvActionIds.DISABLE_AUTO_TM - ) + command = object_id + struct.pack("!I", SupvActionIds.DISABLE_AUTO_TM) command = PusTelecommand(service=8, subservice=128, ssc=63, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "49": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Enable ADC monitor task")) - command = object_id + struct.pack( - "!I", SupvActionIds.ENABLE_ADC_MONITOR_TASK + tc_queue.appendleft( + (QueueCommands.PRINT, "PLOC Supervisor: Enable ADC monitor task") ) + command = object_id + struct.pack("!I", SupvActionIds.ENABLE_ADC_MONITOR_TASK) command = PusTelecommand(service=8, subservice=128, ssc=64, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "50": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Disable ADC monitor task")) - command = object_id + struct.pack( - "!I", SupvActionIds.DISABLE_ADC_MONITOR_TASK + tc_queue.appendleft( + (QueueCommands.PRINT, "PLOC Supervisor: Disable ADC monitor task") ) + command = object_id + struct.pack("!I", SupvActionIds.DISABLE_ADC_MONITOR_TASK) command = PusTelecommand(service=8, subservice=128, ssc=65, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "51": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Logging request event buffers")) + tc_queue.appendleft( + (QueueCommands.PRINT, "PLOC Supervisor: Logging request event buffers") + ) command = pack_logging_buffer_request(object_id) command = PusTelecommand(service=8, subservice=128, ssc=66, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "52": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Logging clear counters")) - command = object_id + struct.pack( - "!I", SupvActionIds.LOGGING_CLEAR_COUNTERS + tc_queue.appendleft( + (QueueCommands.PRINT, "PLOC Supervisor: Logging clear counters") ) + command = object_id + struct.pack("!I", SupvActionIds.LOGGING_CLEAR_COUNTERS) command = PusTelecommand(service=8, subservice=128, ssc=67, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "53": @@ -455,13 +451,17 @@ def pack_ploc_supv_commands( command = PusTelecommand(service=8, subservice=128, ssc=68, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "54": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Logging request counters")) - command = object_id + struct.pack('!I', SupvActionIds.LOGGING_REQUEST_COUNTERS) + tc_queue.appendleft( + (QueueCommands.PRINT, "PLOC Supervisor: Logging request counters") + ) + command = object_id + struct.pack("!I", SupvActionIds.LOGGING_REQUEST_COUNTERS) command = PusTelecommand(service=8, subservice=128, ssc=69, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "55": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Request ADC report")) - command = object_id + struct.pack('!I', SupvActionIds.REQUEST_ADC_REPORT) + tc_queue.appendleft( + (QueueCommands.PRINT, "PLOC Supervisor: Request ADC report") + ) + command = object_id + struct.pack("!I", SupvActionIds.REQUEST_ADC_REPORT) command = PusTelecommand(service=8, subservice=128, ssc=70, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) @@ -714,30 +714,30 @@ def pack_update_command(object_id: bytearray) -> bytearray: start_address = int(input("Specify start address: 0x"), 16) update_file = get_update_file() command += object_id - command += struct.pack('!I', SupvActionIds.PERFORM_UPDATE) - command += bytearray(update_file, 'utf-8') + command += struct.pack("!I", SupvActionIds.PERFORM_UPDATE) + command += bytearray(update_file, "utf-8") # Adding null terminator - command += struct.pack('!B', 0) - command += struct.pack('!B', memory_id) - command += struct.pack('!I', start_address) + command += struct.pack("!B", 0) + command += struct.pack("!B", memory_id) + command += struct.pack("!I", start_address) return command def pack_set_shutdown_timeout_command(object_id: bytearray) -> bytearray: command = bytearray() command += object_id - command += struct.pack('!I', SupvActionIds.SET_SHUTDOWN_TIMEOUT) + command += struct.pack("!I", SupvActionIds.SET_SHUTDOWN_TIMEOUT) timeout = int(input("Specify shutdown timeout (ms): ")) - command += struct.pack('!I', timeout) + command += struct.pack("!I", timeout) return command def pack_logging_buffer_request(object_id: bytearray) -> bytearray: command = bytearray() command += object_id - command += struct.pack('!I', SupvActionIds.LOGGING_REQUEST_EVENT_BUFFERS) + command += struct.pack("!I", SupvActionIds.LOGGING_REQUEST_EVENT_BUFFERS) path = get_event_buffer_path() - command += bytearray(path, 'utf-8') + command += bytearray(path, "utf-8") return command @@ -789,9 +789,9 @@ def pack_read_gpio_cmd(object_id: bytearray) -> bytearray: def pack_logging_set_topic(objetc_id: bytearray) -> bytearray: - command = objetc_id + struct.pack('!I', SupvActionIds.LOGGING_SET_TOPIC) + command = objetc_id + struct.pack("!I", SupvActionIds.LOGGING_SET_TOPIC) tpc = int(input("Specify logging topic: ")) - command += struct.pack('!B', tpc) + command += struct.pack("!B", tpc) return command diff --git a/pus_tc/devs/syrlinks_hk_handler.py b/pus_tc/devs/syrlinks_hk_handler.py index b5f3099..0e4848c 100644 --- a/pus_tc/devs/syrlinks_hk_handler.py +++ b/pus_tc/devs/syrlinks_hk_handler.py @@ -64,17 +64,17 @@ def pack_syrlinks_command( tc_queue.appendleft(command.pack_command_tuple()) if op_code == "3": tc_queue.appendleft((QueueCommands.PRINT, "syrlinks: Set TX mode standby")) - command = object_id + struct.pack('!I', CommandIds.SET_TX_MODE_STANDBY) + command = object_id + struct.pack("!I", CommandIds.SET_TX_MODE_STANDBY) command = PusTelecommand(service=8, subservice=128, ssc=10, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) if op_code == "4": tc_queue.appendleft((QueueCommands.PRINT, "syrlinks: Set TX mode modulation")) - command = object_id + struct.pack('!I', CommandIds.SET_TX_MODE_MODULATION) + command = object_id + struct.pack("!I", CommandIds.SET_TX_MODE_MODULATION) command = PusTelecommand(service=8, subservice=128, ssc=11, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) if op_code == "5": tc_queue.appendleft((QueueCommands.PRINT, "syrlinks: Set TX mode CW")) - command = object_id + struct.pack('!I', CommandIds.SET_TX_MODE_CW) + command = object_id + struct.pack("!I", CommandIds.SET_TX_MODE_CW) command = PusTelecommand(service=8, subservice=128, ssc=12, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) if op_code == "6": @@ -89,26 +89,26 @@ def pack_syrlinks_command( tc_queue.appendleft(command.pack_command_tuple()) if op_code == "8": tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Read TX status")) - command = object_id + struct.pack('!I', CommandIds.READ_TX_STATUS) + command = object_id + struct.pack("!I", CommandIds.READ_TX_STATUS) command = PusTelecommand(service=8, subservice=128, ssc=13, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) if op_code == "9": tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Read TX waveform")) - command = object_id + struct.pack('!I', CommandIds.READ_TX_WAVEFORM) + command = object_id + struct.pack("!I", CommandIds.READ_TX_WAVEFORM) command = PusTelecommand(service=8, subservice=128, ssc=14, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) if op_code == "10": tc_queue.appendleft( (QueueCommands.PRINT, "Syrlinks: Read TX AGC value high byte") ) - command = object_id + struct.pack('!I', CommandIds.READ_TX_AGC_VALUE_HIGH_BYTE) + command = object_id + struct.pack("!I", CommandIds.READ_TX_AGC_VALUE_HIGH_BYTE) command = PusTelecommand(service=8, subservice=128, ssc=15, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) if op_code == "11": tc_queue.appendleft( (QueueCommands.PRINT, "Syrlinks: Read TX AGC value low byte") ) - command = object_id + struct.pack('!I', CommandIds.READ_TX_AGC_VALUE_LOW_BYTE) + command = object_id + struct.pack("!I", CommandIds.READ_TX_AGC_VALUE_LOW_BYTE) command = PusTelecommand(service=8, subservice=128, ssc=16, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) if op_code == "12": diff --git a/pus_tc/tc_packer_hook.py b/pus_tc/tc_packer_hook.py index 00b50f6..e3bba4a 100644 --- a/pus_tc/tc_packer_hook.py +++ b/pus_tc/tc_packer_hook.py @@ -238,7 +238,7 @@ def pack_service_queue_user( CustomServiceList.REACTION_WHEEL_1, CustomServiceList.REACTION_WHEEL_2, CustomServiceList.REACTION_WHEEL_3, - CustomServiceList.REACTION_WHEEL_4 + CustomServiceList.REACTION_WHEEL_4, ]: return pack_rw_cmds(op_code=op_code) LOGGER.warning("Invalid Service !")