Telecommand Handling Package

TC Handler Submodule

class tmtccmd.tmtc.handler.FeedWrapper(queue_wrapper: QueueWrapper, auto_dispatch: bool)

Bases: object

This class wraps the queue and some additional information and useful fields which can be set by the user.

Variables:

queue_helper – Can be used to simplify insertion of queue entries like telecommands into the queue
dispatch_next_queue – Can be used to prevent the dispatch of the queue
modes – Currently contains the current TC Mode and TM mode of the calling handler class

class tmtccmd.tmtc.handler.SendCbParams(info: ProcedureWrapper, entry: QueueEntryHelper, com_if: ComInterface)

Bases: object

Wrapper for all important parameters passed to the TC send callback.

Variables:

info – Procedure info about the procedure this queue entry is related too
entry – Queue entry base type. The user can cast this back to the concrete type or just use duck typing if the concrete type is known
com_if – Communication interface. Will generally be used to send the packet, using the tmtccmd.com_if.ComInterface.send() method

Creates the parameters passed to the send callback.

class tmtccmd.tmtc.handler.TcHandlerBase

Bases: ABC

Generic abstract class for a TC handler object. Should be implemented by the user. This object then takes care of sending packets by providing the send_cb() send-callback. It also provides telecommand queues by providing the feed_cb() queue feeder callback.

abstract feed_cb(info: ProcedureWrapper, wrapper: FeedWrapper)

This function will be called to retrieve a telecommand queue from the user code, based on a procedure. The passed feed wrapper can be used to set the TC queue or other parameter like the inter-packet delay.

Parameters:

info – Generic base class for a procedure. For example, the py:class:tmtccmd.tmtc.DefaultProcedureInfo class uses a service string and op code string which can be used in the user code to select between different telecommand queues being packed
wrapper – Wrapper type around the queue. It also contains a queue helper class to simplify adding entries to the telecommand queue

Returns:

abstract queue_finished_cb(info: ProcedureWrapper)

abstract send_cb(send_params: SendCbParams)

This function callback will be called for each queue entry. This also includes miscellaneous queue entries, for example the ones used to log additional information. It is up to the user code implementation to determine the concrete queue entry and what to do with it.

In general, an implementation will perform the following steps:

Determine the queue entry and what to do with it
If applicable, retrieve the raw data to send from the queue entry and send it using the generic communication interface

All delay related entries will generally be handled by the send queue consumer so there is no need to manually delay the application in this callback. However, the queue consumer will not handle log entries so the user needs to take care of handling these entries and log the content to a console, file logger or any other system used to log something.

Parameters:: send_params

TC Queue Submodule

class tmtccmd.tmtc.queue.DefaultPusQueueHelper(queue_wrapper: QueueWrapper, tc_sched_timestamp_len: int, seq_cnt_provider: ProvidesSeqCount | None, pus_verificator: PusVerificator | None, default_pus_apid: int | None)

Bases: QueueHelperBase

Default PUS Queue Helper which simplifies inserting PUS telecommands into the queue. It also provides a way to optionally stamp common PUS TC fields which would otherwise add boilerplate code during the packet creation process. This includes the following packet properties and it is also able to add the telecommand into a provided PUS verificator.

This queue helper also has special support for PUS 11 time tagged PUS telecommands and will perform its core functionality for the time-tagged telecommands as well.

Parameters:

queue_wrapper – Queue Wrapper. All entries are inserted here
default_pus_apid – Default APID which will be stamped onto all provided PUS TC packets
seq_cnt_provider – The sequence count will be stamped onto all provided PUS TC packets
pus_verificator – All provided PUS TCs will be added to this verificator

add_ccsds_tc(space_packet: SpacePacket)

add_pus_tc(pus_tc: PusTc)

pre_add_cb(entry: TcQueueEntryBase)

class tmtccmd.tmtc.queue.LogQueueEntry(log_str: str): Bases: TcQueueEntryBase

class tmtccmd.tmtc.queue.PacketDelayEntry(delay_time: timedelta)

Bases: TcQueueEntryBase

classmethod from_millis(millis: int) → PacketDelayEntry

class tmtccmd.tmtc.queue.PusTcEntry(pus_tc: PusTc): Bases: TcQueueEntryBase

class tmtccmd.tmtc.queue.QueueEntryHelper(base: TcQueueEntryBase | None)

Bases: object

property entry_type: TcQueueEntryType

property is_tc: bool

to_log_entry() → LogQueueEntry

to_packet_delay_entry() → PacketDelayEntry

to_pus_tc_entry() → PusTcEntry

to_raw_tc_entry() → RawTcEntry

to_space_packet_entry() → SpacePacketEntry

to_wait_entry() → WaitEntry

class tmtccmd.tmtc.queue.QueueHelperBase(queue_wrapper: QueueWrapper)

Bases: ABC

add_log_cmd(print_str: str)

add_packet_delay(delay: timedelta)

add_packet_delay_ms(delay_ms: int)

add_raw_tc(tc: bytes)

add_wait(wait_time: timedelta)

add_wait_ms(wait_ms: int)

add_wait_seconds(wait_seconds: float)

empty() → bool

abstract pre_add_cb(entry: TcQueueEntryBase)

class tmtccmd.tmtc.queue.QueueWrapper(info: TcProcedureBase, queue: deque[TcQueueEntryBase], inter_cmd_delay: timedelta = datetime.timedelta(0))

Bases: object

classmethod empty()

class tmtccmd.tmtc.queue.RawTcEntry(tc: bytes): Bases: TcQueueEntryBase

class tmtccmd.tmtc.queue.SpacePacketEntry(space_packet: SpacePacket): Bases: TcQueueEntryBase

class tmtccmd.tmtc.queue.TcQueueEntryBase(etype: TcQueueEntryType)

Bases: object

Generic TC queue entry abstraction. This allows filling the TC queue with custom objects

is_tc() → bool: Check whether concrete object is an actual telecommand

class tmtccmd.tmtc.queue.TcQueueEntryType(value)

Bases: Enum

An enumeration.

CCSDS_TC = 'ccsds-tc'

CUSTOM = 'custom'

LOG = 'log'

PACKET_DELAY = 'set-delay'

PUS_TC = 'pus-tc'

RAW_TC = 'raw-tc'

WAIT = 'wait'

class tmtccmd.tmtc.queue.WaitEntry(wait_time: timedelta)

Bases: TcQueueEntryBase

classmethod from_millis(millis: int) → WaitEntry

TC Procedure Submodule

class tmtccmd.tmtc.procedure.CfdpProcedure

Bases: TcProcedureBase

property cfdp_request_type

class tmtccmd.tmtc.procedure.CustomProcedureInfo(procedure: Any): Bases: TcProcedureBase

class tmtccmd.tmtc.procedure.ProcedureWrapper(procedure: TcProcedureBase | None)

Bases: object

Procedure helper class. It wraps the concrete procedure object but allows easily casting it to concrete types supported by the framework.

property proc_type

to_cfdp_procedure() → CfdpProcedure

to_custom_procedure() → CustomProcedureInfo

to_tree_commanding_procedure() → TreeCommandingProcedure

class tmtccmd.tmtc.procedure.TcProcedureBase(procedure_type: TcProcedureType): Bases: object

class tmtccmd.tmtc.procedure.TcProcedureType(value)

Bases: Enum

An enumeration.

CFDP = 1

CUSTOM = 2

TREE_COMMANDING = 0

class tmtccmd.tmtc.procedure.TreeCommandingProcedure(cmd_path: str | None)

Bases: TcProcedureBase

Generic abstraction for procedures. A procedure can be a single command or a sequence of commands. Generally, one procedure is mapped to a specific TC queue which is packed during run-time

classmethod empty()

Sequential CCSDS Sender Submodule

Used to send multiple TCs in sequence

class tmtccmd.tmtc.ccsds_seq_sender.SenderMode(value)

Bases: IntEnum

An enumeration.

BUSY = 0

DONE = 1

class tmtccmd.tmtc.ccsds_seq_sender.SeqResultWrapper(mode: SenderMode): Bases: object

class tmtccmd.tmtc.ccsds_seq_sender.SequentialCcsdsSender(queue_wrapper: QueueWrapper, tc_handler: TcHandlerBase)

Bases: object

Specific implementation of CommandSenderReceiver to send multiple telecommands in sequence

Parameters:

queue_wrapper – Wrapper object containing the queue and queue handling properties
tc_handler

handle_new_queue_forced(queue_wrapper: QueueWrapper)

handle_non_tc_entry(queue_entry: TcQueueEntryBase) → bool: Checks whether the entry in the pus_tc queue is a telecommand. :param queue_entry: Generic queue entry :return: True if queue entry is telecommand, False if it is not

property mode

no_delay_remaining() → bool

operation(com_if: ComInterface) → SeqResultWrapper

Primary function which should be called periodically to consume a TC queue.

Parameters:: com_if – Communication interface used to send telecommands. Will be passed to the user send function

property queue_wrapper

resume(): Can be used to resume a finished sequential sender it the provided queue is not empty anymore