import enum from typing import Tuple, Dict from spacepackets.ecss import PusTelecommand from .common import pack_mode_cmd_with_info from config.object_ids import ACS_SUBSYSTEM_ID from config.definitions import CustomServiceList from tmtccmd.config.tmtc import ( tmtc_definitions_provider, TmtcDefinitionWrapper, OpCodeEntry, ) from tmtccmd.tc.pus_200_fsfw_modes import Subservices as ModeSubservices from tmtccmd.tc import service_provider from tmtccmd.tc.decorator import ServiceProviderParams class OpCodes(str, enum.Enum): OFF = "off" SAFE = "safe" DETUMBLE = "detumble" IDLE = "idle" TARGET_PT = "target" REPORT_ALL_MODES = "all_modes" class AcsModes(enum.IntEnum): OFF = 0 SAFE = 1 << 24 DETUMBLE = 2 << 24 IDLE = 3 << 24 TARGET_PT = 4 << 24 class Info(str, enum.Enum): OFF = "Off Command" SAFE = "Safe Mode Command" DETUMBLE = "Detumble Mode Command" IDLE = "Idle Mode Command" TARGET_PT = "Target Pointing Mode Command" REPORT_ALL_MODES = "Report All Modes Recursively" HANDLER_LIST: Dict[str, Tuple[int, str]] = { OpCodes.OFF: (AcsModes.OFF, Info.OFF), OpCodes.IDLE: (AcsModes.IDLE, Info.IDLE), OpCodes.SAFE: (AcsModes.SAFE, Info.SAFE), OpCodes.DETUMBLE: (AcsModes.DETUMBLE, Info.DETUMBLE), } @service_provider(CustomServiceList.ACS_SS.value) def build_acs_subsystem_cmd(p: ServiceProviderParams): op_code = p.op_code q = p.queue_helper info_prefix = "ACS Subsystem" if op_code in OpCodes.REPORT_ALL_MODES: q.add_log_cmd(f"{info_prefix}: {Info.REPORT_ALL_MODES}") q.add_pus_tc( PusTelecommand( service=200, subservice=ModeSubservices.TC_MODE_ANNOUNCE_RECURSIVE, app_data=ACS_SUBSYSTEM_ID, ) ) mode_info_tup = HANDLER_LIST.get(op_code) if mode_info_tup is None: return pack_mode_cmd_with_info( object_id=ACS_SUBSYSTEM_ID, info=f"{info_prefix}: {mode_info_tup[1]}", submode=0, mode=mode_info_tup[0], q=q, ) @tmtc_definitions_provider def add_acs_subsystem_cmds(defs: TmtcDefinitionWrapper): oce = OpCodeEntry() oce.add(OpCodes.OFF, Info.OFF) oce.add(OpCodes.SAFE, Info.SAFE) oce.add(OpCodes.IDLE, Info.IDLE) oce.add(OpCodes.REPORT_ALL_MODES, Info.REPORT_ALL_MODES) defs.add_service(CustomServiceList.ACS_SS, "ACS Subsystem", oce)