dcpmessage package

Submodules

dcpmessage.credentials module

class HashAlgo(algorithm)

Bases: object

A class representing a hashing algorithm.

This class serves as a base class for different hashing algorithms, such as SHA-1 and SHA-256, providing a common interface to initialize and create new hash objects.

algorithm

The name of the hashing algorithm.

Type:

str

new()

Create a new hash object using the specified algorithm.

Returns:

A new hash object from the hashlib library.

class Sha1

Bases: HashAlgo

A class representing the SHA-1 hashing algorithm.

Inherits from the HashAlgo class and is pre-configured with the “sha1” algorithm.

class Sha256

Bases: HashAlgo

A class representing the SHA-256 hashing algorithm.

Inherits from the HashAlgo class and is pre-configured with the “sha256” algorithm.

class Credentials(username: str = None, password: str = None)

Bases: object

get_preliminary_hash(password: str) → bytes

Generate the preliminary hash for the password.

This method creates a hash by combining the username and password multiple times.

Parameters:

password – The password to hash.

Returns:

The resulting hash as bytes.

get_authenticator_hash(time: datetime, hash_algo: HashAlgo) → str

Generate an authenticator hash using a specified hash algorithm.

This hash is used for authenticating the user based on the current time and the user’s credentials.

Parameters:

  • time – The current time as a datetime object.

  • hash_algo – The hashing algorithm to use (e.g., Sha1, Sha256).

Returns:

The authenticator hash as a hexadecimal string.

get_authenticated_hello(time: datetime, hash_algo: HashAlgo)

Create an authenticated hello message for the user.

This method combines the username, current time, authenticator hash, and protocol version into a single string used for authentication with the server.

Parameters:

  • time – The current time as a datetime object.

  • hash_algo – The hashing algorithm to use for the authenticator hash (e.g., Sha1, Sha256).

Returns:

The authenticated hello message as a string.

dcpmessage.dcp_message module

class DcpMessage

Bases: object

Class for handling DCP messages, including fetching and processing them from a remote server using the LDDS protocol.

Parameters:

  • DATA_LENGTH – Standard length of the data field in a DCP message.

  • HEADER_LENGTH – Standard length of the header in a DCP message.

DATA_LENGTH = 32
HEADER_LENGTH = 37
static get(username: str, password: str, search_criteria: dict | str | Path, host: str, port: int = 16003, timeout: int = 30, tls_mode: int = 1)

Fetches DCP messages from a server based on provided search criteria.

This method handles the complete process of connecting to the server, authenticating, sending search criteria, retrieving DCP messages, and finally disconnecting.

Parameters:

  • username – Username for server authentication.

  • password – Password for server authentication.

  • search_criteria – File path to search criteria or search criteria as a string.

  • host – Hostname or IP address of the server.

  • port – Port number for server connection (default: 16003).

  • timeout – Connection timeout in seconds (default: 30 seconds). Will be passed to socket.settimeout

  • tls_mode –

    TLS configuration level for the connection (default: 1 - DISABLED). Must be one of the following values from the TlsMode enum:

    • 1: Do not use TLS.

    • 2: Try to upgrade to TLS via STARTTLS. Continue without TLS if upgrade fails.

    • 3: Try to upgrade via STARTTLS. Fail if TLS cannot be established.

    • 4: Require full TLS before any communication begins.

Returns:

List of DCP messages retrieved from the server.

Raises:

  • ValueError – If tls_mode is not valid value.

  • TypeError – If search_criteria is not a valid type (dict, str, or Path).

  • Exception – If connection, authentication, or message retrieval fails.

static explode(message_blocks: list[LddsMessage]) → list[str]

Splits a message block bytes containing multiple DCP messages into individual messages.

Parameters:

message_blocks – message block (concatenated response from the server).

Returns:

A list of individual DCP messages.

dcpmessage.exceptions module

class ServerErrorCode(*values)

Bases: Enum

Enum for server error codes, each with a numeric value and description.

Each member has:

  • A unique numeric value (int)

  • A human-readable description (str), accessible via .description

DSUCCESS = 0
DNOFLAG = 1
DDUMMY = 2
DLONGLIST = 3
DARCERROR = 4
DNOCONFIG = 5
DNOSRCHSHM = 6
DNODIRLOCK = 7
DNODIRFILE = 8
DNOMSGFILE = 9
DDIRSEMERR = 10
DMSGTIMEOUT = 11
DNONETLIST = 12
DNOSRCHCRIT = 13
DBADSINCE = 14
DBADUNTIL = 15
DBADNLIST = 16
DBADADDR = 17
DBADEMAIL = 18
DBADRTRAN = 19
DNLISTXCD = 20
DADDRXCD = 21
DNOLRGSLAST = 22
DWRONGMSG = 23
DNOMOREPROC = 24
DBADDAPSSTAT = 25
DBADTIMEOUT = 26
DCANTIOCTL = 27
DUNTILDRS = 28
DBADCHANNEL = 29
DCANTOPENSER = 30
DBADDCPNAME = 31
DNONAMELIST = 32
DIDXFILEIO = 33
DBADSEARCHCRIT = 34
DUNTIL = 35
DJAVAIF = 36
DNOTATTACHED = 37
DBADKEYWORD = 38
DPARSEERROR = 39
DNONAMELISTSEM = 40
DBADINPUTFILE = 41
DARCFILEIO = 42
DNOARCFILE = 43
DICPIOCTL = 44
DICPIOERR = 45
DINVALIDUSER = 46
DDDSAUTHFAILED = 47
DDDSINTERNAL = 48
DDDSFATAL = 49
DNOSUCHSOURCE = 50
DALREADYATTACHED = 51
DNOSUCHFILE = 52
DTOOMANYDCPS = 53
DBADPASSWORD = 54
DSTRONGREQUIRED = 55
property description

Returns the description of the error code.

Returns:

Human-readable description.

class ServerError(message: str, server_code_no: int = 0, system_code_no: int = 0)

Bases: object

is_end_of_message()
property description
static parse(message: bytes)

Parse ServerError from error_string

Parameters:

message – bytes response returned from server

Returns:

object of ServerError class

raise_exception()
exception ProtocolError

Bases: Exception

exception LddsMessageError

Bases: Exception

dcpmessage.ldds_client module

class TlsMode(*values)

Bases: Enum

Enum representing levels of Transport Layer Security (TLS) to apply during the connection to the server.

DISABLED = 1
OPTIONAL_STARTTLS = 2
REQUIRED_STARTTLS = 3
IMMEDIATE_TLS = 4
class LddsClient(host: str, port: int, timeout: float | int, tls_mode: TlsMode, ssl_context: SSLContext | None)

Bases: object

A client for communicating with an LDDS (Low Data Rate Demodulation System) server.

connect()

Establish a socket connection to the server using the provided host and port. Sets the socket to blocking mode and applies the specified timeout.

Raises:

IOError – If the connection attempt times out or fails for any reason.

Returns:

None

disconnect()

Close the established socket connection.

Returns:

None

send_data(data: bytes)

Send data over the established socket connection.

Parameters:

data – The byte data to send over the socket.

Raises:

IOError – If the socket is not connected.

Returns:

None

receive_data(buffer_size: int = 1024) → bytes

Receive data from the socket.

Parameters:

buffer_size – The size of the buffer to use when receiving data.

Returns:

The received byte data.

Raises:

IOError – If the socket is not connected.

authenticate_user(user_name: str, password: str)

Authenticate a user with the LDDS server using the provided username and password.

Parameters:

  • user_name – The username to authenticate with.

  • password – The password to authenticate with.

Raises:

Exception – If authentication fails.

Returns:

None

request_dcp_message(message_id, message_data: str | bytes | bytearray = '') → LddsMessage

Request a DCP (Data Collection Platform) message from the LDDS server.

Parameters:

  • message_id – The ID of the message to request.

  • message_data – The data to include in the message request.

Returns:

The response from the server as bytes.

send_search_criteria(search_criteria: SearchCriteria)

Send search criteria to the LDDS server.

Parameters:

search_criteria – The search criteria to send.

Returns:

None

request_dcp_blocks() → list[LddsMessage]

Request a block of DCP messages from the LDDS server.

Returns:

The received DCP block as bytearray.

send_goodbye()

Send a goodbye message to the LDDS server to terminate the session.

Returns:

None

dcpmessage.ldds_message module

class LddsMessageConstants(VALID_HEADER_LENGTH: int = 10, SYNC_LENGTH: int = 4, VALID_SYNC_CODE: bytes = b'FAF0', MAX_DATA_LENGTH: int = 99000, VALID_IDS: frozenset[str] = frozenset({'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u'}))

Bases: object

Constants related to LDDS messages.

VALID_HEADER_LENGTH: int = 10
SYNC_LENGTH: int = 4
VALID_SYNC_CODE: bytes = b'FAF0'
MAX_DATA_LENGTH: int = 99000
VALID_IDS: frozenset[str] = frozenset({'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u'})
class LddsMessageIds(hello: str = 'a', goodbye: str = 'b', status: str = 'c', start: str = 'd', stop: str = 'e', dcp: str = 'f', search_criteria: str = 'g', get_outages: str = 'h', idle: str = 'i', put_netlist: str = 'j', get_netlist: str = 'k', assert_outages: str = 'l', auth_hello: str = 'm', dcp_block: str = 'n', events: str = 'o', ret_config: str = 'p', inst_config: str = 'q', dcp_block_ext: str = 'r', unused_6: str = 's', start_tls: str = 't', user: str = 'u')

Bases: object

Ids associated with LDDS messages.

hello: str = 'a'
goodbye: str = 'b'
status: str = 'c'
start: str = 'd'
stop: str = 'e'
dcp: str = 'f'
search_criteria: str = 'g'
get_outages: str = 'h'
idle: str = 'i'
put_netlist: str = 'j'
get_netlist: str = 'k'
assert_outages: str = 'l'
auth_hello: str = 'm'
dcp_block: str = 'n'
events: str = 'o'
ret_config: str = 'p'
inst_config: str = 'q'
dcp_block_ext: str = 'r'
unused_6: str = 's'
start_tls: str = 't'
user: str = 'u'
class LddsMessage(message_id: str = None, message_length: int = None, message_data: bytes = None, header: bytes = None)

Bases: object

check_server_errors()
check_other_errors()
static parse(message: bytes)

Parse bytes into an LddsMessage instance.

Parameters:

message – The message in bytes to parse.

Returns:

A tuple containing an LddsMessage instance and a ServerError if any.

Raises:

ProtocolError – If there is an issue with the message header.

static get_message_length(message: bytes) → int
static get_total_length(message: bytes) → int
static create(message_id: str, message_data: bytes = b'')

Create a LDDS message from scratch.

Parameters:

  • message_id – The ID of the message.

  • message_data – The data of the message in bytes.

Returns:

A new LddsMessage instance.

to_bytes()

Convert the LDDS message to bytes.

Returns:

The message in bytes.

is_end_of_message()
is_success()

dcpmessage.search_criteria module

class DcpMessageSource(*values)

Bases: Enum

Enumeration for DCP message sources with unique values.

Parameters:

  • GOES – Represents a standard GOES message source.

  • GOES_SELFTIMED – Represents a self-timed GOES message source.

  • GOES_RANDOM – Represents a random GOES message source.

GOES = 0
GOES_SELFTIMED = 65536
GOES_RANDOM = 131072
class SearchCriteriaConstants(max_sources: int = 12)

Bases: object

Constants used in SearchCriteria.

Parameters:

max_sources – The maximum number of sources that can be added to the search criteria.

max_sources: int = 12
class DcpAddress(address: str)

Bases: object

Class representing a DCP address.

Parameters:

address – (str) The DCP address, which must be 8 characters long.

class SearchCriteria(lrgs_since: str, lrgs_until: str, dcp_address: list[DcpAddress], sources: list[int])

Bases: object

classmethod from_file(file: str) → SearchCriteria

Create a SearchCriteria object from a JSON file.

Parameters:

file – Path to the JSON file containing search criteria.

Returns:

A SearchCriteria object.

Raises:

Exception – If there is an issue parsing the JSON file.

classmethod from_dict(data: dict) → SearchCriteria

Create a SearchCriteria object from a dict.

Parameters:

data – dict containing search criteria data.

Returns:

A SearchCriteria object.

Raises:

Exception – If there is an issue parsing the JSON file.

dcpmessage.utils module

class ByteUtil

Bases: object

static extract_string(b: bytes | bytearray, offset: int = 0) → str

Pull a null-terminated string out of a byte array starting at given offset.

Parameters:

  • b – The byte array to extract string from.

  • offset – The offset to start reading from. Default is 0.

Returns:

The extracted string.

static to_hex_string(b: bytes | bytearray)

Convert byte array to hex string.

Parameters:

b – The byte array to convert to hex string.

Returns:

Hex string of byte array.

Module contents