Merge branch 'mueller/major-dependency-update' into irini

This commit is contained in:
2022-07-08 16:21:01 +02:00
114 changed files with 2901 additions and 4017 deletions

View File

@ -1,11 +1,10 @@
import enum
import json
import struct
from config.definitions import CustomServiceList
from tmtccmd.tc import QueueHelper
from tmtccmd.tc.pus_8_funccmd import generate_action_command
from tmtccmd.config import QueueCommands, ServiceOpCodeDictT, add_op_code_entry, add_service_op_code_entry
from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.config import OpCodeEntry, TmTcDefWrapper
from config.object_ids import SCEX_HANDLER_ID
@ -45,74 +44,51 @@ class Info:
FRAM = "Read FRAM"
def add_scex_cmds(cmd_dict: ServiceOpCodeDictT):
op_code_dict = dict()
add_op_code_entry(
op_code_dict=op_code_dict, keys=OpCodes.PING, info=Info.PING
)
def add_scex_cmds(defs: TmTcDefWrapper):
oce = OpCodeEntry()
oce.add(keys=OpCodes.PING, info=Info.PING)
oce.add(keys=OpCodes.ION_CMD, info=Info.ION_CMD)
oce.add(keys=OpCodes.TEMP_CMD, info=Info.TEMP_CMD)
oce.add(keys=OpCodes.EXP_STATUS_CMD, info=Info.EXP_STATUS_CMD)
oce.add(keys=OpCodes.ONE_CELLS_CMD, info=Info.ONE_CELLS_CMD)
add_op_code_entry(
op_code_dict=op_code_dict, keys=OpCodes.ION_CMD, info=Info.ION_CMD
)
oce.add(keys=OpCodes.ALL_CELLS_CMD, info=Info.ALL_CELLS_CMD)
oce.add(keys=OpCodes.FRAM, info=Info.FRAM)
add_op_code_entry(
op_code_dict=op_code_dict, keys=OpCodes.TEMP_CMD, info=Info.TEMP_CMD
)
add_op_code_entry(
op_code_dict=op_code_dict, keys=OpCodes.EXP_STATUS_CMD, info=Info.EXP_STATUS_CMD
)
add_op_code_entry(
op_code_dict=op_code_dict, keys=OpCodes.ONE_CELLS_CMD, info=Info.ONE_CELLS_CMD
)
add_op_code_entry(
op_code_dict=op_code_dict, keys=OpCodes.ALL_CELLS_CMD, info=Info.ALL_CELLS_CMD
)
add_op_code_entry(
op_code_dict=op_code_dict, keys=OpCodes.FRAM, info=Info.FRAM
)
add_service_op_code_entry(
srv_op_code_dict=cmd_dict,
defs.add_service(
name=CustomServiceList.SCEX.value,
info="SCEX Device",
op_code_entry=op_code_dict
op_code_entry=oce
)
def pack_scex_cmds(tc_queue: TcQueueT, op_code: str):
def pack_scex_cmds(q: QueueHelper, op_code: str):
if op_code in OpCodes.PING:
tc_queue.appendleft((QueueCommands.PRINT, Info.PING))
q.add_log_cmd(Info.PING)
app_data = bytes([0])
command = generate_action_command(SCEX_HANDLER_ID, ActionIds.PING, app_data)
tc_queue.appendleft(command.pack_command_tuple())
q.add_pus_tc(generate_action_command(SCEX_HANDLER_ID, ActionIds.PING, app_data))
if op_code in OpCodes.ION_CMD:
tc_queue.appendleft((QueueCommands.PRINT, Info.ION_CMD))
q.add_log_cmd(Info.ION_CMD)
app_data = bytes([0])
command = generate_action_command(SCEX_HANDLER_ID, ActionIds.ION_CMD, app_data)
tc_queue.appendleft(command.pack_command_tuple())
q.add_pus_tc(generate_action_command(SCEX_HANDLER_ID, ActionIds.ION_CMD, app_data))
if op_code in OpCodes.TEMP_CMD:
tc_queue.appendleft((QueueCommands.PRINT, Info.TEMP_CMD))
q.add_log_cmd(Info.TEMP_CMD)
app_data = bytes([0])
command = generate_action_command(SCEX_HANDLER_ID, ActionIds.TEMP_CMD, app_data)
tc_queue.appendleft(command.pack_command_tuple())
q.add_pus_tc(generate_action_command(SCEX_HANDLER_ID, ActionIds.TEMP_CMD, app_data))
if op_code in OpCodes.EXP_STATUS_CMD:
tc_queue.appendleft((QueueCommands.PRINT, Info.EXP_STATUS_CMD))
q.add_log_cmd(Info.EXP_STATUS_CMD)
app_data = bytes([0])
command = generate_action_command(SCEX_HANDLER_ID, ActionIds.EXP_STATUS_CMD, app_data)
tc_queue.appendleft(command.pack_command_tuple())
q.add_pus_tc(generate_action_command(SCEX_HANDLER_ID, ActionIds.EXP_STATUS_CMD, app_data))
# one cell
if op_code in OpCodes.ONE_CELLS_CMD:
tc_queue.appendleft((QueueCommands.PRINT, Info.ONE_CELLS_CMD))
q.add_log_cmd(Info.ONE_CELLS_CMD)
app_data = bytearray()
# cell number
cn = 0
while True:
cell_select = input("Which solar cell should be measured? (1-10): ")
if not cell_select.isdigit():
@ -150,11 +126,10 @@ def pack_scex_cmds(tc_queue: TcQueueT, op_code: str):
app_data.append(dac_weight2[cn])
app_data.append(dac_weight3[cn])
command = generate_action_command(SCEX_HANDLER_ID, ActionIds.ONE_CELLS_CMD, app_data)
tc_queue.appendleft(command.pack_command_tuple())
q.add_pus_tc(generate_action_command(SCEX_HANDLER_ID, ActionIds.ONE_CELLS_CMD, app_data))
if op_code in OpCodes.ALL_CELLS_CMD:
tc_queue.appendleft((QueueCommands.PRINT, Info.ALL_CELLS_CMD))
q.add_log_cmd(Info.ALL_CELLS_CMD)
app_data = bytearray()
# cell number
@ -181,15 +156,14 @@ def pack_scex_cmds(tc_queue: TcQueueT, op_code: str):
app_data.append(dac_weight2[cn])
app_data.append(dac_weight3[cn])
command = generate_action_command(SCEX_HANDLER_ID, ActionIds.ALL_CELLS_CMD, app_data)
tc_queue.appendleft(command.pack_command_tuple())
q.add_pus_tc(generate_action_command(SCEX_HANDLER_ID, ActionIds.ALL_CELLS_CMD, app_data))
if op_code in OpCodes.FRAM:
tc_queue.appendleft((QueueCommands.PRINT, Info.FRAM))
q.add_log_cmd(Info.FRAM)
app_data = bytes([0])
command = generate_action_command(SCEX_HANDLER_ID, ActionIds.FRAM, app_data)
tc_queue.appendleft(command.pack_command_tuple())
q.add_pus_tc(generate_action_command(SCEX_HANDLER_ID, ActionIds.FRAM, app_data))
def append_16_bit_val(packet: bytearray, val: int):
packet.append((val >> 8) & 0xFF)
packet.append(val & 0xFF)
packet.append(val & 0xFF)