Connection

icspacket.proto.cotp.connection.COTP_DEFAULT_SRC_REF = 0

Default source reference identifier for COTP connections.

class icspacket.proto.cotp.connection.COTP_Connection(src_ref: int | None = None, sock: socket | None = None, sock_cls: type[socket] | None = None, protocol_class: TPDU_Class = TPDU_Class.CLASS0, max_tpdu_size: TPDU_Size = TPDU_Size.SIZE_1024, src_tsap: str = '0000', dst_tsap: str = '0001', parameters: Iterable[Parameter] | None = None, timeout: float | None = None)[source]

Manages a COTP (Connection-Oriented Transport Protocol) connection over a TCP/IP socket.

This class provides a high-level interface to establish, maintain, and terminate a COTP session on top of a stream-oriented transport such as TCP. It encapsulates the connection establishment (CR/CC exchange), error handling, data segmentation, and orderly connection release via DR TPDU messages.

Example:

>>> conn = COTP_Connection(sock_cls=tpktsock)
>>> conn.connect(("127.0.0.1", 1234))
>>> conn.send_data(b"Hello, world!")
>>> data = conn.recv_data()
>>> conn.close()
Parameters:

  • src_ref (int | None) – Source reference identifier (defaults to COTP_DEFAULT_SRC_REF).

  • sock (socket.socket | None) – Existing socket instance to use for communication.

  • sock_cls (type[socket.socket] | None) – Socket class to use if sock is not provided.

  • protocol_class (TPDU_Class) – The COTP protocol class to use (default: TPDU_Class.CLASS0).

  • max_tpdu_size (TPDU_Size) – Negotiated maximum TPDU size capability (default: TPDU_Size.SIZE_1024).

  • src_tsap (str) – Calling transport selector as a hex string (default: “0000”).

  • dst_tsap (str) – Called transport selector as a hex string (default: “0001”).

  • parameters (Iterable[Parameter] | None) – Additional transport parameters to include in the CR TPDU.

property protocol_class: TPDU_Class

Configured transport protocol class.

connect(address: tuple[str, int]) → None[source]

Establish a COTP connection to a remote endpoint.

Parameters:

address (tuple[str, int]) – Tuple specifying the (host, port) for the TCP connection.

Raises:

ConnectionError – If the TCP connection fails or the CC TPDU is invalid.

connect_raw(address: tuple[str, int], tpdu_cr: TPDU_ConnectionRequest) → None[source]

Perform raw COTP connection establishment with a provided CR TPDU.

Parameters:

  • address (tuple[str, int]) – (host, port) tuple for the remote TCP endpoint.

  • tpdu_cr (TPDU_ConnectionRequest) – Pre-constructed CR TPDU to send.

Raises:

  • ConnectionRefusedError – If a Disconnect Request TPDU is received.

  • ConnectionError – If an unexpected or invalid TPDU is received.

close() → None[source]

Close the connection gracefully with a normal disconnect reason.

close_with_reason(reason: TPDU_DisconnectReason = TPDU_DisconnectReason.NORMAL) → None[source]

Close the connection with an explicit disconnect reason.

Parameters:

reason (TPDU_DisconnectReason) – Disconnect reason code (default: NORMAL).

close_raw(dr_tdpu: TPDU_DisconnectRequest) → None[source]

Perform low-level connection close with a provided Disconnect Request TPDU.

Parameters:

dr_tdpu (TPDU_DisconnectRequest) – Pre-constructed DR TPDU to send.

send_tpdu(tpdu: TPDU | TPDU_ConnectionConfirm | TPDU_ConnectionRequest | TPDU_Data | TPDU_DataAcknowledgement | TPDU_DisconnectConfirm | TPDU_DisconnectRequest | TPDU_Error | TPDU_ExpeditedData | TPDU_ExpeditedDataAcknowledgement | TPDU_Reject) → None[source]

Serialize and send a TPDU over the transport connection.

Parameters:

tpdu (TPDU) – TPDU instance to send.

Raises:

ConnectionError – If the connection is not established.

receive_tpdu()[source]

Receive and parse a TPDU from the transport connection.

Returns:

Parsed TPDU instance.

Return type:

TPDU

Raises:

ConnectionError – If the connection is not established or closed.

send_data(data: bytes) → None[source]

Send user data segmented into DT (Data Transfer) TPDUs.

Parameters:

data (bytes) – User data to transmit.

Raises:

ConnectionError – If the connection is not valid or established.

recv_data() → bytes[source]

Receive user data segmented across multiple DT TPDUs (or just a single DT TPDU).

Returns:

The complete reassembled user data stream.

Return type:

bytes

Raises:

ConnectionError – If the connection is not valid or a non-DT TPDU is received.