Init tv test procedure #74

Merged
muellerr merged 7 commits from kranz/master into develop 2022-05-19 18:16:43 +02:00
16 changed files with 472 additions and 61 deletions
Showing only changes of commit 2557662c4d - Show all commits

View File

@ -49,3 +49,4 @@ class CustomServiceList(enum.Enum):
TIME = "time" TIME = "time"
PROCEDURE = "proc" PROCEDURE = "proc"
TVTTESTPROCEDURE = "tvtestproc" TVTTESTPROCEDURE = "tvtestproc"
CONTROLLERS = "controllers"

View File

@ -0,0 +1,55 @@
"""
@brief This file transfers control of the custom definitions like modes to the user.
@details Template configuration file. Copy this folder to the TMTC commander root and adapt
it to your needs.
"""
import enum
PUS_APID = 0x65
SPACE_PACKET_IDS = (0x08 << 8 | PUS_APID,)
class CustomServiceList(enum.Enum):
TEST_DEVICE = "test"
P60DOCK = "p60dock"
PDU1 = "pdu1"
PDU2 = "pdu2"
ACU = "acu"
ACS = "acs"
BPX_BATTERY = "bpx"
TMP1075_1 = "tmp1075_1"
TMP1075_2 = "tmp1075_2"
HEATER = "heater"
IMTQ = "imtq"
PLOC_SUPV = "ploc_supv"
PLOC_MPSOC = "ploc_mpsoc"
PCDU = "pcdu"
PL_PCDU = "plpcdu"
SA_DEPLYOMENT = "sa_depl"
REACTION_WHEEL_1 = "rw-1"
REACTION_WHEEL_2 = "rw-2"
REACTION_WHEEL_3 = "rw-3"
REACTION_WHEEL_4 = "rw-4"
RW_ASSEMBLY = "rw-ass"
RAD_SENSOR = "rad_sensor"
GPS_0 = "gps0"
GPS_1 = "gps1"
PLOC_MEMORY_DUMPER = "ploc_memory_dumper"
CORE = "core"
STAR_TRACKER = "star_tracker"
CCSDS_HANDLER = "ccsds_handler"
PDEC_HANDLER = "pdec_handler"
STR_IMG_HELPER = "str_img_helper"
SYRLINKS = "syrlinks"
ACS_ASS = "acs-ass"
SUS_ASS = "sus-ass"
TCS_ASS = "tcs-ass"
TIME = "time"
<<<<<<< HEAD
PROCEDURE = "proc"
TVTTESTPROCEDURE = "tvtestproc"
=======
CONTROLLERS = "controllers"
>>>>>>> origin/develop

View File

@ -0,0 +1,49 @@
"""
@brief This file transfers control of the custom definitions like modes to the user.
@details Template configuration file. Copy this folder to the TMTC commander root and adapt
it to your needs.
"""
import enum
PUS_APID = 0x65
SPACE_PACKET_IDS = (0x08 << 8 | PUS_APID,)
class CustomServiceList(enum.Enum):
TEST_DEVICE = "test"
P60DOCK = "p60dock"
PDU1 = "pdu1"
PDU2 = "pdu2"
ACU = "acu"
ACS = "acs"
BPX_BATTERY = "bpx"
TMP1075_1 = "tmp1075_1"
TMP1075_2 = "tmp1075_2"
HEATER = "heater"
IMTQ = "imtq"
PLOC_SUPV = "ploc_supv"
PLOC_MPSOC = "ploc_mpsoc"
PCDU = "pcdu"
PL_PCDU = "plpcdu"
SA_DEPLYOMENT = "sa_depl"
REACTION_WHEEL_1 = "rw-1"
REACTION_WHEEL_2 = "rw-2"
REACTION_WHEEL_3 = "rw-3"
REACTION_WHEEL_4 = "rw-4"
RW_ASSEMBLY = "rw-ass"
RAD_SENSOR = "rad_sensor"
GPS_0 = "gps0"
GPS_1 = "gps1"
PLOC_MEMORY_DUMPER = "ploc_memory_dumper"
CORE = "core"
STAR_TRACKER = "star_tracker"
CCSDS_HANDLER = "ccsds_handler"
PDEC_HANDLER = "pdec_handler"
STR_IMG_HELPER = "str_img_helper"
SYRLINKS = "syrlinks"
ACS_ASS = "acs-ass"
SUS_ASS = "sus-ass"
TCS_ASS = "tcs-ass"
TIME = "time"

View File

@ -0,0 +1,51 @@
"""
@brief This file transfers control of the custom definitions like modes to the user.
@details Template configuration file. Copy this folder to the TMTC commander root and adapt
it to your needs.
"""
import enum
PUS_APID = 0x65
SPACE_PACKET_IDS = (0x08 << 8 | PUS_APID,)
class CustomServiceList(enum.Enum):
TEST_DEVICE = "test"
P60DOCK = "p60dock"
PDU1 = "pdu1"
PDU2 = "pdu2"
ACU = "acu"
ACS = "acs"
BPX_BATTERY = "bpx"
TMP1075_1 = "tmp1075_1"
TMP1075_2 = "tmp1075_2"
HEATER = "heater"
IMTQ = "imtq"
PLOC_SUPV = "ploc_supv"
PLOC_MPSOC = "ploc_mpsoc"
PCDU = "pcdu"
PL_PCDU = "plpcdu"
SA_DEPLYOMENT = "sa_depl"
REACTION_WHEEL_1 = "rw-1"
REACTION_WHEEL_2 = "rw-2"
REACTION_WHEEL_3 = "rw-3"
REACTION_WHEEL_4 = "rw-4"
RW_ASSEMBLY = "rw-ass"
RAD_SENSOR = "rad_sensor"
GPS_0 = "gps0"
GPS_1 = "gps1"
PLOC_MEMORY_DUMPER = "ploc_memory_dumper"
CORE = "core"
STAR_TRACKER = "star_tracker"
CCSDS_HANDLER = "ccsds_handler"
PDEC_HANDLER = "pdec_handler"
STR_IMG_HELPER = "str_img_helper"
SYRLINKS = "syrlinks"
ACS_ASS = "acs-ass"
SUS_ASS = "sus-ass"
TCS_ASS = "tcs-ass"
TIME = "time"
PROCEDURE = "proc"
TVTTESTPROCEDURE = "tvtestproc"

View File

@ -0,0 +1,50 @@
"""
@brief This file transfers control of the custom definitions like modes to the user.
@details Template configuration file. Copy this folder to the TMTC commander root and adapt
it to your needs.
"""
import enum
PUS_APID = 0x65
SPACE_PACKET_IDS = (0x08 << 8 | PUS_APID,)
class CustomServiceList(enum.Enum):
TEST_DEVICE = "test"
P60DOCK = "p60dock"
PDU1 = "pdu1"
PDU2 = "pdu2"
ACU = "acu"
ACS = "acs"
BPX_BATTERY = "bpx"
TMP1075_1 = "tmp1075_1"
TMP1075_2 = "tmp1075_2"
HEATER = "heater"
IMTQ = "imtq"
PLOC_SUPV = "ploc_supv"
PLOC_MPSOC = "ploc_mpsoc"
PCDU = "pcdu"
PL_PCDU = "plpcdu"
SA_DEPLYOMENT = "sa_depl"
REACTION_WHEEL_1 = "rw-1"
REACTION_WHEEL_2 = "rw-2"
REACTION_WHEEL_3 = "rw-3"
REACTION_WHEEL_4 = "rw-4"
RW_ASSEMBLY = "rw-ass"
RAD_SENSOR = "rad_sensor"
GPS_0 = "gps0"
GPS_1 = "gps1"
PLOC_MEMORY_DUMPER = "ploc_memory_dumper"
CORE = "core"
STAR_TRACKER = "star_tracker"
CCSDS_HANDLER = "ccsds_handler"
PDEC_HANDLER = "pdec_handler"
STR_IMG_HELPER = "str_img_helper"
SYRLINKS = "syrlinks"
ACS_ASS = "acs-ass"
SUS_ASS = "sus-ass"
TCS_ASS = "tcs-ass"
TIME = "time"
CONTROLLERS = "controllers"

View File

@ -27,6 +27,7 @@ ACU_HANDLER_ID = bytes([0x44, 0x25, 0x00, 0x03])
BPX_HANDLER_ID = bytes([0x44, 0x26, 0x00, 0x00]) BPX_HANDLER_ID = bytes([0x44, 0x26, 0x00, 0x00])
# Thermal Object IDs # Thermal Object IDs
THERMAL_CONTROLLER_ID = bytes([0x43, 0x40, 0x00, 0x01])
HEATER_ID = bytes([0x44, 0x41, 0x00, 0xA4]) HEATER_ID = bytes([0x44, 0x41, 0x00, 0xA4])
TMP_1075_1_HANDLER_ID = bytes([0x44, 0x42, 0x00, 0x04]) TMP_1075_1_HANDLER_ID = bytes([0x44, 0x42, 0x00, 0x04])
TMP_1075_2_HANDLER_ID = bytes([0x44, 0x42, 0x00, 0x05]) TMP_1075_2_HANDLER_ID = bytes([0x44, 0x42, 0x00, 0x05])

View File

@ -971,6 +971,7 @@ def add_proc_cmds(cmd_dict: ServiceOpCodeDictT):
def add_system_cmds(cmd_dict: ServiceOpCodeDictT): def add_system_cmds(cmd_dict: ServiceOpCodeDictT):
from pus_tc.system.acs import AcsOpCodes, SusOpCodes from pus_tc.system.acs import AcsOpCodes, SusOpCodes
import pus_tc.system.tcs as tcs import pus_tc.system.tcs as tcs
import pus_tc.system.controllers as controllers
default_opts = generate_op_code_options( default_opts = generate_op_code_options(
enter_listener_mode=False, custom_timeout=8.0 enter_listener_mode=False, custom_timeout=8.0
@ -1076,3 +1077,23 @@ def add_system_cmds(cmd_dict: ServiceOpCodeDictT):
info="TCS Board Assembly", info="TCS Board Assembly",
op_code_entry=op_code_dict, op_code_entry=op_code_dict,
) )
op_code_dict = dict()
add_op_code_entry(
op_code_dict=op_code_dict,
keys=controllers.OpCodes.THERMAL_CONTROLLER,
info=controllers.Info.THERMAL_CONTROLLER,
options=default_opts,
)
add_op_code_entry(
op_code_dict=op_code_dict,
keys=controllers.OpCodes.CORE_CONTROLLER,
info=controllers.Info.CORE_CONTROLLER,
options=default_opts,
)
add_service_op_code_entry(
srv_op_code_dict=cmd_dict,
name=CustomServiceList.CONTROLLERS.value,
info="Controllers",
op_code_entry=op_code_dict,
)

View File

@ -3,8 +3,7 @@ from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.tc.pus_200_fsfw_modes import Modes from tmtccmd.tc.pus_200_fsfw_modes import Modes
from config.object_ids import ACS_BOARD_ASS_ID, SUS_BOARD_ASS_ID from config.object_ids import ACS_BOARD_ASS_ID, SUS_BOARD_ASS_ID
from .common import command_assembly from .common import command_mode
class AcsOpCodes: class AcsOpCodes:
ACS_ASS_A_SIDE = ["0", "acs-a"] ACS_ASS_A_SIDE = ["0", "acs-a"]
@ -31,7 +30,7 @@ class DualSideSubmodes(enum.IntEnum):
def pack_acs_command(tc_queue: TcQueueT, op_code: str): def pack_acs_command(tc_queue: TcQueueT, op_code: str):
if op_code in AcsOpCodes.ACS_ASS_A_SIDE: if op_code in AcsOpCodes.ACS_ASS_A_SIDE:
command_assembly( command_mode(
object_id=ACS_BOARD_ASS_ID, object_id=ACS_BOARD_ASS_ID,
mode=Modes.NORMAL, mode=Modes.NORMAL,
submode=DualSideSubmodes.A_SIDE, submode=DualSideSubmodes.A_SIDE,
@ -39,7 +38,7 @@ def pack_acs_command(tc_queue: TcQueueT, op_code: str):
info="Switching to ACS board assembly A side", info="Switching to ACS board assembly A side",
) )
if op_code in AcsOpCodes.ACS_ASS_B_SIDE: if op_code in AcsOpCodes.ACS_ASS_B_SIDE:
command_assembly( command_mode(
object_id=ACS_BOARD_ASS_ID, object_id=ACS_BOARD_ASS_ID,
mode=Modes.NORMAL, mode=Modes.NORMAL,
submode=DualSideSubmodes.B_SIDE, submode=DualSideSubmodes.B_SIDE,
@ -47,7 +46,7 @@ def pack_acs_command(tc_queue: TcQueueT, op_code: str):
info="Switching to ACS board assembly B side", info="Switching to ACS board assembly B side",
) )
if op_code in AcsOpCodes.ACS_ASS_DUAL_MODE: if op_code in AcsOpCodes.ACS_ASS_DUAL_MODE:
command_assembly( command_mode(
object_id=ACS_BOARD_ASS_ID, object_id=ACS_BOARD_ASS_ID,
mode=Modes.NORMAL, mode=Modes.NORMAL,
submode=DualSideSubmodes.DUAL_SIDE, submode=DualSideSubmodes.DUAL_SIDE,
@ -55,7 +54,7 @@ def pack_acs_command(tc_queue: TcQueueT, op_code: str):
info="Switching to ACS board assembly dual mode", info="Switching to ACS board assembly dual mode",
) )
if op_code in AcsOpCodes.ACS_ASS_A_ON: if op_code in AcsOpCodes.ACS_ASS_A_ON:
command_assembly( command_mode(
object_id=ACS_BOARD_ASS_ID, object_id=ACS_BOARD_ASS_ID,
mode=Modes.ON, mode=Modes.ON,
submode=DualSideSubmodes.A_SIDE, submode=DualSideSubmodes.A_SIDE,
@ -63,7 +62,7 @@ def pack_acs_command(tc_queue: TcQueueT, op_code: str):
info="Switching ACS board assembly A side on", info="Switching ACS board assembly A side on",
) )
if op_code in AcsOpCodes.ACS_ASS_B_ON: if op_code in AcsOpCodes.ACS_ASS_B_ON:
command_assembly( command_mode(
object_id=ACS_BOARD_ASS_ID, object_id=ACS_BOARD_ASS_ID,
mode=Modes.ON, mode=Modes.ON,
submode=DualSideSubmodes.B_SIDE, submode=DualSideSubmodes.B_SIDE,
@ -71,7 +70,7 @@ def pack_acs_command(tc_queue: TcQueueT, op_code: str):
info="Switching ACS board assembly B side on", info="Switching ACS board assembly B side on",
) )
if op_code in AcsOpCodes.ACS_ASS_DUAL_ON: if op_code in AcsOpCodes.ACS_ASS_DUAL_ON:
command_assembly( command_mode(
object_id=ACS_BOARD_ASS_ID, object_id=ACS_BOARD_ASS_ID,
mode=Modes.ON, mode=Modes.ON,
submode=DualSideSubmodes.B_SIDE, submode=DualSideSubmodes.B_SIDE,
@ -79,7 +78,7 @@ def pack_acs_command(tc_queue: TcQueueT, op_code: str):
info="Switching ACS board assembly dual side on", info="Switching ACS board assembly dual side on",
) )
if op_code in AcsOpCodes.ACS_ASS_OFF: if op_code in AcsOpCodes.ACS_ASS_OFF:
command_assembly( command_mode(
object_id=ACS_BOARD_ASS_ID, object_id=ACS_BOARD_ASS_ID,
mode=Modes.OFF, mode=Modes.OFF,
submode=0, submode=0,
@ -90,7 +89,7 @@ def pack_acs_command(tc_queue: TcQueueT, op_code: str):
def pack_sus_cmds(tc_queue: TcQueueT, op_code: str): def pack_sus_cmds(tc_queue: TcQueueT, op_code: str):
if op_code in SusOpCodes.SUS_ASS_NOM_SIDE: if op_code in SusOpCodes.SUS_ASS_NOM_SIDE:
command_assembly( command_mode(
object_id=SUS_BOARD_ASS_ID, object_id=SUS_BOARD_ASS_ID,
mode=Modes.NORMAL, mode=Modes.NORMAL,
submode=DualSideSubmodes.A_SIDE, submode=DualSideSubmodes.A_SIDE,
@ -98,7 +97,7 @@ def pack_sus_cmds(tc_queue: TcQueueT, op_code: str):
info="Switching to SUS board to nominal side", info="Switching to SUS board to nominal side",
) )
if op_code in SusOpCodes.SUS_ASS_RED_SIDE: if op_code in SusOpCodes.SUS_ASS_RED_SIDE:
command_assembly( command_mode(
object_id=SUS_BOARD_ASS_ID, object_id=SUS_BOARD_ASS_ID,
mode=Modes.NORMAL, mode=Modes.NORMAL,
submode=DualSideSubmodes.B_SIDE, submode=DualSideSubmodes.B_SIDE,
@ -106,7 +105,7 @@ def pack_sus_cmds(tc_queue: TcQueueT, op_code: str):
info="Switching to SUS board to redundant side", info="Switching to SUS board to redundant side",
) )
if op_code in SusOpCodes.SUS_ASS_OFF: if op_code in SusOpCodes.SUS_ASS_OFF:
command_assembly( command_mode(
object_id=SUS_BOARD_ASS_ID, object_id=SUS_BOARD_ASS_ID,
mode=Modes.OFF, mode=Modes.OFF,
submode=0, submode=0,
@ -114,7 +113,7 @@ def pack_sus_cmds(tc_queue: TcQueueT, op_code: str):
info="Switching SUS board off", info="Switching SUS board off",
) )
if op_code in SusOpCodes.SUS_ASS_DUAL_MODE: if op_code in SusOpCodes.SUS_ASS_DUAL_MODE:
command_assembly( command_mode(
object_id=SUS_BOARD_ASS_ID, object_id=SUS_BOARD_ASS_ID,
mode=Modes.NORMAL, mode=Modes.NORMAL,
submode=DualSideSubmodes.DUAL_SIDE, submode=DualSideSubmodes.DUAL_SIDE,

View File

@ -3,7 +3,7 @@ from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes, Subservices from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes, Subservices
def command_assembly( def command_mode(
object_id: bytes, mode: Modes, submode: int, tc_queue: TcQueueT, info: str object_id: bytes, mode: Modes, submode: int, tc_queue: TcQueueT, info: str
): ):
tc_queue.appendleft((QueueCommands.PRINT, info)) tc_queue.appendleft((QueueCommands.PRINT, info))

View File

@ -0,0 +1,44 @@
from ast import Pass
from tmtccmd.tc.definitions import TcQueueT
from .common import command_mode
import config.object_ids as obj_ids
class OpCodes:
THERMAL_CONTROLLER = [obj_ids.THERMAL_CONTROLLER_ID.hex(), "thermal_controller"]
CORE_CONTROLLER = [obj_ids.CORE_CONTROLLER_ID.hex(), "core_controller"]
class Info:
THERMAL_CONTROLLER = "Thermal controller"
CORE_CONTROLLER = "ACS controller"
def pack_controller_commands(tc_queue: TcQueueT, op_code: str):
mode = int(input("Specify mode: (OFF = 0; ON = 1; NORMAL = 2) [2] ") or "2")
print(mode)
if mode < 0 or mode > 2:
print("Invalid Mode, defaulting to OFF")
mode = 0
submode = int(input("Specify submode [0]: ") or "0")
command_mode(
object_id=get_object_from_op_code(op_code),
mode=mode,
submode=submode,
tc_queue=tc_queue,
info=op_code + " to " + str(mode) + "," + str(submode),
)
def get_object_from_op_code(op_code: str):
try:
return bytes.fromhex(op_code)
except:
pass
if op_code in OpCodes.THERMAL_CONTROLLER:
return obj_ids.THERMAL_CONTROLLER_ID
if op_code in OpCodes.CORE_CONTROLLER:
return obj_ids.CORE_CONTROLLER_ID

View File

@ -1,7 +1,7 @@
from tmtccmd.tc.definitions import TcQueueT, QueueCommands from tmtccmd.tc.definitions import TcQueueT, QueueCommands
from tmtccmd.tc.pus_200_fsfw_modes import Modes from tmtccmd.tc.pus_200_fsfw_modes import Modes
from .common import command_assembly from .common import command_mode
from config.object_ids import TCS_BOARD_ASS_ID from config.object_ids import TCS_BOARD_ASS_ID
@ -17,7 +17,7 @@ class Info:
def pack_tcs_sys_commands(tc_queue: TcQueueT, op_code: str): def pack_tcs_sys_commands(tc_queue: TcQueueT, op_code: str):
if op_code in OpCodes.TCS_BOARD_ASS_NORMAL: if op_code in OpCodes.TCS_BOARD_ASS_NORMAL:
command_assembly( command_mode(
object_id=TCS_BOARD_ASS_ID, object_id=TCS_BOARD_ASS_ID,
mode=Modes.NORMAL, mode=Modes.NORMAL,
submode=0, submode=0,
@ -25,7 +25,7 @@ def pack_tcs_sys_commands(tc_queue: TcQueueT, op_code: str):
info=Info.TCS_BOARD_ASS_NORMAL, info=Info.TCS_BOARD_ASS_NORMAL,
) )
if op_code in OpCodes.TCS_BOARD_ASS_OFF: if op_code in OpCodes.TCS_BOARD_ASS_OFF:
command_assembly( command_mode(
object_id=TCS_BOARD_ASS_ID, object_id=TCS_BOARD_ASS_ID,
mode=Modes.OFF, mode=Modes.OFF,
submode=0, submode=0,

View File

@ -41,6 +41,7 @@ from pus_tc.devs.plpcdu import pack_pl_pcdu_commands
from pus_tc.devs.str_img_helper import pack_str_img_helper_command from pus_tc.devs.str_img_helper import pack_str_img_helper_command
from pus_tc.system.tcs import pack_tcs_sys_commands from pus_tc.system.tcs import pack_tcs_sys_commands
from pus_tc.system.proc import pack_proc_commands from pus_tc.system.proc import pack_proc_commands
from pus_tc.system.controllers import pack_controller_commands
from config.definitions import CustomServiceList from config.definitions import CustomServiceList
from config.object_ids import ( from config.object_ids import (
P60_DOCK_HANDLER, P60_DOCK_HANDLER,
@ -237,6 +238,8 @@ def pack_service_queue_user(
return pack_rw_ass_cmds( return pack_rw_ass_cmds(
tc_queue=service_queue, object_id=RW_ASSEMBLY, op_code=op_code tc_queue=service_queue, object_id=RW_ASSEMBLY, op_code=op_code
) )
if service == CustomServiceList.CONTROLLERS.value:
return pack_controller_commands(tc_queue=service_queue, op_code=op_code)
LOGGER.warning("Invalid Service !") LOGGER.warning("Invalid Service !")

View File

@ -33,6 +33,7 @@ def ccsds_tm_handler(apid: int, raw_tm_packet: bytes, _user_args: any) -> None:
pus_factory_hook(raw_tm_packet=raw_tm_packet) pus_factory_hook(raw_tm_packet=raw_tm_packet)
def pus_factory_hook(raw_tm_packet: bytes): def pus_factory_hook(raw_tm_packet: bytes):
if len(raw_tm_packet) < 8: if len(raw_tm_packet) < 8:
LOGGER.warning("Detected packet shorter than 8 bytes!") LOGGER.warning("Detected packet shorter than 8 bytes!")

View File

@ -3,6 +3,8 @@ import struct
import os import os
import datetime import datetime
from tmtccmd.config.definitions import HkReplyUnpacked from tmtccmd.config.definitions import HkReplyUnpacked
from tmtccmd.tm.pus_3_fsfw_hk import ( from tmtccmd.tm.pus_3_fsfw_hk import (
Service3Base, Service3Base,
@ -20,8 +22,14 @@ import config.object_ids as obj_ids
from pus_tm.devs.reaction_wheels import handle_rw_hk_data from pus_tm.devs.reaction_wheels import handle_rw_hk_data
from pus_tm.defs import FsfwTmTcPrinter, log_to_both from pus_tm.defs import FsfwTmTcPrinter, log_to_both
from pus_tm.tm_tcp_server import TmTcpServer
LOGGER = get_console_logger() LOGGER = get_console_logger()
TM_TCP_SERVER = TmTcpServer.getInstance()
def handle_hk_packet( def handle_hk_packet(
raw_tm: bytes, raw_tm: bytes,
@ -34,6 +42,9 @@ def handle_hk_packet(
named_obj_id = tm_packet.object_id named_obj_id = tm_packet.object_id
if tm_packet.subservice == 25 or tm_packet.subservice == 26: if tm_packet.subservice == 25 or tm_packet.subservice == 26:
hk_data = tm_packet.tm_data[8:] hk_data = tm_packet.tm_data[8:]
TM_TCP_SERVER.report_raw_hk_data(object_id=named_obj_id,
set_id=tm_packet.set_id,
hk_data=hk_data)
printer.generic_hk_tm_print( printer.generic_hk_tm_print(
content_type=HkContentType.HK, content_type=HkContentType.HK,
object_id=named_obj_id, object_id=named_obj_id,
@ -49,7 +60,6 @@ def handle_hk_packet(
if tm_packet.subservice == 10 or tm_packet.subservice == 12: if tm_packet.subservice == 10 or tm_packet.subservice == 12:
LOGGER.warning("HK definitions printout not implemented yet") LOGGER.warning("HK definitions printout not implemented yet")
def handle_regular_hk_print( def handle_regular_hk_print(
printer: FsfwTmTcPrinter, printer: FsfwTmTcPrinter,
object_id: ObjectId, object_id: ObjectId,
@ -97,12 +107,13 @@ def handle_regular_hk_print(
handle_p60_hk_data(printer=printer, set_id=set_id, hk_data=hk_data) handle_p60_hk_data(printer=printer, set_id=set_id, hk_data=hk_data)
if objb == obj_ids.PL_PCDU_ID: if objb == obj_ids.PL_PCDU_ID:
log_to_both(printer, "Received PL PCDU HK data") log_to_both(printer, "Received PL PCDU HK data")
if objb == obj_ids.THERMAL_CONTROLLER_ID:
handle_thermal_controller_hk_data( object_id=object_id, printer=printer, set_id=set_id, hk_data=hk_data)
else: else:
LOGGER.info("Service 3 TM: Parsing for this SID has not been implemented.") LOGGER.info("Service 3 TM: Parsing for this SID has not been implemented.")
return HkReplyUnpacked() return HkReplyUnpacked()
def handle_syrlinks_rx_registers_dataset( printer: FsfwTmTcPrinter, hk_data: bytes):
def handle_syrlinks_rx_registers_dataset(printer: FsfwTmTcPrinter, hk_data: bytes):
reply = HkReplyUnpacked() reply = HkReplyUnpacked()
header_list = [ header_list = [
"RX Status", "RX Status",
@ -137,7 +148,6 @@ def handle_syrlinks_rx_registers_dataset(printer: FsfwTmTcPrinter, hk_data: byte
log_to_both(printer, str(content_list)) log_to_both(printer, str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=8) printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=8)
def handle_syrlinks_tx_registers_dataset( def handle_syrlinks_tx_registers_dataset(
printer: FsfwTmTcPrinter, printer: FsfwTmTcPrinter,
hk_data: bytes, hk_data: bytes,
@ -153,7 +163,6 @@ def handle_syrlinks_tx_registers_dataset(
log_to_both(printer, str(content_list)) log_to_both(printer, str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=3) printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=3)
def handle_self_test_data(printer: FsfwTmTcPrinter, hk_data: bytes): def handle_self_test_data(printer: FsfwTmTcPrinter, hk_data: bytes):
header_list = [ header_list = [
"Init Err", "Init Err",
@ -258,7 +267,7 @@ def handle_self_test_data(printer: FsfwTmTcPrinter, hk_data: bytes):
init_coil_z_temperature, init_coil_z_temperature,
err, err,
raw_mag_x, raw_mag_x,
init_raw_mag_y, raw_mag_y,
raw_mag_z, raw_mag_z,
cal_mag_x, cal_mag_x,
cal_mag_y, cal_mag_y,
@ -288,6 +297,34 @@ def handle_self_test_data(printer: FsfwTmTcPrinter, hk_data: bytes):
log_to_both(printer, str(content_list)) log_to_both(printer, str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars) printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)
def handle_thermal_controller_hk_data(object_id: ObjectId, printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
if set_id == 0:
LOGGER.info("Received Sensor Temperature data")
# get all the floats
tm_data = struct.unpack("!ffffffffffffffff", hk_data[:16 * 4])
parsed_data = {}
# put them into an list with their names
parsed_data = []
parsed_data.append({"SENSOR_PLOC_HEATSPREADER": tm_data[0]})
parsed_data.append({"SENSOR_PLOC_MISSIONBOARD": tm_data[1]})
parsed_data.append({"SENSOR_4K_CAMERA": tm_data[2]})
parsed_data.append({"SENSOR_DAC_HEATSPREADER": tm_data[3]})
parsed_data.append({"SENSOR_STARTRACKER": tm_data[4]})
parsed_data.append({"SENSOR_RW1": tm_data[5]})
parsed_data.append({"SENSOR_DRO": tm_data[6]})
parsed_data.append({"SENSOR_SCEX": tm_data[7]})
parsed_data.append({"SENSOR_X8": tm_data[8]})
parsed_data.append({"SENSOR_HPA": tm_data[9]})
parsed_data.append({"SENSOR_TX_MODUL": tm_data[10]})
parsed_data.append({"SENSOR_MPA": tm_data[11]})
parsed_data.append({"SENSOR_ACU": tm_data[12]})
parsed_data.append({"SENSOR_PLPCDU_HEATSPREADER": tm_data[13]})
parsed_data.append({"SENSOR_TCS_BOARD": tm_data[14]})
parsed_data.append({"SENSOR_MAGNETTORQUER": tm_data[15]})
TM_TCP_SERVER.report_parsed_hk_data(object_id, set_id, parsed_data)
def handle_gps_data(printer: FsfwTmTcPrinter, hk_data: bytes): def handle_gps_data(printer: FsfwTmTcPrinter, hk_data: bytes):
LOGGER.info(f"Received GPS data, HK data length {len(hk_data)}") LOGGER.info(f"Received GPS data, HK data length {len(hk_data)}")
@ -342,7 +379,6 @@ def handle_gps_data(printer: FsfwTmTcPrinter, hk_data: bytes):
log_to_both(printer, str(content_list)) log_to_both(printer, str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=10) printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=10)
def handle_bpx_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes): def handle_bpx_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
if set_id == BpxSetIds.GET_HK_SET: if set_id == BpxSetIds.GET_HK_SET:
fmt_str = "!HHHHhhhhIB" fmt_str = "!HHHHhhhhIB"
@ -402,13 +438,12 @@ def handle_bpx_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
log_to_both(printer, str(content_list)) log_to_both(printer, str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=10) printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=10)
def handle_core_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes): def handle_core_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes):
fmt_str = "!fffH" fmt_str = "!fffH"
inc_len = struct.calcsize(fmt_str) inc_len = struct.calcsize(fmt_str)
(temperature, ps_voltage, pl_voltage, tx_agc_value) = struct.unpack( (temperature, ps_voltage, pl_voltage, tx_agc_value) = struct.unpack(
fmt_str, hk_data[0 : 0 + inc_len] fmt_str, hk_data[0: 0 + inc_len]
) )
printout = ( printout = (
f"Chip Temperature [°C] {temperature} | PS Voltage [mV] {ps_voltage} | " f"Chip Temperature [°C] {temperature} | PS Voltage [mV] {ps_voltage} | "
@ -417,7 +452,6 @@ def handle_core_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes):
log_to_both(printer, printout) log_to_both(printer, printout)
printer.print_validity_buffer(validity_buffer=hk_data[inc_len:], num_vars=4) printer.print_validity_buffer(validity_buffer=hk_data[inc_len:], num_vars=4)
P60_INDEX_LIST = [ P60_INDEX_LIST = [
"ACU VCC", "ACU VCC",
"PDU1 VCC", "PDU1 VCC",
@ -462,7 +496,6 @@ PDU2_CHANNELS_NAMES = [
PDU_CHANNEL_NAMES = [PDU1_CHANNELS_NAMES, PDU2_CHANNELS_NAMES] PDU_CHANNEL_NAMES = [PDU1_CHANNELS_NAMES, PDU2_CHANNELS_NAMES]
class WdtInfo: class WdtInfo:
def __init__(self): def __init__(self):
self.wdt_reboots_list = [] self.wdt_reboots_list = []
@ -474,7 +507,7 @@ class WdtInfo:
for idx in range(len(self.wdt_reboots_list)): for idx in range(len(self.wdt_reboots_list)):
log_to_both( log_to_both(
printer, printer,
f"{WDT_LIST[idx].ljust(5)} | " f"{TmHandler.WDT_LIST[idx].ljust(5)} | "
f"{self.wdt_reboots_list[idx]:010} | {self.time_pings_left_list[idx]:010}", f"{self.wdt_reboots_list[idx]:010} | {self.time_pings_left_list[idx]:010}",
) )
@ -484,13 +517,13 @@ class WdtInfo:
self.time_pings_left_list = [] self.time_pings_left_list = []
for idx in range(5): for idx in range(5):
self.wdt_reboots_list.append( self.wdt_reboots_list.append(
struct.unpack("!I", wdt_data[priv_idx : priv_idx + 4])[0] struct.unpack("!I", wdt_data[priv_idx: priv_idx + 4])[0]
) )
priv_idx += 4 priv_idx += 4
current_idx += 4 current_idx += 4
for idx in range(3): for idx in range(3):
self.time_pings_left_list.append( self.time_pings_left_list.append(
struct.unpack("!I", wdt_data[priv_idx : priv_idx + 4])[0] struct.unpack("!I", wdt_data[priv_idx: priv_idx + 4])[0]
) )
priv_idx += 4 priv_idx += 4
current_idx += 4 current_idx += 4
@ -500,7 +533,6 @@ class WdtInfo:
priv_idx += 1 priv_idx += 1
return current_idx return current_idx
def handle_pdu_data( def handle_pdu_data(
printer: FsfwTmTcPrinter, pdu_idx: int, set_id: int, hk_data: bytes printer: FsfwTmTcPrinter, pdu_idx: int, set_id: int, hk_data: bytes
): ):
@ -518,7 +550,7 @@ def handle_pdu_data(
boot_cause, boot_cause,
uptime, uptime,
reset_cause, reset_cause,
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) ) = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
log_to_both(printer, f"VCC {vcc} mV | VBAT {vbat} mV") log_to_both(printer, f"VCC {vcc} mV | VBAT {vbat} mV")
log_to_both( log_to_both(
printer, f"Converter Enables [{conv_enb_0},{conv_enb_1},{conv_enb_2}]" printer, f"Converter Enables [{conv_enb_0},{conv_enb_1},{conv_enb_2}]"
@ -530,21 +562,21 @@ def handle_pdu_data(
current_idx += inc_len current_idx += inc_len
latchup_list = [] latchup_list = []
log_to_both(printer, "Latchups") log_to_both(printer, "Latchups")
for idx in range(len(PDU1_CHANNELS_NAMES)): for idx in range(len(TmHandler.PDU1_CHANNELS_NAMES)):
latchup_list.append( latchup_list.append(
struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0] struct.unpack("!H", hk_data[current_idx: current_idx + 2])[0]
) )
content_line = ( content_line = (
f"{PDU_CHANNEL_NAMES[priv_idx][idx].ljust(24)} | {latchup_list[idx]}" f"{TmHandler.PDU_CHANNEL_NAMES[priv_idx][idx].ljust(24)} | {latchup_list[idx]}"
) )
log_to_both(printer, content_line) log_to_both(printer, content_line)
current_idx += 2 current_idx += 2
device_types = [] device_types = []
for idx in range(len(PDU1_CHANNELS_NAMES)): for idx in range(len(TmHandler.PDU1_CHANNELS_NAMES)):
device_types.append(hk_data[current_idx]) device_types.append(hk_data[current_idx])
current_idx += 1 current_idx += 1
device_statuses = [] device_statuses = []
for idx in range(len(PDU1_CHANNELS_NAMES)): for idx in range(len(TmHandler.PDU1_CHANNELS_NAMES)):
device_statuses.append(hk_data[current_idx]) device_statuses.append(hk_data[current_idx])
current_idx += 1 current_idx += 1
wdt = WdtInfo() wdt = WdtInfo()
@ -553,35 +585,35 @@ def handle_pdu_data(
if set_id == SetIds.PDU_1_CORE or set_id == SetIds.PDU_2_CORE: if set_id == SetIds.PDU_1_CORE or set_id == SetIds.PDU_2_CORE:
log_to_both(printer, f"Received PDU HK from PDU {pdu_idx}") log_to_both(printer, f"Received PDU HK from PDU {pdu_idx}")
current_list = [] current_list = []
for idx in range(len(PDU1_CHANNELS_NAMES)): for idx in range(len(TmHandler.PDU1_CHANNELS_NAMES)):
current_list.append( current_list.append(
struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0] struct.unpack("!h", hk_data[current_idx: current_idx + 2])[0]
) )
current_idx += 2 current_idx += 2
voltage_list = [] voltage_list = []
for idx in range(len(PDU1_CHANNELS_NAMES)): for idx in range(len(TmHandler.PDU1_CHANNELS_NAMES)):
voltage_list.append( voltage_list.append(
struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0] struct.unpack("!H", hk_data[current_idx: current_idx + 2])[0]
) )
current_idx += 2 current_idx += 2
output_enb_list = [] output_enb_list = []
for idx in range(len(PDU1_CHANNELS_NAMES)): for idx in range(len(TmHandler.PDU1_CHANNELS_NAMES)):
output_enb_list.append(hk_data[current_idx]) output_enb_list.append(hk_data[current_idx])
current_idx += 1 current_idx += 1
header_str = f"{'Name'.ljust(24)} | OutEnb | U [mV] | I [mA]" header_str = f"{'Name'.ljust(24)} | OutEnb | U [mV] | I [mA]"
print(header_str) print(header_str)
printer.file_logger.info(header_str) printer.file_logger.info(header_str)
for idx in range(len(PDU1_CHANNELS_NAMES)): for idx in range(len(TmHandler.PDU1_CHANNELS_NAMES)):
out_enb = f"{output_enb_list[idx]}".ljust(6) out_enb = f"{output_enb_list[idx]}".ljust(6)
content_line = ( content_line = (
f"{PDU_CHANNEL_NAMES[priv_idx][idx].ljust(24)} | {out_enb} | " f"{TmHandler.PDU_CHANNEL_NAMES[priv_idx][idx].ljust(24)} | {out_enb} | "
f"{voltage_list[idx]:05} | {current_list[idx]:04}" f"{voltage_list[idx]:05} | {current_list[idx]:04}"
) )
log_to_both(printer, content_line) log_to_both(printer, content_line)
fmt_str = "!IBh" fmt_str = "!IBh"
inc_len = struct.calcsize(fmt_str) inc_len = struct.calcsize(fmt_str)
(boot_count, batt_mode, temperature) = struct.unpack( (boot_count, batt_mode, temperature) = struct.unpack(
fmt_str, hk_data[current_idx : current_idx + inc_len] fmt_str, hk_data[current_idx: current_idx + inc_len]
) )
info = ( info = (
f"Boot Count {boot_count} | Battery Mode {batt_mode} | " f"Boot Count {boot_count} | Battery Mode {batt_mode} | "
@ -589,7 +621,6 @@ def handle_pdu_data(
) )
log_to_both(printer, info) log_to_both(printer, info)
def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes): def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
if set_id == SetIds.P60_CORE: if set_id == SetIds.P60_CORE:
log_to_both(printer, "Received P60 Core HK. Voltages in mV, currents in mA") log_to_both(printer, "Received P60 Core HK. Voltages in mV, currents in mA")
@ -597,13 +628,13 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
current_list = [] current_list = []
for idx in range(13): for idx in range(13):
current_list.append( current_list.append(
struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0] struct.unpack("!h", hk_data[current_idx: current_idx + 2])[0]
) )
current_idx += 2 current_idx += 2
voltage_list = [] voltage_list = []
for idx in range(13): for idx in range(13):
voltage_list.append( voltage_list.append(
struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0] struct.unpack("!H", hk_data[current_idx: current_idx + 2])[0]
) )
current_idx += 2 current_idx += 2
out_enb_list = [] out_enb_list = []
@ -616,7 +647,7 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
for idx in range(13): for idx in range(13):
out_enb = f"{out_enb_list[idx]}".ljust(6) out_enb = f"{out_enb_list[idx]}".ljust(6)
content_line = ( content_line = (
f"{P60_INDEX_LIST[idx].ljust(24)} | {out_enb} | " f"{TmHandler.P60_INDEX_LIST[idx].ljust(24)} | {out_enb} | "
f"{voltage_list[idx]:05} | {current_list[idx]:04}" f"{voltage_list[idx]:05} | {current_list[idx]:04}"
) )
log_to_both(printer, content_line) log_to_both(printer, content_line)
@ -629,7 +660,7 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
batt_voltage, batt_voltage,
temp_0, temp_0,
temp_1, temp_1,
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) ) = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
current_idx += inc_len current_idx += inc_len
batt_info = ( batt_info = (
f"Batt: Mode {batt_mode} | Boot Count {boot_count} | " f"Batt: Mode {batt_mode} | Boot Count {boot_count} | "
@ -646,9 +677,9 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
log_to_both(printer, "P60 Dock Latchups") log_to_both(printer, "P60 Dock Latchups")
for idx in range(0, 13): for idx in range(0, 13):
latchup_list.append( latchup_list.append(
struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0] struct.unpack("!H", hk_data[current_idx: current_idx + 2])[0]
) )
content_line = f"{P60_INDEX_LIST[idx].ljust(24)} | {latchup_list[idx]}" content_line = f"{TmHandler.P60_INDEX_LIST[idx].ljust(24)} | {latchup_list[idx]}"
log_to_both(printer, content_line) log_to_both(printer, content_line)
current_idx += 2 current_idx += 2
fmt_str = "!IIHBBHHhhB" fmt_str = "!IIHBBHHhhB"
@ -664,7 +695,7 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
batt_temp_0, batt_temp_0,
batt_temp_1, batt_temp_1,
dearm_status, dearm_status,
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) ) = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
current_idx += inc_len current_idx += inc_len
wdt = WdtInfo() wdt = WdtInfo()
current_idx = wdt.parse(wdt_data=hk_data[current_idx:], current_idx=current_idx) current_idx = wdt.parse(wdt_data=hk_data[current_idx:], current_idx=current_idx)
@ -675,7 +706,7 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
batt_discharge_current, batt_discharge_current,
ant6_depl, ant6_depl,
ar6_depl, ar6_depl,
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) ) = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
current_idx += inc_len current_idx += inc_len
device_types = [] device_types = []
device_statuses = [] device_statuses = []

102
pus_tm/tm_tcp_server.py Normal file
View File

@ -0,0 +1,102 @@
import socket
from typing import Optional
import json
import base64
from tmtccmd.logging import get_console_logger
from tmtccmd.utility.obj_id import ObjectId
from dle_encoder import DleEncoder
# TODO add to configuration parameters
SERVER_HOST = ""
SERVER_PORT = 7305
LOGGER = get_console_logger()
class TmTcpServer:
_Instance = None
def __init__(
self):
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server_socket.bind((SERVER_HOST, SERVER_PORT))
# for now, only accept one connection
self.server_socket.listen(0)
self.server_socket.setblocking(False)
self.client_connection: Optional[socket.socket] = None
self.dle_encoder = DleEncoder()
def __del__(self):
try:
self.close()
except:
LOGGER.warning("Could not close sockets!")
def close(self):
self.server_socket.close()
if self.client_connection != None:
self.client_connection.close()
def getInstance():
if TmTcpServer._Instance == None:
TmTcpServer._Instance = TmTcpServer()
return TmTcpServer._Instance
def _send_dictionary_over_socket(self, dictionary):
# keep listeners current
if self.client_connection == None:
# no running connection, see if a client wants to connect
try:
(self.client_connection, _) = self.server_socket.accept()
self.client_connection.setblocking(False)
except:
# no client waiting
return
data_json_bytes = json.dumps(dictionary).encode()
# dle encode the bytes
# adding a newline because someone might want to look at it in a console
data_json_bytes = self.dle_encoder.encode(data_json_bytes + b'\n')
try:
sent_length = self.client_connection.send(data_json_bytes)
except:
self.client_connection = None
return
if sent_length == 0:
self.client_connection.close()
self.client_connection = None
def report_raw_hk_data(self, object_id: ObjectId,
set_id: int,
hk_data: bytes):
data_dict = {}
data_dict["type"] = "TM"
data_dict["tmType"] = "Raw HK"
data_dict["objectId"] = object_id.as_string
data_dict["setId"] = set_id
data_dict["rawData"] = base64.b64encode(hk_data).decode()
self._send_dictionary_over_socket(data_dict)
def report_parsed_hk_data(self, object_id: ObjectId,
set_id: int,
data_dictionary):
data_dict = {}
data_dict["type"] = "TM"
data_dict["tmType"] = "Parsed HK"
data_dict["objectId"] = object_id.as_string
data_dict["setId"] = set_id
data_dict["content"] = data_dictionary
self._send_dictionary_over_socket(data_dict)

View File

@ -1,5 +1,6 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""EIVE TMTC Commander""" """EIVE TMTC Commander"""
from distutils.log import debug
import sys import sys
import traceback import traceback
@ -11,6 +12,7 @@ try:
create_default_args_parser, create_default_args_parser,
add_default_tmtccmd_args, add_default_tmtccmd_args,
parse_default_input_arguments, parse_default_input_arguments,
handle_unspecified_args,
) )
from tmtccmd.ccsds.handler import CcsdsTmHandler, ApidHandler from tmtccmd.ccsds.handler import CcsdsTmHandler, ApidHandler
from tmtccmd.logging import get_console_logger from tmtccmd.logging import get_console_logger
@ -62,7 +64,7 @@ def main():
tmtc_backend.set_mode(CoreModeList.CONTINUOUS_MODE) tmtc_backend.set_mode(CoreModeList.CONTINUOUS_MODE)
get_console_logger().info("Disabling console logger for continuous operation") get_console_logger().info("Disabling console logger for continuous operation")
get_console_logger().disabled = True get_console_logger().setLevel("ERROR")
tmtccmd.init_and_start_daemons(tmtc_backend=tmtc_backend) tmtccmd.init_and_start_daemons(tmtc_backend=tmtc_backend)
tmtccmd.performOperation(tmtc_backend=tmtc_backend) tmtccmd.performOperation(tmtc_backend=tmtc_backend)
@ -71,11 +73,12 @@ def main():
sys.argv = sys.argv[:1] sys.argv = sys.argv[:1]
while True: while True:
args = parse_default_input_arguments(arg_parser, hook_obj) args.service = None
setup_args = SetupArgs( args.op_code = None
hook_obj=hook_obj, use_gui=False, apid=PUS_APID, cli_args=args handle_unspecified_args(args, hook_obj.get_service_op_code_dictionary())
)
tmtccmd.setup(setup_args=setup_args) tmtc_backend.set_service(args.service)
tmtc_backend.set_opcode(args.op_code)
tmtc_backend.set_mode(CoreModeList.CONTINUOUS_MODE) tmtc_backend.set_mode(CoreModeList.CONTINUOUS_MODE)
tmtccmd.performOperation(tmtc_backend=tmtc_backend) tmtccmd.performOperation(tmtc_backend=tmtc_backend)