Compare commits

...

22 Commits

Author SHA1 Message Date
0c67ce5b08 bump release 2023-01-18 11:50:40 +01:00
9cccefc938 another small bump for tmtccmd 2023-01-18 11:48:51 +01:00
404082c38d this version uses a concrete spacepackets version 2023-01-18 11:38:15 +01:00
5021a88112 update setup.cfg again 2023-01-18 11:35:07 +01:00
15fbfd0f95 Merge branch 'main' of https://egit.irs.uni-stuttgart.de/eive/eive-tmtc 2023-01-18 11:32:21 +01:00
4a02e284b3 changes for updated tmtccmd 2023-01-18 11:12:07 +01:00
bd9c134729 bump tmtccmd 2023-01-17 19:08:00 +01:00
44f15eced7 bump tmtccmd 2023-01-17 18:37:16 +01:00
ff72c02508 annoying 2023-01-17 18:32:53 +01:00
f14bb81c9b and yet another bump 2023-01-17 11:16:16 +01:00
136aaecb77 should be the last of renamings 2023-01-17 11:10:52 +01:00
32db4fabdc bump tmtccmd again 2023-01-17 10:48:57 +01:00
484deb0f97 bump tmtccmd again 2023-01-17 10:44:26 +01:00
ead4c91769 bump tmtccmd again 2023-01-16 17:45:46 +01:00
e5e19a8c04 Merge branch 'main' of https://egit.irs.uni-stuttgart.de/eive/eive-tmtc 2023-01-16 17:11:29 +01:00
92ef500f42 restructure TCS module a bit 2023-01-16 17:11:20 +01:00
43ca7de786 bump setup.cfg again 2023-01-16 15:37:32 +01:00
4839ab4365 update tmtccmd dependency 2023-01-16 15:05:33 +01:00
7bc49f1ff7 Merge branch 'main' of https://egit.irs.uni-stuttgart.de/eive/eive-tmtc 2023-01-16 14:53:46 +01:00
74a439955f use specific commit 2023-01-16 14:53:38 +01:00
aff3ced2c2 start adding acs ctrl modes 2023-01-16 14:48:12 +01:00
e1cd0e9b5c update author field 2023-01-16 14:22:25 +01:00
39 changed files with 350 additions and 307 deletions

View File

@ -8,6 +8,11 @@ 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.
# [v2.3.0] and [v2.3.1] 18.01.2023
- Bumped `tmtccmd` version to new alpha release, use some new features
and fix the breaking changes
# [v2.2.0] 16.01.2023
- A lot of consistency renaming: Single name for enums which are not flag enums,

View File

@ -2,10 +2,10 @@ from pathlib import Path
SW_NAME = "eive-tmtc"
VERSION_MAJOR = 2
VERSION_MINOR = 2
VERSION_REVISION = 0
VERSION_MINOR = 3
VERSION_REVISION = 1
__version__ = "2.2.0"
__version__ = "2.3.1"
EIVE_TMTC_ROOT = Path(__file__).parent
PACKAGE_ROOT = EIVE_TMTC_ROOT.parent

View File

@ -3,7 +3,7 @@ import os
from eive_tmtc import EIVE_TMTC_ROOT
from tmtccmd import get_console_logger
from tmtccmd.fsfw import parse_fsfw_events_csv
from tmtccmd.pus.pus_5_event import EventDictT
from tmtccmd.pus.s5_event import EventDictT
LOGGER = get_console_logger()
DEFAULT_EVENTS_CSV_PATH = EIVE_TMTC_ROOT / "config/events.csv"

View File

@ -12,7 +12,7 @@ from typing import Union
from spacepackets.ecss import PusTelecommand
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd
from tmtccmd.util import ObjectIdU32, ObjectIdBase
@ -63,7 +63,7 @@ class Channel:
def pack_request_config_command(object_id: bytes) -> PusTelecommand:
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id, action_id=GomspaceDeviceActionId.REQUEST_CONFIG_TABLE
)
@ -88,7 +88,7 @@ def pack_get_param_command(
else:
app_data += memory_address
app_data += struct.pack("!B", parameter_size)
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id,
action_id=GomspaceDeviceActionId.PARAM_GET,
user_data=app_data,
@ -103,7 +103,7 @@ def pack_set_float_param_command(
app_data += memory_address
app_data.append(4)
app_data += struct.pack("!f", parameter)
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id, action_id=action_id, user_data=app_data
)
@ -116,7 +116,7 @@ def pack_set_u8_param_command(
app_data += memory_address
app_data.append(1)
app_data.append(parameter)
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id, action_id=action_id, user_data=app_data
)
@ -129,7 +129,7 @@ def pack_set_i8_param_command(
app_data += memory_address
app_data.append(1)
app_data += struct.pack("!b", parameter)
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id, action_id=action_id, user_data=app_data
)
@ -142,7 +142,7 @@ def pack_set_u16_param_command(
app_data += memory_address
app_data.append(2)
app_data += struct.pack("!H", parameter)
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id, action_id=action_id, user_data=app_data
)
@ -155,7 +155,7 @@ def pack_set_i16_param_command(
app_data += memory_address
app_data.append(2)
app_data += struct.pack("!h", parameter)
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id, action_id=action_id, user_data=app_data
)
@ -166,7 +166,7 @@ def pack_set_u32_param_command(object_id: bytes, memory_address: bytes, paramete
app_data += memory_address
app_data.append(4)
app_data += struct.pack("!I", parameter)
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id, action_id=action_id, user_data=app_data
)
@ -177,7 +177,7 @@ def pack_set_i32_param_command(object_id: bytes, memory_address: bytes, paramete
app_data += memory_address
app_data.append(4)
app_data += struct.pack("!i", parameter)
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id, action_id=action_id, user_data=app_data
)
@ -224,7 +224,7 @@ def pack_ping_command(object_id: ObjectIdU32, data: bytearray) -> PusTelecommand
@note The ping request sends the specified data to a gompsace device. These
data are simply copied by the device and then sent back.
"""
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id.as_bytes,
action_id=GomspaceDeviceActionId.PING,
user_data=data,
@ -235,7 +235,7 @@ def pack_gnd_wdt_reset_command(object_id: ObjectIdBase) -> PusTelecommand:
""" " Function to generate the command to reset the watchdog of a gomspace device.
@param object_id Object Id of the gomspace device handler.
"""
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id.as_bytes, action_id=GomspaceDeviceActionId.WDT_RESET
)
@ -244,7 +244,7 @@ def pack_reboot_command(object_id: ObjectIdU32) -> PusTelecommand:
"""Function to generate the command which triggers a reboot of a gomspace device
@param object_id The object id of the gomspace device handler.
"""
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id.as_bytes, action_id=GomspaceDeviceActionId.REBOOT
)
@ -254,6 +254,6 @@ def pack_request_full_hk_table_command(object_id: ObjectIdU32) -> PusTelecommand
device.
@param object_id The object id of the gomspace device handler.
"""
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id.as_bytes, action_id=GomspaceDeviceActionId.REQUEST_HK_TABLE
)

View File

@ -9,10 +9,10 @@ from tmtccmd.config.tmtc import (
)
from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd
from tmtccmd.tc.pus_3_fsfw_hk import generate_one_hk_command, make_sid
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
from tmtccmd.tc.pus_200_fsfw_modes import Subservices as ModeSubservices
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Mode
from tmtccmd.tc.pus_200_fsfw_modes import Subservice as ModeSubservices
class BpxSetId:
@ -66,7 +66,7 @@ def pack_bpx_commands(p: ServiceProviderParams):
q.add_pus_tc(generate_one_hk_command(sid=sid))
if op_code in BpxOpCode.OFF:
q.add_log_cmd("Off mode")
mode_cmd = pack_mode_data(BPX_HANDLER_ID, Modes.OFF, 0)
mode_cmd = pack_mode_data(BPX_HANDLER_ID, Mode.OFF, 0)
q.add_pus_tc(
PusTelecommand(
service=200,
@ -76,7 +76,7 @@ def pack_bpx_commands(p: ServiceProviderParams):
)
if op_code in BpxOpCode.ON:
q.add_log_cmd("On mode")
mode_cmd = pack_mode_data(BPX_HANDLER_ID, Modes.ON, 0)
mode_cmd = pack_mode_data(BPX_HANDLER_ID, Mode.ON, 0)
q.add_pus_tc(
PusTelecommand(
service=200,
@ -87,16 +87,14 @@ def pack_bpx_commands(p: ServiceProviderParams):
if op_code in BpxOpCode.RST_BOOT_CNT:
q.add_log_cmd("Resetting reboot counters")
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=BPX_HANDLER_ID, action_id=BpxActionId.RESET_COUNTERS
)
)
if op_code in BpxOpCode.REQUEST_CFG:
q.add_log_cmd("Requesting configuration struct")
q.add_pus_tc(
make_fsfw_action_cmd(
object_id=BPX_HANDLER_ID, action_id=BpxActionId.GET_CFG
)
create_action_cmd(object_id=BPX_HANDLER_ID, action_id=BpxActionId.GET_CFG)
)
if op_code in BpxOpCode.REQUEST_CFG_HK:
q.add_log_cmd("Requesting configuration struct HK")
@ -105,5 +103,5 @@ def pack_bpx_commands(p: ServiceProviderParams):
if op_code in BpxOpCode.REBOOT:
q.add_log_cmd("Rebooting BPX battery")
q.add_pus_tc(
make_fsfw_action_cmd(object_id=BPX_HANDLER_ID, action_id=BpxActionId.REBOOT)
create_action_cmd(object_id=BPX_HANDLER_ID, action_id=BpxActionId.REBOOT)
)

View File

@ -11,12 +11,12 @@ from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.util.obj_id import ObjectIdU32
from tmtccmd.tc.pus_201_fsfw_health import (
from tmtccmd.pus.s201_fsfw_health import (
pack_set_health_cmd_data,
FsfwHealth,
Subservices,
Subservice,
)
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd
from spacepackets.ecss.tc import PusTelecommand
@ -183,7 +183,7 @@ def health_cmd(
app_data = pack_set_health_cmd_data(object_id=object_id.as_bytes, health=health)
q.add_pus_tc(
PusTelecommand(
service=201, subservice=Subservices.TC_SET_HEALTH, app_data=app_data
service=201, subservice=Subservice.TC_SET_HEALTH, app_data=app_data
)
)
@ -200,6 +200,6 @@ def pack_switch_heater_command(
command.append(switch_nr)
command.append(switch_action)
command.append(COMMAND_SOURCE_PARAM_EXTERNAL)
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=object_id, action_id=ActionIds.SWITCH_HEATER, user_data=command
)

View File

@ -15,10 +15,10 @@ from tmtccmd.tc.pus_3_fsfw_hk import (
disable_periodic_hk_command,
)
from tmtccmd.tc.pus_11_tc_sched import (
generate_enable_tc_sched_cmd,
generate_time_tagged_cmd,
create_enable_tc_sched_cmd,
create_time_tagged_cmd,
)
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes, Subservices
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Mode, Subservice
from tmtccmd.tc.pus_20_params import (
pack_scalar_double_param_app_data,
pack_fsfw_load_param_cmd,
@ -161,9 +161,9 @@ def add_pl_pcdu_cmds(defs: TmtcDefinitionWrapper):
def pack_pl_pcdu_commands(q: DefaultPusQueueHelper, op_code: str):
if op_code in OpCode.SWITCH_ON:
pack_pl_pcdu_mode_cmd(q=q, info=Info.SWITCH_ON, mode=Modes.ON, submode=0)
pack_pl_pcdu_mode_cmd(q=q, info=Info.SWITCH_ON, mode=Mode.ON, submode=0)
if op_code in OpCode.SWITCH_OFF:
pack_pl_pcdu_mode_cmd(q=q, info=Info.SWITCH_OFF, mode=Modes.OFF, submode=0)
pack_pl_pcdu_mode_cmd(q=q, info=Info.SWITCH_OFF, mode=Mode.OFF, submode=0)
if op_code in OpCode.ENABLE_HK:
interval = float(
input("Please enter HK collection interval in floating point seconds: ")
@ -184,7 +184,7 @@ def pack_pl_pcdu_commands(q: DefaultPusQueueHelper, op_code: str):
pack_pl_pcdu_mode_cmd(
q=q,
info=Info.NORMAL_SSR,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=submode_mask_to_submode(
NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON
),
@ -193,35 +193,35 @@ def pack_pl_pcdu_commands(q: DefaultPusQueueHelper, op_code: str):
pack_pl_pcdu_mode_cmd(
q=q,
info=Info.NORMAL_DRO,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=submode_mask_to_submode(NormalSubmodesMask.DRO_ON),
)
if op_code in OpCode.NORMAL_X8:
pack_pl_pcdu_mode_cmd(
q=q,
info=Info.NORMAL_X8,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=submode_mask_to_submode(NormalSubmodesMask.X8_ON),
)
if op_code in OpCode.NORMAL_TX:
pack_pl_pcdu_mode_cmd(
q=q,
info=Info.NORMAL_TX,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=submode_mask_to_submode(NormalSubmodesMask.TX_ON),
)
if op_code in OpCode.NORMAL_MPA:
pack_pl_pcdu_mode_cmd(
q=q,
info=Info.NORMAL_MPA,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=submode_mask_to_submode(NormalSubmodesMask.MPA_ON),
)
if op_code in OpCode.NORMAL_HPA:
pack_pl_pcdu_mode_cmd(
q=q,
info=Info.NORMAL_HPA,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=submode_mask_to_submode(NormalSubmodesMask.HPA_ON),
)
if op_code in OpCode.REQ_OS_HK:
@ -251,15 +251,15 @@ def hpa_on_procedure(q: DefaultPusQueueHelper):
)
pl_pcdu_on = PusTelecommand(
service=200,
subservice=Subservices.TC_MODE_COMMAND,
app_data=pack_mode_data(object_id=PL_PCDU_ID, mode=Modes.ON, submode=0),
subservice=Subservice.TC_MODE_COMMAND,
app_data=pack_mode_data(object_id=PL_PCDU_ID, mode=Mode.ON, submode=0),
)
ssr_on = PusTelecommand(
service=200,
subservice=Subservices.TC_MODE_COMMAND,
subservice=Subservice.TC_MODE_COMMAND,
app_data=pack_mode_data(
object_id=PL_PCDU_ID,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=submode_mask_to_submode(
NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON
),
@ -267,95 +267,95 @@ def hpa_on_procedure(q: DefaultPusQueueHelper):
)
dro_on = PusTelecommand(
service=200,
subservice=Subservices.TC_MODE_COMMAND,
subservice=Subservice.TC_MODE_COMMAND,
app_data=pack_mode_data(
object_id=PL_PCDU_ID,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=submode_mask_to_submode(NormalSubmodesMask.DRO_ON),
),
)
x8_on = PusTelecommand(
service=200,
subservice=Subservices.TC_MODE_COMMAND,
subservice=Subservice.TC_MODE_COMMAND,
app_data=pack_mode_data(
object_id=PL_PCDU_ID,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=submode_mask_to_submode(NormalSubmodesMask.X8_ON),
),
)
tx_on = PusTelecommand(
service=200,
subservice=Subservices.TC_MODE_COMMAND,
subservice=Subservice.TC_MODE_COMMAND,
app_data=pack_mode_data(
object_id=PL_PCDU_ID,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=submode_mask_to_submode(NormalSubmodesMask.TX_ON),
),
)
mpa_on = PusTelecommand(
service=200,
subservice=Subservices.TC_MODE_COMMAND,
subservice=Subservice.TC_MODE_COMMAND,
app_data=pack_mode_data(
object_id=PL_PCDU_ID,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=submode_mask_to_submode(NormalSubmodesMask.MPA_ON),
),
)
hpa_on = PusTelecommand(
service=200,
subservice=Subservices.TC_MODE_COMMAND,
subservice=Subservice.TC_MODE_COMMAND,
app_data=pack_mode_data(
object_id=PL_PCDU_ID,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=submode_mask_to_submode(NormalSubmodesMask.HPA_ON),
),
)
current_time = time.time()
enb_sched = generate_enable_tc_sched_cmd()
enb_sched = create_enable_tc_sched_cmd()
sched_time = int(round(current_time + 10))
q.add_pus_tc(enb_sched)
tagged_on_cmd = generate_time_tagged_cmd(
tagged_on_cmd = create_time_tagged_cmd(
release_time=struct.pack("!I", sched_time),
tc_to_insert=pl_pcdu_on,
)
q.add_pus_tc(tagged_on_cmd)
sched_time += 5
tagged_ssr_cmd = generate_time_tagged_cmd(
tagged_ssr_cmd = create_time_tagged_cmd(
release_time=struct.pack("!I", sched_time),
tc_to_insert=ssr_on,
)
q.add_pus_tc(tagged_ssr_cmd)
sched_time += 5
tagged_dro_cmd = generate_time_tagged_cmd(
tagged_dro_cmd = create_time_tagged_cmd(
release_time=struct.pack("!I", sched_time), tc_to_insert=dro_on
)
q.add_pus_tc(tagged_dro_cmd)
sched_time += delay_dro_to_x8
sched_time = int(round(sched_time))
tagged_x8_cmd = generate_time_tagged_cmd(
tagged_x8_cmd = create_time_tagged_cmd(
release_time=struct.pack("!I", sched_time), tc_to_insert=x8_on
)
q.add_pus_tc(tagged_x8_cmd)
sched_time += 5
tagged_tx_cmd = generate_time_tagged_cmd(
tagged_tx_cmd = create_time_tagged_cmd(
release_time=struct.pack("!I", sched_time), tc_to_insert=tx_on
)
q.add_pus_tc(tagged_tx_cmd)
sched_time += 5
tagged_mpa_cmd = generate_time_tagged_cmd(
tagged_mpa_cmd = create_time_tagged_cmd(
release_time=struct.pack("!I", sched_time), tc_to_insert=mpa_on
)
q.add_pus_tc(tagged_mpa_cmd)
sched_time += 5
tagged_hpa_cmd = generate_time_tagged_cmd(
tagged_hpa_cmd = create_time_tagged_cmd(
release_time=struct.pack("!I", sched_time), tc_to_insert=hpa_on
)
q.add_pus_tc(tagged_hpa_cmd)
@ -439,12 +439,12 @@ def pack_failure_injection_cmd(q: DefaultPusQueueHelper, param_id: int, print_st
def pack_pl_pcdu_mode_cmd(
q: DefaultPusQueueHelper, info: str, mode: Modes, submode: int
q: DefaultPusQueueHelper, info: str, mode: Mode, submode: int
):
q.add_log_cmd(info)
mode_data = pack_mode_data(object_id=PL_PCDU_ID, mode=mode, submode=submode)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=mode_data
service=200, subservice=Subservice.TC_MODE_COMMAND, app_data=mode_data
)
)

View File

@ -10,7 +10,7 @@ import struct
from eive_tmtc.config.definitions import CustomServiceList
from spacepackets.ecss.tc import PusTelecommand
from eive_tmtc.pus_tc.service_200_mode import pack_mode_data, Modes
from eive_tmtc.pus_tc.service_200_mode import pack_mode_data, Mode
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
@ -69,11 +69,11 @@ def pack_rad_sensor_test_into(
q.add_log_cmd(f"Commanding Radiation sensor handler {object_id}")
if op_code in OpCode.ON:
rad_sensor_mode_cmd(object_id, Modes.ON, Info.ON, q)
rad_sensor_mode_cmd(object_id, Mode.ON, Info.ON, q)
if op_code in OpCode.NORMAL:
rad_sensor_mode_cmd(object_id, Modes.NORMAL, Info.NORMAL, q)
rad_sensor_mode_cmd(object_id, Mode.NORMAL, Info.NORMAL, q)
if op_code in OpCode.OFF:
rad_sensor_mode_cmd(object_id, Modes.OFF, Info.OFF, q)
rad_sensor_mode_cmd(object_id, Mode.OFF, Info.OFF, q)
if op_code in OpCode.REQ_HK_ONCE:
q.add_log_cmd(f"Rad sensor: {Info.REQ_OS_HK}")
q.add_pus_tc(
@ -90,7 +90,7 @@ def pack_rad_sensor_test_into(
def rad_sensor_mode_cmd(
object_id: ObjectIdU32, mode: Modes, info: str, q: DefaultPusQueueHelper
object_id: ObjectIdU32, mode: Mode, info: str, q: DefaultPusQueueHelper
):
q.add_log_cmd(f"Rad sensor: {info}")
mode_data = pack_mode_data(object_id.as_bytes, mode, 0)

View File

@ -8,7 +8,7 @@ from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.util import ObjectIdU32
from tmtccmd.tc.pus_200_fsfw_modes import Modes, pack_mode_data, Subservices
from tmtccmd.tc.pus_200_fsfw_modes import Mode, pack_mode_data, Subservice
import eive_tmtc.config.object_ids as oids
from eive_tmtc.config.object_ids import get_object_ids
@ -68,30 +68,28 @@ def pack_rtd_commands(
object_id_dict = get_object_ids()
object_id = object_id_dict.get(RTD_IDS[tgt_rtd_idx])
if op_code in OpCode.ON:
app_data = pack_mode_data(
object_id=object_id.as_bytes, mode=Modes.ON, submode=0
)
app_data = pack_mode_data(object_id=object_id.as_bytes, mode=Mode.ON, submode=0)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=app_data
service=200, subservice=Subservice.TC_MODE_COMMAND, app_data=app_data
)
)
if op_code in OpCode.NORMAL:
app_data = pack_mode_data(
object_id=object_id.as_bytes, mode=Modes.NORMAL, submode=0
object_id=object_id.as_bytes, mode=Mode.NORMAL, submode=0
)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=app_data
service=200, subservice=Subservice.TC_MODE_COMMAND, app_data=app_data
)
)
if op_code in OpCode.OFF:
app_data = pack_mode_data(
object_id=object_id.as_bytes, mode=Modes.OFF, submode=0
object_id=object_id.as_bytes, mode=Mode.OFF, submode=0
)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=app_data
service=200, subservice=Subservice.TC_MODE_COMMAND, app_data=app_data
)
)
if op_code in OpCode.CONFIG_CMD:

View File

@ -5,10 +5,10 @@ from spacepackets.ecss import PusTelecommand
from eive_tmtc.config.definitions import CustomServiceList
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc.pus_200_fsfw_modes import Modes, pack_mode_data, Subservices
from tmtccmd.tc.pus_200_fsfw_modes import Mode, pack_mode_data, Subservice
from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd
from tmtccmd.config import OpCodeEntry, TmtcDefinitionWrapper
from eive_tmtc.config.object_ids import SCEX_HANDLER_ID
@ -83,8 +83,8 @@ def pack_scex_cmds(p: ServiceProviderParams):
q.add_pus_tc(
PusTelecommand(
service=200,
subservice=Subservices.TC_MODE_COMMAND,
app_data=pack_mode_data(SCEX_HANDLER_ID, Modes.ON, 0),
subservice=Subservice.TC_MODE_COMMAND,
app_data=pack_mode_data(SCEX_HANDLER_ID, Mode.ON, 0),
)
)
if op_code in OpCode.SWITCH_OFF:
@ -92,28 +92,28 @@ def pack_scex_cmds(p: ServiceProviderParams):
q.add_pus_tc(
PusTelecommand(
service=200,
subservice=Subservices.TC_MODE_COMMAND,
app_data=pack_mode_data(SCEX_HANDLER_ID, Modes.OFF, 0),
subservice=Subservice.TC_MODE_COMMAND,
app_data=pack_mode_data(SCEX_HANDLER_ID, Mode.OFF, 0),
)
)
if op_code in OpCode.PING:
q.add_log_cmd(Info.PING)
app_data = bytes([0])
q.add_pus_tc(make_fsfw_action_cmd(SCEX_HANDLER_ID, ActionId.PING, app_data))
q.add_pus_tc(create_action_cmd(SCEX_HANDLER_ID, ActionId.PING, app_data))
if op_code in OpCode.ION_CMD:
q.add_log_cmd(Info.ION_CMD)
app_data = bytes([0])
q.add_pus_tc(make_fsfw_action_cmd(SCEX_HANDLER_ID, ActionId.ION_CMD, app_data))
q.add_pus_tc(create_action_cmd(SCEX_HANDLER_ID, ActionId.ION_CMD, app_data))
if op_code in OpCode.TEMP_CMD:
q.add_log_cmd(Info.TEMP_CMD)
app_data = bytes([0])
q.add_pus_tc(make_fsfw_action_cmd(SCEX_HANDLER_ID, ActionId.TEMP_CMD, app_data))
q.add_pus_tc(create_action_cmd(SCEX_HANDLER_ID, ActionId.TEMP_CMD, app_data))
if op_code in OpCode.EXP_STATUS_CMD:
q.add_log_cmd(Info.EXP_STATUS_CMD)
app_data = bytes([0])
q.add_pus_tc(
make_fsfw_action_cmd(SCEX_HANDLER_ID, ActionId.EXP_STATUS_CMD, app_data)
create_action_cmd(SCEX_HANDLER_ID, ActionId.EXP_STATUS_CMD, app_data)
)
# one cell
@ -162,7 +162,7 @@ def pack_scex_cmds(p: ServiceProviderParams):
app_data.append(dac_weight3[cn])
q.add_pus_tc(
make_fsfw_action_cmd(SCEX_HANDLER_ID, ActionId.ONE_CELLS_CMD, app_data)
create_action_cmd(SCEX_HANDLER_ID, ActionId.ONE_CELLS_CMD, app_data)
)
if op_code in OpCode.ALL_CELLS_CMD:
@ -194,13 +194,13 @@ def pack_scex_cmds(p: ServiceProviderParams):
app_data.append(dac_weight3[cn])
q.add_pus_tc(
make_fsfw_action_cmd(SCEX_HANDLER_ID, ActionId.ALL_CELLS_CMD, app_data)
create_action_cmd(SCEX_HANDLER_ID, ActionId.ALL_CELLS_CMD, app_data)
)
if op_code in OpCode.FRAM:
q.add_log_cmd(Info.FRAM)
app_data = bytes([0])
q.add_pus_tc(make_fsfw_action_cmd(SCEX_HANDLER_ID, ActionId.FRAM, app_data))
q.add_pus_tc(create_action_cmd(SCEX_HANDLER_ID, ActionId.FRAM, app_data))
def append_16_bit_val(packet: bytearray, val: int):

View File

@ -9,7 +9,7 @@ import enum
import struct
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Mode
from tmtccmd.logging import get_console_logger
from tmtccmd.tc import DefaultPusQueueHelper
@ -164,23 +164,23 @@ def pack_star_tracker_commands(
obyt = object_id.as_bytes
if op_code == "0":
q.add_log_cmd("Star tracker: Mode On, Submode Bootloader")
data = pack_mode_data(obyt, Modes.ON, Submode.BOOTLOADER)
data = pack_mode_data(obyt, Mode.ON, Submode.BOOTLOADER)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data))
if op_code == "1":
q.add_log_cmd("Star tracker: Mode On, Submode Firmware")
data = pack_mode_data(obyt, Modes.ON, Submode.FIRMWARE)
data = pack_mode_data(obyt, Mode.ON, Submode.FIRMWARE)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data))
if op_code == "2":
q.add_log_cmd("Star tracker: Mode Normal")
data = pack_mode_data(obyt, Modes.NORMAL, 0)
data = pack_mode_data(obyt, Mode.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data))
if op_code == "3":
q.add_log_cmd("Star tracker: Mode Off")
data = pack_mode_data(obyt, Modes.OFF, 0)
data = pack_mode_data(obyt, Mode.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data))
if op_code == "4":
q.add_log_cmd("Star tracker: Mode Raw")
data = pack_mode_data(obyt, Modes.RAW, 0)
data = pack_mode_data(obyt, Mode.RAW, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data))
if op_code == "5":
q.add_log_cmd("Star tracker: Ping")

View File

@ -16,7 +16,7 @@ from tmtccmd.config.tmtc import (
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import make_sid, generate_one_hk_command
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Mode
import struct
from tmtccmd.util import ObjectIdU32
@ -100,15 +100,15 @@ def pack_syrlinks_command(
q.add_log_cmd(f"Testing Syrlinks with object id: {object_id.as_hex_string}")
if op_code == OpCode.OFF:
q.add_log_cmd(f"{prefix}: Set mode off")
data = pack_mode_data(obyt, Modes.OFF, 0)
data = pack_mode_data(obyt, Mode.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data))
if op_code == OpCode.ON:
q.add_log_cmd(f"{prefix}: Set mode on")
data = pack_mode_data(obyt, Modes.ON, 0)
data = pack_mode_data(obyt, Mode.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data))
if op_code == OpCode.NORMAL:
q.add_log_cmd(f"{prefix}: Mode Normal")
data = pack_mode_data(obyt, Modes.NORMAL, 0)
data = pack_mode_data(obyt, Mode.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data))
if op_code == OpCode.STANDBY:
q.add_log_cmd(f"{prefix}: Set TX mode standby")

View File

@ -5,10 +5,13 @@
@author J. Meier
@date 06.01.2021
"""
import enum
from spacepackets.ecss.tc import PusTelecommand
from eive_tmtc.pus_tc.service_200_mode import pack_mode_data
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_200_fsfw_modes import Modes
from tmtccmd.tc.pus_200_fsfw_modes import Mode
from tmtccmd.pus.s8_fsfw_funccmd import make_action_id
from tmtccmd.util import ObjectIdU32
@ -28,9 +31,9 @@ class Tmp1075TestProcedure:
set_mode_on = False # If mode is MODE_ON, temperature will only be read on command
class Tmp1075ActionId:
get_temp = bytearray([0x0, 0x0, 0x0, 0x01])
start_adc_conversion = bytearray([0x0, 0x0, 0x0, 0x02])
class Tmp1075ActionId(enum.IntEnum):
GET_TEMP = 1
START_ADC_CONV = 2
def pack_tmp1075_test_into(
@ -42,21 +45,21 @@ def pack_tmp1075_test_into(
obyt = object_id.as_bytes
if Tmp1075TestProcedure.all or Tmp1075TestProcedure.start_adc_conversion:
q.add_log_cmd("TMP1075: Starting new temperature conversion")
command = obyt + Tmp1075ActionId.start_adc_conversion
command = obyt + make_action_id(Tmp1075ActionId.GET_TEMP)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if Tmp1075TestProcedure.all or Tmp1075TestProcedure.get_temp:
q.add_log_cmd("TMP1075: Read temperature")
command = obyt + Tmp1075ActionId.get_temp
command = obyt + make_action_id(Tmp1075ActionId.START_ADC_CONV)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if Tmp1075TestProcedure.set_mode_normal:
q.add_log_cmd("TMP1075: Set Mode Normal")
mode_data = pack_mode_data(obyt, Modes.NORMAL, 0)
mode_data = pack_mode_data(obyt, Mode.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
if Tmp1075TestProcedure.set_mode_on:
q.add_log_cmd("TMP1075: Set Mode On")
mode_data = pack_mode_data(obyt, Modes.ON, 0)
mode_data = pack_mode_data(obyt, Mode.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
return q

View File

@ -9,6 +9,7 @@ from eive_tmtc.pus_tc.system.controllers import (
pack_cmd_ctrl_to_prompted_mode,
get_object_from_op_code,
)
from eive_tmtc.tmtc.tcs import pack_tcs_sys_commands
from tmtccmd import DefaultProcedureInfo, TcHandlerBase
from tmtccmd.config import CoreServiceList
from tmtccmd.logging import get_console_logger
@ -20,7 +21,6 @@ from tmtccmd.tc.decorator import (
from tmtccmd.tc.pus_5_event import (
pack_generic_service_5_test_into,
)
from tmtccmd.pus.pus_17_test import pack_service_17_ping_command
from eive_tmtc.pus_tc.service_200_mode import pack_service_200_test_into
from eive_tmtc.tmtc.power.p60dock import pack_p60dock_cmds
@ -42,10 +42,8 @@ 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.pus_tc.devs.gps import pack_gps_command
from eive_tmtc.tmtc.acs.acs_board import pack_acs_command
from eive_tmtc.tmtc.acs.sus_board import pack_sus_cmds
from eive_tmtc.pus_tc.devs.plpcdu import pack_pl_pcdu_commands
from eive_tmtc.pus_tc.devs.str_img_helper import pack_str_img_helper_command
from eive_tmtc.pus_tc.system.tcs import pack_tcs_sys_commands
from eive_tmtc.pus_tc.system.proc import pack_proc_commands
from eive_tmtc.config.definitions import CustomServiceList
from eive_tmtc.config.object_ids import (

View File

@ -7,7 +7,7 @@
"""
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Mode
from eive_tmtc.config.object_ids import TEST_DEVICE_ID
TEST_DEVICE_OBJ_ID = TEST_DEVICE_ID
@ -19,17 +19,17 @@ def pack_service_200_test_into(q: DefaultPusQueueHelper):
obj_id = TEST_DEVICE_OBJ_ID
# Set On Mode
q.add_log_cmd("Testing Service 200: Set Mode On")
mode_data = pack_mode_data(obj_id, Modes.ON, 0)
mode_data = pack_mode_data(obj_id, Mode.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
# Set Normal mode
q.add_log_cmd("Testing Service 200: Set Mode Normal")
mode_data = pack_mode_data(obj_id, Modes.NORMAL, 0)
mode_data = pack_mode_data(obj_id, Mode.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
# Set Raw Mode
q.add_log_cmd("Testing Service 200: Set Mode Raw")
mode_data = pack_mode_data(obj_id, Modes.RAW, 0)
mode_data = pack_mode_data(obj_id, Mode.RAW, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
# Set Off Mode
q.add_log_cmd("Testing Service 200: Set Mode Off")
mode_data = pack_mode_data(obj_id, Modes.OFF, 0)
mode_data = pack_mode_data(obj_id, Mode.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))

View File

@ -1,7 +1,7 @@
from typing import Union
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_200_fsfw_modes import Modes
from tmtccmd.tc.pus_200_fsfw_modes import Mode
from tmtccmd.util import ObjectIdU32, ObjectIdBase
from eive_tmtc.tmtc.common import pack_mode_cmd_with_info
@ -54,7 +54,7 @@ def pack_cmd_ctrl_to_off(
):
pack_mode_cmd_with_info(
object_id=object_id.as_bytes,
mode=Modes.OFF,
mode=Mode.OFF,
submode=0,
q=q,
info=f"Commanding {object_id} OFF",
@ -64,7 +64,7 @@ def pack_cmd_ctrl_to_off(
def pack_cmd_ctrl_to_on(q: DefaultPusQueueHelper, object_id: ObjectIdU32):
pack_mode_cmd_with_info(
object_id=object_id.as_bytes,
mode=Modes.ON,
mode=Mode.ON,
submode=0,
q=q,
info=f"Commanding {object_id} ON",
@ -76,7 +76,7 @@ def pack_cmd_ctrl_to_nml(
):
pack_mode_cmd_with_info(
object_id=object_id.as_bytes,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=0,
q=q,
info=f"Commanding {object_id} NORMAL",

View File

@ -5,21 +5,19 @@ from datetime import timedelta
from eive_tmtc.config.definitions import CustomServiceList
from eive_tmtc.config.object_ids import get_object_ids
from eive_tmtc.pus_tc.system.tcs import pack_tcs_sys_commands
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.decorator import ServiceProviderParams
from tmtccmd.tc.pus_11_tc_sched import (
generate_time_tagged_cmd,
generate_enable_tc_sched_cmd,
generate_reset_tc_sched_cmd,
from tmtccmd.pus.s11_tc_sched import (
create_time_tagged_cmd,
create_enable_tc_sched_cmd,
create_reset_tc_sched_cmd,
)
from tmtccmd.tc.pus_3_fsfw_hk import *
import eive_tmtc.config.object_ids as oids
from eive_tmtc.pus_tc.system.tcs import OpCode as TcsOpCodes
from eive_tmtc.tmtc.tcs import OpCode as TcsOpCodes, pack_tcs_sys_commands
from eive_tmtc.pus_tc.devs.bpx_batt import BpxSetId
from eive_tmtc.tmtc.core import SetId as CoreSetIds
from eive_tmtc.tmtc.power.common_power import SetId as GsSetIds
@ -209,7 +207,7 @@ def pack_generic_hk_listening_cmds(
else:
for cmd in disable_cmd_list:
q.add_pus_tc(
generate_time_tagged_cmd(
create_time_tagged_cmd(
release_time=struct.pack("!I", int(current_time)), tc_to_insert=cmd
)
)
@ -225,7 +223,7 @@ def pack_proc_commands(q: DefaultPusQueueHelper, op_code: str):
obj_id_dict = get_object_ids()
if op_code in OpCode.RESET_SCHED:
q.add_log_cmd("Resetting/Clearing TC schedule")
q.add_pus_tc(generate_reset_tc_sched_cmd())
q.add_pus_tc(create_reset_tc_sched_cmd())
if op_code in OpCode.BAT_FT:
key = KAI.BAT_FT[0]
sid_list.append(make_sid(oids.BPX_HANDLER_ID, BpxSetId.GET_HK_SET))
@ -300,7 +298,7 @@ def pack_proc_commands(q: DefaultPusQueueHelper, op_code: str):
if op_code in OpCode.TV_SETUP_TCS_FT_ON:
# Enable scheduling
q.add_pus_tc(generate_enable_tc_sched_cmd())
q.add_pus_tc(create_enable_tc_sched_cmd())
# check whether tcs_assembly also has to be commanded to NORMAL Mode
pack_tcs_sys_commands(q=q, op_code=TcsOpCodes.TCS_BOARD_ASS_NORMAL[0])
pack_cmd_ctrl_to_nml(q=q, object_id=obj_id_dict.get(oids.THERMAL_CONTROLLER_ID))
@ -736,7 +734,7 @@ def enable_listen_to_hk_for_x_seconds(
)
for cmd in cmd_tuple:
q.add_pus_tc(cmd)
generate_time_tagged_cmd(
create_time_tagged_cmd(
struct.pack("!I", int(round(time.time() + interval_seconds))),
gen_disable_listen_to_hk_for_x_seconds(q, diag, device, sid),
)

View File

@ -1,78 +0,0 @@
import enum
from eive_tmtc.config.definitions import CustomServiceList
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_200_fsfw_modes import Modes
from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
generate_one_hk_command,
)
from eive_tmtc.tmtc.common import pack_mode_cmd_with_info
from eive_tmtc.config.object_ids import TCS_BOARD_ASS_ID, TCS_CONTROLLER
class OpCode:
TCS_BOARD_ASS_NORMAL = ["0", "normal"]
TCS_BOARD_ASS_OFF = ["1", "off"]
REQUEST_SENSOR_TEMP_SET = ["2", "temps"]
class Info:
REQUEST_SENSOR_TEMP_SET = "Request HK set of primary sensor temperatures"
TCS_BOARD_ASS_NORMAL = "Switching TCS board assembly on"
TCS_BOARD_ASS_OFF = "Switching TCS board assembly off"
class SetId(enum.IntEnum):
PRIMARY_SENSORS = 0
DEVICE_SENSORS = 1
SUS_TEMP_SENSORS = 2
@tmtc_definitions_provider
def add_tcs_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
oce.add(
keys=OpCode.TCS_BOARD_ASS_NORMAL,
info=Info.TCS_BOARD_ASS_NORMAL,
)
oce.add(
keys=OpCode.TCS_BOARD_ASS_OFF,
info=Info.TCS_BOARD_ASS_OFF,
)
oce.add(keys=OpCode.REQUEST_SENSOR_TEMP_SET, info=Info.REQUEST_SENSOR_TEMP_SET)
defs.add_service(
name=CustomServiceList.TCS.value,
info="TCS Board Assembly",
op_code_entry=oce,
)
def pack_tcs_sys_commands(q: DefaultPusQueueHelper, op_code: str):
if op_code in OpCode.REQUEST_SENSOR_TEMP_SET:
sensor_set_sid = make_sid(TCS_CONTROLLER, SetId.PRIMARY_SENSORS)
q.add_log_cmd(Info.REQUEST_SENSOR_TEMP_SET)
q.add_pus_tc(generate_one_hk_command(sensor_set_sid))
pack_tcs_ass_cmds(q, op_code)
def pack_tcs_ass_cmds(q: DefaultPusQueueHelper, op_code: str):
if op_code in OpCode.TCS_BOARD_ASS_NORMAL:
pack_mode_cmd_with_info(
object_id=TCS_BOARD_ASS_ID,
mode=Modes.NORMAL,
submode=0,
q=q,
info=Info.TCS_BOARD_ASS_NORMAL,
)
if op_code in OpCode.TCS_BOARD_ASS_OFF:
pack_mode_cmd_with_info(
object_id=TCS_BOARD_ASS_ID,
mode=Modes.OFF,
submode=0,
q=q,
info=Info.TCS_BOARD_ASS_OFF,
)

View File

@ -9,6 +9,7 @@ from eive_tmtc.tmtc.power.tm import handle_get_param_data_reply
from tmtccmd.logging import get_console_logger
from tmtccmd.tm import Service8FsfwTm
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
from spacepackets.ccsds.time import CdsShortTimestamp
LOGGER = get_console_logger()
@ -19,7 +20,9 @@ def handle_action_reply(
"""Core Action reply handler
:return:
"""
tm_packet = Service8FsfwTm.unpack(raw_telemetry=raw_tm)
tm_packet = Service8FsfwTm.unpack(
raw_telemetry=raw_tm, time_reader=CdsShortTimestamp.empty()
)
printer.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
object_id = obj_id_dict.get(tm_packet.source_object_id_as_bytes)
pw = PrintWrapper(printer)

View File

@ -5,19 +5,20 @@ from eive_tmtc.config.object_ids import get_object_ids
from eive_tmtc.pus_tm.defs import PrintWrapper
from eive_tmtc.pus_tm.verification_handler import generic_retval_printout
from eive_tmtc.tmtc.acs.acs_subsystem import AcsMode
from tmtccmd.tc.pus_200_fsfw_modes import Modes
from tmtccmd.tc.pus_200_fsfw_modes import Mode
from tmtccmd.tm import Service5Tm
from tmtccmd.logging import get_console_logger
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
from tmtccmd.fsfw import EventInfo
from spacepackets.ccsds.time import CdsShortTimestamp
LOGGER = get_console_logger()
def handle_event_packet(raw_tm: bytes, printer: FsfwTmTcPrinter):
pw = PrintWrapper(printer)
tm = Service5Tm.unpack(raw_telemetry=raw_tm)
tm = Service5Tm.unpack(raw_telemetry=raw_tm, time_reader=CdsShortTimestamp.empty())
event_dict = get_event_dict()
info = event_dict.get(tm.event_id)
if info is None:
@ -55,7 +56,7 @@ def handle_event_packet(raw_tm: bytes, printer: FsfwTmTcPrinter):
if info.name == "MODE_INFO":
mode_name = "Unknown"
if obj_name == "ACS_SUBSYSTEM":
if tm.param_1 == Modes.OFF:
if tm.param_1 == Mode.OFF:
mode_name = "Off"
elif tm.param_1 == AcsMode.IDLE:
mode_name = "Idle"
@ -66,13 +67,13 @@ def handle_event_packet(raw_tm: bytes, printer: FsfwTmTcPrinter):
elif tm.param_1 == AcsMode.TARGET_PT:
mode_name = "Target Pointing"
else:
if tm.param_1 == Modes.OFF:
if tm.param_1 == Mode.OFF:
mode_name = "Off"
elif tm.param_1 == Modes.ON:
elif tm.param_1 == Mode.ON:
mode_name = "On"
elif tm.param_1 == Modes.NORMAL:
elif tm.param_1 == Mode.NORMAL:
mode_name = "Normal"
elif tm.param_1 == Modes.RAW:
elif tm.param_1 == Mode.RAW:
mode_name = "Raw"
pw.dlog(f"Mode Number {tm.param_1}, Mode Name {mode_name}")
pw.dlog(f"Submode: {tm.param_2}")

View File

@ -4,11 +4,12 @@ from eive_tmtc.config.object_ids import get_object_ids
from spacepackets.ecss import PusTelemetry
from spacepackets.ecss.pus_17_test import Service17Tm
from spacepackets.util import PrintFormats
from spacepackets.ccsds.time import CdsShortTimestamp
from tmtccmd import get_console_logger
from tmtccmd.logging.pus import RawTmtcTimedLogWrapper
from tmtccmd.pus import VerificationWrapper
from tmtccmd.tm import Service20FsfwTm, Service200FsfwTm
from tmtccmd.tm.pus_200_fsfw_modes import Subservices as ModeSubservices
from tmtccmd.tm.pus_200_fsfw_modes import Subservice as ModeSubservices
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
from .defs import PrintWrapper
@ -30,7 +31,7 @@ def pus_factory_hook(
LOGGER.warning("Detected packet shorter than 8 bytes!")
return
try:
tm_packet = PusTelemetry.unpack(packet)
tm_packet = PusTelemetry.unpack(packet, CdsShortTimestamp.empty())
except ValueError:
LOGGER.warning("Could not generate PUS TM object from raw data")
LOGGER.warning(f"Raw Packet: [{packet.hex(sep=',')}], REPR: {packet!r}")
@ -48,15 +49,21 @@ def pus_factory_hook(
elif service == 8:
handle_action_reply(raw_tm=packet, printer=printer, obj_id_dict=obj_id_dict)
elif service == 17:
tm_packet = Service17Tm.unpack(raw_telemetry=packet)
tm_packet = Service17Tm.unpack(
raw_telemetry=packet, time_reader=CdsShortTimestamp.empty()
)
if tm_packet.subservice == 2:
verif_wrapper.dlog("Received Ping Reply TM[17,2]")
dedicated_handler = True
elif service == 20:
tm_packet = Service20FsfwTm.unpack(raw_telemetry=packet)
tm_packet = Service20FsfwTm.unpack(
raw_telemetry=packet, time_reader=CdsShortTimestamp.empty()
)
dedicated_handler = False
elif service == 200:
tm_packet = Service200FsfwTm.unpack(raw_telemetry=packet)
tm_packet = Service200FsfwTm.unpack(
raw_telemetry=packet, time_reader=CdsShortTimestamp.empty()
)
if tm_packet.subservice == ModeSubservices.TM_CANT_REACH_MODE:
obj_id = tm_packet.object_id
obj_id_obj = obj_id_dict.get(obj_id)

View File

@ -4,9 +4,9 @@ from eive_tmtc.tmtc.acs.acs_ctrl import handle_raw_mgm_data, handle_acs_ctrl_hk_
from eive_tmtc.pus_tm.devs.plpcdu import handle_plpcdu_hk
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.pus_tm.system.tcs import handle_thermal_controller_hk_data
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.tcs import handle_thermal_controller_hk_data
from tmtccmd.tm.pus_3_fsfw_hk import (
Service3Base,
HkContentType,

View File

@ -1,5 +1,6 @@
from typing import List, Optional
from spacepackets.ccsds import CdsShortTimestamp
from spacepackets.ecss.pus_1_verification import UnpackParams, Service1Tm
from tmtccmd.logging import get_console_logger
from tmtccmd.pus import VerificationWrapper
@ -15,7 +16,9 @@ def handle_service_1_fsfw_packet(wrapper: VerificationWrapper, raw_tm: bytes):
"Console logger or file logger not valid. Please set a valid one"
)
# Error code with length 2 is FSFW specific
tm_packet = Service1Tm.unpack(data=raw_tm, params=UnpackParams(1, 2))
tm_packet = Service1Tm.unpack(
data=raw_tm, params=UnpackParams(CdsShortTimestamp.empty(), 1, 2)
)
fsfw_wrapper = Service1FsfwWrapper(tm_packet)
res = wrapper.verificator.add_tm(tm_packet)
if res is None:

View File

@ -8,7 +8,7 @@ from tmtccmd.config.tmtc import (
)
from tmtccmd.tc import service_provider, DefaultPusQueueHelper
from tmtccmd.tc.decorator import ServiceProviderParams
from tmtccmd.tc.pus_200_fsfw_modes import Modes
from tmtccmd.tc.pus_200_fsfw_modes import Mode
from eive_tmtc.config.object_ids import ACS_BOARD_ASS_ID
from eive_tmtc.tmtc.common import pack_mode_cmd_with_info
@ -34,7 +34,7 @@ def pack_acs_command(q: DefaultPusQueueHelper, op_code: str):
if op_code in AcsOpCodes.ACS_ASS_A_SIDE:
pack_mode_cmd_with_info(
object_id=ACS_BOARD_ASS_ID,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=DualSideSubmode.A_SIDE,
q=q,
info="Switching to ACS board assembly A side",
@ -42,7 +42,7 @@ def pack_acs_command(q: DefaultPusQueueHelper, op_code: str):
if op_code in AcsOpCodes.ACS_ASS_B_SIDE:
pack_mode_cmd_with_info(
object_id=ACS_BOARD_ASS_ID,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=DualSideSubmode.B_SIDE,
q=q,
info="Switching to ACS board assembly B side",
@ -50,7 +50,7 @@ def pack_acs_command(q: DefaultPusQueueHelper, op_code: str):
if op_code in AcsOpCodes.ACS_ASS_DUAL_MODE:
pack_mode_cmd_with_info(
object_id=ACS_BOARD_ASS_ID,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=DualSideSubmode.DUAL_SIDE,
q=q,
info="Switching to ACS board assembly dual mode",
@ -58,7 +58,7 @@ def pack_acs_command(q: DefaultPusQueueHelper, op_code: str):
if op_code in AcsOpCodes.ACS_ASS_A_ON:
pack_mode_cmd_with_info(
object_id=ACS_BOARD_ASS_ID,
mode=Modes.ON,
mode=Mode.ON,
submode=DualSideSubmode.A_SIDE,
q=q,
info="Switching ACS board assembly A side on",
@ -66,7 +66,7 @@ def pack_acs_command(q: DefaultPusQueueHelper, op_code: str):
if op_code in AcsOpCodes.ACS_ASS_B_ON:
pack_mode_cmd_with_info(
object_id=ACS_BOARD_ASS_ID,
mode=Modes.ON,
mode=Mode.ON,
submode=DualSideSubmode.B_SIDE,
q=q,
info="Switching ACS board assembly B side on",
@ -74,7 +74,7 @@ def pack_acs_command(q: DefaultPusQueueHelper, op_code: str):
if op_code in AcsOpCodes.ACS_ASS_DUAL_ON:
pack_mode_cmd_with_info(
object_id=ACS_BOARD_ASS_ID,
mode=Modes.ON,
mode=Mode.ON,
submode=DualSideSubmode.B_SIDE,
q=q,
info="Switching ACS board assembly dual side on",
@ -82,7 +82,7 @@ def pack_acs_command(q: DefaultPusQueueHelper, op_code: str):
if op_code in AcsOpCodes.ACS_ASS_OFF:
pack_mode_cmd_with_info(
object_id=ACS_BOARD_ASS_ID,
mode=Modes.OFF,
mode=Mode.OFF,
submode=0,
q=q,
info="Switching to ACS board assembly off",

View File

@ -15,6 +15,7 @@ from tmtccmd.config.tmtc import (
)
from tmtccmd.tc.pus_20_params import pack_scalar_boolean_parameter_app_data
from tmtccmd.tc import service_provider
from tmtccmd.tc.pus_200_fsfw_modes import Mode, pack_mode_command
from tmtccmd.tc.decorator import ServiceProviderParams
from tmtccmd.tc.pus_3_fsfw_hk import (
generate_one_hk_command,
@ -42,6 +43,8 @@ class SetId(enum.IntEnum):
class OpCodes:
OFF = ["off"]
NML = ["normal"]
REQUEST_RAW_MGM_HK = ["0", "mgm_raw_hk"]
ENABLE_RAW_MGM_HK = ["1", "enable_mgm_raw_hk"]
DISABLE_RAW_MGM_HK = ["2", "disable_mgm_raw_hk"]
@ -75,6 +78,8 @@ class OpCodes:
class Info:
OFF = "Switch ACS CTRL off"
NML = "Switch ACS CTRL normal"
REQUEST_RAW_MGM_HK = "Request Raw MGM HK once"
ENABLE_RAW_MGM_HK = "Enable Raw MGM HK data generation"
DISABLE_RAW_MGM_HK = "Disable Raw MGM HK data generation"
@ -122,6 +127,8 @@ if PERFORM_MGM_CALIBRATION:
@tmtc_definitions_provider
def acs_cmd_defs(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
oce.add(keys=OpCodes.OFF, info=Info.OFF)
oce.add(keys=OpCodes.NML, info=Info.NML)
oce.add(keys=OpCodes.REQUEST_RAW_MGM_HK, info=Info.REQUEST_RAW_MGM_HK)
oce.add(keys=OpCodes.ENABLE_RAW_MGM_HK, info=Info.ENABLE_RAW_MGM_HK)
oce.add(keys=OpCodes.DISABLE_RAW_MGM_HK, info=Info.DISABLE_RAW_MGM_HK)
@ -161,6 +168,12 @@ def acs_cmd_defs(defs: TmtcDefinitionWrapper):
def pack_acs_ctrl_command(p: ServiceProviderParams):
op_code = p.op_code
q = p.queue_helper
if op_code in OpCodes.OFF:
q.add_log_cmd(f"{Info.OFF}")
q.add_pus_tc(pack_mode_command(ACS_CONTROLLER, Mode.OFF, 0))
if op_code in OpCodes.NML:
q.add_log_cmd(f"{Info.NML}")
q.add_pus_tc(pack_mode_command(ACS_CONTROLLER, Mode.NORMAL, 0))
if op_code in OpCodes.REQUEST_RAW_MGM_HK:
q.add_log_cmd(Info.REQUEST_RAW_MGM_HK)
q.add_pus_tc(

View File

@ -10,7 +10,7 @@ from tmtccmd.config.tmtc import (
TmtcDefinitionWrapper,
OpCodeEntry,
)
from tmtccmd.tc.pus_200_fsfw_modes import Subservices as ModeSubservices
from tmtccmd.tc.pus_200_fsfw_modes import Subservice as ModeSubservices
from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams

View File

@ -22,7 +22,7 @@ from tmtccmd.tc.pus_3_fsfw_hk import (
generate_one_diag_command,
generate_one_hk_command,
)
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Mode
from tmtccmd.util import ObjectIdU32
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
@ -87,15 +87,15 @@ def pack_imtq_test_into(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_cod
if op_code in OpCode.OFF:
q.add_log_cmd("IMTQ: Set mode off")
command = pack_mode_data(object_id.as_bytes, Modes.OFF, 0)
command = pack_mode_data(object_id.as_bytes, Mode.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code in OpCode.ON:
q.add_log_cmd("IMTQ: Set mode on")
command = pack_mode_data(object_id.as_bytes, Modes.ON, 0)
command = pack_mode_data(object_id.as_bytes, Mode.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code in OpCode.NORMAL:
q.add_log_cmd("IMTQ: Mode Normal")
command = pack_mode_data(object_id.as_bytes, Modes.NORMAL, 0)
command = pack_mode_data(object_id.as_bytes, Mode.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code == "3":
q.add_log_cmd("IMTQ: Perform positive x self test")

View File

@ -20,7 +20,7 @@ from tmtccmd.tc.pus_3_fsfw_hk import (
disable_periodic_hk_command,
)
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes, Subservices
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Mode, Subservice
from eive_tmtc.config.definitions import CustomServiceList
from tmtccmd.util import ObjectIdU32
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
@ -148,17 +148,17 @@ def pack_single_rw_test_into(
if op_code in OpCodesDev.ON:
q.add_log_cmd(f"RW {rw_idx}: {InfoDev.ON}")
mode_data = pack_mode_data(object_id, Modes.ON, 0)
mode_data = pack_mode_data(object_id, Mode.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
if op_code in OpCodesDev.NML:
q.add_log_cmd(f"RW {rw_idx}: {InfoDev.NML}")
mode_data = pack_mode_data(object_id, Modes.NORMAL, 0)
mode_data = pack_mode_data(object_id, Mode.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
if op_code in OpCodesDev.OFF:
q.add_log_cmd(f"RW {rw_idx}: {InfoDev.OFF}")
mode_data = pack_mode_data(object_id, Modes.OFF, 0)
mode_data = pack_mode_data(object_id, Mode.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
if op_code in OpCodesDev.GET_TM:
@ -194,24 +194,24 @@ def pack_single_rw_test_into(
def pack_rw_ass_cmds(q: DefaultPusQueueHelper, object_id: bytes, op_code: str):
if op_code in OpCodesAss.OFF:
data = pack_mode_data(object_id=object_id, mode=Modes.OFF, submode=0)
data = pack_mode_data(object_id=object_id, mode=Mode.OFF, submode=0)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=data
service=200, subservice=Subservice.TC_MODE_COMMAND, app_data=data
)
)
if op_code in OpCodesAss.ON:
data = pack_mode_data(object_id=object_id, mode=Modes.ON, submode=0)
data = pack_mode_data(object_id=object_id, mode=Mode.ON, submode=0)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=data
service=200, subservice=Subservice.TC_MODE_COMMAND, app_data=data
)
)
if op_code in OpCodesAss.NML:
data = pack_mode_data(object_id=object_id, mode=Modes.NORMAL, submode=0)
data = pack_mode_data(object_id=object_id, mode=Mode.NORMAL, submode=0)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=data
service=200, subservice=Subservice.TC_MODE_COMMAND, app_data=data
)
)
if op_code in OpCodesAss.ALL_SPEED_UP:

View File

@ -9,7 +9,7 @@ from tmtccmd.config.tmtc import (
)
from tmtccmd.tc import service_provider, DefaultPusQueueHelper
from tmtccmd.tc.decorator import ServiceProviderParams
from tmtccmd.tc.pus_200_fsfw_modes import Modes
from tmtccmd.tc.pus_200_fsfw_modes import Mode
class SusOpCode:
@ -23,7 +23,7 @@ def pack_sus_cmds(q: DefaultPusQueueHelper, op_code: str):
if op_code in SusOpCode.SUS_ASS_NOM_SIDE:
pack_mode_cmd_with_info(
object_id=SUS_BOARD_ASS_ID,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=DualSideSubmode.A_SIDE,
q=q,
info="Switching to SUS board to nominal side",
@ -31,7 +31,7 @@ def pack_sus_cmds(q: DefaultPusQueueHelper, op_code: str):
if op_code in SusOpCode.SUS_ASS_RED_SIDE:
pack_mode_cmd_with_info(
object_id=SUS_BOARD_ASS_ID,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=DualSideSubmode.B_SIDE,
q=q,
info="Switching to SUS board to redundant side",
@ -39,7 +39,7 @@ def pack_sus_cmds(q: DefaultPusQueueHelper, op_code: str):
if op_code in SusOpCode.SUS_ASS_OFF:
pack_mode_cmd_with_info(
object_id=SUS_BOARD_ASS_ID,
mode=Modes.OFF,
mode=Mode.OFF,
submode=0,
q=q,
info="Switching SUS board off",
@ -47,7 +47,7 @@ def pack_sus_cmds(q: DefaultPusQueueHelper, op_code: str):
if op_code in SusOpCode.SUS_ASS_DUAL_MODE:
pack_mode_cmd_with_info(
object_id=SUS_BOARD_ASS_ID,
mode=Modes.NORMAL,
mode=Mode.NORMAL,
submode=DualSideSubmode.DUAL_SIDE,
q=q,
info="Switching to SUS board to dual side",

View File

@ -2,13 +2,13 @@ from typing import Union
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes, Subservices
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Mode, Subservice
from tmtccmd.util import ObjectIdU32
def pack_mode_cmd_with_info(
object_id: Union[ObjectIdU32, bytes],
mode: Union[int, Modes],
mode: Union[int, Mode],
submode: int,
q: DefaultPusQueueHelper,
info: str,
@ -27,6 +27,6 @@ def pack_mode_cmd_with_info(
)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=mode_data
service=200, subservice=Subservice.TC_MODE_COMMAND, app_data=mode_data
)
)

View File

@ -8,7 +8,7 @@ from spacepackets.ecss import PusTelecommand
from tmtccmd.config import TmtcDefinitionWrapper
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd
from tmtccmd.logging import get_console_logger
from tmtccmd.tc.pus_3_fsfw_hk import make_sid, generate_one_hk_command
from tmtccmd.config.tmtc import OpCodeEntry, tmtc_definitions_provider
@ -152,7 +152,11 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
)
if op_code in OpCode.REBOOT_FULL:
q.add_log_cmd(f"Core Command: {Info.REBOOT_FULL}")
q.add_pus_tc(create_full_reboot_cmds())
q.add_pus_tc(
create_action_cmd(
object_id=CORE_CONTROLLER_ID, action_id=ActionId.FULL_REBOOT
)
)
if op_code in OpCode.XSC_REBOOT_SELF:
add_xsc_reboot_cmd(q=q, reboot_self=True)
if op_code in OpCode.XSC_REBOOT_0_0:
@ -181,7 +185,7 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
q.add_log_cmd("Disabling reboot file handling")
user_data = bytearray([0])
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=CORE_CONTROLLER_ID,
action_id=ActionId.SWITCH_REBOOT_FILE_HANDLING,
user_data=user_data,
@ -191,7 +195,7 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
q.add_log_cmd("Enabling reboot file handling")
user_data = bytearray([1])
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=CORE_CONTROLLER_ID,
action_id=ActionId.SWITCH_REBOOT_FILE_HANDLING,
user_data=user_data,
@ -200,7 +204,7 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
if op_code in OpCode.RESET_ALL_REBOOT_COUNTERS:
q.add_log_cmd("Resetting all reboot counters")
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=CORE_CONTROLLER_ID,
action_id=ActionId.RESET_REBOOT_COUNTER,
)
@ -225,14 +229,14 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
if op_code in OpCode.SWITCH_TO_SD_0:
q.add_log_cmd(Info.SWITCH_TO_SD_0)
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=CORE_CONTROLLER_ID, action_id=ActionId.SWITCH_TO_SD_0
)
)
if op_code in OpCode.SWITCH_TO_SD_1:
q.add_log_cmd(Info.SWITCH_TO_SD_1)
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=CORE_CONTROLLER_ID, action_id=ActionId.SWITCH_TO_SD_1
)
)
@ -244,7 +248,7 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
break
q.add_log_cmd(Info.SWITCH_TO_BOTH_SD_CARDS)
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=CORE_CONTROLLER_ID,
action_id=ActionId.SWITCH_TO_BOTH_SD_CARDS,
user_data=bytes([active_sd_card]),
@ -259,7 +263,7 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
def reset_specific_boot_counter(q: DefaultPusQueueHelper, chip: int, copy: int):
q.add_log_cmd(f"Resetting boot counter {chip} {copy}")
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=CORE_CONTROLLER_ID,
action_id=ActionId.RESET_REBOOT_COUNTER,
user_data=bytes([chip, copy]),
@ -268,7 +272,7 @@ def reset_specific_boot_counter(q: DefaultPusQueueHelper, chip: int, copy: int):
def create_full_reboot_cmds() -> PusTelecommand:
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=CORE_CONTROLLER_ID, action_id=ActionId.FULL_REBOOT
)
@ -311,7 +315,7 @@ def determine_chip_and_copy() -> (int, int):
def pack_obsw_update_cmd(action_id: int) -> PusTelecommand:
chip, copy = determine_chip_and_copy()
user_data = bytes([chip, copy])
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=CORE_CONTROLLER_ID, action_id=action_id, user_data=user_data
)
@ -341,7 +345,7 @@ def create_xsc_reboot_cmds(
tc_data.append(False)
tc_data.append(chip)
tc_data.append(copy)
return make_fsfw_action_cmd(
return create_action_cmd(
object_id=CORE_CONTROLLER_ID, action_id=ActionId.XSC_REBOOT, user_data=tc_data
)

View File

@ -9,7 +9,7 @@ from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams
from tmtccmd.tc.pus_200_fsfw_modes import Subservices as ModeSubservices
from tmtccmd.tc.pus_200_fsfw_modes import Subservice as ModeSubservice
class OpCode(str, enum.Enum):
@ -41,7 +41,7 @@ def build_acs_subsystem_cmd(p: ServiceProviderParams):
q.add_pus_tc(
PusTelecommand(
service=200,
subservice=ModeSubservices.TC_MODE_ANNOUNCE_RECURSIVE,
subservice=ModeSubservice.TC_MODE_ANNOUNCE_RECURSIVE,
app_data=PL_SUBSYSTEM_ID,
)
)

View File

@ -21,7 +21,7 @@ from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams
from eive_tmtc.utility.input_helper import InputHelper
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Mode
LOGGER = get_console_logger()
@ -138,15 +138,15 @@ def pack_ploc_mpsoc_commands(p: ServiceProviderParams):
obyt = object_id.as_bytes
if op_code in OpCode.OFF:
q.add_log_cmd(f"{prefix}: {Info.OFF}")
command = pack_mode_data(obyt, Modes.OFF, 0)
command = pack_mode_data(obyt, Mode.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code in OpCode.ON:
q.add_log_cmd(f"{prefix}: {Info.ON}")
data = pack_mode_data(obyt, Modes.ON, 0)
data = pack_mode_data(obyt, Mode.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data))
if op_code in OpCode.NORMAL:
q.add_log_cmd(f"{prefix}: {Info.NORMAL}")
data = pack_mode_data(object_id.as_bytes, Modes.NORMAL, 0)
data = pack_mode_data(object_id.as_bytes, Mode.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data))
if op_code == "3":
q.add_log_cmd("PLOC MPSoC: TC mem write test")

View File

@ -19,8 +19,8 @@ from tmtccmd.config.tmtc import tmtc_definitions_provider, OpCodeEntry
from tmtccmd.logging import get_console_logger
from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.pus.s200_fsfw_mode import pack_mode_data, Mode
from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
from eive_tmtc.utility.input_helper import InputHelper
@ -232,15 +232,15 @@ def pack_ploc_supv_commands(p: ServiceProviderParams):
prefix = "PLOC Supervisor"
if op_code in OpCodes.OFF:
q.add_log_cmd(f"{prefix}: {Info.OFF}")
command = pack_mode_data(object_id.as_bytes, Modes.OFF, 0)
command = pack_mode_data(object_id.as_bytes, Mode.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code in OpCodes.ON:
q.add_log_cmd(f"{prefix}: {Info.ON}")
command = pack_mode_data(object_id.as_bytes, Modes.ON, 0)
command = pack_mode_data(object_id.as_bytes, Mode.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code in OpCodes.NORMAL:
q.add_log_cmd(f"{prefix}: {Info.NML}")
command = pack_mode_data(object_id.as_bytes, Modes.NORMAL, 0)
command = pack_mode_data(object_id.as_bytes, Mode.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code in OpCodes.HK_TO_OBC:
q.add_log_cmd(f"{prefix}: {Info.HK_TO_OBC}")
@ -279,7 +279,7 @@ def pack_ploc_supv_commands(p: ServiceProviderParams):
print("Key invalid!")
break
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=PLOC_SUPV_ID,
action_id=SupvActionId.FACTORY_RESET,
user_data=bytes([key]),
@ -475,7 +475,7 @@ def pack_ploc_supv_commands(p: ServiceProviderParams):
f"{prefix}: {Info.MEM_CHECK} for file {update_file} at memory ID {memory_id} at start "
f"address {start_address}"
)
command = make_fsfw_action_cmd(
command = create_action_cmd(
object_id.as_bytes, SupvActionId.MEM_CHECK, custom_data
)
q.add_pus_tc(command)

View File

@ -23,7 +23,7 @@ from tmtccmd.tc.pus_3_fsfw_hk import (
enable_periodic_hk_command_with_interval,
disable_periodic_hk_command,
)
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd
from tmtccmd.util import ObjectIdU32, ObjectIdBase
@ -143,7 +143,7 @@ def pack_common_gomspace_cmds(
if op_code in PowerOpCodes.PRINT_SWITCH_V_I:
q.add_log_cmd(f"{prefix}: {PowerInfo.PRINT_SWITCH_V_I}")
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=objb, action_id=GomspaceDeviceActionId.PRINT_SWITCH_V_I
)
)
@ -153,7 +153,7 @@ def pack_common_gomspace_cmds(
if op_code in PowerOpCodes.PRINT_LATCHUPS:
q.add_log_cmd(f"{prefix}: {PowerInfo.PRINT_LATCHUPS}")
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=objb, action_id=GomspaceDeviceActionId.PRINT_LATCHUPS
)
)
@ -187,7 +187,7 @@ def pack_common_gomspace_cmds(
))
"""
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=object_id.as_bytes,
action_id=GomspaceDeviceActionId.SAVE_TABLE,
user_data=bytes([source_table]),
@ -202,7 +202,7 @@ def pack_common_gomspace_cmds(
if source_table not in [0, 1, 2]:
raise ValueError("Invalid source table index")
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=object_id.as_bytes,
action_id=GomspaceDeviceActionId.SAVE_TABLE_DEFAULT,
user_data=bytes([source_table]),
@ -230,7 +230,7 @@ def pack_common_gomspace_cmds(
# Will be ignored
source_table = 4
q.add_pus_tc(
make_fsfw_action_cmd(
create_action_cmd(
object_id=object_id.as_bytes,
action_id=GomspaceDeviceActionId.LOAD_TABLE,
user_data=bytes([source_table, target_table]),

View File

@ -7,15 +7,15 @@
"""
import struct
from config.definitions import CustomServiceList
from config.object_ids import SOLAR_ARRAY_DEPLOYMENT_ID
from eive_tmtc.config.definitions import CustomServiceList
from eive_tmtc.config.object_ids import SOLAR_ARRAY_DEPLOYMENT_ID
from tmtccmd.config.tmtc import (
tmtc_definitions_provider,
TmtcDefinitionWrapper,
OpCodeEntry,
)
from tmtccmd.tc import service_provider
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd
from tmtccmd.tc.decorator import ServiceProviderParams
from tmtccmd import get_console_logger
@ -72,7 +72,7 @@ def pack_solar_array_deployment_test_into(p: ServiceProviderParams):
else:
dry_run_str = ""
q.add_log_cmd(f"Testing S/A Deployment with burn time {burn_time}{dry_run_str}")
command = make_fsfw_action_cmd(
command = create_action_cmd(
SOLAR_ARRAY_DEPLOYMENT_ID, ActionId.MANUAL_DEPLOYMENT, user_data
)
q.add_pus_tc(command)

View File

@ -3,14 +3,103 @@ import pprint
import struct
from eive_tmtc.pus_tm.defs import PrintWrapper
from eive_tmtc.config.definitions import CustomServiceList
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_200_fsfw_modes import Mode
from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
generate_one_hk_command,
)
from eive_tmtc.tmtc.common import pack_mode_cmd_with_info
from eive_tmtc.config.object_ids import TCS_BOARD_ASS_ID, TCS_CONTROLLER
from tmtccmd.util import ObjectIdU32
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
class OpCode:
TCS_BOARD_ASS_NORMAL = ["0", "normal"]
TCS_BOARD_ASS_OFF = ["1", "off"]
REQUEST_PRIMARY_TEMP_SET = ["temp"]
REQUEST_DEVICE_TEMP_SET = ["temp_devs"]
REQUEST_DEVICE_SUS_SET = ["temp_sus"]
class Info:
REQUEST_PRIMARY_TEMP_SET = "Request HK set of primary sensor temperatures"
REQUEST_DEVICE_TEMP_SET = (
"Request HK set of device sensor (not dedicated sensors) temperatures"
)
REQUEST_DEVICE_SUS_SET = "Request HK set of the SUS temperatures"
TCS_BOARD_ASS_NORMAL = "Switching TCS board assembly on"
TCS_BOARD_ASS_OFF = "Switching TCS board assembly off"
class SetId(enum.IntEnum):
SENSOR_TEMPERATURE_SET = 0
DEVICE_TEMPERATURE_SET = 1
SUS_TEMPERATURE_SET = 2
PRIMARY_SENSORS = 0
DEVICE_SENSORS = 1
SUS_TEMP_SENSORS = 2
@tmtc_definitions_provider
def add_tcs_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
oce.add(
keys=OpCode.TCS_BOARD_ASS_NORMAL,
info=Info.TCS_BOARD_ASS_NORMAL,
)
oce.add(
keys=OpCode.TCS_BOARD_ASS_OFF,
info=Info.TCS_BOARD_ASS_OFF,
)
oce.add(keys=OpCode.REQUEST_PRIMARY_TEMP_SET, info=Info.REQUEST_PRIMARY_TEMP_SET)
oce.add(keys=OpCode.REQUEST_DEVICE_TEMP_SET, info=Info.REQUEST_DEVICE_TEMP_SET)
oce.add(keys=OpCode.REQUEST_DEVICE_SUS_SET, info=Info.REQUEST_DEVICE_SUS_SET)
defs.add_service(
name=CustomServiceList.TCS.value,
info="TCS Board Assembly",
op_code_entry=oce,
)
def pack_tcs_sys_commands(q: DefaultPusQueueHelper, op_code: str):
if op_code in OpCode.REQUEST_PRIMARY_TEMP_SET:
sensor_set_sid = make_sid(TCS_CONTROLLER, SetId.PRIMARY_SENSORS)
q.add_log_cmd(Info.REQUEST_PRIMARY_TEMP_SET)
q.add_pus_tc(generate_one_hk_command(sensor_set_sid))
if op_code in OpCode.REQUEST_DEVICE_TEMP_SET:
q.add_log_cmd(Info.REQUEST_DEVICE_TEMP_SET)
q.add_pus_tc(
generate_one_hk_command(make_sid(TCS_CONTROLLER, SetId.DEVICE_SENSORS))
)
if op_code in OpCode.REQUEST_DEVICE_SUS_SET:
q.add_log_cmd(Info.REQUEST_DEVICE_SUS_SET)
q.add_pus_tc(
generate_one_hk_command(make_sid(TCS_CONTROLLER, SetId.SUS_TEMP_SENSORS))
)
pack_tcs_ass_cmds(q, op_code)
def pack_tcs_ass_cmds(q: DefaultPusQueueHelper, op_code: str):
if op_code in OpCode.TCS_BOARD_ASS_NORMAL:
pack_mode_cmd_with_info(
object_id=TCS_BOARD_ASS_ID,
mode=Mode.NORMAL,
submode=0,
q=q,
info=Info.TCS_BOARD_ASS_NORMAL,
)
if op_code in OpCode.TCS_BOARD_ASS_OFF:
pack_mode_cmd_with_info(
object_id=TCS_BOARD_ASS_ID,
mode=Mode.OFF,
submode=0,
q=q,
info=Info.TCS_BOARD_ASS_OFF,
)
def handle_thermal_controller_hk_data(
@ -23,7 +112,7 @@ def handle_thermal_controller_hk_data(
if TCP_TEMP_DEV_SERVER:
TCP_TEMP_DEV_SERVER = TmTcpServer("localhost", 7306)
"""
if set_id == SetId.SENSOR_TEMPERATURE_SET:
if set_id == SetId.PRIMARY_SENSORS:
pw = PrintWrapper(printer)
pw.dlog("Received sensor temperature data")
@ -58,7 +147,7 @@ def handle_thermal_controller_hk_data(
# tcp_server_sensor_temperatures.report_parsed_hk_data(
# object_id, set_id, parsed_data
# )
elif set_id == SetId.DEVICE_TEMPERATURE_SET:
elif set_id == SetId.DEVICE_SENSORS:
pw = PrintWrapper(printer)
pw.dlog("Received device temperature data")
fmt_str = "!fhhhhiiiifffhffffffffffffff"
@ -99,7 +188,7 @@ def handle_thermal_controller_hk_data(
# tcp_server_device_temperatures.report_parsed_hk_data(
# object_id, set_id, parsed_data
# )
elif set_id == SetId.SUS_TEMPERATURE_SET:
elif set_id == SetId.SUS_TEMP_SENSORS:
pass
# pw = PrintWrapper(printer)
# pw.dlog("Received SUS temperature data")

View File

@ -1,11 +1,11 @@
from spacepackets.ecss import PusTelecommand, PusServices
from spacepackets.ecss import PusTelecommand, PusService
from tmtccmd.config import CoreServiceList
from tmtccmd.config.tmtc import (
tmtc_definitions_provider,
TmtcDefinitionWrapper,
OpCodeEntry,
)
from tmtccmd.pus.pus_17_test import pack_service_17_ping_command
from tmtccmd.pus.s17_test import pack_service_17_ping_command
from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams
@ -45,7 +45,7 @@ def pack_test_command(p: ServiceProviderParams):
q.add_pus_tc(pack_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=PusServices.S17_TEST, subservice=128))
q.add_pus_tc(PusTelecommand(service=PusService.S17_TEST, subservice=128))
if info.op_code == OpCodes.PING_WITH_DATA:
q.add_log_cmd("Sending Ping With Data, Size Reported Back [17, 129]")
while True:
@ -60,6 +60,6 @@ def pack_test_command(p: ServiceProviderParams):
next_byte = not next_byte
q.add_pus_tc(
PusTelecommand(
service=PusServices.S17_TEST, subservice=130, app_data=dummy_data
service=PusService.S17_TEST, subservice=130, app_data=dummy_data
)
)

View File

@ -5,7 +5,7 @@ version = attr: eive_tmtc.__version__
long_description = file: README.md
long_description_content_type = text/markdown
license = Apache-2.0
author = Robin Mueller
author = Robin Mueller, Jakob Meier
author_email = muellerr@irs.uni-stuttgart.de
platform = any
@ -27,7 +27,8 @@ classifiers =
[options]
install_requires =
tmtccmd @ git+https://github.com/robamu-org/tmtccmd@v4.0.0a1
# tmtccmd @ git+https://github.com/robamu-org/tmtccmd@v4.0.0a1
tmtccmd @ git+https://github.com/robamu-org/tmtccmd@667913bd7dd#egg=tmtccmd
packages = find:
python_requires = >=3.10
include_package_data = True