eive-tmtc/pus_tc/devs/plpcdu.py

240 lines
7.2 KiB
Python
Raw Normal View History

2022-02-25 19:21:32 +01:00
import enum
2022-03-01 18:00:39 +01:00
from typing import Optional
2022-02-25 19:21:32 +01:00
from tmtccmd.config import QueueCommands
from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.tc.service_200_mode import pack_mode_data, Modes, Subservices
2022-03-01 17:31:36 +01:00
from tmtccmd.tc.service_20_parameter import (
pack_scalar_double_param_app_data,
2022-03-01 18:12:31 +01:00
pack_fsfw_load_param_cmd,
2022-03-04 10:44:55 +01:00
pack_boolean_parameter_app_data,
2022-03-01 17:31:36 +01:00
)
from tmtccmd.utility.logger import get_console_logger
2022-02-25 19:21:32 +01:00
from spacepackets.ecss.tc import PusTelecommand
from config.object_ids import PL_PCDU_ID
2022-03-01 17:31:36 +01:00
LOGGER = get_console_logger()
2022-02-25 19:21:32 +01:00
2022-03-01 18:00:39 +01:00
2022-02-25 19:21:32 +01:00
class OpCodes:
2022-03-03 18:55:11 +01:00
SWITCH_ON = ["0", "on"]
SWITCH_OFF = ["1", "off"]
NORMAL_SSR = ["2", "nml-ssr"]
NORMAL_DRO = ["3", "nml-dro"]
NORMAL_X8 = ["4", "nml-x8"]
NORMAL_TX = ["5", "nml-tx"]
NORMAL_MPA = ["6", "nml-mpa"]
NORMAL_HPA = ["7", "nml-hpa"]
2022-03-03 18:24:17 +01:00
INJECT_SSR_TO_DRO_FAILURE = ["10", "inject-ssr-dro-fault"]
INJECT_DRO_TO_X8_FAILURE = ["11", "inject-dro-x8-fault"]
INJECT_X8_TO_TX_FAILURE = ["12", "inject-x8-tx-fault"]
INJECT_TX_TO_MPA_FAILURE = ["13", "inject-tx-mpa-fault"]
INJECT_MPA_TO_HPA_FAILURE = ["14", "inject-mpa-hpa-fault"]
INJECT_ALL_ON_FAILURE = ["15", "inject-all-on-fault"]
2022-02-25 19:21:32 +01:00
# The following commands might become deprecated in the future
UPDATE_DRO_TO_X8_WAIT = ["128", "dro-to-x8-wait"]
UPDATE_X8_TO_TX_WAIT_TIME = ["129", "x8-to-tx-wait"]
UPDATE_TX_TO_MPA_WAIT_TIME = ["130", "tx-to-mpa-wait"]
UPDATE_MPA_TO_HPA_WAIT_TIME = ["131", "mpa-to-hpa-wait"]
2022-02-25 19:21:32 +01:00
class Info:
NORMAL = "PL PCDU ADC modules normal"
SWITCH_ON = "Switching PL PCDU on"
SWITCH_OFF = "Switching PL PCDU off"
NORMAL_SSR = f"{NORMAL}, SSR on"
NORMAL_DRO = f"{NORMAL},DRO on"
NORMAL_X8 = f"{NORMAL}, X8 on"
NORMAL_TX = f"{NORMAL}, TX on"
NORMAL_MPA = f"{NORMAL}, MPA on"
NORMAL_HPA = f"{NORMAL}, HPA on"
class NormalSubmodes(enum.IntEnum):
ALL_OFF = 0
SOLID_STATE_RELAYS_ADC_ON = 1
DRO_ON = 2
X8_ON = 3
TX_ON = 4
MPA_ON = 5
HPA_ON = 6
2022-02-25 19:21:32 +01:00
2022-03-01 17:31:36 +01:00
class ParamIds(enum.IntEnum):
2022-03-02 11:26:18 +01:00
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
2022-03-01 19:41:24 +01:00
SSR_TO_DRO_WAIT_TIME = 17
DRO_TO_X8_WAIT_TIME = 18
2022-03-02 11:26:18 +01:00
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
2022-03-01 17:31:36 +01:00
2022-02-25 19:21:32 +01:00
def pack_pl_pcdu_commands(tc_queue: TcQueueT, op_code: str):
2022-03-03 18:24:17 +01:00
if op_code in OpCodes.SWITCH_ON:
pack_pl_pcdu_mode_cmd(
tc_queue=tc_queue, info=Info.SWITCH_ON, mode=Modes.ON, submode=0
2022-03-03 18:55:11 +01:00
)
if op_code in OpCodes.SWITCH_OFF:
pack_pl_pcdu_mode_cmd(
tc_queue=tc_queue, info=Info.SWITCH_OFF, mode=Modes.OFF, submode=0
2022-03-03 18:24:17 +01:00
)
if op_code in OpCodes.NORMAL_SSR:
pack_pl_pcdu_mode_cmd(
tc_queue=tc_queue,
info=Info.NORMAL_SSR,
mode=Modes.NORMAL,
submode=NormalSubmodes.SOLID_STATE_RELAYS_ADC_ON,
2022-03-04 10:44:55 +01:00
)
if op_code in OpCodes.NORMAL_DRO:
pack_pl_pcdu_mode_cmd(
tc_queue=tc_queue,
info=Info.NORMAL_DRO,
mode=Modes.NORMAL,
submode=NormalSubmodes.DRO_ON,
2022-02-25 19:21:32 +01:00
)
if op_code in OpCodes.NORMAL_X8:
pack_pl_pcdu_mode_cmd(
tc_queue=tc_queue,
info=Info.NORMAL_X8,
mode=Modes.NORMAL,
submode=NormalSubmodes.X8_ON,
2022-02-25 19:21:32 +01:00
)
if op_code in OpCodes.NORMAL_TX:
pack_pl_pcdu_mode_cmd(
tc_queue=tc_queue,
info=Info.NORMAL_TX,
mode=Modes.NORMAL,
submode=NormalSubmodes.TX_ON,
2022-03-04 10:44:55 +01:00
)
if op_code in OpCodes.NORMAL_MPA:
pack_pl_pcdu_mode_cmd(
tc_queue=tc_queue,
info=Info.NORMAL_MPA,
mode=Modes.NORMAL,
submode=NormalSubmodes.MPA_ON,
2022-02-25 19:21:32 +01:00
)
if op_code in OpCodes.NORMAL_HPA:
pack_pl_pcdu_mode_cmd(
tc_queue=tc_queue,
info=Info.NORMAL_HPA,
mode=Modes.NORMAL,
submode=NormalSubmodes.HPA_ON,
2022-02-25 19:21:32 +01:00
)
2022-03-01 17:25:50 +01:00
if op_code in OpCodes.UPDATE_DRO_TO_X8_WAIT:
2022-03-02 11:44:39 +01:00
pack_wait_time_cmd(
tc_queue=tc_queue,
param_id=ParamIds.DRO_TO_X8_WAIT_TIME,
print_str="DRO to X8",
2022-03-01 17:31:36 +01:00
)
2022-03-02 11:44:39 +01:00
if op_code in OpCodes.UPDATE_X8_TO_TX_WAIT_TIME:
2022-03-02 15:06:50 +01:00
pack_wait_time_cmd(
tc_queue=tc_queue,
param_id=ParamIds.X8_TO_TX_WAIT_TIME,
print_str="X8 to TX",
)
if op_code in OpCodes.UPDATE_TX_TO_MPA_WAIT_TIME:
pack_wait_time_cmd(
tc_queue=tc_queue,
param_id=ParamIds.TX_TO_MPA_WAIT_TIME,
print_str="TX to MPA",
)
if op_code in OpCodes.UPDATE_MPA_TO_HPA_WAIT_TIME:
pack_wait_time_cmd(
tc_queue=tc_queue,
param_id=ParamIds.MPA_TO_HPA_WAIT_TIME,
print_str="MPA to HPA",
)
if op_code in OpCodes.INJECT_ALL_ON_FAILURE:
pack_failure_injection_cmd(
tc_queue=tc_queue,
param_id=ParamIds.INJECT_ALL_ON_FAILURE,
2022-03-04 10:44:55 +01:00
print_str="All On",
2022-03-02 15:06:50 +01:00
)
2022-03-01 18:00:39 +01:00
def request_wait_time() -> Optional[float]:
while True:
2022-03-01 19:41:24 +01:00
wait_time = input("Please enter DRO to X8 wait time in seconds, x to cancel: ")
2022-03-01 18:00:39 +01:00
if wait_time.lower() == "x":
return None
2022-03-01 19:41:24 +01:00
try:
wait_time = float(wait_time)
except ValueError:
2022-03-01 18:00:39 +01:00
LOGGER.warning("Invalid input")
continue
if wait_time <= 0:
LOGGER.warning("Invalid input")
else:
return wait_time
2022-03-02 11:44:39 +01:00
def pack_wait_time_cmd(tc_queue: TcQueueT, param_id: int, print_str: str):
wait_time = request_wait_time()
tc_queue.appendleft(
(QueueCommands.PRINT, f"Updating {print_str} 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=param_id,
parameter=wait_time,
)
cmd = pack_fsfw_load_param_cmd(ssc=0, app_data=param_data)
tc_queue.appendleft(cmd.pack_command_tuple())
2022-03-02 15:06:50 +01:00
def pack_failure_injection_cmd(tc_queue: TcQueueT, param_id: int, print_str: str):
tc_queue.appendleft((QueueCommands.PRINT, f"Inserting {print_str} error"))
param_data = pack_boolean_parameter_app_data(
2022-03-04 10:44:55 +01:00
object_id=PL_PCDU_ID, domain_id=0, unique_id=param_id, parameter=True
2022-03-02 15:06:50 +01:00
)
cmd = pack_fsfw_load_param_cmd(ssc=0, app_data=param_data)
tc_queue.appendleft(cmd.pack_command_tuple())
def pack_pl_pcdu_mode_cmd(tc_queue: TcQueueT, info: str, mode: int, submode: int):
tc_queue.appendleft(
(
QueueCommands.PRINT,
info,
)
)
mode_data = pack_mode_data(object_id=PL_PCDU_ID, mode=mode, submode=submode)
mode_cmd = PusTelecommand(
service=200, subservice=Subservices.COMMAND_MODE_COMMAND, app_data=mode_data
)
tc_queue.appendleft(mode_cmd.pack_command_tuple())