Compare commits

...

5 Commits

Author SHA1 Message Date
8a96e21d1a bump changelog 2023-02-07 16:48:48 +01:00
eed6b82353 add some gps cmds 2023-02-07 16:48:23 +01:00
2bd6caa3c2 add tm store module 2023-02-07 15:21:48 +01:00
2766a5de7b bump changelog 2023-02-07 14:43:16 +01:00
895987ebf3 added store object IDS 2023-02-07 14:42:53 +01:00
6 changed files with 62 additions and 14 deletions

View File

@@ -10,6 +10,11 @@ list yields a list of all related PRs for each release.
# [unreleased]
## Added
- Added persistent TM store object IDs
- GPS enable and disable HK commands
# [v2.12.0] 2023-02-06
## Changed

View File

@@ -73,3 +73,4 @@ class CustomServiceList(str, enum.Enum):
TVTTESTPROCEDURE = "tvtestproc"
CONTROLLERS = "controllers"
SCEX = "scex"
TM_STORE = "tm_store"

View File

@@ -141,6 +141,11 @@ TCS_CONTROLLER = bytes([0x43, 0x40, 0x00, 0x01])
ACS_CONTROLLER = bytes([0x43, 0x00, 0x00, 0x02])
CORE_CONTROLLER_ID = bytes([0x43, 0x00, 0x00, 0x03])
MISC_TM_STORE = bytes([0x73, 0x02, 0x00, 0x01])
OK_TM_STORE = bytes([0x73, 0x02, 0x00, 0x02])
NOT_OK_TM_STORE = bytes([0x73, 0x02, 0x00, 0x03])
HK_TM_STORE = bytes([0x73, 0x02, 0x00, 0x04])
CFDP_TM_STORE = bytes([0x73, 0x03, 0x00, 0x00])
ObjectIdDict = Dict[bytes, ObjectIdU32]

View File

@@ -528,40 +528,40 @@ def handle_mgm_data_processed(pw: PrintWrapper, hk_data: bytes):
current_idx = 0
fmt_str = "!fff"
inc_len = struct.calcsize(fmt_str)
mgm_0 = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
mgm_0 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
mgm_0_str = [f"{val:8.3f}" for val in mgm_0]
pw.dlog(f"MGM 0 Vec: {mgm_0_str}")
current_idx += inc_len
mgm_1 = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
mgm_1 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
mgm_1_str = [f"{val:8.3f}" for val in mgm_1]
pw.dlog(f"MGM 1 Vec: {mgm_1_str}")
current_idx += inc_len
mgm_2 = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
mgm_2 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
mgm_2_str = [f"{val:8.3f}" for val in mgm_2]
pw.dlog(f"MGM 2 Vec: {mgm_2_str}")
current_idx += inc_len
mgm_3 = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
mgm_3 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
mgm_3_str = [f"{val:8.3f}" for val in mgm_3]
pw.dlog(f"MGM 3 Vec: {mgm_3_str}")
current_idx += inc_len
mgm_4 = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
mgm_4 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
mgm_4_str = [f"{val:8.3f}" for val in mgm_4]
pw.dlog(f"MGM 4 Vec: {mgm_4_str}")
current_idx += inc_len
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 = 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]
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]
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}")

View File

@@ -1,3 +1,4 @@
import datetime
import logging
import struct
@@ -6,7 +7,12 @@ from eive_tmtc.pus_tm.defs import PrintWrapper
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import make_sid, generate_one_hk_command
from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
create_request_one_hk_command,
create_enable_periodic_hk_command_with_interval,
create_disable_periodic_hk_command,
)
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
_LOGGER = logging.getLogger(__name__)
@@ -14,11 +20,15 @@ _LOGGER = logging.getLogger(__name__)
class OpCode:
REQ_OS_HK = ["hk"]
ENABLE_HK = ["enable_hk"]
DISABLE_HK = ["disable_hk"]
RESET_GNSS = ["reset"]
class Info:
REQ_OS_HK = "Request One-Shot HK"
ENABLE_HK = "Enable HK"
DISABLE_HK = "Disable HK"
RESET_GNSS = "Reset GNSS using reset pin"
@@ -31,6 +41,8 @@ def add_gps_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
oce.add(keys=OpCode.RESET_GNSS, info=Info.RESET_GNSS)
oce.add(keys=OpCode.REQ_OS_HK, info=Info.REQ_OS_HK)
oce.add(keys=OpCode.ENABLE_HK, info=Info.ENABLE_HK)
oce.add(keys=OpCode.DISABLE_HK, info=Info.DISABLE_HK)
defs.add_service(
name=CustomServiceList.GPS_CTRL.value,
info="GPS/GNSS Controller",
@@ -39,14 +51,26 @@ def add_gps_cmds(defs: TmtcDefinitionWrapper):
def pack_gps_command(object_id: bytes, q: DefaultPusQueueHelper, op_code: str):
sid = make_sid(object_id=object_id, set_id=SetId.HK)
if op_code in OpCode.RESET_GNSS:
# TODO: This needs to be re-implemented
_LOGGER.warning("Reset pin handling needs to be re-implemented")
if op_code in OpCode.REQ_OS_HK:
q.add_log_cmd(f"GMSS: {Info.REQ_OS_HK}")
q.add_pus_tc(
generate_one_hk_command(sid=make_sid(object_id=object_id, set_id=SetId.HK))
if op_code in OpCode.ENABLE_HK:
interval = float(input("Please specify interval in floating point seconds: "))
if interval <= 0:
raise ValueError("invalid interval")
q.add_log_cmd(f"GPS: {Info.ENABLE_HK}")
cmds = create_enable_periodic_hk_command_with_interval(
diag=False, sid=sid, interval_seconds=interval
)
for cmd in cmds:
q.add_pus_tc(cmd)
if op_code in OpCode.DISABLE_HK:
q.add_log_cmd(f"gps: {Info.DISABLE_HK}")
q.add_pus_tc(create_disable_periodic_hk_command(diag=False, sid=sid))
if op_code in OpCode.REQ_OS_HK:
q.add_log_cmd(f"GPS: {Info.REQ_OS_HK}")
q.add_pus_tc(create_request_one_hk_command(sid=sid))
def handle_gps_data(printer: FsfwTmTcPrinter, hk_data: bytes):
@@ -72,7 +96,12 @@ def handle_gps_data(printer: FsfwTmTcPrinter, hk_data: bytes):
unix_seconds,
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
current_idx += inc_len
date_string = f"{day}.{month}.{year} {hours}:{minutes}:{seconds}"
if year == 0:
date_string = "No date string, year is 0"
else:
date_string = datetime.datetime(
year=year, month=month, day=day, hour=hours, minute=minutes, second=seconds
)
pw.dlog(f"Lat: {lat} deg")
pw.dlog(f"Long: {long} deg")
pw.dlog(f"Altitude: {alt} m | Speed: {speed} m/s")

View File

@@ -0,0 +1,8 @@
from eive_tmtc.config.definitions import CustomServiceList
from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams
@service_provider(CustomServiceList.TM_STORE)
def pack_tm_store_commands(p: ServiceProviderParams):
pass