eive/eive-tmtc
eive/eive-tmtc
14
0
Fork 0
Code Issues Pull Requests 2 Projects Releases 97 Wiki Activity

Merge branch 'main' into meier/rtd

Browse Source
This commit is contained in:
Robin Müller 2023-01-19 16:49:08 +01:00
commit 485f0015d7
13 changed files with 110 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.run/tmtccli.run.xml
View File

@ -1,7 +1,7 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="tmtccli" type="PythonConfigurationType" factoryName="Python">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="INTERPRETER_OPTIONS" value="-X dev" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />

4
CHANGELOG.md
View File

@ -8,6 +8,10 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/).
The [milestone](https://egit.irs.uni-stuttgart.de/eive/eive-obsw/milestones)
list yields a list of all related PRs for each release.
# [unreleased]
- Unify and move Syrlinks module, some bugfixes
# [v2.3.0] and [v2.3.1] 18.01.2023
- Bumped `tmtccmd` version to new alpha release, use some new features

2
eive_tmtc/pus_tc/procedure_packer.py
View File

@ -39,7 +39,7 @@ from eive_tmtc.tmtc.payload.ploc_memory_dumper import pack_ploc_memory_dumper_cm
from eive_tmtc.pus_tc.devs.ccsds_handler import pack_ccsds_handler_test
from eive_tmtc.tmtc.core import pack_core_commands
from eive_tmtc.pus_tc.devs.star_tracker import pack_star_tracker_commands
from eive_tmtc.pus_tc.devs.syrlinks_hk_handler import pack_syrlinks_command
from eive_tmtc.tmtc.syrlinks_hk_handler import pack_syrlinks_command
from eive_tmtc.pus_tc.devs.gps import pack_gps_command
from eive_tmtc.tmtc.acs.acs_board import pack_acs_command
from eive_tmtc.pus_tc.devs.plpcdu import pack_pl_pcdu_commands

2
eive_tmtc/pus_tc/system/proc.py
View File

@ -26,7 +26,7 @@ from eive_tmtc.pus_tc.devs.mgms import MgmLis3SetId as MgmLis3SetIds_0_2
from eive_tmtc.pus_tc.devs.mgms import MgmRm3100SetId as MgmRm3100SetIds_1_3
from eive_tmtc.pus_tc.devs.gyros import AdisGyroSetId as AdisGyroSetIds_0_2
from eive_tmtc.pus_tc.devs.gyros import L3gGyroSetId as L3gGyroSetIds_1_3
from eive_tmtc.pus_tc.devs.syrlinks_hk_handler import SetId as SyrlinksSetIds
from eive_tmtc.tmtc.syrlinks_hk_handler import SetId as SyrlinksSetIds
from eive_tmtc.pus_tc.devs.gps import SetId as GpsSetIds
from eive_tmtc.tmtc.acs.imtq import ImtqSetId
from eive_tmtc.pus_tc.devs.sus import SetId

67
eive_tmtc/pus_tm/devs/syrlinks.py
View File

@ -1,67 +0,0 @@
import struct
from eive_tmtc.pus_tm.defs import PrintWrapper
from eive_tmtc.pus_tc.devs.syrlinks_hk_handler import SetId
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
def handle_syrlinks_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
if set_id == SetId.RX_REGISTERS_DATASET:
return handle_syrlinks_rx_registers_dataset(printer, hk_data)
elif set_id == SetId.TX_REGISTERS_DATASET:
return handle_syrlinks_tx_registers_dataset(printer, hk_data)
else:
pw = PrintWrapper(printer)
pw.dlog(f"Service 3 TM: Syrlinks handler reply with unknown set ID {set_id}")
def handle_syrlinks_rx_registers_dataset(printer: FsfwTmTcPrinter, hk_data: bytes):
pw = PrintWrapper(printer)
header_list = [
"RX Status",
"RX Sensitivity",
"RX Frequency Shift",
"RX IQ Power",
"RX AGC Value",
"RX Demod Eb",
"RX Demod N0",
"RX Datarate",
]
rx_status = hk_data[0]
rx_sensitivity = struct.unpack("!I", hk_data[1:5])
rx_frequency_shift = struct.unpack("!I", hk_data[5:9])
rx_iq_power = struct.unpack("!H", hk_data[9:11])
rx_agc_value = struct.unpack("!H", hk_data[11:13])
rx_demod_eb = struct.unpack("!I", hk_data[13:17])
rx_demod_n0 = struct.unpack("!I", hk_data[17:21])
rx_data_rate = hk_data[21]
content_list = [
rx_status,
rx_sensitivity,
rx_frequency_shift,
rx_iq_power,
rx_agc_value,
rx_demod_eb,
rx_demod_n0,
rx_data_rate,
]
validity_buffer = hk_data[22:]
pw.dlog(str(header_list))
pw.dlog(str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=8)
def handle_syrlinks_tx_registers_dataset(
printer: FsfwTmTcPrinter,
hk_data: bytes,
):
pw = PrintWrapper(printer)
header_list = ["TX Status", "TX Waveform", "TX AGC value"]
tx_status = hk_data[0]
tx_waveform = hk_data[1]
tx_agc_value = struct.unpack("!H", hk_data[2:4])
content_list = [tx_status, tx_waveform, tx_agc_value]
validity_buffer = hk_data[4:]
pw.dlog(str(header_list))
pw.dlog(str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=3)

2
eive_tmtc/pus_tm/hk_handling.py
View File

@ -6,6 +6,7 @@ from eive_tmtc.pus_tm.devs.rad_sensor import handle_rad_sensor_data
from eive_tmtc.pus_tm.devs.sus import handle_sus_hk
from eive_tmtc.tmtc.payload.ploc_supervisor import handle_supv_hk_data
from eive_tmtc.tmtc.acs.reaction_wheels import handle_rw_hk_data
from eive_tmtc.tmtc.syrlinks_hk_handler import handle_syrlinks_hk_data
from eive_tmtc.tmtc.tcs import handle_thermal_controller_hk_data
from tmtccmd.tm.pus_3_fsfw_hk import (
Service3Base,
@ -23,7 +24,6 @@ from eive_tmtc.tmtc.power.tm import (
handle_p60_hk_data,
handle_acu_hk_data,
)
from eive_tmtc.pus_tm.devs.syrlinks import handle_syrlinks_hk_data
from eive_tmtc.tmtc.acs.imtq import (
ImtqSetId,
handle_self_test_data,

1
eive_tmtc/tmtc/__init__.py
View File

@ -1,2 +1,3 @@
from .payload.pl_subsystem import add_payload_subsystem_cmds
from .solar_array_deployment import add_sa_depl_cmds
from .test import add_test_defs

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

@ -258,7 +258,7 @@ def pack_acs_ctrl_command(p: ServiceProviderParams):
elif op_code in OpCodes.ENABLE_RAW_GYR_HK:
q.add_log_cmd(Info.ENABLE_RAW_GYR_HK)
cmd_tuple = enable_periodic_hk_command_with_interval(
False, make_sid(ACS_CONTROLLER, SetId.GYR_RAW_SET), 2.0
True, make_sid(ACS_CONTROLLER, SetId.GYR_RAW_SET), 2.0
)
q.add_pus_tc(cmd_tuple[0])
q.add_pus_tc(cmd_tuple[1])
@ -266,7 +266,7 @@ def pack_acs_ctrl_command(p: ServiceProviderParams):
q.add_log_cmd(Info.DISABLE_RAW_GYR_HK)
q.add_pus_tc(
disable_periodic_hk_command(
False, make_sid(ACS_CONTROLLER, SetId.GYR_RAW_SET)
True, make_sid(ACS_CONTROLLER, SetId.GYR_RAW_SET)
)
)
elif op_code in OpCodes.REQUEST_PROC_GYR_HK:
@ -313,7 +313,7 @@ def pack_acs_ctrl_command(p: ServiceProviderParams):
elif op_code in OpCodes.ENABLE_MEKF_HK:
q.add_log_cmd(Info.ENABLE_MEKF_HK)
cmd_tuple = enable_periodic_hk_command_with_interval(
False, make_sid(ACS_CONTROLLER, SetId.MEKF_DATA), 2.0
True, make_sid(ACS_CONTROLLER, SetId.MEKF_DATA), 2.0
)
q.add_pus_tc(cmd_tuple[0])
q.add_pus_tc(cmd_tuple[1])
@ -321,7 +321,7 @@ def pack_acs_ctrl_command(p: ServiceProviderParams):
q.add_log_cmd(Info.DISABLE_MEKF_HK)
q.add_pus_tc(
disable_periodic_hk_command(
False, make_sid(ACS_CONTROLLER, SetId.MEKF_DATA)
True, make_sid(ACS_CONTROLLER, SetId.MEKF_DATA)
)
)
elif op_code in OpCodes.REQUEST_CTRL_VAL_HK:

2
eive_tmtc/tmtc/solar_array_deployment.py
View File

@ -35,7 +35,7 @@ class ActionId:
@tmtc_definitions_provider
def pack_sa_depl_cmds(defs: TmtcDefinitionWrapper):
def add_sa_depl_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
oce.add(keys=OpCode.MANUAL_DEPLOYMENT, info=Info.MANUAL_DEPLOYMENT)
defs.add_service(

75
eive_tmtc/pus_tc/devs/syrlinks_hk_handler.py → eive_tmtc/tmtc/syrlinks_hk_handler.py
View File

@ -7,6 +7,8 @@
"""
import enum
from eive_tmtc.pus_tm.defs import PrintWrapper
from eive_tmtc.config.definitions import CustomServiceList
from tmtccmd.config.tmtc import (
tmtc_definitions_provider,
@ -14,17 +16,22 @@ from tmtccmd.config.tmtc import (
OpCodeEntry,
)
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import make_sid, generate_one_hk_command
from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
create_request_one_diag_command,
)
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Mode
import struct
from tmtccmd.util import ObjectIdU32
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
class SetId:
RX_REGISTERS_DATASET = 1
TX_REGISTERS_DATASET = 2
TEMPERATURE_SET_ID = 3
class OpCode:
@ -125,11 +132,11 @@ def pack_syrlinks_command(
if op_code in OpCode.HK_RX_REGS:
q.add_log_cmd(f"{prefix}: {Info.HK_RX_REGS}")
sid = make_sid(obyt, SetId.RX_REGISTERS_DATASET)
q.add_pus_tc(generate_one_hk_command(sid))
q.add_pus_tc(create_request_one_diag_command(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)
q.add_pus_tc(generate_one_hk_command(sid))
q.add_pus_tc(create_request_one_diag_command(sid))
if op_code in OpCode.TX_STATUS:
q.add_log_cmd(f"{prefix}: {Info.TX_STATUS}")
command = obyt + struct.pack("!I", CommandId.READ_TX_STATUS)
@ -178,3 +185,65 @@ def pack_syrlinks_command(
q.add_log_cmd("Syrlinks: Disable debug printout")
command = obyt + struct.pack("!I", CommandId.DISABLE_DEBUG)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
def handle_syrlinks_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
if set_id == SetId.RX_REGISTERS_DATASET:
return handle_syrlinks_rx_registers_dataset(printer, hk_data)
elif set_id == SetId.TX_REGISTERS_DATASET:
return handle_syrlinks_tx_registers_dataset(printer, hk_data)
else:
pw = PrintWrapper(printer)
pw.dlog(f"Service 3 TM: Syrlinks handler reply with unknown set ID {set_id}")
def handle_syrlinks_rx_registers_dataset(printer: FsfwTmTcPrinter, hk_data: bytes):
pw = PrintWrapper(printer)
header_list = [
"RX Status",
"RX Sensitivity",
"RX Frequency Shift",
"RX IQ Power",
"RX AGC Value",
"RX Demod Eb",
"RX Demod N0",
"RX Datarate",
]
rx_status = hk_data[0]
rx_sensitivity = struct.unpack("!I", hk_data[1:5])[0]
rx_frequency_shift = struct.unpack("!I", hk_data[5:9])[0]
rx_iq_power = struct.unpack("!H", hk_data[9:11])[0]
rx_agc_value = struct.unpack("!H", hk_data[11:13])[0]
rx_demod_eb = struct.unpack("!I", hk_data[13:17])[0]
rx_demod_n0 = struct.unpack("!I", hk_data[17:21])[0]
rx_data_rate = hk_data[21]
content_list = [
rx_status,
rx_sensitivity,
rx_frequency_shift,
rx_iq_power,
rx_agc_value,
rx_demod_eb,
rx_demod_n0,
rx_data_rate,
]
validity_buffer = hk_data[22:]
for header, content in zip(header_list, content_list):
pw.dlog(f"{header}: {content}")
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=8)
def handle_syrlinks_tx_registers_dataset(
printer: FsfwTmTcPrinter,
hk_data: bytes,
):
pw = PrintWrapper(printer)
header_list = ["TX Status", "TX Waveform", "TX AGC value"]
tx_status = hk_data[0]
tx_waveform = hk_data[1]
tx_agc_value = struct.unpack("!H", hk_data[2:4])
content_list = [tx_status, tx_waveform, tx_agc_value]
validity_buffer = hk_data[4:]
for header, content in zip(header_list, content_list):
pw.dlog(f"{header}: {content}")
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=3)

49
eive_tmtc/tmtc/tcs.py
View File

@ -144,9 +144,6 @@ def handle_thermal_controller_hk_data(
printer.file_logger.info(str(parsed_data))
pp = pprint.PrettyPrinter(depth=4)
pp.pprint(parsed_data)
# tcp_server_sensor_temperatures.report_parsed_hk_data(
# object_id, set_id, parsed_data
# )
elif set_id == SetId.DEVICE_SENSORS:
pw = PrintWrapper(printer)
pw.dlog("Received device temperature data")
@ -184,29 +181,25 @@ def handle_thermal_controller_hk_data(
printer.file_logger.info(str(parsed_data))
pp = pprint.PrettyPrinter(depth=4)
pp.pprint(parsed_data)
# print(parsed_data)
# tcp_server_device_temperatures.report_parsed_hk_data(
# object_id, set_id, parsed_data
# )
elif set_id == SetId.SUS_TEMP_SENSORS:
pass
# pw = PrintWrapper(printer)
# pw.dlog("Received SUS temperature data")
# fmt_str = "!ffffffffffffffffff"
# tm_data = struct.unpack(fmt_str, hk_data[:4 * 18])
# parsed_data = {
# "SUS_0": tm_data[0],
# "SUS_1": tm_data[1],
# "SUS_2": tm_data[2],
# "SUS_3": tm_data[3],
# "SUS_4": tm_data[4],
# "SUS_5": tm_data[5],
# "SUS_6": tm_data[6],
# "SUS_7": tm_data[7],
# "SUS_8": tm_data[8],
# "SUS_9": tm_data[9],
# "SUS_10": tm_data[10],
# "SUS_11": tm_data[11],
# }
# print(parsed_data)
# TODO: Forward data to space simulator
pw = PrintWrapper(printer)
pw.dlog("Received SUS temperature data")
fmt_str = "!ffffffffffffffffff"
tm_data = struct.unpack(fmt_str, hk_data[: 4 * 18])
parsed_data = {
"SUS_0": tm_data[0],
"SUS_1": tm_data[1],
"SUS_2": tm_data[2],
"SUS_3": tm_data[3],
"SUS_4": tm_data[4],
"SUS_5": tm_data[5],
"SUS_6": tm_data[6],
"SUS_7": tm_data[7],
"SUS_8": tm_data[8],
"SUS_9": tm_data[9],
"SUS_10": tm_data[10],
"SUS_11": tm_data[11],
}
printer.file_logger.info(str(parsed_data))
pp = pprint.PrettyPrinter(depth=4)
pp.pprint(parsed_data)

4
eive_tmtc/tmtc/test.py
View File

@ -5,7 +5,7 @@ from tmtccmd.config.tmtc import (
TmtcDefinitionWrapper,
OpCodeEntry,
)
from tmtccmd.pus.s17_test import pack_service_17_ping_command
from tmtccmd.pus.s17_test import create_service_17_ping_command
from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams
@ -42,7 +42,7 @@ def pack_test_command(p: ServiceProviderParams):
q = p.queue_helper
if info.op_code == OpCodes.PING:
q.add_log_cmd("Sending PUS TC [17,1]")
q.add_pus_tc(pack_service_17_ping_command())
q.add_pus_tc(create_service_17_ping_command())
if info.op_code == OpCodes.TRIGGER_EVENT:
q.add_log_cmd("Sending PUS TC Event Trigger [17, 128]")
q.add_pus_tc(PusTelecommand(service=PusService.S17_TEST, subservice=128))

2
setup.cfg
View File

@ -28,7 +28,7 @@ classifiers =
[options]
install_requires =
# tmtccmd @ git+https://github.com/robamu-org/tmtccmd@v4.0.0a1
tmtccmd @ git+https://github.com/robamu-org/tmtccmd@667913bd7dd#egg=tmtccmd
tmtccmd @ git+https://github.com/robamu-org/tmtccmd@fa3c006d3ed#egg=tmtccmd
packages = find:
python_requires = >=3.10
include_package_data = True