260 lines
8.3 KiB
Python
260 lines
8.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
@file gomspace_common.py
|
|
@brief PDU2 tests
|
|
@details All functions and classes common for all gomspace devices are defined in this file.
|
|
@author J. Meier
|
|
@date 17.12.2020
|
|
"""
|
|
import enum
|
|
import struct
|
|
from typing import Union
|
|
|
|
from spacepackets.ecss import PusTelecommand
|
|
from tmtccmd.tc import DefaultPusQueueHelper
|
|
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
|
|
from tmtccmd.util import ObjectIdU32, ObjectIdBase
|
|
|
|
|
|
class GomspaceDeviceActionIds(enum.IntEnum):
|
|
PING = 1
|
|
REBOOT = 4
|
|
PARAM_GET = 0
|
|
PARAM_SET = 255
|
|
WDT_RESET = 9
|
|
REQUEST_HK_TABLE = 16
|
|
REQUEST_CONFIG_TABLE = 17
|
|
SAVE_TABLE = 18
|
|
SAVE_TABLE_DEFAULT = 19
|
|
LOAD_TABLE = 20
|
|
PRINT_SWITCH_V_I = 32
|
|
PRINT_LATCHUPS = 33
|
|
|
|
|
|
class ParamTypes(enum.Enum):
|
|
U8 = 0
|
|
U16 = 1
|
|
U32 = 2
|
|
I8 = 3
|
|
I16 = 4
|
|
I32 = 5
|
|
FLOAT = 6
|
|
STR = 7
|
|
|
|
|
|
class TableIds:
|
|
CONFIG = 1
|
|
HK = 4
|
|
|
|
|
|
class TableEntry:
|
|
uint8_size = 1
|
|
uint16_size = 2
|
|
uint32_size = 4
|
|
|
|
def __init__(self, parameter_address: bytes, parameter_size):
|
|
self.parameter_address = parameter_address
|
|
self.parameter_size = parameter_size
|
|
|
|
|
|
class Channel:
|
|
on = 1
|
|
off = 0
|
|
|
|
|
|
def pack_request_config_command(object_id: bytes) -> PusTelecommand:
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id, action_id=GomspaceDeviceActionIds.REQUEST_CONFIG_TABLE
|
|
)
|
|
|
|
|
|
def pack_get_param_command(
|
|
object_id: bytes,
|
|
table_id: int,
|
|
memory_address: Union[int, bytes],
|
|
parameter_size: int,
|
|
) -> PusTelecommand:
|
|
"""Function to generate a command to retrieve parameters like the temperature from a gomspace device.
|
|
@param object_id: The object id of the gomspace device handler.
|
|
@param table_id: The table id of the gomspace device
|
|
@param memory_address: Address offset within table of the value to read.
|
|
@param parameter_size: Size of the value to read. E.g. temperature is uint16_t and thus
|
|
parameter_size is 2
|
|
@return: The command as bytearray.
|
|
"""
|
|
app_data = struct.pack("!B", table_id)
|
|
if isinstance(memory_address, int):
|
|
app_data += struct.pack("!H", memory_address)
|
|
else:
|
|
app_data += memory_address
|
|
app_data += struct.pack("!B", parameter_size)
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id,
|
|
action_id=GomspaceDeviceActionIds.PARAM_GET,
|
|
user_data=app_data,
|
|
)
|
|
|
|
|
|
def pack_set_float_param_command(
|
|
object_id: bytes, memory_address: bytes, parameter: float
|
|
) -> PusTelecommand:
|
|
action_id = GomspaceDeviceActionIds.PARAM_SET
|
|
app_data = bytearray()
|
|
app_data += memory_address
|
|
app_data.append(4)
|
|
app_data += struct.pack("!f", parameter)
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id, action_id=action_id, user_data=app_data
|
|
)
|
|
|
|
|
|
def pack_set_u8_param_command(
|
|
object_id: bytes, memory_address: bytes, parameter: int
|
|
) -> PusTelecommand:
|
|
action_id = GomspaceDeviceActionIds.PARAM_SET
|
|
app_data = bytearray()
|
|
app_data += memory_address
|
|
app_data.append(1)
|
|
app_data.append(parameter)
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id, action_id=action_id, user_data=app_data
|
|
)
|
|
|
|
|
|
def pack_set_i8_param_command(
|
|
object_id: bytes, memory_address: bytes, parameter: int
|
|
) -> PusTelecommand:
|
|
action_id = GomspaceDeviceActionIds.PARAM_SET
|
|
app_data = bytearray()
|
|
app_data += memory_address
|
|
app_data.append(1)
|
|
app_data += struct.pack("!b", parameter)
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id, action_id=action_id, user_data=app_data
|
|
)
|
|
|
|
|
|
def pack_set_u16_param_command(
|
|
object_id: bytes, memory_address: bytes, parameter: int
|
|
) -> PusTelecommand:
|
|
action_id = GomspaceDeviceActionIds.PARAM_SET
|
|
app_data = bytearray()
|
|
app_data += memory_address
|
|
app_data.append(2)
|
|
app_data += struct.pack("!H", parameter)
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id, action_id=action_id, user_data=app_data
|
|
)
|
|
|
|
|
|
def pack_set_i16_param_command(
|
|
object_id: bytes, memory_address: bytes, parameter: int
|
|
) -> PusTelecommand:
|
|
action_id = GomspaceDeviceActionIds.PARAM_SET
|
|
app_data = bytearray()
|
|
app_data += memory_address
|
|
app_data.append(2)
|
|
app_data += struct.pack("!h", parameter)
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id, action_id=action_id, user_data=app_data
|
|
)
|
|
|
|
|
|
def pack_set_u32_param_command(object_id: bytes, memory_address: bytes, parameter: int):
|
|
action_id = GomspaceDeviceActionIds.PARAM_SET
|
|
app_data = bytearray()
|
|
app_data += memory_address
|
|
app_data.append(4)
|
|
app_data += struct.pack("!I", parameter)
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id, action_id=action_id, user_data=app_data
|
|
)
|
|
|
|
|
|
def pack_set_i32_param_command(object_id: bytes, memory_address: bytes, parameter: int):
|
|
action_id = GomspaceDeviceActionIds.PARAM_SET
|
|
app_data = bytearray()
|
|
app_data += memory_address
|
|
app_data.append(4)
|
|
app_data += struct.pack("!i", parameter)
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id, action_id=action_id, user_data=app_data
|
|
)
|
|
|
|
|
|
def prompt_and_pack_get_param_command(q: DefaultPusQueueHelper, object_id: ObjectIdU32):
|
|
table_id = int(input("Specify table ID: "))
|
|
memory_address = int(input("Specify memory address: 0x"), 16)
|
|
parameter_size = int(input("Specify parameter size: "))
|
|
q.add_pus_tc(
|
|
pack_get_param_command(
|
|
object_id.as_bytes, table_id, memory_address, parameter_size
|
|
)
|
|
)
|
|
|
|
|
|
def prompt_and_pack_set_integer_param_command(
|
|
q: DefaultPusQueueHelper, object_id: ObjectIdU32, ptype: ParamTypes
|
|
):
|
|
memory_address = int(input("Specify memory address: 0x"), 16)
|
|
memory_address = struct.pack("!H", memory_address)
|
|
parameter = int(input("Specify parameter: "))
|
|
if ptype == ParamTypes.U8:
|
|
cmd = pack_set_u8_param_command(object_id.as_bytes, memory_address, parameter)
|
|
elif ptype == ParamTypes.U16:
|
|
cmd = pack_set_u16_param_command(object_id.as_bytes, memory_address, parameter)
|
|
elif ptype == ParamTypes.U32:
|
|
cmd = pack_set_u16_param_command(object_id.as_bytes, memory_address, parameter)
|
|
elif ptype == ParamTypes.I8:
|
|
cmd = pack_set_i8_param_command(object_id.as_bytes, memory_address, parameter)
|
|
elif ptype == ParamTypes.I16:
|
|
cmd = pack_set_i16_param_command(object_id.as_bytes, memory_address, parameter)
|
|
elif ptype == ParamTypes.I32:
|
|
cmd = pack_set_i32_param_command(object_id.as_bytes, memory_address, parameter)
|
|
else:
|
|
raise ValueError(f"Invalid parameter type {ptype} for this function")
|
|
q.add_pus_tc(cmd)
|
|
|
|
|
|
def pack_ping_command(object_id: ObjectIdU32, data: bytearray) -> PusTelecommand:
|
|
""" " Function to generate the command to ping a gomspace device
|
|
@param object_id Object Id of the gomspace device handler.
|
|
@param data Bytearray containing the bytes to send to the gomspace device. For now the on board software
|
|
supports only the handling of up to 33 bytes.
|
|
@note The ping request sends the specified data to a gompsace device. These
|
|
data are simply copied by the device and then sent back.
|
|
"""
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id.as_bytes,
|
|
action_id=GomspaceDeviceActionIds.PING,
|
|
user_data=data,
|
|
)
|
|
|
|
|
|
def pack_gnd_wdt_reset_command(object_id: ObjectIdBase) -> PusTelecommand:
|
|
""" " Function to generate the command to reset the watchdog of a gomspace device.
|
|
@param object_id Object Id of the gomspace device handler.
|
|
"""
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id.as_bytes, action_id=GomspaceDeviceActionIds.WDT_RESET
|
|
)
|
|
|
|
|
|
def pack_reboot_command(object_id: ObjectIdU32) -> PusTelecommand:
|
|
"""Function to generate the command which triggers a reboot of a gomspace device
|
|
@param object_id The object id of the gomspace device handler.
|
|
"""
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id.as_bytes, action_id=GomspaceDeviceActionIds.REBOOT
|
|
)
|
|
|
|
|
|
def pack_request_full_hk_table_command(object_id: ObjectIdU32) -> PusTelecommand:
|
|
"""Function to generate the command to request the full housekeeping table from a gomspace
|
|
device.
|
|
@param object_id The object id of the gomspace device handler.
|
|
"""
|
|
return make_fsfw_action_cmd(
|
|
object_id=object_id.as_bytes, action_id=GomspaceDeviceActionIds.REQUEST_HK_TABLE
|
|
)
|