117 lines
3.4 KiB
Python
117 lines
3.4 KiB
Python
import enum
|
|
from typing import Optional
|
|
|
|
from tmtccmd.config import QueueCommands
|
|
from tmtccmd.tc.definitions import TcQueueT
|
|
from tmtccmd.tc.service_200_mode import pack_mode_data, Modes, Subservices
|
|
from tmtccmd.tc.service_20_parameter import (
|
|
pack_scalar_double_param_app_data,
|
|
pack_fsfw_load_param_cmd,
|
|
)
|
|
from tmtccmd.utility.logger import get_console_logger
|
|
from spacepackets.ecss.tc import PusTelecommand
|
|
from config.object_ids import PL_PCDU_ID
|
|
|
|
LOGGER = get_console_logger()
|
|
|
|
|
|
class OpCodes:
|
|
SWITCH_ADC_ON = ["0", "switch-adc-on"]
|
|
SWITCH_ALL_ON = ["1", "switch-all-on"]
|
|
UPDATE_DRO_TO_X8_WAIT = ["2", "dro-to-x8-wait"]
|
|
|
|
|
|
class Submodes(enum.IntEnum):
|
|
ADC_ON = 0
|
|
ALL_ON = 1
|
|
|
|
|
|
class ParamIds(enum.IntEnum):
|
|
NEG_V_LOWER_BOUND = 0
|
|
NEG_V_UPPER_BOUND = 1
|
|
|
|
DRO_U_LOWER_BOUND = 2
|
|
DRO_U_UPPER_BOUND = 3
|
|
DRO_I_UPPER_BOUND = 4
|
|
|
|
X8_U_LOWER_BOUND = 5
|
|
X8_U_UPPER_BOUND = 6
|
|
X8_I_UPPER_BOUND = 7
|
|
|
|
TX_U_LOWER_BOUND = 8
|
|
TX_U_UPPER_BOUND = 9
|
|
TX_I_UPPER_BOUND = 10
|
|
|
|
MPA_U_LOWER_BOUND = 11
|
|
MPA_U_UPPER_BOUND = 12
|
|
MPA_I_UPPER_BOUND = 13
|
|
|
|
HPA_U_LOWER_BOUND = 14
|
|
HPA_U_UPPER_BOUND = 15
|
|
HPA_I_UPPER_BOUND = 16
|
|
|
|
SSR_TO_DRO_WAIT_TIME = 17
|
|
DRO_TO_X8_WAIT_TIME = 18
|
|
X8_TO_TX_WAIT_TIME = 19
|
|
TX_TO_MPA_WAIT_TIME = 20
|
|
MPA_TO_HPA_WAIT_TIME = 21
|
|
|
|
INJECT_SSR_TO_DRO_FAILURE = 30
|
|
INJECT_DRO_TO_X8_FAILURE = 31
|
|
INJECT_X8_TO_TX_FAILURE = 32
|
|
INJECT_TX_TO_MPA_FAILURE = 33
|
|
INJECT_MPA_TO_HPA_FAILURE = 34
|
|
INJECT_ALL_ON_FAILURE = 35
|
|
|
|
|
|
def pack_pl_pcdu_commands(tc_queue: TcQueueT, op_code: str):
|
|
if op_code in OpCodes.SWITCH_ADC_ON:
|
|
tc_queue.appendleft((QueueCommands.PRINT, "Switching PL PCDU ADC module on"))
|
|
mode_data = pack_mode_data(
|
|
object_id=PL_PCDU_ID, mode=Modes.NORMAL, submode=Submodes.ADC_ON
|
|
)
|
|
mode_cmd = PusTelecommand(
|
|
service=200, subservice=Subservices.SWITCH_MODE, app_data=mode_data
|
|
)
|
|
tc_queue.appendleft(mode_cmd.pack_command_tuple())
|
|
if op_code in OpCodes.SWITCH_ALL_ON:
|
|
tc_queue.appendleft((QueueCommands.PRINT, "Switching all PL PCDU modules on"))
|
|
mode_data = pack_mode_data(
|
|
object_id=PL_PCDU_ID, mode=Modes.NORMAL, submode=Submodes.ALL_ON
|
|
)
|
|
mode_cmd = PusTelecommand(
|
|
service=200, subservice=Subservices.SWITCH_MODE, app_data=mode_data
|
|
)
|
|
tc_queue.appendleft(mode_cmd.pack_command_tuple())
|
|
if op_code in OpCodes.UPDATE_DRO_TO_X8_WAIT:
|
|
wait_time = request_wait_time()
|
|
tc_queue.appendleft(
|
|
(QueueCommands.PRINT, f"Updating DRO to X8 wait time to {wait_time}")
|
|
)
|
|
if wait_time is None:
|
|
return
|
|
param_data = pack_scalar_double_param_app_data(
|
|
object_id=PL_PCDU_ID,
|
|
domain_id=0,
|
|
unique_id=ParamIds.DRO_TO_X8_WAIT_TIME,
|
|
parameter=wait_time,
|
|
)
|
|
cmd = pack_fsfw_load_param_cmd(ssc=0, app_data=param_data)
|
|
tc_queue.appendleft(cmd.pack_command_tuple())
|
|
|
|
|
|
def request_wait_time() -> Optional[float]:
|
|
while True:
|
|
wait_time = input("Please enter DRO to X8 wait time in seconds, x to cancel: ")
|
|
if wait_time.lower() == "x":
|
|
return None
|
|
try:
|
|
wait_time = float(wait_time)
|
|
except ValueError:
|
|
LOGGER.warning("Invalid input")
|
|
continue
|
|
if wait_time <= 0:
|
|
LOGGER.warning("Invalid input")
|
|
else:
|
|
return wait_time
|