Core Objects#

The foundational objects present on virtually every CIP device: the object registry/base class, Identity, Message Router, Assembly, and the Connection / Connection Manager object pair.

class icspacket.proto.cip.objects._base.CIPString(prefix: _StructLike[int, int], padded: bool = False)[source]#

Length-prefixed CIP string (SHORT_STRING or STRING).

padded follows the CIP STRING type’s word-alignment rule: a payload with an odd byte length is followed by one pad byte so the field always ends on a 16-bit boundary. SHORT_STRING (CIP_SHORT_STRING) never pads; STRING (CIP_STRING) does.

encode(obj: str | bytes, context: _ContextLike) → bytes[source]#

Encode data using the wrapped _StructLike object.

Parameters:

obj – The original data to be encoded.
context – The current context.

Returns:

The encoded data.

decode(parsed: bytes, context: _ContextLike) → str[source]#

Decode data using the wrapped _StructLike object.

Parameters:

parsed – The parsed data to be decoded.
context – The current context.

Returns:

The decoded data.

pack_single(obj: str | bytes, context: _ContextLike) → None[source]#

Pack a single value into the stream using encoding.

Parameters:

obj – The original data to be encoded and packed.
context – The current context.

property minimum_size: int#: Bytes needed before the length prefix itself can be read.

icspacket.proto.cip.objects._base.CIP_SHORT_STRING = <CIPString>#: SHORT_STRING: USINT length prefix, unpadded.

icspacket.proto.cip.objects._base.CIP_STRING = <CIPString>#: STRING: UINT length prefix, word-padded.

class icspacket.proto.cip.objects._base.CIPPrefixedEPATH[source]#

CIP attribute shaped as UINT path_size (16-bit words) + EPATH bytes.

Unlike CIPString, the prefix here counts words, not bytes, so it cannot be expressed as a plain Prefixed wrapper; the word/byte conversion is done directly against the shared stream instead.

unpack_single(context: _ContextLike) → EPATH[source]#

Abstract method to unpack a single element.

Parameters:: context (_ContextLike) – The current operation context.
Raises:: NotImplementedError – This method must be implemented by subclasses.
Returns:: The unpacked element.

pack_single(obj: EPATH, context: _ContextLike) → None[source]#

Abstract method to pack a single element.

Parameters:

obj (Any) – The element to pack.
context (_ContextLike) – The current operation context.

Raises:

NotImplementedError – This method must be implemented by subclasses.

icspacket.proto.cip.objects._base.CIP_PREFIXED_EPATH = <CIPPrefixedEPATH>#: TCP/IP Interface Object attribute 4 (Physical Link Object): the one core CIP attribute shaped as a word-counted EPATH rather than a raw one.

class icspacket.proto.cip.objects._base.CIPEPATH[source]#

Bare CIP EPATH occupying an entire attribute payload.

Unlike CIPPrefixedEPATH, this has no length prefix of its own - it reads to the end of the current stream, so it is only valid as a whole-payload attribute schema and cannot be sequenced before other fields in a CIPAttributeReader multi-attribute decode.

unpack_single(context: _ContextLike) → EPATH[source]#

Abstract method to unpack a single element.

Parameters:: context (_ContextLike) – The current operation context.
Raises:: NotImplementedError – This method must be implemented by subclasses.
Returns:: The unpacked element.

pack_single(obj: EPATH, context: _ContextLike) → None[source]#

Abstract method to pack a single element.

Parameters:

obj (Any) – The element to pack.
context (_ContextLike) – The current operation context.

Raises:

NotImplementedError – This method must be implemented by subclasses.

icspacket.proto.cip.objects._base.CIP_EPATH = <CIPEPATH>#: Connection Object attributes 14/16 (Produced/Consumed Connection Path).

class icspacket.proto.cip.objects._base.CIPAttribute(id: int, schema: _IT, ~icspacket.proto.cip.objects._base._IT] | type[~icspacket.proto.cip.objects._base._IT] | None=None, *, size: int | None = None, order: _EndianLike = ByteOrder(name='Little Endian', ch='<', alignment=<Alignment.NONE: 0>, size=<Size.STANDARD: 0>))[source]#

Descriptor for one CIP object attribute’s wire schema.

Assign as a class attribute on a CIPObject subclass; reading it issues Get_Attribute_Single and decodes the reply via schema, writing it encodes the value and issues Set_Attribute_Single requiring an empty response. schema is any caterpillar field/struct or None for raw, undecoded bytes.

from_bytes(data: bytes) → _IT[source]#: Decode a Get_Attribute_Single reply for this attribute.

to_bytes(value: _IT) → bytes[source]#: Encode a value for Set_Attribute_Single.

property minimum_size: int | None#

Minimum bytes needed before this attribute might be stream- decodable.

None means it cannot be decoded from a sequential stream at all; 0 means any amount (including none) is acceptable.

property stream_decodable: bool#

Whether this attribute can be decoded from the middle of a shared, multi-attribute stream (e.g. a Get_Attributes_All reply) without over-consuming bytes that belong to a later attribute.

True for any fixed-size schema (including an explicit size override) and for the self-delimiting CIPString/ CIPPrefixedEPATH wrappers, each of which reads its own explicit length prefix and then stops. False for a schema with no fixed size that is not self-delimiting - e.g. a bare CIPEPATH or a greedy schema[...] array sized by another attribute - since those read to the end of the stream regardless of where the next attribute’s bytes begin.

class icspacket.proto.cip.objects._base.CIPAttributeReader(data: bytes)[source]#

Sequential reader for variable CIP attribute payloads.

Use this for Get_Attributes_All replies where fields are optional or followed by vendor-specific trailing data. Fixed single-attribute getters should normally use a CIPAttribute descriptor instead.

property remaining: int#: Number of unread bytes.

read_field(schema: _StructLike[_IT, _IT]) → _IT[source]#: Decode one field from the shared stream using schema.

read_attributes(definitions: dict[int, CIPAttribute[Any]], attributes: tuple[int, ...] | list[int] | None = None, *, missing_ok: bool = True) → dict[int, object][source]#

Decode sequential attributes using declarative definitions.

The method stops before the first missing optional attribute or the first definition that cannot be safely decoded from a stream.

read_remaining() → bytes[source]#: Return all unread bytes.

class icspacket.proto.cip.objects._base.CIPObject(connection: CIP_Connection, instance: int = 1)[source]#

Small typed facade over a CIP_Connection.

Subclasses that assign CLASS_CODE directly in their own body are registered automatically under that class code, for object_for().

path(attribute: int | None = None, *, instance: int | None = None) → EPATH[source]#

Build the EPATH addressing this object, optionally down to one attribute.

Parameters:

attribute – Narrows the path to one specific attribute, if given.
instance – Overrides this object’s own instance, if given.

get(attribute: int, *, instance: int | None = None) → bytes[source]#

Read one attribute’s raw bytes via Get_Attribute_Single.

Parameters:

attribute – The attribute ID to read.
instance – Overrides this object’s own instance, if given.

set(attribute: int, value: bytes, *, instance: int | None = None) → bytes[source]#

Write one attribute’s raw bytes via Set_Attribute_Single.

Parameters:

attribute – The attribute ID to write.
value – The encoded attribute value.
instance – Overrides this object’s own instance, if given.

all(*, instance: int | None = None) → bytes[source]#

Read every attribute’s raw bytes via Get_Attributes_All.

Parameters:: instance – Overrides this object’s own instance, if given.

message(service: int | CommonService, request_data: bytes = b'', *, connected: bool = False, instance: int | None = None) → bytes[source]#

Issue a generic (non-Get/Set_Attribute) service request to this object.

Parameters:

service – The CIP service code to invoke.
request_data – The service-specific request payload.
connected – Send over the connected (Class 3) session if True.
instance – Overrides this object’s own instance, if given.

get_attr(attribute: int, *, instance: int | None = None) → _IT | bytes[source]#

Read and decode one attribute using attribute_definitions.

Falls back to raw bytes if attribute has no registered CIPAttribute definition.

Parameters:

attribute – The attribute ID to read.
instance – Overrides this object’s own instance, if given.

set_empty(attribute: int, value: bytes, *, instance: int | None = None) → bytes[source]#

Write an attribute and require an empty service response.

Parameters:

attribute – The attribute ID to write.
value – The encoded attribute value.
instance – Overrides this object’s own instance, if given.

Registry mapping a CIP object class code to its typed wrapper class.

Every core CIP object wrapper (IdentityObject, ConnectionObject, …) registers itself automatically via CIPObject.__init_subclass__, so object_for() can construct the right wrapper for a class code discovered at runtime (e.g. while walking a device’s object list).

icspacket.proto.cip.objects.registry.object_for(connection: CIP_Connection, class_code: int | ClassCode, instance: int = 1) → CIPObject[source]#

Construct the registered wrapper for a standard CIP object class.

Parameters:

connection – The session-bound connection the wrapper will issue requests over.
class_code – The CIP object class code to look up.
instance – The object instance the wrapper addresses (defaults to the first instance).

Raises:

ValueError – if class_code has no registered wrapper.

[ODVA CIP Vol 1] Wrapper for the CIP Identity Object (class 0x01).

Instance attributes are defined according to §5-2.2, Table 5-2.2.

class icspacket.proto.cip.objects.identity.IdentityAttributes(vendor_id: int, device_type: int, product_code: int, revision_major: int, revision_minor: int, status: int, serial_number: int, product_name: str, state: int)[source]#

Identity Object instance attributes 1-8 (See CIP Vol 1, §5-2.2, Table 5-2.2).

vendor_id: int#: Numeric code identifying which vendor manufactured the device.

device_type: int#: Numeric code indicating the device’s general product category.

product_code: int#: Vendor-specific numeric code that pins down the exact product model within that vendor’s own lineup.

revision_major: int#: Major component of the Identity Object’s revision number.

revision_minor: int#: Minor component of the Identity Object’s revision number, reported alongside revision_major.

status: int#: 16-bit value that summarizes the device’s current status.

serial_number: int#: 32-bit value holding the device’s serial number.

product_name: str#: Human-readable product name, decoded as a CIP short string.

state: int#: Single-byte code reporting the device’s current state.

property revision: tuple[int, int]#: Major/minor revision tuple.

class icspacket.proto.cip.objects.identity.IdentityObject(connection: CIP_Connection, instance: int = 1)[source]#

Typed access to Identity Object instance attributes (See CIP Vol 1, §5-2.2, Table 5-2.2).

vendor_id: CIPAttribute[int]#: Vendor ID (attribute 1).

device_type: CIPAttribute[int]#: Device type (attribute 2).

product_code: CIPAttribute[int]#: Product code (attribute 3).

revision: CIPAttribute[Collection[int]]#: Major/minor revision (attribute 4).

status: CIPAttribute[int]#: Identity status word (attribute 5).

serial_number: CIPAttribute[int]#: Serial number (attribute 6).

product_name: CIPAttribute[str]#: Product name (attribute 7).

state: CIPAttribute[int]#: Present state (attribute 8).

get_attributes() → IdentityAttributes[source]#: Read and decode Get_Attributes_All for Identity.

reset(reset_type: int = 0) → bytes[source]#: Invoke the Identity Reset service.

Wrapper for the CIP Message Router Object (class 0x02).

class icspacket.proto.cip.objects.msgrouter.MessageRouterObject(connection: CIP_Connection, instance: int = 1)[source]#

Access instance-level Message Router attributes (See CIP Vol 1, Table 5-3.2).

object_list: CIPAttribute[Collection[int]]#

Object class codes advertised by the device.

On the wire, attribute 1 (instance 1) leads with a 2-byte count followed by that many UINT class codes; the count is stripped here and only the class codes are returned.

connection_count: CIPAttribute[int]#: Message Router connection count (attribute 2).

Wrapper for the CIP Assembly Object (class 0x04).

class icspacket.proto.cip.objects.assembly.AssemblyObject(connection: CIP_Connection, instance: int = 1)[source]#

Raw access to Assembly instance Data (attribute 3).

data: CIPAttribute[bytes]#: Assembly Data (attribute 3).

[ODVA CIP Vol 1] Wrappers for the CIP Connection Object (class 0x05) and Connection Configuration Object (class 0xF3).

Connection Object instance attributes and diagnostics are defined according to §3-4.4, Table 3-4.9. Connection Configuration Object attributes are defined according to §5-48.

class icspacket.proto.cip.objects.connection.ConnectionDiagnostics(state: int, instance_type: int, transport_class_trigger: int, produced_connection_size: int, consumed_connection_size: int, expected_packet_rate: int, produced_connection_id: int, consumed_connection_id: int, watchdog_timeout_action: int)[source]#

Selected Connection Object diagnostic attributes (See CIP Vol 1, §3-4.4, Table 3-4.9).

state: int#: Current operating state of this Connection instance.

instance_type: int#: Marks this Connection instance as either an I/O connection or a Messaging (explicit) connection.

transport_class_trigger: int#: Governs how this Connection transports its data.

produced_connection_size: int#: Upper bound on how many bytes this Connection transmits.

consumed_connection_size: int#: Upper bound on how many bytes this Connection receives.

expected_packet_rate: int#: Sets the packet-timing expectation for this Connection.

produced_connection_id: int#: Identifier attached to messages this Connection sends out on the subnet.

consumed_connection_id: int#: Identifier expected on incoming messages this Connection reads from the subnet.

watchdog_timeout_action: int#: Specifies the action taken when this Connection’s inactivity/watchdog timer expires.

class icspacket.proto.cip.objects.connection.ConnectionObject(connection: CIP_Connection, instance: int = 1)[source]#

Read-only diagnostic attributes of a Connection instance (See CIP Vol 1, §3-4.4, Table 3-4.9).

state: CIPAttribute[int]#: Connection state (attribute 1).

instance_type: CIPAttribute[int]#: Connection instance type (attribute 2).

transport_class_trigger: CIPAttribute[int]#: Transport class and trigger byte (attribute 3).

produced_connection_size: CIPAttribute[int]#: Produced connection size in bytes (attribute 7).

consumed_connection_size: CIPAttribute[int]#: Consumed connection size in bytes (attribute 8).

expected_packet_rate: CIPAttribute[int]#: Expected packet rate in milliseconds (attribute 9).

produced_connection_id: CIPAttribute[int]#: CIP produced connection ID (attribute 10).

consumed_connection_id: CIPAttribute[int]#: CIP consumed connection ID (attribute 11).

watchdog_timeout_action: CIPAttribute[int]#: Watchdog timeout action (attribute 12).

produced_connection_path: CIPAttribute[EPATH]#: Produced connection path (attribute 14).

consumed_connection_path: CIPAttribute[EPATH]#: Consumed connection path (attribute 16).

get_diagnostics() → ConnectionDiagnostics[source]#: Read selected diagnostic attributes as one structured value.

class icspacket.proto.cip.objects.connection.DeviceID(vendor_id: int, product_type: int, product_code: int, major_rev: int, minor_rev: int)[source]#: Vendor/product identification quintuple (Connection Configuration Object attributes 3 and 11).

class icspacket.proto.cip.objects.connection.ConnectionStatus(gen_status: int, reserved: int, ext_status: int)[source]#

Connection Configuration Object attribute 1 payload (See CIP Vol 1, §5-48.2.2.1).

gen_status: int#: General status code (See Table 5-48.5/5-48.6).

reserved: int#: Reserved, shall be zero.

ext_status: int#: Extended status code.

class icspacket.proto.cip.objects.connection.NetConnectionParameters(conn_timeout: int, transport_class_trigger: int, rpi_ot: int, net_ot: int, rpi_to: int, net_to: int)[source]#

Connection Configuration Object attribute 5 payload: Forward_Open-shaped connection parameters (See CIP Vol 1, §5-48.2.2.5).

conn_timeout: int#: Connection timeout multiplier, as used in the Forward_Open request.

transport_class_trigger: int#: Transport Class and Trigger byte, as used in the Forward_Open request.

rpi_ot: int#: Originator to Target Requested Packet Interval, in microseconds.

net_ot: int#: Originator to Target network connection parameters (size/type bitfield).

rpi_to: int#: Target to Originator Requested Packet Interval, in microseconds.

net_to: int#: Target to Originator network connection parameters (size/type bitfield).

class icspacket.proto.cip.objects.connection.LargeNetConnectionParameters(conn_timeout: int, transport_class_trigger: int, rpi_ot: int, net_ot: int, rpi_to: int, net_to: int)[source]#

Connection Configuration Object attribute 19 payload: large-format variant of NetConnectionParameters (See CIP Vol 1, §5-48.2).

conn_timeout: int#: Connection timeout multiplier, as used in the Forward_Open request.

transport_class_trigger: int#: Transport Class and Trigger byte, as used in the Forward_Open request.

rpi_ot: int#: Originator to Target Requested Packet Interval, in microseconds.

net_ot: int#: Originator to Target large-format network connection parameters (size/type bitfield).

rpi_to: int#: Target to Originator Requested Packet Interval, in microseconds.

net_to: int#: Target to Originator large-format network connection parameters (size/type bitfield).

class icspacket.proto.cip.objects.connection.ConnectionPathAttribute(open_path_size: int, reserved: int, open_connection_path: EPATH)[source]#

Connection Configuration Object attribute 6 payload (See CIP Vol 1, §5-48.2.2.6).

open_path_size: int#: Size of open_connection_path in bytes, as used in the Forward_Open request.

reserved: int#: Reserved, shall be zero.

open_connection_path: EPATH#: Connection path, as used in the Forward_Open request.

class icspacket.proto.cip.objects.connection.ConnectionConfigurationObject(connection: CIP_Connection, instance: int = 1)[source]#

Creates, configures, and controls CIP connections in a device (See CIP Vol 1, §5-48).

Attributes 13-17 and 20-22 (safety-only) are out of scope (See CIP Volume 5, CIP Safety). config_1_data/config_2_data/io_mapping each start with one or two leading count fields followed by vendor/format-specific bytes, and connection_name is a UTF-16 (STRING2) string; all four are exposed as raw bytes rather than a fixed schema.

connection_status: CIPAttribute[ConnectionStatus]#: General/extended connection status (attribute 1, See Table 5-48.5/5-48.6).

connection_flags: CIPAttribute[int]#: Originator/Target role and O->T/T->O real-time transfer format bits (attribute 2, See Table 5-48.7).

target_device_id: CIPAttribute[DeviceID]#: Identity of the connection’s target device, for locating its EDS (attribute 3).

cs_data_index_number: CIPAttribute[int]#: ControlNet Schedule Object connection_index value; ignored for target instances (attribute 4).

net_connection_parameters: CIPAttribute[NetConnectionParameters]#: Forward_Open-shaped connection parameters: timeout, transport, RPIs, and sizes (attribute 5).

connection_path: CIPAttribute[ConnectionPathAttribute]#: Forward_Open connection path, with its own byte-size prefix (attribute 6).

config_1_data: CIPAttribute[bytes]#: UINT config_data_size prefix followed by that many bytes of Config #1 data (attribute 7).

connection_name: CIPAttribute[bytes]#: USINT name_size, USINT reserved(=0) prefix followed by a UTF-16 (STRING2) connection name (attribute 8).

io_mapping: CIPAttribute[bytes]#: UINT format_number, UINT mapping_data_size prefix followed by that many bytes of mapping data; format matches class attribute 8 (attribute 9, See Table 5-48.4).

config_2_data: CIPAttribute[bytes]#: UINT config_data_size prefix followed by that many bytes of Config #2 data (attribute 10).

proxy_device_id: CIPAttribute[DeviceID]#: Identity of the device that owns this instance’s Forward_Open, e.g. for target/proxy setups (attribute 11).

connection_disable: CIPAttribute[int]#: 0 = enabled, 1 = disabled; required iff Open_Connection/Close_Connection are supported (attribute 12).

net_connection_parameters_selection: CIPAttribute[int]#: Selects whether net_connection_parameters (0) or large_net_connection_parameters (1) is active (attribute 18).

large_net_connection_parameters: CIPAttribute[LargeNetConnectionParameters]#: Large-format variant of net_connection_parameters, required iff attribute 18 is supported (attribute 19).

get_class_attribute(attribute: int) → bytes[source]#: Read a Connection Configuration class attribute from instance 0 (See §5-48.1, Table 5-48.1).

open_connection(request_data: bytes = b'') → bytes[source]#: Invoke Open_Connection (service 0x4C), opening this instance’s connection.

close_connection(request_data: bytes = b'') → bytes[source]#: Invoke Close_Connection (service 0x4D), closing this instance’s connection.

stop_connection(request_data: bytes = b'') → bytes[source]#: Invoke Stop_Connection (service 0x4E), stopping without deleting this instance’s connection.

change_start() → bytes[source]#: Invoke Change_Start (service 0x4F), beginning a class-wide configuration edit session.

get_status(request_data: bytes = b'') → bytes[source]#: Invoke Get_Status (service 0x50), reading status for multiple connections.

change_complete(request_data: bytes = b'') → bytes[source]#: Invoke Change_Complete (service 0x51), committing a class-wide configuration edit session.

audit_changes(request_data: bytes = b'') → bytes[source]#: Invoke Audit_Changes (service 0x52), inspecting pending configuration edits.

Wrapper and service helpers for the CIP Connection Manager Object (0x06).

class icspacket.proto.cip.objects.connmgr.ConnectionManagerObject(connection: CIP_Connection, instance: int = 1)[source]#

Access Connection Manager class and instance attributes.

Forward_Open, Forward_Close, and Unconnected_Send wire handling remains in CIP_Connection; these methods only delegate to that implementation.

get_class_attribute(attribute: int) → bytes[source]#: Read a Connection Manager class attribute from instance 0.

get_instance_attribute(attribute: int) → bytes[source]#: Read a Connection Manager instance attribute.

set_instance_attribute(attribute: int, value: bytes) → bytes[source]#: Write a Connection Manager instance attribute.

get_connection_data(request_data: bytes = b'') → bytes[source]#: Invoke Get_Connection_Data with caller-provided request bytes.

forward_open(request: ForwardOpenRequest | LargeForwardOpenRequest | None = None, **kwargs: Any) → ForwardOpenResponse[source]#: Delegate Forward_Open to the underlying connection.

large_forward_open(request: LargeForwardOpenRequest | None = None, **kwargs: Any) → ForwardOpenResponse[source]#: Delegate Large_Forward_Open to the underlying connection.

forward_close(request: ForwardCloseRequest | None = None, **kwargs: Any) → ForwardCloseResponse[source]#: Delegate Forward_Close to the underlying connection.

unconnected_send(message: MessageRouterRequest | bytes | None = None, route_path: EPATH | bytes | Iterable[Any] | None = None, **kwargs: Any) → MessageRouterResponse[source]#: Delegate Unconnected_Send to the underlying connection.