372 lines
13 KiB
Python
372 lines
13 KiB
Python
import enum
|
|
import socket
|
|
import struct
|
|
from socket import AF_INET
|
|
from typing import Tuple
|
|
|
|
from eive_tmtc.config.definitions import CustomServiceList
|
|
from eive_tmtc.config.object_ids import ACS_CONTROLLER
|
|
from eive_tmtc.pus_tm.defs import PrintWrapper
|
|
from tmtccmd import get_console_logger
|
|
from tmtccmd.config.tmtc import (
|
|
tmtc_definitions_provider,
|
|
TmtcDefinitionWrapper,
|
|
OpCodeEntry,
|
|
)
|
|
from tmtccmd.tc import service_provider
|
|
from tmtccmd.tc.decorator import ServiceProviderParams
|
|
from tmtccmd.tc.pus_3_fsfw_hk import (
|
|
generate_one_hk_command,
|
|
make_sid,
|
|
enable_periodic_hk_command_with_interval,
|
|
disable_periodic_hk_command,
|
|
)
|
|
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
|
|
|
|
|
|
LOGGER = get_console_logger()
|
|
|
|
|
|
class SetIds(enum.IntEnum):
|
|
MGM_RAW_SET = 0
|
|
MGM_PROC_SET = 1
|
|
SUS_RAW_SET = 2
|
|
SUS_PROC_SET = 3
|
|
GYR_RAW_SET = 4
|
|
GYR_PROC_SET = 5
|
|
GPS_PROC_SET = 6
|
|
MEKF_DATA = 7
|
|
CTRL_VAL_DATA = 8
|
|
ACTUATOR_CMD_DATA = 9
|
|
|
|
|
|
class OpCodes:
|
|
REQUEST_MGM_HK = ["0", "mgm_raw_hk"]
|
|
ENABLE_MGM_HK = ["1", "enable_raw_hk"]
|
|
DISABLE_MGM_HK = ["1", "disable_raw_hk"]
|
|
REQUEST_PROC_MGM_HK = ["mgm_proc_hk"]
|
|
REQUEST_RAW_GYRO_HK = ["gyro_raw_hk"]
|
|
REQUEST_PROC_GYRO_HK = ["gyro_proc_hk"]
|
|
REQUEST_RAW_SUS_HK = ["sus_raw_hk"]
|
|
REQUEST_PROC_SUS_HK = ["sus_proc_hk"]
|
|
|
|
|
|
class Info:
|
|
REQUEST_MGM_HK = "Request MGM HK once"
|
|
ENABLE_MGM_HK = "Enable MGM HK data generation"
|
|
DISABLE_MGM_HK = "Disable MGM HK data generation"
|
|
REQUEST_PROC_MGM_HK = "Request Processed MGM HK"
|
|
REQUEST_RAW_GYRO_HK = "Request Raw Gyro HK"
|
|
REQUEST_PROC_GYRO_HK = "Request Processed Gyro HK"
|
|
REQUEST_RAW_SUS_HK = "Request Raw SUS HK"
|
|
REQUEST_PROC_SUS_HK = "Request Processed SUS HK"
|
|
|
|
|
|
PERFORM_MGM_CALIBRATION = False
|
|
CALIBRATION_SOCKET_HOST = "localhost"
|
|
CALIBRATION_SOCKET_PORT = 6677
|
|
CALIBRATION_ADDR = (CALIBRATION_SOCKET_HOST, CALIBRATION_SOCKET_PORT)
|
|
|
|
if PERFORM_MGM_CALIBRATION:
|
|
CALIBR_SOCKET = socket.socket(AF_INET, socket.SOCK_STREAM)
|
|
CALIBR_SOCKET.setblocking(False)
|
|
CALIBR_SOCKET.settimeout(0.2)
|
|
CALIBR_SOCKET.connect(CALIBRATION_ADDR)
|
|
|
|
|
|
@tmtc_definitions_provider
|
|
def acs_cmd_defs(defs: TmtcDefinitionWrapper):
|
|
oce = OpCodeEntry()
|
|
oce.add(keys=OpCodes.REQUEST_MGM_HK, info=Info.REQUEST_MGM_HK)
|
|
oce.add(keys=OpCodes.ENABLE_MGM_HK, info=Info.ENABLE_MGM_HK)
|
|
oce.add(keys=OpCodes.DISABLE_MGM_HK, info=Info.DISABLE_MGM_HK)
|
|
defs.add_service(
|
|
name=CustomServiceList.ACS_CTRL.value, info="ACS Controller", op_code_entry=oce
|
|
)
|
|
|
|
|
|
@service_provider(CustomServiceList.ACS_CTRL.value)
|
|
def pack_acs_ctrl_command(p: ServiceProviderParams):
|
|
op_code = p.op_code
|
|
q = p.queue_helper
|
|
sid = make_sid(ACS_CONTROLLER, SetIds.MGM_RAW_SET)
|
|
if op_code in OpCodes.REQUEST_MGM_HK:
|
|
q.add_log_cmd(Info.REQUEST_MGM_HK)
|
|
q.add_pus_tc(generate_one_hk_command(sid))
|
|
elif op_code in OpCodes.ENABLE_MGM_HK:
|
|
q.add_log_cmd(Info.ENABLE_MGM_HK)
|
|
cmd_tuple = enable_periodic_hk_command_with_interval(False, sid, 2.0)
|
|
q.add_pus_tc(cmd_tuple[0])
|
|
q.add_pus_tc(cmd_tuple[1])
|
|
elif op_code in OpCodes.DISABLE_MGM_HK:
|
|
q.add_log_cmd(Info.DISABLE_MGM_HK)
|
|
q.add_pus_tc(disable_periodic_hk_command(False, sid))
|
|
else:
|
|
LOGGER.info(f"Unknown op code {op_code}")
|
|
|
|
|
|
def handle_acs_ctrl_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
|
|
pw = PrintWrapper(printer)
|
|
if set_id == SetIds.MGM_RAW_SET:
|
|
handle_raw_mgm_data(pw, hk_data)
|
|
elif set_id == SetIds.MGM_PROC_SET:
|
|
handle_mgm_data_processed(pw, hk_data)
|
|
elif set_id == SetIds.GYR_RAW_SET:
|
|
handle_gyr_data_raw(pw, hk_data)
|
|
elif set_id == SetIds.GYR_PROC_SET:
|
|
handle_gyr_data_processed(pw, hk_data)
|
|
elif set_id == SetIds.SUS_RAW_SET:
|
|
handle_acs_ctrl_sus_raw_data(pw, hk_data)
|
|
elif set_id == SetIds.SUS_PROC_SET:
|
|
handle_acs_ctrl_sus_processed_data(pw, hk_data)
|
|
|
|
|
|
def handle_acs_ctrl_sus_raw_data(pw: PrintWrapper, hk_data: bytes):
|
|
if len(hk_data) < 6 * 2 * 12:
|
|
pw.dlog(
|
|
f"SUS Raw dataset with size {len(hk_data)} does not have expected size"
|
|
f" of {6 * 2 * 12} bytes"
|
|
)
|
|
return
|
|
current_idx = 0
|
|
for idx in range(12):
|
|
fmt_str = "!HHHHHH"
|
|
length = struct.calcsize(fmt_str)
|
|
sus_list = struct.unpack(fmt_str, hk_data[current_idx : current_idx + length])
|
|
sus_list_formatted = [f"{val:#04x}" for val in sus_list]
|
|
current_idx += length
|
|
pw.dlog(f"SUS {idx} RAW: {sus_list_formatted}")
|
|
pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=12)
|
|
|
|
|
|
def handle_acs_ctrl_sus_processed_data(pw: PrintWrapper, hk_data: bytes):
|
|
if len(hk_data) < 3 * 4 * 12:
|
|
pw.dlog(
|
|
f"SUS Raw dataset with size {len(hk_data)} does not have expected size"
|
|
f" of {3 * 4 * 12} bytes"
|
|
)
|
|
return
|
|
current_idx = 0
|
|
for idx in range(12):
|
|
fmt_str = "!fff"
|
|
length = struct.calcsize(fmt_str)
|
|
sus_list = struct.unpack(fmt_str, hk_data[current_idx : current_idx + length])
|
|
sus_list_formatted = [f"{val:8.3f}" for val in sus_list]
|
|
current_idx += length
|
|
pw.dlog(f"SUS {idx} CALIB: {sus_list_formatted}")
|
|
fmt_str = "!ddd"
|
|
inc_len = struct.calcsize(fmt_str)
|
|
sus_vec_tot = list(
|
|
struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
|
|
)
|
|
sus_vec_tot = [f"{val:8.3f}" for val in {sus_vec_tot}]
|
|
current_idx += inc_len
|
|
pw.dlog(f"SUS Vector Total: {sus_vec_tot}")
|
|
sus_vec_tot_deriv = struct.unpack(
|
|
fmt_str, hk_data[current_idx : current_idx + inc_len]
|
|
)
|
|
sus_vec_tot_deriv = [f"{val:8.3f}" for val in {sus_vec_tot_deriv}]
|
|
current_idx += inc_len
|
|
pw.dlog(f"SUS Vector Derivative: {sus_vec_tot_deriv}")
|
|
sun_ijk_model = list(
|
|
struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
|
|
)
|
|
sun_ijk_model = [f"{val:8.3f}" for val in {sun_ijk_model}]
|
|
current_idx += inc_len
|
|
pw.dlog(f"SUS ijk Model: {sun_ijk_model}")
|
|
pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=12)
|
|
|
|
|
|
def handle_raw_mgm_data(pw: PrintWrapper, hk_data: bytes):
|
|
current_idx = 0
|
|
|
|
if len(hk_data) < 61:
|
|
pw.dlog(
|
|
f"ACS CTRL HK: MGM HK data with length {len(hk_data)} shorter than expected 61 bytes"
|
|
)
|
|
pw.dlog(f"Raw Data: {hk_data.hex(sep=',')}")
|
|
return
|
|
|
|
def unpack_float_tuple(idx: int) -> (tuple, int):
|
|
f_tuple = struct.unpack(
|
|
float_tuple_fmt_str,
|
|
hk_data[idx : idx + struct.calcsize(float_tuple_fmt_str)],
|
|
)
|
|
idx += struct.calcsize(float_tuple_fmt_str)
|
|
return f_tuple, idx
|
|
|
|
float_tuple_fmt_str = "!fff"
|
|
mgm_0_lis3_floats_ut, current_idx = unpack_float_tuple(current_idx)
|
|
mgm_1_rm3100_floats_ut, current_idx = unpack_float_tuple(current_idx)
|
|
mgm_2_lis3_floats_ut, current_idx = unpack_float_tuple(current_idx)
|
|
mgm_3_rm3100_floats_ut, current_idx = unpack_float_tuple(current_idx)
|
|
isis_floats_nt, current_idx = unpack_float_tuple(current_idx)
|
|
imtq_mgm_ut = tuple(val / 1000.0 for val in isis_floats_nt)
|
|
pw.dlog("ACS CTRL HK: MGM values [X,Y,Z] in floating point uT: ")
|
|
mgm_lists = [
|
|
mgm_0_lis3_floats_ut,
|
|
mgm_1_rm3100_floats_ut,
|
|
mgm_2_lis3_floats_ut,
|
|
mgm_3_rm3100_floats_ut,
|
|
imtq_mgm_ut,
|
|
]
|
|
formatted_list = []
|
|
# Reserve 8 decimal digits, use precision 3
|
|
float_str_fmt = "[{:8.3f}, {:8.3f}, {:8.3f}]"
|
|
for mgm_entry in mgm_lists[0:4]:
|
|
formatted_list.append(float_str_fmt.format(*mgm_entry))
|
|
formatted_list.append(hk_data[current_idx])
|
|
formatted_list.append(float_str_fmt.format(*mgm_lists[4]))
|
|
print_str_list = [
|
|
"ACS Board MGM 0 LIS3MDL",
|
|
"ACS Board MGM 1 RM3100",
|
|
"ACS Board MGM 2 LIS3MDL",
|
|
"ACS Board MGM 3 RM3100",
|
|
"IMTQ Actuation Status:",
|
|
"IMTQ MGM:",
|
|
]
|
|
for entry in zip(print_str_list, formatted_list):
|
|
pw.dlog(f"{entry[0].ljust(28)}: {entry[1]}")
|
|
current_idx += 1
|
|
if PERFORM_MGM_CALIBRATION:
|
|
perform_mgm_calibration(pw, mgm_0_lis3_floats_ut)
|
|
assert current_idx == 61
|
|
|
|
|
|
def handle_mgm_data_processed(pw: PrintWrapper, hk_data: bytes):
|
|
pw.dlog("Received Processed MGM Set")
|
|
fmt_str = "!fffffddd"
|
|
inc_len = struct.calcsize(fmt_str)
|
|
if len(hk_data) < inc_len:
|
|
pw.dlog("Recieved HK set too small")
|
|
return
|
|
current_idx = 0
|
|
for i in range(5):
|
|
fmt_str = "!fff"
|
|
inc_len = struct.calcsize(fmt_str)
|
|
mgm_vec = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
|
|
mgm_vec = [f"{val:8.3f}" for val in mgm_vec]
|
|
pw.dlog(f"MGM {i}: {mgm_vec}")
|
|
fmt_str = "!ddd"
|
|
inc_len = struct.calcsize(fmt_str)
|
|
mgm_vec_tot = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
|
|
mgm_vec_tot = [f"{val:8.3f}" for val in mgm_vec_tot]
|
|
current_idx += inc_len
|
|
pw.dlog(f"MGM Total Vec: {mgm_vec_tot}")
|
|
mgm_vec_tot_deriv = struct.unpack(
|
|
fmt_str, hk_data[current_idx : current_idx + inc_len]
|
|
)
|
|
mgm_vec_tot_deriv = [f"{val:8.3f}" for val in mgm_vec_tot_deriv]
|
|
pw.dlog(f"MGM Total Vec Deriv: {mgm_vec_tot_deriv}")
|
|
current_idx += inc_len
|
|
mag_igrf_model = struct.unpack(
|
|
fmt_str, hk_data[current_idx : current_idx + inc_len]
|
|
)
|
|
mag_igrf_model = [f"{val:8.3f}" for val in mag_igrf_model]
|
|
pw.dlog(f"MAG IGRF Model: {mag_igrf_model}")
|
|
current_idx += inc_len
|
|
pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=8)
|
|
|
|
|
|
def handle_gyr_data_raw(pw: PrintWrapper, hk_data: bytes):
|
|
pw.dlog("Received Gyro Raw Set with rotation rates in deg per second")
|
|
float_fmt = "!fff"
|
|
double_fmt = "!ddd"
|
|
inc_len_flt = struct.calcsize(float_fmt)
|
|
inc_len_double = struct.calcsize(double_fmt)
|
|
if len(hk_data) < 2 * inc_len_double + 2 * inc_len_flt:
|
|
pw.dlog("HK data too small")
|
|
return
|
|
current_idx = 0
|
|
gyr_0_adis = [
|
|
f"{val:8.3f}" for val in struct.unpack(double_fmt, hk_data[:inc_len_double])
|
|
]
|
|
current_idx += inc_len_double
|
|
gyr_1_l3 = [
|
|
f"{val:8.3f}" for val in struct.unpack(float_fmt, hk_data[:inc_len_flt])
|
|
]
|
|
current_idx += inc_len_flt
|
|
gyr_2_adis = [
|
|
f"{val:8.3f}" for val in struct.unpack(double_fmt, hk_data[:inc_len_double])
|
|
]
|
|
current_idx += inc_len_flt
|
|
gyr_3_l3 = [
|
|
f"{val:8.3f}" for val in struct.unpack(float_fmt, hk_data[:inc_len_flt])
|
|
]
|
|
current_idx += inc_len_flt
|
|
pw.dlog(f"Gyro 0 ADIS: {gyr_0_adis}")
|
|
pw.dlog(f"Gyro 1 L3: {gyr_1_l3}")
|
|
pw.dlog(f"Gyro 2 ADIS: {gyr_2_adis}")
|
|
pw.dlog(f"Gyro 3 L3: {gyr_3_l3}")
|
|
pw.printer.print_validity_buffer(hk_data[current_idx:], 4)
|
|
|
|
|
|
GYRO_NAMES = ["Gyro 0 ADIS", "Gyro 1 L3", "Gyro 2 ADIS", "Gyro 3 L3"]
|
|
|
|
|
|
def handle_gyr_data_processed(pw: PrintWrapper, hk_data: bytes):
|
|
pw.dlog("Received Gyro Processed Set with rotation rates in deg per second")
|
|
fmt_str = "!ddd"
|
|
inc_len = struct.calcsize(fmt_str)
|
|
current_idx = 0
|
|
for i in range(4):
|
|
gyr_vec = [
|
|
f"{val:8.3f}"
|
|
for val in struct.unpack(
|
|
fmt_str, hk_data[current_idx : current_idx + inc_len]
|
|
)
|
|
]
|
|
pw.dlog(f"{GYRO_NAMES[i]}: {gyr_vec}")
|
|
current_idx += inc_len
|
|
gyr_vec_tot = [
|
|
f"{val:8.3f}"
|
|
for val in struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
|
|
]
|
|
pw.dlog(f"Gyro Vec Total: {gyr_vec_tot}")
|
|
current_idx += inc_len
|
|
|
|
|
|
def perform_mgm_calibration(pw: PrintWrapper, mgm_tuple: Tuple):
|
|
global CALIBR_SOCKET, CALIBRATION_ADDR
|
|
try:
|
|
declare_api_cmd = "declare_api_version 2"
|
|
CALIBR_SOCKET.sendall(f"{declare_api_cmd}\n".encode())
|
|
reply = CALIBR_SOCKET.recv(1024)
|
|
if len(reply) != 2:
|
|
pw.dlog(
|
|
f"MGM calibration: Reply received command {declare_api_cmd} has"
|
|
f" invalid length {len(reply)}"
|
|
)
|
|
return
|
|
else:
|
|
if str(reply[0]) == "0":
|
|
pw.dlog(f"MGM calibration: API version 2 was not accepted")
|
|
return
|
|
if len(mgm_tuple) != 3:
|
|
pw.dlog(f"MGM tuple has invalid length {len(mgm_tuple)}")
|
|
mgm_list = [mgm / 1e6 for mgm in mgm_tuple]
|
|
command = (
|
|
f"magnetometer_field {mgm_list[0]} {mgm_list[1]} {mgm_list[2]}\n".encode()
|
|
)
|
|
CALIBR_SOCKET.sendall(command)
|
|
reply = CALIBR_SOCKET.recv(1024)
|
|
if len(reply) != 2:
|
|
pw.dlog(
|
|
f"MGM calibration: Reply received command magnetometer_field has invalid "
|
|
f"length {len(reply)}"
|
|
)
|
|
return
|
|
else:
|
|
if str(reply[0]) == "0":
|
|
pw.dlog(f"MGM calibration: magnetmeter field format was not accepted")
|
|
return
|
|
pw.dlog(f"Sent data {mgm_list} to Helmholtz Testbench successfully")
|
|
except socket.timeout:
|
|
pw.dlog("Socket timeout")
|
|
except BlockingIOError as e:
|
|
pw.dlog(f"Error {e}")
|
|
except ConnectionResetError as e:
|
|
pw.dlog("Socket was closed")
|
|
except ConnectionRefusedError or OSError:
|
|
pw.dlog("Connecting to Calibration Socket on addrss {} failed")
|