CDTP Module

SPDX-FileCopyrightText: 2024 DESY and the Constellation authors SPDX-License-Identifier: EUPL-1.2

Module implementing the Constellation Data Transmission Protocol.

class core.cdtp.CDTPMessage

Bases: object

Class holding details of a received CDTP command.

meta: dict[str, Any] = {}

msgtype: CDTPMessageIdentifier | None = None

name: str = ''

payload: Any = None

sequence_number: int = -1

set_header(name: str, msgtype: int, seqno: int, meta: dict[str, Any]) → None

Sets information retrieved from a message header.

class core.cdtp.CDTPMessageIdentifier(*values)

Bases: Enum

Defines the message types of the CDTP.

Part of the Constellation Satellite Data Protocol, see docs/protocols/cdtp.md for details.

BOR = 1

DAT = 0

EOR = 2

class core.cdtp.DataTransmitter(name: str, socket: Socket | None)

Bases: object

Base class for sending Constellation data packets via ZMQ.

property BOR_timeout: int

The beginning-of-run sending timeout value (in ms).

property EOR_timeout: int

The end-of-run sending timeout value (in ms).

property data_timeout: int

The data sending timeout value (in ms).

decode(binmsg: list[bytes]) → CDTPMessage

Decode a binary message into a CTDPMessage.

recv(flags: int = 0) → CDTPMessage | None

Receive a multi-part data transmission.

Follows the Constellation Data Transmission Protocol.

flags: additional ZMQ socket flags to use during transmission.

Returns: CTDPMessage

send_data(payload: Any, meta: dict[str, Any] | None = None, flags: int = 0) → None

Send data message of data run over a ZMQ socket.

Follows the Constellation Data Transmission Protocol.

payload: meta information about the beginning of run.

meta: optional dictionary that is sent as a map of string/value pairs with the header.

flags: additional ZMQ socket flags to use during transmission.

send_end(payload: Any, meta: dict[str, Any] | None = None, flags: int = 0) → None

Send ending message of data run over a ZMQ socket.

Follows the Constellation Data Transmission Protocol.

payload: meta information about the end of run.

flags: additional ZMQ socket flags to use during transmission.

send_start(payload: Any, meta: dict[str, Any] | None = None, flags: int = 0) → None

Send starting message of data run over a ZMQ socket.

Follows the Constellation Data Transmission Protocol.

payload: meta information about the beginning of run.

flags: additional ZMQ socket flags to use during transmission.