fixed conflicts
This commit is contained in:
@ -53,13 +53,17 @@ class ACUHkTable:
|
||||
wdt_gnd_left = TableEntry(bytearray([0x00, 0x74]), TableEntry.uint32_size)
|
||||
|
||||
|
||||
def pack_acu_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
|
||||
def pack_acu_test_into(object_id: bytearray, tc_queue: TcQueueT):
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "Testing ACU"))
|
||||
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "P60 Dock: Enabling ACU connected to X1 slot (channel 0)"))
|
||||
tc_queue.appendleft(
|
||||
(QueueCommands.PRINT, "P60 Dock: Enabling ACU connected to X1 slot (channel 0)")
|
||||
)
|
||||
p60dock_object_id = P60_DOCK_HANDLER
|
||||
command = pack_set_param_command(p60dock_object_id, P60DockConfigTable.out_en_0.parameter_address,
|
||||
P60DockConfigTable.out_en_0.parameter_size, Channel.on)
|
||||
command = pack_set_param_command(
|
||||
p60dock_object_id, P60DockConfigTable.out_en_0.parameter_address,
|
||||
P60DockConfigTable.out_en_0.parameter_size, Channel.on
|
||||
)
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
|
||||
@ -70,8 +74,10 @@ def pack_acu_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
if ACUTestProcedure.all or ACUTestProcedure.read_gnd_wdt:
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "ACU: Reading ground watchdog timer value"))
|
||||
command = pack_get_param_command(object_id, TableIds.hk, ACUHkTable.wdt_gnd_left.parameter_address,
|
||||
ACUHkTable.wdt_gnd_left.parameter_size)
|
||||
command = pack_get_param_command(
|
||||
object_id, TableIds.hk, ACUHkTable.wdt_gnd_left.parameter_address,
|
||||
ACUHkTable.wdt_gnd_left.parameter_size
|
||||
)
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=20, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
if ACUTestProcedure.all or ACUTestProcedure.gnd_wdt_reset:
|
||||
|
112
pus_tc/core.py
Normal file
112
pus_tc/core.py
Normal file
@ -0,0 +1,112 @@
|
||||
import enum
|
||||
|
||||
from tmtccmd.config.definitions import QueueCommands
|
||||
from tmtccmd.tc.definitions import TcQueueT
|
||||
from tmtccmd.tc.service_8_functional_cmd import generate_action_command
|
||||
from tmtccmd.utility.logger import get_console_logger
|
||||
from config.object_ids import CORE_CONTROLLER_ID
|
||||
|
||||
LOGGER = get_console_logger()
|
||||
|
||||
|
||||
class ActionIds(enum.IntEnum):
|
||||
REBOOT = 32
|
||||
|
||||
|
||||
class OpCodes(enum.Enum):
|
||||
REBOOT = 'reboot'
|
||||
REBOOT_SELF = 'reboot_self'
|
||||
REBOOT_0_0 = 'reboot_0_0'
|
||||
REBOOT_0_1 = 'reboot_0_1'
|
||||
REBOOT_1_0 = 'reboot_1_0'
|
||||
REBOOT_1_1 = 'reboot_1_1'
|
||||
|
||||
|
||||
class Chip(enum.IntEnum):
|
||||
CHIP_0 = 0
|
||||
CHIP_1 = 1
|
||||
NONE = 2
|
||||
|
||||
|
||||
class Copy(enum.IntEnum):
|
||||
COPY_0_NOM = 0
|
||||
COPY_1_GOLD = 1
|
||||
NONE = 2
|
||||
|
||||
|
||||
def pack_core_commands(tc_queue: TcQueueT, op_code: str):
|
||||
if op_code == OpCodes.REBOOT.value:
|
||||
reboot_self, chip_select, copy_select = determine_reboot_params()
|
||||
perform_reboot_cmd(
|
||||
tc_queue=tc_queue, reboot_self=reboot_self, chip=chip_select, copy=copy_select
|
||||
)
|
||||
elif op_code == OpCodes.REBOOT_SELF.value:
|
||||
perform_reboot_cmd(tc_queue=tc_queue, reboot_self=True)
|
||||
elif op_code == OpCodes.REBOOT_0_0.value:
|
||||
perform_reboot_cmd(
|
||||
tc_queue=tc_queue, reboot_self=False, chip=Chip.CHIP_0, copy=Copy.COPY_0_NOM
|
||||
)
|
||||
elif op_code == OpCodes.REBOOT_0_1.value:
|
||||
perform_reboot_cmd(
|
||||
tc_queue=tc_queue, reboot_self=False, chip=Chip.CHIP_0, copy=Copy.COPY_1_GOLD
|
||||
)
|
||||
elif op_code == OpCodes.REBOOT_1_0.value:
|
||||
perform_reboot_cmd(
|
||||
tc_queue=tc_queue, reboot_self=False, chip=Chip.CHIP_1, copy=Copy.COPY_0_NOM
|
||||
)
|
||||
elif op_code == OpCodes.REBOOT_1_1.value:
|
||||
perform_reboot_cmd(
|
||||
tc_queue=tc_queue, reboot_self=False, chip=Chip.CHIP_1, copy=Copy.COPY_1_GOLD
|
||||
)
|
||||
|
||||
|
||||
def determine_reboot_params() -> (bool, Chip, Copy):
|
||||
chip_select = -1
|
||||
copy_select = -1
|
||||
reboot_self = input('Reboot self? [y/n]: ')
|
||||
if reboot_self in ['y', 'yes', '1']:
|
||||
LOGGER.info('Rebooting currently running image')
|
||||
return True, chip_select, copy_select
|
||||
LOGGER.info('Rebooting image specified by chip and copy')
|
||||
while True:
|
||||
chip_select = input('Chip select [0/1]: ')
|
||||
if chip_select in ['0', '1']:
|
||||
if chip_select == '0':
|
||||
chip_select = Chip.CHIP_0
|
||||
else:
|
||||
chip_select = Chip.CHIP_1
|
||||
break
|
||||
else:
|
||||
LOGGER.warning('Invalid chip select value. Try again')
|
||||
while True:
|
||||
copy_select = input('Copy select [0/1]: ')
|
||||
if copy_select in ['0', '1']:
|
||||
if copy_select == '0':
|
||||
copy_select = Copy.COPY_0_NOM
|
||||
else:
|
||||
copy_select = Copy.COPY_1_GOLD
|
||||
break
|
||||
else:
|
||||
LOGGER.warning('Invalid copy select value. Try again')
|
||||
return False, chip_select, copy_select
|
||||
|
||||
|
||||
def perform_reboot_cmd(
|
||||
tc_queue: TcQueueT, reboot_self: bool, chip: Chip = Chip.NONE, copy: Copy = Copy.NONE
|
||||
):
|
||||
tc_data = bytearray()
|
||||
if reboot_self:
|
||||
tc_queue.appendleft((QueueCommands.PRINT, 'Packing reboot command for current image'))
|
||||
tc_data.append(True)
|
||||
else:
|
||||
tc_data.append(False)
|
||||
tc_data.append(chip)
|
||||
tc_data.append(copy)
|
||||
tc_queue.append(
|
||||
(QueueCommands.PRINT, f'Packing reboot command for chip {chip} and copy {copy}')
|
||||
)
|
||||
action_cmd = generate_action_command(
|
||||
object_id=CORE_CONTROLLER_ID, action_id=ActionIds.REBOOT, app_data=tc_data, ssc=0
|
||||
)
|
||||
tc_queue.appendleft(action_cmd.pack_command_tuple())
|
||||
|
@ -27,6 +27,7 @@ from pus_tc.heater import pack_heater_test_into
|
||||
from pus_tc.reaction_wheels import pack_single_rw_test_into
|
||||
from pus_tc.rad_sensor import pack_rad_sensor_test_into
|
||||
from pus_tc.ploc_upater import pack_ploc_updater_test_into
|
||||
from pus_tc.core import pack_core_commands
|
||||
from config.definitions import CustomServiceList
|
||||
from config.object_ids import P60_DOCK_HANDLER, PDU_1_HANDLER_ID, PDU_2_HANDLER_ID, ACU_HANDLER_ID, \
|
||||
TMP_1075_1_HANDLER_ID, TMP_1075_2_HANDLER_ID, HEATER_ID, IMTQ_HANDLER_ID, PLOC_MPSOC_ID, RW1_ID, RW2_ID, RW3_ID, RW4_ID, \
|
||||
@ -97,7 +98,8 @@ def pack_service_queue_user(service: Union[str, int], op_code: str, service_queu
|
||||
if service == CustomServiceList.PLOC_UPDATER.value:
|
||||
object_id = PLOC_UPDATER_ID
|
||||
return pack_ploc_updater_test_into(object_id=object_id, tc_queue=service_queue, op_code=op_code)
|
||||
|
||||
if service == CustomServiceList.CORE.value:
|
||||
return pack_core_commands(tc_queue=service_queue, op_code=op_code)
|
||||
LOGGER.warning("Invalid Service !")
|
||||
|
||||
|
||||
|
Reference in New Issue
Block a user