eive/eive-tmtc
eive
/
eive-tmtc
13
0
Fork 0
Code Issues Pull Requests 1 Projects Releases 96 Wiki Activity

Merge remote-tracking branch 'origin/main' into cfdp-file-downlink
EIVE/-/pipeline/pr-main This commit looks good Details

This commit is contained in:
Robin Müller 2023-09-12 14:17:39 +02:00
parent 1860b754ce b20abc5bfb
commit 8b45dd8bff
Signed by: muellerr
GPG Key ID: A649FB78196E3849
27 changed files with 140 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
CHANGELOG.md
View File

@ -10,9 +10,17 @@ list yields a list of all related PRs for each release.
# [unreleased]
# [v5.5.0] 2023-09-04
# [v5.5.1] 2023-09-12
- Bumped `tmtccmd` to v6.0.0rc0
## Fixed
- Some API usage fixes related to `tmtccmd` update.
# [v5.5.0] 2023-09-12
- Version is not specfied dynamically anymore and can be updated in `pyproject.toml`
- New events and returnvalues
- Bump `tmtccmd` to v6.0.0rc0
# [v5.4.3] 2023-08-15

7
eive_tmtc/__init__.py
View File

@ -1,13 +1,6 @@
__version__ = "5.4.3"
import logging
from pathlib import Path
SW_NAME = "eive-tmtc"
VERSION_MAJOR = 5
VERSION_MINOR = 4
VERSION_REVISION = 3
EIVE_TMTC_ROOT = Path(__file__).parent
PACKAGE_ROOT = EIVE_TMTC_ROOT.parent

4
eive_tmtc/cfdp/user.py
View File

@ -1,7 +1,6 @@
import logging
from spacepackets.cfdp import ConditionCode
from spacepackets.cfdp.pdu import DeliveryCode
from tmtccmd.cfdp import CfdpUserBase, TransactionId
from tmtccmd.cfdp.user import (
TransactionFinishedParams,
@ -30,7 +29,8 @@ class EiveCfdpUser(CfdpUserBase):
def file_segment_recv_indication(self, params: FileSegmentRecvdParams):
_LOGGER.info(
f"CFDP User: Received File Data PDU for {params.transaction_id} | Offset: {params.offset} | Segment Length: {params.length}"
f"CFDP User: Received File Data PDU for {params.transaction_id} | Offset:"
f" {params.offset} | Segment Length: {params.length}"
)
def report_indication(self, transaction_id: TransactionId, status_report: any):

7
eive_tmtc/pus_tc/tc_handler.py
View File

@ -140,7 +140,7 @@ class TcHandler(TcHandlerBase):
if pdu.pdu_directive_type == DirectiveType.METADATA_PDU:
metadata = pdu.to_metadata_pdu()
self.queue_helper.add_log_cmd(
f"CFDP Source: Sending Metadata PDU for file with size "
"CFDP Source: Sending Metadata PDU for file with size "
f"{metadata.file_size}"
)
elif pdu.pdu_directive_type == DirectiveType.EOF_PDU:
@ -150,7 +150,7 @@ class TcHandler(TcHandlerBase):
else:
fd_pdu = pdu.to_file_data_pdu()
self.queue_helper.add_log_cmd(
f"CFDP Source: Sending File Data PDU for segment at offset "
"CFDP Source: Sending File Data PDU for segment at offset "
f"{fd_pdu.offset} with length {len(fd_pdu.file_data)}"
)
self.queue_helper.add_ccsds_tc(sp)
@ -162,7 +162,8 @@ class TcHandler(TcHandlerBase):
if info.proc_type == TcQueueEntryType.PUS_TC:
def_proc = info.to_def_procedure()
_LOGGER.info(
f"Finished queue for service {def_proc.service} and op code {def_proc.op_code}"
f"Finished queue for service {def_proc.service} and op code"
f" {def_proc.op_code}"
)
elif info.proc_type == TcProcedureType.CFDP:
_LOGGER.info("Finished CFDP queue")

11
eive_tmtc/pus_tm/event_handler.py
View File

@ -37,12 +37,13 @@ def handle_event_packet( # noqa C901: Complexity okay here
else:
obj_name = obj_id_obj.name
generic_event_string = (
f"Object {obj_name} generated Event {info.name} (ID: {event_def.event_id:#04x}) "
f"at {tm.time_provider.as_date_time()}"
f"Object {obj_name} generated Event {info.name} (ID: {event_def.event_id:#04x})"
f" at {tm.time_provider.as_date_time()}"
)
_LOGGER.info(generic_event_string)
pw.file_logger.info(
f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}: {generic_event_string}"
f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}:"
f" {generic_event_string}"
)
specific_handler = False
if info.name == "MODE_TRANSITION_FAILED":
@ -54,8 +55,8 @@ def handle_event_packet( # noqa C901: Complexity okay here
if info.name == "SUPV_UPDATE_PROGRESS" or info.name == "WRITE_MEMORY_FAILED":
additional_event_info = f"Additional info: {info.info}"
context = (
f"Progress Percent: {event_def.param1 >> 24 & 0xff} | "
f"Sequence Count: {event_def.param1 & 0xffff} | Bytes Written: {event_def.param2}"
f"Progress Percent: {event_def.param1 >> 24 & 0xff} | Sequence Count:"
f" {event_def.param1 & 0xffff} | Bytes Written: {event_def.param2}"
)
pw.dlog(additional_event_info)
pw.dlog(context)

2
eive_tmtc/pus_tm/hk_handler.py
View File

@ -188,5 +188,5 @@ def handle_regular_hk_print( # noqa C901: Complexity okay here
else:
_LOGGER.info(
f"Service 3 TM: Parsing for object {object_id} and set ID {set_id} "
f"has not been implemented."
"has not been implemented."
)

3
eive_tmtc/pus_tm/pus_demux.py
View File

@ -87,7 +87,8 @@ def pus_factory_hook( # noqa C901 : Complexity okay here
# TODO: Could improve display further by actually displaying a matrix as a
# matrix using row and column information
pw.dlog(
f"Received vector or matrix data: {param.param_raw.hex(sep=',')}"
"Received vector or matrix data:"
f" {param.param_raw.hex(sep=',')}"
)
except ValueError as e:
pw.dlog(f"received {e} when trying to parse parameters")

6
eive_tmtc/pus_tm/verification_handler.py
View File

@ -48,7 +48,7 @@ def generic_retval_printout(
if retval_info is None:
raw_err = retval
return [
f"No returnvalue information found for error code with "
"No returnvalue information found for error code with "
f"subsystem ID {(raw_err >> 8) & 0xff} and unique ID {raw_err & 0xff}"
]
else:
@ -58,9 +58,9 @@ def generic_retval_printout(
)
string_list = [retval_string]
if p1:
error_param_1_str = f"Error Parameter 1: hex {p1:#010x} " f"dec {p1} "
error_param_1_str = f"Error Parameter 1: hex {p1:#010x} dec {p1} "
string_list.append(error_param_1_str)
if p2:
error_param_2_str = f"Error Parameter 2: hex {p2:#010x} " f"dec {p2}"
error_param_2_str = f"Error Parameter 2: hex {p2:#010x} dec {p2}"
string_list.append(error_param_2_str)
return string_list

15
eive_tmtc/tmtc/acs/acs_ctrl.py
View File

@ -571,8 +571,8 @@ def pack_acs_ctrl_command(p: ServiceProviderParams): # noqa C901
def set_acs_ctrl_param_scalar(q: DefaultPusQueueHelper):
pt = int(
input(
'Specify parameter type to set {0: "uint8", 1: "uint16", 2: "int32", 3: "float", '
'4: "double"}: '
'Specify parameter type to set {0: "uint8", 1: "uint16", 2: "int32", 3:'
' "float", 4: "double"}: '
)
)
sid = int(input("Specify parameter struct ID to set: "))
@ -808,8 +808,8 @@ def handle_acs_ctrl_sus_raw_data(pw: PrintWrapper, hk_data: bytes):
def handle_acs_ctrl_sus_processed_data(pw: PrintWrapper, hk_data: bytes):
if len(hk_data) < 3 * 4 * 12 + 3 * 8 * 3:
pw.dlog(
f"SUS Processed dataset with size {len(hk_data)} does not have expected size"
f" of {3 * 4 * 12 + 3 * 8 * 3} bytes"
f"SUS Processed dataset with size {len(hk_data)} does not have expected"
f" size of {3 * 4 * 12 + 3 * 8 * 3} bytes"
)
return
current_idx = 0
@ -846,7 +846,8 @@ def handle_raw_mgm_data(pw: PrintWrapper, hk_data: bytes):
if len(hk_data) < 61:
pw.dlog(
f"ACS CTRL HK: MGM HK data with length {len(hk_data)} shorter than expected 61 bytes"
f"ACS CTRL HK: MGM HK data with length {len(hk_data)} shorter than expected"
" 61 bytes"
)
pw.dlog(f"Raw Data: {hk_data.hex(sep=',')}")
return
@ -1276,8 +1277,8 @@ def perform_mgm_calibration( # noqa C901: Complexity okay
reply = CALIBR_SOCKET.recv(1024)
if len(reply) != 2:
pw.dlog(
f"MGM calibration: Reply received command magnetometer_field has invalid "
f"length {len(reply)}"
"MGM calibration: Reply received command magnetometer_field has"
f" invalid length {len(reply)}"
)
return
else:

16
eive_tmtc/tmtc/acs/gps.py
View File

@ -12,8 +12,8 @@ from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
create_request_one_hk_command,
create_enable_periodic_hk_command_with_interval,
create_disable_periodic_hk_command,
create_enable_periodic_hk_command_with_interval_with_diag,
create_disable_periodic_hk_command_with_diag,
)
from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
@ -72,7 +72,9 @@ def add_gps_cmds(defs: TmtcDefinitionWrapper):
)
def pack_gps_command(object_id: bytes, q: DefaultPusQueueHelper, op_code: str):
def pack_gps_command( # noqa: C901
object_id: bytes, q: DefaultPusQueueHelper, op_code: str
): # noqa: C901:
if op_code in OpCode.RESET_GNSS:
# TODO: This needs to be re-implemented
_LOGGER.warning("Reset pin handling needs to be re-implemented")
@ -81,7 +83,7 @@ def pack_gps_command(object_id: bytes, q: DefaultPusQueueHelper, op_code: str):
if interval <= 0:
raise ValueError("invalid interval")
q.add_log_cmd(f"GPS: {Info.ENABLE_CORE_HK}")
cmds = create_enable_periodic_hk_command_with_interval(
cmds = create_enable_periodic_hk_command_with_interval_with_diag(
diag=False,
sid=make_sid(object_id=object_id, set_id=SetId.CORE_HK),
interval_seconds=interval,
@ -91,7 +93,7 @@ def pack_gps_command(object_id: bytes, q: DefaultPusQueueHelper, op_code: str):
if op_code in OpCode.DISABLE_CORE_HK:
q.add_log_cmd(f"gps: {Info.DISABLE_CORE_HK}")
q.add_pus_tc(
create_disable_periodic_hk_command(
create_disable_periodic_hk_command_with_diag(
diag=False, sid=make_sid(object_id=object_id, set_id=SetId.CORE_HK)
)
)
@ -107,7 +109,7 @@ def pack_gps_command(object_id: bytes, q: DefaultPusQueueHelper, op_code: str):
if interval <= 0:
raise ValueError("invalid interval")
q.add_log_cmd(f"GPS: {Info.ENABLE_SKYVIEW_HK}")
cmds = create_enable_periodic_hk_command_with_interval(
cmds = create_enable_periodic_hk_command_with_interval_with_diag(
diag=False,
sid=make_sid(object_id=object_id, set_id=SetId.SKYVIEW_HK),
interval_seconds=interval,
@ -117,7 +119,7 @@ def pack_gps_command(object_id: bytes, q: DefaultPusQueueHelper, op_code: str):
if op_code in OpCode.DISABLE_SKYVIEW_HK:
q.add_log_cmd(f"gps: {Info.DISABLE_SKYVIEW_HK}")
q.add_pus_tc(
create_disable_periodic_hk_command(
create_disable_periodic_hk_command_with_diag(
diag=False, sid=make_sid(object_id=object_id, set_id=SetId.SKYVIEW_HK)
)
)

31
eive_tmtc/tmtc/acs/imtq.py
View File

@ -24,8 +24,8 @@ from tmtccmd.tc.pus_3_fsfw_hk import (
generate_one_diag_command,
generate_one_hk_command,
create_request_one_diag_command,
create_disable_periodic_hk_command,
create_enable_periodic_hk_command_with_interval,
create_enable_periodic_hk_command_with_interval_with_diag,
create_disable_periodic_hk_command_with_diag,
)
from tmtccmd.tc.pus_200_fsfw_mode import pack_mode_data, Mode
from tmtccmd.util import ObjectIdU32
@ -227,12 +227,13 @@ def pack_imtq_test_into( # noqa C901
duration = int(
input(
f"Specify torque duration [range [0, {pow(2, 16) - 1}, "
f"0 for continuous generation until update]: "
"0 for continuous generation until update]: "
)
)
dur_str = "infinite" if duration == 0 else str(duration)
q.add_log_cmd(
f"IMTQ: Commanding dipole X={x_dipole}, Y={y_dipole}, Z={y_dipole}, duration={dur_str}"
f"IMTQ: Commanding dipole X={x_dipole}, Y={y_dipole}, Z={y_dipole},"
f" duration={dur_str}"
)
q.add_pus_tc(
pack_dipole_command(
@ -248,7 +249,7 @@ def pack_imtq_test_into( # noqa C901
if op_code == OpCode.ENABLE_ENG_HK_NO_TORQUE:
q.add_log_cmd("IMTQ: Enable ENG HK")
interval = float(input("Please enter collection interval in seconds: "))
cmds = create_enable_periodic_hk_command_with_interval(
cmds = create_enable_periodic_hk_command_with_interval_with_diag(
diag=True,
sid=make_sid(object_id.as_bytes, ImtqSetId.ENG_HK_NO_TORQUE),
interval_seconds=interval,
@ -258,7 +259,7 @@ def pack_imtq_test_into( # noqa C901
if op_code == OpCode.DISABLE_ENG_HK_NO_TORQUE:
q.add_log_cmd("IMTQ: Disable ENG HK (No Torque)")
q.add_pus_tc(
create_disable_periodic_hk_command(
create_disable_periodic_hk_command_with_diag(
True, make_sid(object_id.as_bytes, ImtqSetId.ENG_HK_NO_TORQUE)
)
)
@ -275,7 +276,7 @@ def pack_imtq_test_into( # noqa C901
if op_code == OpCode.ENABLE_ENG_HK_WITH_TORQUE:
q.add_log_cmd("IMTQ: Enable ENG HK with torque")
interval = float(input("Please enter collection interval in seconds: "))
cmds = create_enable_periodic_hk_command_with_interval(
cmds = create_enable_periodic_hk_command_with_interval_with_diag(
diag=True,
sid=make_sid(object_id.as_bytes, ImtqSetId.ENG_HK_SET_WITH_TORQUE),
interval_seconds=interval,
@ -285,7 +286,7 @@ def pack_imtq_test_into( # noqa C901
if op_code == OpCode.DISABLE_ENG_HK_WITH_TORQUE:
q.add_log_cmd("IMTQ: Disable ENG HK with Torque")
q.add_pus_tc(
create_disable_periodic_hk_command(
create_disable_periodic_hk_command_with_diag(
True, make_sid(object_id.as_bytes, ImtqSetId.ENG_HK_SET_WITH_TORQUE)
)
)
@ -320,14 +321,14 @@ def pack_imtq_test_into( # noqa C901
if op_code == OpCode.DISABLE_MGM_RAW_NO_TORQUE:
q.add_log_cmd("IMTQ: Disable MGM RAW HK (No Torque)")
q.add_pus_tc(
create_disable_periodic_hk_command(
create_disable_periodic_hk_command_with_diag(
True, make_sid(object_id.as_bytes, ImtqSetId.RAW_MTM_NO_TORQUE)
)
)
if op_code == OpCode.ENABLE_MGM_RAW_NO_TORQUE:
q.add_log_cmd("IMTQ: Enable MGM RAW HK (No Torque)")
interval = float(input("Please enter collection interval in seconds: "))
cmds = create_enable_periodic_hk_command_with_interval(
cmds = create_enable_periodic_hk_command_with_interval_with_diag(
diag=True,
sid=make_sid(object_id.as_bytes, ImtqSetId.RAW_MTM_NO_TORQUE),
interval_seconds=interval,
@ -346,7 +347,7 @@ def pack_imtq_test_into( # noqa C901
if op_code == OpCode.ENABLE_MGM_RAW_WITH_TORQUE:
q.add_log_cmd("IMTQ: Enable MGM RAW HK (No Torque)")
interval = float(input("Please enter collection interval in seconds: "))
cmds = create_enable_periodic_hk_command_with_interval(
cmds = create_enable_periodic_hk_command_with_interval_with_diag(
diag=True,
sid=make_sid(object_id.as_bytes, ImtqSetId.RAW_MTM_WITH_TORQUE),
interval_seconds=interval,
@ -356,7 +357,7 @@ def pack_imtq_test_into( # noqa C901
if op_code == OpCode.DISABLE_MGM_RAW_WITH_TORQUE:
q.add_log_cmd("IMTQ: Disable MGM RAW HK (No Torque)")
q.add_pus_tc(
create_disable_periodic_hk_command(
create_disable_periodic_hk_command_with_diag(
True, make_sid(object_id.as_bytes, ImtqSetId.RAW_MTM_WITH_TORQUE)
)
)
@ -388,7 +389,8 @@ def pack_dipole_command(
duration = int(round(duration))
if duration < 0 or duration > pow(2, 16) - 1:
raise ValueError(
f"Duration in ms of {duration} smaller than 0 or larger than allowed {pow(2, 16) - 1}"
f"Duration in ms of {duration} smaller than 0 or larger than allowed"
f" {pow(2, 16) - 1}"
)
command += struct.pack("!h", x_dipole)
command += struct.pack("!h", y_dipole)
@ -400,7 +402,8 @@ def pack_dipole_command(
def raise_dipole_error(dipole_str: str, value: int):
raise ValueError(
f"{dipole_str} {value} negative or larger than maximum allowed 2000 * 10^-4*Am^2"
f"{dipole_str} {value} negative or larger than maximum allowed 2000 *"
" 10^-4*Am^2"
)

34
eive_tmtc/tmtc/acs/reaction_wheels.py
View File

@ -261,14 +261,12 @@ def pack_set_speed_command(
if speed > 0:
if speed < 1000 or speed > 65000:
raise ValueError(
"Invalid RW speed specified. "
"Allowed range is [1000, 65000] 0.1 * RPM"
"Invalid RW speed specified. Allowed range is [1000, 65000] 0.1 * RPM"
)
elif speed < 0:
if speed < -65000 or speed > -1000:
raise ValueError(
"Invalid RW speed specified. "
"Allowed range is [-65000, -1000] 0.1 * RPM"
"Invalid RW speed specified. Allowed range is [-65000, -1000] 0.1 * RPM"
)
else:
# Speed is 0
@ -304,15 +302,16 @@ def handle_rw_hk_data(
speed_rpm = speed / 10.0
ref_speed_rpm = ref_speed / 10.0
pw.dlog(
f"Temperature {temp} C | Speed {speed_rpm} rpm | Reference Speed {ref_speed_rpm} rpm"
f"Temperature {temp} C | Speed {speed_rpm} rpm | Reference Speed"
f" {ref_speed_rpm} rpm"
)
pw.dlog(
f"State {state}. 0: Error, 1: Idle, 2: Coasting, 3: Running, speed stable, "
f"4: Running, speed changing"
"4: Running, speed changing"
)
pw.dlog(
f"Current Limit Control mode {clc_mode}. 0: Low Current Mode (0.3 A), "
f"1: High Current Mode (0.6 A)"
"1: High Current Mode (0.6 A)"
)
pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], 5))
if set_id == RwSetId.LAST_RESET:
@ -361,22 +360,24 @@ def handle_rw_hk_data(
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
pw.dlog(
f"MCU Temperature {mcu_temp} | Pressure Sensore Temperature {pressure_sens_temp} C"
f"MCU Temperature {mcu_temp} | Pressure Sensore Temperature"
f" {pressure_sens_temp} C"
)
pw.dlog(f"Last Reset Status {last_reset_status}")
pw.dlog(
f"Current Limit Control mode {clc_mode}. 0: Low Current Mode (0.3 A), "
f"1: High Current Mode (0.6 A)"
"1: High Current Mode (0.6 A)"
)
pw.dlog(f"Speed {current_speed} rpm | Reference Speed {ref_speed} rpm")
pw.dlog(
f"State {state}. 0: Error, 1: Idle, 2: Coasting, 3: Running, speed stable, "
f"4: Running, speed changing"
"4: Running, speed changing"
)
pw.dlog("Number Of Invalid Packets:")
pw.dlog("CRC | Length | CMD")
pw.dlog(
f"{num_invalid_crc_packets} | {num_invalid_len_packets} | {num_invalid_cmd_packets}"
f"{num_invalid_crc_packets} | {num_invalid_len_packets} |"
f" {num_invalid_cmd_packets}"
)
pw.dlog(
f"Num Of CMD Executed Requests {num_of_cmd_executed_requests} | "
@ -388,15 +389,16 @@ def handle_rw_hk_data(
"RegOverrunErrs | TotalErrs"
)
pw.dlog(
f"{uart_num_of_bytes_written} | {uart_num_of_bytes_read} | {uart_num_parity_errors} | "
f"{uart_num_noise_errors} | {uart_num_frame_errors} | {uart_num_reg_overrun_errors} | "
f"{uart_total_num_errors}"
f"{uart_num_of_bytes_written} | {uart_num_of_bytes_read} |"
f" {uart_num_parity_errors} | {uart_num_noise_errors} |"
f" {uart_num_frame_errors} | {uart_num_reg_overrun_errors} |"
f" {uart_total_num_errors}"
)
pw.dlog("SPI COM Info:")
pw.dlog("NumBytesWritten | NumBytesRead | RegOverrunErrs | TotalErrs")
pw.dlog(
f"{spi_num_bytes_written} | {spi_num_bytes_read} | {spi_num_reg_overrun_errors} | "
f"{spi_total_num_errors}"
f"{spi_num_bytes_written} | {spi_num_bytes_read} |"
f" {spi_num_reg_overrun_errors} | {spi_total_num_errors}"
)
if current_idx > 0:
pw.dlog(

4
eive_tmtc/tmtc/com/pdec_handler.py
View File

@ -83,7 +83,7 @@ def pack_pdec_handler_test(
0,
ParameterId.POSITIVE_WINDOW,
pw,
).pack()
)
)
)
if op_code == OpCode.NEGATIVE_WINDOW:
@ -96,7 +96,7 @@ def pack_pdec_handler_test(
0,
ParameterId.NEGATIVE_WINDOW,
nw,
).pack()
)
)
)
if op_code == OpCode.RESET_NO_INIT:

8
eive_tmtc/tmtc/com/subsystem.py
View File

@ -3,6 +3,7 @@ import enum
from eive_tmtc.config.definitions import CustomServiceList
from eive_tmtc.config.object_ids import COM_SUBSYSTEM_ID
from eive_tmtc.tmtc.com.syrlinks_handler import Datarate
from tmtccmd.pus.s20_fsfw_param_defs import create_scalar_u8_parameter
from .defs import Mode as ComMode
@ -21,7 +22,6 @@ from tmtccmd.tc.pus_200_fsfw_mode import (
)
from tmtccmd.tc.pus_20_fsfw_param import (
create_load_param_cmd,
pack_scalar_u8_parameter_app_data,
)
from tmtccmd.pus.s20_fsfw_param import create_scalar_u32_parameter
@ -87,7 +87,7 @@ def build_com_subsystem_cmd(p: ServiceProviderParams): # noqa C901
q.add_log_cmd(f"{prefix}: {Info.UPDATE_DEFAULT_DATARATE_LOW}")
q.add_pus_tc(
create_load_param_cmd(
pack_scalar_u8_parameter_app_data(
create_scalar_u8_parameter(
COM_SUBSYSTEM_ID,
0,
ParameterId.DATARATE,
@ -99,7 +99,7 @@ def build_com_subsystem_cmd(p: ServiceProviderParams): # noqa C901
q.add_log_cmd(f"{prefix}: {Info.UPDATE_DEFAULT_DATARATE_HIGH}")
q.add_pus_tc(
create_load_param_cmd(
pack_scalar_u8_parameter_app_data(
create_scalar_u8_parameter(
COM_SUBSYSTEM_ID,
0,
ParameterId.DATARATE,
@ -122,7 +122,7 @@ def build_com_subsystem_cmd(p: ServiceProviderParams): # noqa C901
0,
ParameterId.TRANSMITTER_TIMEOUT,
timeout,
).pack()
)
)
)
elif o == OpCode.READ_MODE:

16
eive_tmtc/tmtc/com/syrlinks_handler.py
View File

@ -21,9 +21,9 @@ from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
create_request_one_diag_command,
create_enable_periodic_hk_command_with_interval,
create_disable_periodic_hk_command,
create_request_one_hk_command,
create_enable_periodic_hk_command_with_interval_with_diag,
create_disable_periodic_hk_command_with_diag,
)
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.pus_200_fsfw_mode import Mode, create_mode_command
@ -182,24 +182,28 @@ def pack_syrlinks_command( # noqa C901: Complexity okay here.
q.add_log_cmd(f"{prefix}: {Info.ENABLE_HK_RX_REGS}")
sid = make_sid(obyt, SetId.RX_REGISTERS_DATASET)
interval = float(input("HK interval in floating point seconds"))
cmds = create_enable_periodic_hk_command_with_interval(True, sid, interval)
cmds = create_enable_periodic_hk_command_with_interval_with_diag(
True, sid, interval
)
for cmd in cmds:
q.add_pus_tc(cmd)
if op_code in OpCode.DISABLE_HK_RX_REGS:
q.add_log_cmd(f"{prefix}: {Info.DISABLE_HK_RX_REGS}")
sid = make_sid(obyt, SetId.RX_REGISTERS_DATASET)
q.add_pus_tc(create_disable_periodic_hk_command(True, sid))
q.add_pus_tc(create_disable_periodic_hk_command_with_diag(True, sid))
if op_code in OpCode.ENABLE_HK_TX_REGS:
q.add_log_cmd(f"{prefix}: {Info.ENABLE_HK_TX_REGS}")
sid = make_sid(obyt, SetId.TX_REGISTERS_DATASET)
interval = float(input("HK interval in floating point seconds"))
cmds = create_enable_periodic_hk_command_with_interval(True, sid, interval)
cmds = create_enable_periodic_hk_command_with_interval_with_diag(
True, sid, interval
)
for cmd in cmds:
q.add_pus_tc(cmd)
if op_code in OpCode.DISABLE_HK_TX_REGS:
q.add_log_cmd(f"{prefix}: {Info.DISABLE_HK_TX_REGS}")
sid = make_sid(obyt, SetId.TX_REGISTERS_DATASET)
q.add_pus_tc(create_disable_periodic_hk_command(True, sid))
q.add_pus_tc(create_disable_periodic_hk_command_with_diag(True, sid))
if op_code in OpCode.HK_TX_REGS:
q.add_log_cmd(f"{prefix}: {Info.HK_TX_REGS}")
sid = make_sid(obyt, SetId.TX_REGISTERS_DATASET)

13
eive_tmtc/tmtc/payload/ploc_mpsoc.py
View File

@ -570,12 +570,12 @@ def handle_ploc_mpsoc_hk_data(pw: PrintWrapper, hk_data: bytes, set_id: int):
pw.dlog(f"CAM SoC Temperature: {cam_soc_temp}")
pw.dlog(f"System Monitor Temperature: {sysmon_temp}")
pw.dlog(
f"SYSMON VCC INT {sysmon_vcc_int:.3f} | SYSMON VCC AUX {sysmon_vcc_aux:.3f} | "
f"SYSMON VCC BRAM {sysmon_vcc_bram:.3f}"
f"SYSMON VCC INT {sysmon_vcc_int:.3f} | SYSMON VCC AUX"
f" {sysmon_vcc_aux:.3f} | SYSMON VCC BRAM {sysmon_vcc_bram:.3f}"
)
pw.dlog(
f"SYSMON VCC PAUX {sysmon_vcc_paux:.3f} | SYSMON VCC PINT {sysmon_vcc_pint:.3f} | "
f"SYSMON VCC PDRO {sysmon_vcc_pdro:.3f}"
f"SYSMON VCC PAUX {sysmon_vcc_paux:.3f} | SYSMON VCC PINT"
f" {sysmon_vcc_pint:.3f} | SYSMON VCC PDRO {sysmon_vcc_pdro:.3f}"
)
fmt_str = "!fffffffffffff"
@ -602,8 +602,9 @@ def handle_ploc_mpsoc_hk_data(pw: PrintWrapper, hk_data: bytes, set_id: int):
f"SYSMON MBA 1V8 {sysmon_mb_1v8:.3f}"
)
pw.dlog(
f"SYSMON VCC 12V {sysmon_vcc_12v:.3f} | SYSMON VCC 5V {sysmon_vcc_5v:.3f} | "
f"SYSMON VCC 3V3 {sysmon_vcc_3v3:.3f} | SYSMON VCC 3V3VA {sysmon_vcc_3v3va}"
f"SYSMON VCC 12V {sysmon_vcc_12v:.3f} | SYSMON VCC 5V {sysmon_vcc_5v:.3f} |"
f" SYSMON VCC 3V3 {sysmon_vcc_3v3:.3f} | SYSMON VCC 3V3VA"
f" {sysmon_vcc_3v3va}"
)
pw.dlog(
f"SYSMON VCC 2V5DDR {sysmon_vcc_2v5ddr:.3f} | "

7
eive_tmtc/tmtc/payload/ploc_supervisor.py
View File

@ -471,8 +471,8 @@ def pack_ploc_supv_commands(p: ServiceProviderParams): # noqa C901
custom_data.extend(struct.pack("!B", memory_id))
custom_data.extend(struct.pack("!I", start_address))
q.add_log_cmd(
f"{prefix}: {Info.MEM_CHECK} for file {update_file} at memory ID {memory_id} at start "
f"address {start_address}"
f"{prefix}: {Info.MEM_CHECK} for file {update_file} at memory ID"
f" {memory_id} at start address {start_address}"
)
command = create_action_cmd(
object_id.as_bytes, SupvActionId.MEM_CHECK, custom_data
@ -775,7 +775,8 @@ def handle_supv_hk_data(set_id: int, hk_data: bytes, pw: PrintWrapper):
) = struct.unpack(fmt_str, hk_data[0 : 0 + inc_len])
current_idx += inc_len
pw.dlog(
f"SoC state (0:off, 1:booting, 2:update, 3:operating, 4:shutdown, 5:reset): {soc_state}"
"SoC state (0:off, 1:booting, 2:update, 3:operating, 4:shutdown, 5:reset):"
f" {soc_state}"
)
pw.dlog(f"Power Cycles {power_cycles}")
pw.dlog(f"Boot after {boot_after_ms} ms | Boot timeout {boot_timeout_ms} ms")

3
eive_tmtc/tmtc/payload/scex.py
View File

@ -131,8 +131,7 @@ def pack_scex_cmds(p: ServiceProviderParams): # noqa C901
cell_select = int(cell_select)
if cell_select < 1 or cell_select > 10:
print(
f"Invalid cell number {cell_select}, "
f"Please enter a valid number: "
f"Invalid cell number {cell_select}, Please enter a valid number: "
)
continue
cn = cell_select - 1

11
eive_tmtc/tmtc/power/common_power.py
View File

@ -200,7 +200,8 @@ def pack_common_gomspace_cmds( # noqa C901: Complexity is okay here.
if op_code in GomspaceOpCode.SAVE_TABLE_DEFAULT:
source_table = int(
input(
"Source table [0: Board Config, 1: Module Config, 2: Calibration Parameter]: "
"Source table [0: Board Config, 1: Module Config, 2: Calibration"
" Parameter]: "
)
)
if source_table not in [0, 1, 2]:
@ -215,8 +216,8 @@ def pack_common_gomspace_cmds( # noqa C901: Complexity is okay here.
if op_code in GomspaceOpCode.LOAD_TABLE:
target_table = int(
input(
"Target table ID [0: Board Config, 1: Module Config, 2: Calibration Parameter, "
"4: HK TM]: "
"Target table ID [0: Board Config, 1: Module Config, 2: Calibration"
" Parameter, 4: HK TM]: "
)
)
if target_table not in [0, 1, 2, 4]:
@ -224,8 +225,8 @@ def pack_common_gomspace_cmds( # noqa C901: Complexity is okay here.
if target_table != 4:
source_table = int(
input(
"Source table (file or default) [0: Board Config, 1: Module Config, "
"2: Calibration Parameter, value + 4 for default table]: "
"Source table (file or default) [0: Board Config, 1: Module Config,"
" 2: Calibration Parameter, value + 4 for default table]: "
)
)
if source_table not in [0, 1, 2, 4, 5, 6]:

2
eive_tmtc/tmtc/power/plpcdu.py
View File

@ -284,7 +284,7 @@ def hpa_on_procedure(q: DefaultPusQueueHelper):
if delay_dro_to_x8 is None:
delay_dro_to_x8 = 900
q.add_log_cmd(
f"Starting procedure to switch on PL PCDU HPA with DRO to X8 "
"Starting procedure to switch on PL PCDU HPA with DRO to X8 "
f"delay of {delay_dro_to_x8} seconds"
)
pl_pcdu_on = PusTelecommand(

18
eive_tmtc/tmtc/power/tm.py
View File

@ -75,8 +75,8 @@ class WdtInfo:
self.pw.dlog(wdt_info)
for idx in range(len(self.wdt_reboots_list)):
self.pw.dlog(
f"{WDT_LIST[idx].ljust(5)} | "
f"{self.wdt_reboots_list[idx]:010} | {self.time_pings_left_list[idx]:010}",
f"{WDT_LIST[idx].ljust(5)} | {self.wdt_reboots_list[idx]:010} |"
f" {self.time_pings_left_list[idx]:010}",
)
def parse(self, wdt_data: bytes, current_idx: int) -> int:
@ -327,8 +327,9 @@ def handle_p60_hk_data(pw: PrintWrapper, set_id: int, hk_data: bytes):
)
pw.dlog(misc_info)
batt_info = (
f"Batt Temp 0 {batt_temp_0 / 10.0} | Batt Temp 1 {batt_temp_1 / 10.0} | "
f"Charge Current {batt_charge_current} | Discharge Current {batt_discharge_current}"
f"Batt Temp 0 {batt_temp_0 / 10.0} | Batt Temp 1 {batt_temp_1 / 10.0} |"
f" Charge Current {batt_charge_current} | Discharge Current"
f" {batt_discharge_current}"
)
pw.dlog(batt_info)
pw.dlog(
@ -419,12 +420,12 @@ def handle_acu_hk_data(pw: PrintWrapper, set_id: int, hk_data: bytes):
)
pw.dlog(f"Boot Cause {boot_cause} | Reset Cause {reset_cause}")
pw.dlog(
f"Ground WDT: Reboot Count {wdt_cnt_gnd} | Time Left {wdt_gnd_time_left} sec"
f"Ground WDT: Reboot Count {wdt_cnt_gnd} | Time Left"
f" {wdt_gnd_time_left} sec"
)
pw.dlog(
"ACU Dev Types: 0:FRAM|1:ADC|2:ADC|3:DAC|4:DAC|"
"5:DAC|6:TempSens|7:Reserved"
"ACU Dev Types: 0:FRAM|1:ADC|2:ADC|3:DAC|4:DAC|5:DAC|6:TempSens|7:Reserved"
)
dev_parser.print(pw=pw)
FsfwTmTcPrinter.get_validity_buffer(
@ -474,7 +475,8 @@ def pdu_config_table_handler(
pw.dlog("[tcs, syrlinks, str, mgt, sus-n, scex, ploc, acs-a, unused]")
elif obj_id.as_bytes == PDU_2_HANDLER_ID:
pw.dlog(
"[obc, pl-pcdu-bat-nom, rw, heaters, sus-r, sa-depl, pl-pcdu-bat-red, acs-b, pl-cam]"
"[obc, pl-pcdu-bat-nom, rw, heaters, sus-r, sa-depl, pl-pcdu-bat-red,"
" acs-b, pl-cam]"
)
out_on_cnt = unpack_array_in_data(custom_data, 0x52, 2, 9, "H")
out_off_cnt = unpack_array_in_data(custom_data, 0x64, 2, 9, "H")

4
eive_tmtc/tmtc/tcs/ctrl.py
View File

@ -12,8 +12,8 @@ from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
generate_one_hk_command,
create_request_one_diag_command,
create_enable_periodic_hk_command_with_interval,
create_request_one_hk_command,
create_enable_periodic_hk_command_with_interval_with_diag,
)
@ -68,7 +68,7 @@ def pack_tcs_ctrl_commands(q: DefaultPusQueueHelper, op_code: str):
)
if op_code == OpCode.ENABLE_TEMP_SET:
interval_seconds = float(input("Please specify interval in seconds: "))
cmds = create_enable_periodic_hk_command_with_interval(
cmds = create_enable_periodic_hk_command_with_interval_with_diag(
False, make_sid(TCS_CONTROLLER, CtrlSetId.PRIMARY_SENSORS), interval_seconds
)
for cmd in cmds:

4
eive_tmtc/tmtc/tm_store.py
View File

@ -148,8 +148,8 @@ def time_prompt_offset_from_now() -> datetime.datetime:
seconds_offset = math.floor(
float(
input(
"Please enter the time as a offset from now in seconds. Negative offset is "
"allowed: "
"Please enter the time as a offset from now in seconds. Negative offset"
" is allowed: "
)
)
)

4
lint.py
View File

@ -32,8 +32,8 @@ def main():
if not os.path.exists("setup.cfg"):
additional_flags_second_step += " --max-line-length=100"
flake8_second_step_cmd = (
f"{python_exe} flake8 . {additional_flags_both_steps} {additional_flags_second_step}"
f" {exclude_dirs_flag}"
f"{python_exe} flake8 . {additional_flags_both_steps} "
f" {additional_flags_second_step} {exclude_dirs_flag}"
)
os.system(flake8_second_step_cmd)

7
pyproject.toml
View File

@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
name = "eive-tmtc"
description = "TMTC Commander EIVE"
readme = "README.md"
dynamic = ["version"]
version = "5.5.1"
requires-python = ">=3.10"
license = {text = "Apache-2.0"}
authors = [
@ -29,7 +29,7 @@ classifiers = [
"Topic :: Scientific/Engineering"
]
dependencies = [
# "tmtccmd ~= 5.0",
# "tmtccmd == 6.0.0rc0",
"python-dateutil ~= 2.8",
"tmtccmd @ git+https://github.com/robamu-org/tmtccmd@main"
]
@ -40,9 +40,6 @@ dependencies = [
[tool.setuptools]
include-package-data = true
[tool.setuptools.dynamic]
version = {attr = "eive_tmtc.__version__"}
# Auto-Discovery is problematic for some reason, so use custom-discovery
[tool.setuptools.packages]
find = {}

2
release-checklist.md
View File

@ -3,7 +3,7 @@ Checklist for new releases
# Pre-Release
1. Bump version inside the `eive_tmtc/__init__.py` file.
1. Bump version inside the `pyproject.toml` file.
2. Update `CHANGELOG.md`: Convert `unreleased` section into version section
with date and new `unreleased`section.
3. Run auto-formatter with `black .`

33
tmtcc.py
View File

@ -2,13 +2,14 @@
import logging
import sys
import time
import traceback
from pathlib import Path
import tmtccmd
from eive_tmtc.cfdp.fault_handler import EiveCfdpFaultHandler
from eive_tmtc.cfdp.tm import CfdpInCcsdsWrapper
from eive_tmtc.cfdp.user import EiveCfdpUser
from spacepackets.version import get_version as get_spacepackets_version
from spacepackets.ecss import PusVerificator
from spacepackets.version import get_version as get_sp_version
from spacepackets.cfdp import (
ChecksumType,
TransmissionMode,
@ -22,28 +23,6 @@ from tmtccmd.cfdp.mib import (
IndicationCfg,
RemoteEntityCfg,
)
try:
import spacepackets
except ImportError as error:
print(error)
print("Python spacepackets module could not be imported")
print(
'Install with "cd spacepackets && python3 -m pip intall -e ." for interative installation'
)
sys.exit(1)
try:
import tmtccmd
except ImportError:
run_tmtc_commander = None
initialize_tmtc_commander = None
tb = traceback.format_exc()
print(tb)
print("Python tmtccmd submodule could not be imported")
sys.exit(1)
from spacepackets.ecss import PusVerificator
from tmtccmd import BackendBase
from tmtccmd.util import FileSeqCountProvider, PusFileSeqCountProvider
from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
@ -67,7 +46,7 @@ from tmtccmd.config.args import (
ProcedureParamsWrapper,
)
from eive_tmtc import APP_LOGGER
from eive_tmtc import __version__
from importlib.metadata import version
from eive_tmtc.config.definitions import (
PUS_APID,
CFDP_APID,
@ -241,8 +220,8 @@ def setup_backend(
def main(): # noqa C901: Complexity okay here.
print(f"-- eive tmtc v{__version__} --")
print(f"-- spacepackets v{get_spacepackets_version()} --")
print(f"-- eive tmtc v{version('eive-tmtc')} --")
print(f"-- spacepackets v{get_sp_version()} --")
add_colorlog_console_logger(_LOGGER)
# TODO: -V CLI argument to enable this?
_LOGGER.setLevel(_LOG_LEVEL)