eive/eive-tmtc
eive
/
eive-tmtc
13
0
Fork 0
Code Issues Pull Requests 1 Projects Releases 96 Wiki Activity

refactor logging usage

This commit is contained in:
Robin Müller 2023-02-01 11:17:04 +01:00
parent 3eeb58182b
commit 7b5df331ef
No known key found for this signature in database
GPG Key ID: 71B58F8A3CDFA9AC
22 changed files with 105 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGELOG.md
View File

@ -10,6 +10,9 @@ list yields a list of all related PRs for each release.
# [unreleased]
- Bump tmtccmd to include more pythonic log usage. All related changes.
It is recommended to use `logging.getLogger(__name__)` for module level loggers now.
# [v2.7.0] 2023-01-31
- tmtccmd v4.0.0a3

5
eive_tmtc/config/events.py
View File

@ -1,11 +1,10 @@
import logging
import os
from eive_tmtc import EIVE_TMTC_ROOT
from tmtccmd import get_console_logger
from tmtccmd.fsfw import parse_fsfw_events_csv
from tmtccmd.pus.s5_event import EventDictT
LOGGER = get_console_logger()
DEFAULT_EVENTS_CSV_PATH = EIVE_TMTC_ROOT / "config/events.csv"
__EVENT_DICT = None
@ -16,6 +15,6 @@ def get_event_dict() -> EventDictT:
if os.path.exists(DEFAULT_EVENTS_CSV_PATH):
__EVENT_DICT = parse_fsfw_events_csv(DEFAULT_EVENTS_CSV_PATH)
else:
LOGGER.warning(f"No Event CSV file found at {DEFAULT_EVENTS_CSV_PATH}")
logging.getLogger(__name__).warning(f"No Event CSV file found at {DEFAULT_EVENTS_CSV_PATH}")
__EVENT_DICT = dict()
return __EVENT_DICT

5
eive_tmtc/config/object_ids.py
View File

@ -3,15 +3,14 @@
@details Template configuration file. Copy this folder to the TMTC commander root and adapt
it to your needs.
"""
import logging
import os.path
from eive_tmtc import EIVE_TMTC_ROOT
from tmtccmd.util.obj_id import ObjectIdDictT
from tmtccmd.fsfw import parse_fsfw_objects_csv
from tmtccmd.logging import get_console_logger
LOGGER = get_console_logger()
DEFAULT_OBJECTS_CSV_PATH = EIVE_TMTC_ROOT / "config/objects.csv"
__OBJECT_ID_DICT = None
@ -143,7 +142,7 @@ CORE_CONTROLLER_ID = bytes([0x43, 0x00, 0x00, 0x03])
def get_object_ids() -> ObjectIdDictT:
global __OBJECT_ID_DICT
if not os.path.exists(DEFAULT_OBJECTS_CSV_PATH):
LOGGER.warning(f"No Objects CSV file found at {DEFAULT_OBJECTS_CSV_PATH}")
logging.getLogger(__name__).warning(f"No Objects CSV file found at {DEFAULT_OBJECTS_CSV_PATH}")
if __OBJECT_ID_DICT is None:
if os.path.exists(DEFAULT_OBJECTS_CSV_PATH):
__OBJECT_ID_DICT = parse_fsfw_objects_csv(csv_file=DEFAULT_OBJECTS_CSV_PATH)

5
eive_tmtc/config/retvals.py
View File

@ -1,12 +1,11 @@
import os
import logging
from eive_tmtc import EIVE_TMTC_ROOT
from tmtccmd.fsfw import parse_fsfw_returnvalues_csv, RetvalDictT
from tmtccmd.logging import get_console_logger
DEFAULT_RETVAL_CSV_NAME = EIVE_TMTC_ROOT / "config/returnvalues.csv"
__RETVAL_DICT = None
LOGGER = get_console_logger()
def get_retval_dict() -> RetvalDictT:
@ -17,7 +16,7 @@ def get_retval_dict() -> RetvalDictT:
csv_file=DEFAULT_RETVAL_CSV_NAME
)
else:
LOGGER.warning(
logging.getLogger(__name__).warning(
f"No Return Value CSV file found at {DEFAULT_RETVAL_CSV_NAME}"
)
__RETVAL_DICT = dict()

7
eive_tmtc/pus_tc/devs/gps.py
View File

@ -1,14 +1,13 @@
import enum
import logging
from eive_tmtc.config.definitions import CustomServiceList
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.logging import get_console_logger
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
class OpCode:
@ -40,7 +39,7 @@ def add_gps_cmds(defs: TmtcDefinitionWrapper):
def pack_gps_command(object_id: bytes, q: DefaultPusQueueHelper, op_code: str):
if op_code in OpCode.RESET_GNSS:
# TODO: This needs to be re-implemented
LOGGER.warning("Reset pin handling 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(

8
eive_tmtc/pus_tc/devs/plpcdu.py
View File

@ -1,4 +1,5 @@
import enum
import logging
import struct
import time
from typing import Optional
@ -24,11 +25,10 @@ from tmtccmd.tc.pus_20_fsfw_param import (
create_load_param_cmd,
pack_boolean_parameter_app_data,
)
from tmtccmd.logging import get_console_logger
from spacepackets.ecss.tc import PusTelecommand
from eive_tmtc.config.object_ids import PL_PCDU_ID
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
class OpCode:
@ -369,10 +369,10 @@ def request_wait_time() -> Optional[float]:
try:
wait_time = float(wait_time)
except ValueError:
LOGGER.warning("Invalid input")
_LOGGER.warning("Invalid input")
continue
if wait_time <= 0:
LOGGER.warning("Invalid input")
_LOGGER.warning("Invalid input")
else:
return wait_time

5
eive_tmtc/pus_tc/procedure_packer.py
View File

@ -1,5 +1,6 @@
"""Hook function which packs telecommands based on service and operation code string
"""
import logging
from typing import cast
from eive_tmtc.tmtc.power.power import pack_power_commands
@ -74,8 +75,6 @@ from eive_tmtc.pus_tc.system.proc import pack_proc_commands
import eive_tmtc.config.object_ids as oids
from tmtccmd.util import ObjectIdU32
LOGGER = get_console_logger()
def handle_default_procedure(
tc_base: TcHandlerBase,
@ -216,4 +215,4 @@ def handle_default_procedure(
queue_helper=queue_helper,
),
):
LOGGER.warning(f"Invalid Service {service}")
logging.getLogger(__name__).warning(f"Invalid Service {service}")

8
eive_tmtc/pus_tc/system/time.py
View File

@ -1,14 +1,12 @@
import logging
from datetime import datetime
from eive_tmtc.config.definitions import CustomServiceList
from spacepackets.ecss import PusTelecommand
from tmtccmd import DefaultProcedureInfo, TcHandlerBase
from tmtccmd.logging import get_console_logger
from tmtccmd.tc import DefaultPusQueueHelper, service_provider
from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams
LOGGER = get_console_logger()
class OpCode:
@ -24,6 +22,6 @@ def pack_set_current_time_ascii_command(p: ServiceProviderParams):
q = p.queue_helper
time_test_current_time = datetime.utcnow().isoformat() + "Z" + "\0"
current_time_ascii = time_test_current_time.encode("ascii")
LOGGER.info(f"Current time in ASCII format: {current_time_ascii}")
logging.getLogger(__name__).info(f"Current time in ASCII format: {current_time_ascii}")
q.add_log_cmd(Info.SET_CURRENT_TIME)
q.add_pus_tc(PusTelecommand(service=9, subservice=128, app_data=current_time_ascii))

5
eive_tmtc/pus_tm/action_reply_handler.py
View File

@ -6,12 +6,11 @@ from eive_tmtc.tmtc.payload.ploc_mpsoc import PlocReplyIds
from eive_tmtc.tmtc.payload.ploc_supervisor import SupvActionId
from eive_tmtc.tmtc.acs.star_tracker import StarTrackerActionId
from eive_tmtc.tmtc.power.tm import handle_get_param_data_reply
from tmtccmd.logging import get_console_logger
from tmtccmd.tm import Service8FsfwTm
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
from spacepackets.ccsds.time import CdsShortTimestamp
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
def handle_action_reply(
@ -123,7 +122,7 @@ def handle_startracker_replies(
):
if action_id == StarTrackerActionId.CHECKSUM:
if len(custom_data) != 5:
LOGGER.warning(
_LOGGER.warning(
"Star tracker reply has invalid length {0}".format(len(custom_data))
)
return

10
eive_tmtc/pus_tm/event_handler.py
View File

@ -1,3 +1,4 @@
import logging
from datetime import datetime
from eive_tmtc.config.events import get_event_dict
@ -8,12 +9,11 @@ from eive_tmtc.tmtc.acs.subsystem import AcsMode
from tmtccmd.tc.pus_200_fsfw_mode import Mode
from tmtccmd.tm import Service5Tm
from tmtccmd.logging import get_console_logger
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
from tmtccmd.fsfw import EventInfo
from spacepackets.ccsds.time import CdsShortTimestamp
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
def handle_event_packet(raw_tm: bytes, printer: FsfwTmTcPrinter):
@ -23,13 +23,13 @@ def handle_event_packet(raw_tm: bytes, printer: FsfwTmTcPrinter):
event_def = tm.event_definition
info = event_dict.get(event_def.event_id)
if info is None:
LOGGER.warning(f"Event ID {event_def.event_id} has no information")
_LOGGER.warning(f"Event ID {event_def.event_id} has no information")
info = EventInfo()
info.name = "Unknown event"
obj_ids = get_object_ids()
obj_id_obj = obj_ids.get(event_def.reporter_id)
if obj_id_obj is None:
LOGGER.warning(f"Object ID 0x{event_def.reporter_id.hex(sep=',')} has no name")
_LOGGER.warning(f"Object ID 0x{event_def.reporter_id.hex(sep=',')} has no name")
obj_name = event_def.reporter_id.hex(sep=",")
else:
obj_name = obj_id_obj.name
@ -37,7 +37,7 @@ def handle_event_packet(raw_tm: bytes, printer: FsfwTmTcPrinter):
pw.printer.file_logger.info(
f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}: {generic_event_string}"
)
LOGGER.info(generic_event_string)
_LOGGER.info(generic_event_string)
specific_handler = True
if info.name == "MODE_TRANSITION_FAILED":
reason = generic_retval_printout(event_def.param1)

17
eive_tmtc/pus_tm/factory_hook.py
View File

@ -1,11 +1,12 @@
"""Core EIVE TM handler module
"""
import logging
from eive_tmtc.config.object_ids import get_object_ids
from spacepackets.ecss import PusTelemetry
from spacepackets.ecss.pus_17_test import Service17Tm
from spacepackets.util import PrintFormats
from spacepackets.ccsds.time import CdsShortTimestamp
from tmtccmd import get_console_logger
from tmtccmd.logging.pus import RawTmtcTimedLogWrapper
from tmtccmd.pus import VerificationWrapper
from tmtccmd.tm import Service20FsfwTm, Service200FsfwTm
@ -20,7 +21,7 @@ from .verification_handler import handle_service_1_fsfw_packet, generic_retval_p
from .hk_handling import handle_hk_packet
from .action_reply_handler import handle_action_reply
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
def pus_factory_hook(
@ -30,13 +31,13 @@ def pus_factory_hook(
raw_logger: RawTmtcTimedLogWrapper,
):
if len(packet) < 8:
LOGGER.warning("Detected packet shorter than 8 bytes!")
_LOGGER.warning("Detected packet shorter than 8 bytes!")
return
try:
tm_packet = PusTelemetry.unpack(packet, CdsShortTimestamp.empty())
except ValueError:
LOGGER.warning("Could not generate PUS TM object from raw data")
LOGGER.warning(f"Raw Packet: [{packet.hex(sep=',')}], REPR: {packet!r}")
_LOGGER.warning("Could not generate PUS TM object from raw data")
_LOGGER.warning(f"Raw Packet: [{packet.hex(sep=',')}], REPR: {packet!r}")
return
service = tm_packet.service
obj_id_dict = get_object_ids()
@ -68,7 +69,7 @@ def pus_factory_hook(
obj = obj_id_dict.get(param_wrapper.param_tm.object_id)
pw.dlog(f"Received parameter dump TM from {obj}")
pw.dlog(f"Parameter: {param}")
if param.row == 1 and param.column == 1:
if param.rows == 1 and param.columns == 1:
try:
scalar_param = param.parse_scalar_param()
if isinstance(scalar_param, int):
@ -81,7 +82,7 @@ def pus_factory_hook(
# TODO: Could improve display further by actually displaying a matrix as a
# matrix using row and column information
pw.dlog(
f"Received vector or matrix data: {param.param_data.hex(sep=',')}"
f"Received vector or matrix data: {param.param_raw.hex(sep=',')}"
)
except ValueError as e:
pw.dlog(f"received {e} when trying to parse parameters")
@ -105,7 +106,7 @@ def pus_factory_hook(
else:
dedicated_handler = False
else:
LOGGER.info(f"The service {service} is not implemented in Telemetry Factory")
_LOGGER.info(f"The service {service} is not implemented in Telemetry Factory")
tm_packet.print_source_data(PrintFormats.HEX)
dedicated_handler = True
if not dedicated_handler and tm_packet is not None:

13
eive_tmtc/pus_tm/hk_handling.py
View File

@ -1,4 +1,6 @@
"""HK Handling for EIVE OBSW"""
import logging
# from pus_tm.tcp_server_objects import TCP_SEVER_SENSOR_TEMPERATURES
from eive_tmtc.tmtc.acs.acs_ctrl import handle_raw_mgm_data, handle_acs_ctrl_hk_data
from eive_tmtc.pus_tm.devs.plpcdu import handle_plpcdu_hk
@ -14,7 +16,6 @@ from tmtccmd.tm.pus_3_fsfw_hk import (
Service3FsfwTm,
)
from tmtccmd.util.obj_id import ObjectIdU32, ObjectIdDictT
from tmtccmd.logging import get_console_logger
from eive_tmtc.pus_tm.devs.bpx_bat import handle_bpx_hk_data
from eive_tmtc.pus_tm.devs.gps import handle_gps_data
@ -37,7 +38,7 @@ from eive_tmtc.pus_tm.devs.mgms import handle_mgm_hk_data
import eive_tmtc.config.object_ids as obj_ids
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
FORWARD_SENSOR_TEMPS = False
@ -74,11 +75,11 @@ def handle_hk_packet(
hk_data=hk_data,
)
except ValueError as e:
LOGGER.exception(
_LOGGER.exception(
f"{e} error when parsing HK data coming from {named_obj_id}"
)
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(
@ -106,7 +107,7 @@ def handle_regular_hk_print(
elif set_id == ImtqSetId.RAW_MTM_SET:
return handle_raw_mtm_measurement(printer, hk_data)
else:
LOGGER.info("Service 3 TM: IMTQ handler reply with unknown set id")
_LOGGER.info("Service 3 TM: IMTQ handler reply with unknown set id")
elif objb == obj_ids.GPS_CONTROLLER:
return handle_gps_data(printer=printer, hk_data=hk_data)
elif objb == obj_ids.BPX_HANDLER_ID:
@ -177,7 +178,7 @@ def handle_regular_hk_print(
elif objb == obj_ids.ACS_CONTROLLER:
handle_acs_ctrl_hk_data(printer, set_id, hk_data)
else:
LOGGER.info(
_LOGGER.info(
f"Service 3 TM: Parsing for object {object_id} and set ID {set_id} "
f"has not been implemented."
)

8
eive_tmtc/pus_tm/verification_handler.py
View File

@ -1,13 +1,13 @@
import logging
from typing import List, Optional
from spacepackets.ccsds import CdsShortTimestamp
from spacepackets.ecss.pus_1_verification import UnpackParams, Service1Tm
from tmtccmd.logging import get_console_logger
from tmtccmd.pus import VerificationWrapper
from tmtccmd.tm.pus_1_verification import Service1FsfwWrapper
from eive_tmtc.config.retvals import get_retval_dict
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
def handle_service_1_fsfw_packet(wrapper: VerificationWrapper, raw_tm: bytes):
@ -22,11 +22,11 @@ def handle_service_1_fsfw_packet(wrapper: VerificationWrapper, raw_tm: bytes):
fsfw_wrapper = Service1FsfwWrapper(tm_packet)
res = wrapper.verificator.add_tm(tm_packet)
if res is None:
LOGGER.info(
_LOGGER.info(
f"Received Verification TM[{tm_packet.service}, {tm_packet.subservice}] "
f"with Request ID {tm_packet.tc_req_id.as_u32():#08x}"
)
LOGGER.warning(f"No matching telecommand found for {tm_packet.tc_req_id}")
_LOGGER.warning(f"No matching telecommand found for {tm_packet.tc_req_id}")
else:
wrapper.log_to_console(tm_packet, res)
wrapper.log_to_file(tm_packet, res)

7
eive_tmtc/tmtc/acs/acs_ctrl.py
View File

@ -1,4 +1,5 @@
import enum
import logging
import socket
import struct
from socket import AF_INET
@ -7,7 +8,6 @@ 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,
@ -25,9 +25,6 @@ from tmtccmd.tc.pus_3_fsfw_hk import (
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
LOGGER = get_console_logger()
class SetId(enum.IntEnum):
MGM_RAW_SET = 0
MGM_PROC_SET = 1
@ -378,7 +375,7 @@ def pack_acs_ctrl_command(p: ServiceProviderParams):
)
)
else:
LOGGER.info(f"Unknown op code {op_code}")
logging.getLogger(__name__).info(f"Unknown op code {op_code}")
def handle_acs_ctrl_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):

10
eive_tmtc/tmtc/acs/star_tracker.py
View File

@ -6,18 +6,18 @@
@date 14.08.2021
"""
import enum
import logging
import struct
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.pus_200_fsfw_mode import pack_mode_data, Mode
from tmtccmd.logging import get_console_logger
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.util import ObjectIdU32
from eive_tmtc.utility.input_helper import InputHelper
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
class StarTrackerActionId(enum.IntEnum):
@ -647,7 +647,7 @@ def pack_checksum_command(object_id: bytes) -> bytearray:
def get_config_file() -> str:
LOGGER.info("Specify json file")
_LOGGER.info("Specify json file")
input_helper = InputHelper(json_dict)
key = input_helper.get_key()
json_file = json_dict[key][1]
@ -655,7 +655,7 @@ def get_config_file() -> str:
def get_firmware() -> str:
LOGGER.info("Specify firmware file")
_LOGGER.info("Specify firmware file")
input_helper = InputHelper(firmware_dict)
key = input_helper.get_key()
firmware = firmware_dict[key][1]
@ -663,7 +663,7 @@ def get_firmware() -> str:
def get_upload_image() -> str:
LOGGER.info("Specify image to upload")
_LOGGER.info("Specify image to upload")
input_helper = InputHelper(upload_image_dict)
key = input_helper.get_key()
image = upload_image_dict[key][1]

13
eive_tmtc/tmtc/core.py
View File

@ -1,4 +1,5 @@
import enum
import logging
import struct
from eive_tmtc.pus_tm.defs import PrintWrapper
@ -15,7 +16,7 @@ from tmtccmd.config.tmtc import OpCodeEntry, tmtc_definitions_provider
from eive_tmtc.config.object_ids import CORE_CONTROLLER_ID
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
class ActionId(enum.IntEnum):
@ -244,7 +245,7 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
while True:
active_sd_card = int(input("Please specify active SD cqrd [0/1]: "))
if active_sd_card not in [0, 1]:
LOGGER.warning("Invalid SD card specified. Try again")
_LOGGER.warning("Invalid SD card specified. Try again")
break
q.add_log_cmd(Info.SWITCH_TO_BOTH_SD_CARDS)
q.add_pus_tc(
@ -282,9 +283,9 @@ def determine_reboot_params() -> (bool, Chip, Copy):
copy_select = -1
reboot_self = input("Reboot self? [y/n]: ")
if reboot_self in ["y", "yes", "1"]:
LOGGER.info("Rebooting currently running image")
_LOGGER.info("Rebooting currently running image")
return True, chip_select, copy_select
LOGGER.info("Rebooting image specified by chip and copy")
_LOGGER.info("Rebooting image specified by chip and copy")
return False, determine_chip_and_copy()
@ -298,7 +299,7 @@ def determine_chip_and_copy() -> (int, int):
chip_select = Chip.CHIP_1
break
else:
LOGGER.warning("Invalid chip select value. Try again")
_LOGGER.warning("Invalid chip select value. Try again")
while True:
copy_select = input("Copy select [0/1]: ")
if copy_select in ["0", "1"]:
@ -308,7 +309,7 @@ def determine_chip_and_copy() -> (int, int):
copy_select = Copy.COPY_1_GOLD
break
else:
LOGGER.warning("Invalid copy select value. Try again")
_LOGGER.warning("Invalid copy select value. Try again")
return chip_select, copy_select

10
eive_tmtc/tmtc/payload/ploc_mpsoc.py
View File

@ -6,6 +6,7 @@
@author J. Meier
@date 06.03.2021
"""
import logging
import struct
import enum
@ -16,7 +17,6 @@ from tmtccmd.config.tmtc import (
OpCodeEntry,
TmtcDefinitionWrapper,
)
from tmtccmd.logging import get_console_logger
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams
@ -24,7 +24,7 @@ from eive_tmtc.utility.input_helper import InputHelper
from tmtccmd.tc.pus_200_fsfw_mode import pack_mode_data, Mode
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
MANUAL_INPUT = "1"
@ -324,7 +324,7 @@ def prepare_replay_write_sequence_cmd(object_id: bytes) -> bytearray:
def get_obc_file() -> str:
LOGGER.info("Specify OBC file ")
_LOGGER.info("Specify OBC file ")
input_helper = InputHelper(flash_write_file_dict)
key = input_helper.get_key()
if key == MANUAL_INPUT:
@ -335,7 +335,7 @@ def get_obc_file() -> str:
def get_mpsoc_file() -> str:
LOGGER.info("Specify MPSoC file")
_LOGGER.info("Specify MPSoC file")
input_helper = InputHelper(mpsoc_file_dict)
key = input_helper.get_key()
if key == MANUAL_INPUT:
@ -346,7 +346,7 @@ def get_mpsoc_file() -> str:
def get_sequence_file() -> str:
LOGGER.info("Specify sequence file")
_LOGGER.info("Specify sequence file")
input_helper = InputHelper(SEQ_FILE_DICT)
key = input_helper.get_key()
if key == MANUAL_INPUT:

8
eive_tmtc/tmtc/payload/ploc_supervisor.py
View File

@ -7,6 +7,7 @@
@date 10.07.2021
"""
import enum
import logging
import struct
from eive_tmtc.config.object_ids import PLOC_SUPV_ID, get_object_ids
@ -16,7 +17,6 @@ from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.pus_3_fsfw_hk import generate_one_hk_command, make_sid
from tmtccmd.config import TmtcDefinitionWrapper
from tmtccmd.config.tmtc import tmtc_definitions_provider, OpCodeEntry
from tmtccmd.logging import get_console_logger
from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams
from tmtccmd.pus.s200_fsfw_mode import pack_mode_data, Mode
@ -24,7 +24,7 @@ from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
from eive_tmtc.utility.input_helper import InputHelper
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
latchup_id_dict = {
"0": "0.85V",
@ -698,7 +698,7 @@ def pack_logging_set_topic(object_id: bytes) -> bytearray:
def get_update_file() -> str:
LOGGER.info("Specify update file ")
_LOGGER.info("Specify update file ")
input_helper = InputHelper(update_file_dict)
key = input_helper.get_key()
if key == HARDCODED:
@ -711,7 +711,7 @@ def get_update_file() -> str:
def get_event_buffer_path() -> str:
LOGGER.info("Specify path where to store event buffer file ")
_LOGGER.info("Specify path where to store event buffer file ")
input_helper = InputHelper(event_buffer_path_dict)
key = input_helper.get_key()
if key == MANUAL_INPUT:

7
eive_tmtc/tmtc/power/power.py
View File

@ -1,4 +1,5 @@
import enum
import logging
from eive_tmtc.tmtc.power.common_power import (
PowerOpCodes,
@ -27,7 +28,6 @@ from eive_tmtc.tmtc.power.pdu2 import (
pdu2_switch_cmds,
add_pdu2_cmds,
)
from tmtccmd import get_console_logger
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from eive_tmtc.tmtc.power.p60dock import P60OpCode, P60Info, p60_dock_req_hk_cmds
@ -37,9 +37,6 @@ from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
LOGGER = get_console_logger()
class SetId(enum.IntEnum):
SWITCHER_SET = 0
@ -84,7 +81,7 @@ def pack_power_commands(q: DefaultPusQueueHelper, op_code: str):
)
)
if q.empty():
LOGGER.info(f"Queue is empty, no stack for op code {op_code}")
logging.getLogger(__name__).info(f"Queue is empty, no stack for op code {op_code}")
@tmtc_definitions_provider

8
eive_tmtc/tmtc/solar_array_deployment.py
View File

@ -5,6 +5,7 @@
@author J. Meier
@date 15.02.2021
"""
import logging
import struct
from eive_tmtc.config.definitions import CustomServiceList
@ -18,9 +19,6 @@ from tmtccmd.config.tmtc import (
from tmtccmd.tc import service_provider, DefaultPusQueueHelper
from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd
from tmtccmd.tc.decorator import ServiceProviderParams
from tmtccmd import get_console_logger
LOGGER = get_console_logger()
class OpCode:
@ -90,7 +88,7 @@ def pack_solar_array_deployment_test_into(p: ServiceProviderParams):
def prompt_burn_time() -> int:
burn_time = int(input("Please specify burn time in seconds [0-120 secs]: "))
if burn_time < 0 or burn_time > 120:
LOGGER.warning(f"Invalid burn time {burn_time}")
logging.getLogger(__name__).warning(f"Invalid burn time {burn_time}")
return -1
return burn_time
@ -102,7 +100,7 @@ def prompt_dry_run() -> int:
elif dry_run in ["no", "n", "0"]:
return 0
else:
LOGGER.warning("Invalid input for dry run parameter")
logging.getLogger(__name__).warning("Invalid input for dry run parameter")
return -1

12
eive_tmtc/utility/input_helper.py
View File

@ -5,10 +5,10 @@
@author J. Meier
@date 13.02.2021
"""
import logging
from tmtccmd.logging import get_console_logger
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
class InputHelper:
@ -25,7 +25,7 @@ class InputHelper:
"""
key = self.menu_handler()
while key not in self.menu:
LOGGER.info("Invalid key specified, try again.")
_LOGGER.info("Invalid key specified, try again.")
key = self.menu_handler()
return key
@ -36,11 +36,11 @@ class InputHelper:
separator_string = separator_width * "-"
key_string = "Key".ljust(key_column_width)
description_string = "Description".ljust(description_column_width)
LOGGER.info(f"{key_string} | {description_string}")
LOGGER.info(separator_string)
_LOGGER.info(f"{key_string} | {description_string}")
_LOGGER.info(separator_string)
for key in self.menu:
key_string = key.ljust(key_column_width)
description_string = self.menu[key][0].ljust(description_column_width)
LOGGER.info(f"{key_string} | {description_string}")
_LOGGER.info(f"{key_string} | {description_string}")
key = input("Specify key: ")
return key

44
tmtcc.py
View File

@ -6,7 +6,6 @@ import traceback
from pathlib import Path
from typing import cast
from spacepackets import SpacePacketHeader, SpacePacket
from spacepackets.ccsds import SPACE_PACKET_HEADER_SIZE
from spacepackets.cfdp import (
ConditionCode,
@ -17,6 +16,7 @@ from spacepackets.cfdp import (
PduFactory,
PduType,
)
from tmtccmd.logging import add_colorlog_console_logger
from tmtccmd.cfdp import CfdpUserBase, TransactionId
from tmtccmd.cfdp.defs import CfdpRequestType
from tmtccmd.cfdp.handler import CfdpInCcsdsHandler
@ -54,7 +54,7 @@ except ImportError as error:
sys.exit(1)
from spacepackets.ecss import PusVerificator
from tmtccmd import get_console_logger, TcHandlerBase, BackendBase
from tmtccmd import TcHandlerBase, BackendBase
from tmtccmd.util import FileSeqCountProvider, PusFileSeqCountProvider
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
@ -96,7 +96,7 @@ from eive_tmtc.config.hook import EiveHookObject
from eive_tmtc.pus_tm.factory_hook import pus_factory_hook
from eive_tmtc.pus_tc.procedure_packer import handle_default_procedure
LOGGER = get_console_logger()
_LOGGER = logging.getLogger(__name__)
# Put rotating file logger parameters here for quick changes
ROTATING_TIMED_LOGGER_INTERVAL_WHEN = TimedLogWhen.PER_MINUTE
@ -119,13 +119,13 @@ class EiveCfdpFaultHandler(DefaultFaultHandlerBase):
class EiveCfdpUser(CfdpUserBase):
def transaction_indication(self, transaction_id: TransactionId):
LOGGER.info(f"CFDP User: Start of File {transaction_id}")
_LOGGER.info(f"CFDP User: Start of File {transaction_id}")
def eof_sent_indication(self, transaction_id: TransactionId):
LOGGER.info(f"CFDP User: EOF sent for {transaction_id}")
_LOGGER.info(f"CFDP User: EOF sent for {transaction_id}")
def transaction_finished_indication(self, params: TransactionFinishedParams):
LOGGER.info(f"CFDP User: {params.transaction_id} finished")
_LOGGER.info(f"CFDP User: {params.transaction_id} finished")
def metadata_recv_indication(self, params: MetadataRecvParams):
pass
@ -178,7 +178,7 @@ class PusHandler(SpecificApidHandlerBase):
class UnknownApidHandler(GenericApidHandlerBase):
def handle_tm(self, apid: int, _packet: bytes, _user_args: any):
LOGGER.warning(f"Packet with unknwon APID {apid} detected")
_LOGGER.warning(f"Packet with unknwon APID {apid} detected")
class CfdpInCcsdsWrapper(SpecificApidHandlerBase):
@ -192,12 +192,12 @@ class CfdpInCcsdsWrapper(SpecificApidHandlerBase):
pdu = packet[SPACE_PACKET_HEADER_SIZE:]
pdu_base = PduFactory.from_raw(pdu)
if pdu_base.pdu_type == PduType.FILE_DATA:
LOGGER.info("Received File Data PDU TM")
_LOGGER.info("Received File Data PDU TM")
else:
if pdu_base.directive_type == DirectiveType.FINISHED_PDU:
LOGGER.info(f"Received Finished PDU TM")
_LOGGER.info(f"Received Finished PDU TM")
else:
LOGGER.info(
_LOGGER.info(
f"Received File Directive PDU with type {pdu_base.directive_type!r} TM"
)
self.handler.pass_pdu_packet(pdu_base)
@ -253,7 +253,7 @@ class TcHandler(TcHandlerBase):
raw_tc = pus_tc_wrapper.pus_tc.pack()
self.pus_raw_logger.log_tc(pus_tc_wrapper.pus_tc)
tc_info_string = f"Sent {pus_tc_wrapper.pus_tc}"
LOGGER.info(tc_info_string)
_LOGGER.info(tc_info_string)
self.high_level_file_logger.info(
f"{get_current_time_string(True)}: {tc_info_string}"
)
@ -273,7 +273,7 @@ class TcHandler(TcHandlerBase):
# self.cfdp_counter += 1
elif entry_helper.entry_type == TcQueueEntryType.LOG:
log_entry = entry_helper.to_log_entry()
LOGGER.info(log_entry.log_str)
_LOGGER.info(log_entry.log_str)
self.high_level_file_logger.info(log_entry.log_str)
def handle_cfdp_procedure(self, info: ProcedureWrapper):
@ -285,7 +285,7 @@ class TcHandler(TcHandlerBase):
):
put_req = cfdp_procedure.request_wrapper.to_put_request()
put_req.cfg.destination_id = self.cfdp_dest_id
LOGGER.info(
_LOGGER.info(
f"CFDP: Starting file put request with parameters:\n{put_req}"
)
self.cfdp_in_ccsds_wrapper.handler.cfdp_handler.put_request(put_req)
@ -319,16 +319,14 @@ class TcHandler(TcHandlerBase):
if info is not None:
if info.proc_type == TcQueueEntryType.PUS_TC:
def_proc = info.to_def_procedure()
LOGGER.info(
_LOGGER.info(
f"Finished queue for service {def_proc.service} and op code {def_proc.op_code}"
)
elif info.proc_type == TcProcedureType.CFDP:
LOGGER.info(f"Finished CFDP queue")
_LOGGER.info(f"Finished CFDP queue")
def setup_params() -> SetupWrapper:
print(f"-- eive tmtc v{__version__} --")
print(f"-- spacepackets v{spacepackets.__version__} --")
hook_obj = EiveHookObject(default_json_path())
params = SetupParams()
parser_wrapper = PreArgsParsingWrapper()
@ -395,7 +393,7 @@ def setup_tmtc_handlers(
gui: bool,
) -> (CcsdsTmHandler, TcHandler):
cfdp_in_ccsds_wrapper = setup_cfdp_handler()
verification_wrapper = VerificationWrapper(verificator, LOGGER, printer.file_logger)
verification_wrapper = VerificationWrapper(verificator, _LOGGER, printer.file_logger)
pus_handler = PusHandler(verification_wrapper, printer, raw_logger)
ccsds_handler = CcsdsTmHandler(generic_handler=UnknownApidHandler(None))
ccsds_handler.add_apid_handler(pus_handler)
@ -429,11 +427,15 @@ def setup_backend(
def main():
print(f"-- eive tmtc v{__version__} --")
print(f"-- spacepackets v{spacepackets.__version__} --")
add_colorlog_console_logger(_LOGGER)
try:
setup_wrapper = setup_params()
except KeyboardInterrupt as e:
LOGGER.info(f"{e}. Exiting")
_LOGGER.info(f"{e}. Exiting")
sys.exit(0)
tmtc_logger = RegularTmtcLogWrapper()
printer = FsfwTmTcPrinter(tmtc_logger.logger)
raw_logger = RawTmtcTimedLogWrapper(
@ -456,11 +458,11 @@ def main():
if state.request == BackendRequest.TERMINATION_NO_ERROR:
sys.exit(0)
elif state.request == BackendRequest.DELAY_IDLE:
LOGGER.info("TMTC Client in IDLE mode")
_LOGGER.info("TMTC Client in IDLE mode")
time.sleep(3.0)
elif state.request == BackendRequest.DELAY_LISTENER:
if tc_handler.cfdp_done():
LOGGER.info("CFDP transaction done, closing client")
_LOGGER.info("CFDP transaction done, closing client")
sys.exit(0)
time.sleep(0.5)
elif state.request == BackendRequest.DELAY_CUSTOM: