Skip to content

EtherNet/IP¤

EtherNet/IP config files declare an Allen-Bradley PLC endpoint, an optional local backplane route, and the scalar tags Nominal reads or writes. The client lives under instro.ethernetip and is provided by the instro-ethernetip package, installable via the instro[ethernetip] extra (also included in instro[all]).

from instro.ethernetip import EtherNetIPDevice

connection = {
    "host": "192.168.1.10",
    "port": 44818,
    "route_path": {"hops": [{"type": "backplane", "slot": 0}]},
}

plc = EtherNetIPDevice("compactlogix.json", connection=connection, autostart=True)
plc.read_tag("line_speed")
plc.write_tag("line_speed", 1200.0)
plc.close()

Current scope¤

Area Supported today
Tested PLC Allen-Bradley CompactLogix 5332E 1769-L32E
Transport EtherNet/IP explicit messaging over TCP
Route paths Direct connection or local backplane slot hops only
Polling Automatic batched reads for poll: true scalar tags
Streaming values Boolean and numeric scalar tags
Manual native operations Single-tag reads, batched reads, and writes
Unsigned integer validation usint, uint, udint, and ulint are implemented, but not validated
Tag discovery Not supported
UDTs Not supported in the config-driven API
Arrays Not supported in the config-driven API

JSON config reference¤

Connection¤

Provide the connection in the config or pass it to the EtherNetIPDevice constructor. The constructor parameter takes precedence, so one tag map can target multiple environments.

plc = EtherNetIPDevice(
    "compactlogix.json",
    connection={
        "host": "192.168.1.10",
        "port": 44818,
        "route_path": {"hops": [{"type": "backplane", "slot": 0}]},
    },
)

{
    "connection": {
        "host": "192.168.1.10",
        "port": 44818,
        "route_path": {
            "hops": [
                {"type": "backplane", "slot": 0}
            ]
        }
    }
}
Field Type Default Description
host string required PLC IP address or hostname
port int 44818 EtherNet/IP TCP port
route_path object null Optional local backplane route path

Route paths accept local backplane hops only:

Field Type Default Description
hops list [] Ordered local backplane hops
hops[].type string required Must be "backplane"
hops[].slot int required Backplane slot number, 0-255

Network hops to another PLC, remote chassis, or IP address are not supported. The schema accepts multiple local backplane hops, but current testing has covered one backplane hop.

Timing¤

The timing section controls background polling:

{
    "timing": {
        "poll_interval": 1.0
    }
}
Field Type Default Description
poll_interval float required Seconds between polling cycles (0.01-10.0)

When polling is running, EtherNetIPDevice reads every poll: true tag in one batched native request at the configured interval. A per-tag failure skips that tag for the current measurement; successful values from the same batch are still published.

Tags¤

Each tag entry defines one named PLC tag:

Field Type Default Description
alias string required Alias used in read_tag() and write_tag()
tag_name string required PLC tag name
description string null Optional description
data_type string required Expected PLC scalar type
poll bool true Include in background polling
write_min number null Minimum allowed write value for numeric tags
write_max number null Maximum allowed write value for numeric tags

Supported data_type values:

Data type Streamable Notes
bool Yes Published as numeric 0 or 1
sint Yes 8-bit signed integer
int Yes 16-bit signed integer
dint Yes 32-bit signed integer
lint Yes 64-bit signed integer
usint Yes 8-bit unsigned integer. Implemented, but not validated
uint Yes 16-bit unsigned integer. Implemented, but not validated
udint Yes 32-bit unsigned integer. Implemented, but not validated
ulint Yes 64-bit unsigned integer. Implemented, but not validated
real Yes 32-bit floating point
lreal Yes 64-bit floating point

PLC string tags are not part of the Python EtherNet/IP API.

Native batched reads¤

instro.ethernetip._ethernetip.EtherNetIpSession.read_tags() reads several PLC tags in one native request and preserves input order:

from instro.ethernetip._ethernetip import EtherNetIpBatchError, EtherNetIpSession

with EtherNetIpSession("192.168.1.10:44818", route_path_slots=[0]) as session:
    for name, result in session.read_tags(["MotorRunning", "LineSpeed"]):
        if isinstance(result, EtherNetIpBatchError):
            print(name, result)
            continue
        print(name, result.kind, result.value)

The call raises EtherNetIpError when the whole batch cannot be dispatched or parsed. Individual tag failures are returned as typed EtherNetIpBatchError instances, including TagNotFoundError, DataTypeMismatchError, NetworkBatchError, CipError, TagPathError, SerializationError, BatchTimeoutError, and OtherBatchError.

Write limits¤

Reject writes outside the configured range before they reach the PLC:

{
    "alias": "line_speed",
    "tag_name": "LineSpeed",
    "data_type": "real",
    "write_min": 0.0,
    "write_max": 2500.0
}

plc.write_tag("line_speed", 1200.0)  # OK
plc.write_tag("line_speed", 9999.0)  # raises ValueError: above write_max (2500.0)

EtherNetIPDevice checks limits before sending the write to the PLC.

Validation rules¤

  • protocol must be "ethernetip".
  • Tag aliases must be unique.
  • Every tag must declare data_type.
  • write_min and write_max are only valid for numeric tags.
  • write_min must be less than or equal to write_max.
  • Integer write limits must fit in the configured PLC integer type.
  • Route path hops must use type: "backplane" with slot from 0 to 255.
  • Tag discovery, UDTs, and arrays are not supported.

API reference¤

EtherNetIPDevice¤

Bases: Instrument

EtherNet/IP device with config-driven tag access.

background_interval instance-attribute ¤

background_interval = poll_interval

address property ¤

address: str

EtherNet/IP endpoint address.

name instance-attribute ¤

name = name

legacy_naming instance-attribute ¤

legacy_naming = legacy_naming

publishers instance-attribute ¤

publishers = publishers or []

default_tags instance-attribute ¤

default_tags: dict[str, str] = {}

start ¤

start() -> None

Start background polling for poll-enabled EtherNet/IP tags.

open ¤

open() -> None

Open the EtherNet/IP session.

close ¤

close() -> None

Close the EtherNet/IP session and stop background polling.

read_tag ¤

read_tag(alias: str, **kwargs) -> Measurement

Read one configured tag by alias and publish the result.

read ¤

read(alias: str, **kwargs) -> Measurement

Read one configured tag by alias and publish the result.

write_tag ¤

write_tag(alias: str, value: bool | int | float, **kwargs) -> Command

Write one configured tag by alias and publish the command.

write ¤

write(alias: str, value: bool | int | float, **kwargs) -> Command

Write one configured tag by alias and publish the command.

add_publisher ¤

add_publisher(publisher: Publisher)

Register a publisher to receive this instrument's Measurement/Command data.

publish ¤

publish(data: Measurement | Command, **kwargs)

Fan data out to every configured publisher; kwargs pass through.

add_background_daemon_function ¤

add_background_daemon_function(method: Callable, *args, **kwargs)

Append method to the daemon's call list. Use define_background_daemon to replace instead.

stop ¤

stop()

Signal the background daemon to stop and join it. No-op if not running.

get_channel ¤

get_channel(
    channel_name: str,
    length: int = 1,
    wait_for_new_samples: bool = False,
    timeout: float = 10.0,
) -> Measurement

Return the most recent length samples for channel_name from the in-memory buffer.

If the channel does not exist yet, or no sample is available, the code will always block until timeout expires.

Parameters:

  • channel_name ¤

    (str) –

    Name of the channel to retrieve.

  • length ¤

    (int, default: 1 ) –

    Number of trailing samples to return.

  • wait_for_new_samples ¤

    (bool, default: False ) –

    Block until at least length new values arrive.

  • timeout ¤

    (float, default: 10.0 ) –

    Seconds to wait when insufficient data exists or wait_for_new_samples=True.

Raises:

  • RuntimeError –

    No background buffer; start() was not called.

  • ChannelNotFoundError –

    channel had no values and no data appeared before timeout. wait_for_new_samples=True and channel did not appear within timeout.

  • ChannelValueTimeoutError –

    wait_for_new_samples=True and values did not arrive within timeout.

get_single_channel_value ¤

get_single_channel_value(channel_name: str) -> float | None

Return the most recent sample for channel_name from the in-memory buffer.

This will not wait, if data is not available then None is returned.

Parameters:

  • channel_name ¤

    (str) –

    Name of the channel to retrieve.

Raises:

  • RuntimeError –

    No background buffer; start() was not called.

define_background_daemon ¤

define_background_daemon(method: Callable, *args, **kwargs)

Replace all daemon functions with a single method (called with the given args).

Configuration types¤

Configuration types for the EtherNet/IP instrument.

EtherNetIPConfig ¤

Bases: EtherNetIPBaseModel

Complete EtherNet/IP instrument configuration.

version class-attribute instance-attribute ¤

version: int = 1

protocol class-attribute instance-attribute ¤

protocol: str = 'ethernetip'

device instance-attribute ¤

device: DeviceInfo

timing class-attribute instance-attribute ¤

timing: TimingConfig | None = None

connection class-attribute instance-attribute ¤

connection: EtherNetIPConnectionInfo | None = None

tags class-attribute instance-attribute ¤

tags: list[TagDef] = Field(default_factory=list)

polled_tags property ¤

polled_tags: list[TagDef]

Tags included in background polling.

model_config class-attribute instance-attribute ¤

model_config = ConfigDict(extra='forbid')

model_post_init ¤

model_post_init(__context) -> None

Validate the top-level config.

from_json classmethod ¤

from_json(path: Path | str) -> 'EtherNetIPConfig'

Load configuration from a JSON file.

get_tag ¤

get_tag(alias: str) -> TagDef

Get a tag definition by local alias.

TimingConfig ¤

Bases: EtherNetIPBaseModel

Timing configuration for EtherNet/IP polling.

poll_interval class-attribute instance-attribute ¤

poll_interval: float = Field(
    ge=0.01, le=10.0, description="Polling interval in seconds"
)

model_config class-attribute instance-attribute ¤

model_config = ConfigDict(extra='forbid')

EtherNetIPConnectionInfo ¤

Bases: EtherNetIPBaseModel

EtherNet/IP TCP endpoint configuration.

host instance-attribute ¤

host: str

port class-attribute instance-attribute ¤

port: int = Field(default=44818, ge=1, le=65535)

route_path class-attribute instance-attribute ¤

route_path: EtherNetIPRoutePath | None = None

address property ¤

address: str

Endpoint string accepted by the native EtherNet/IP session.

model_config class-attribute instance-attribute ¤

model_config = ConfigDict(extra='forbid')

EtherNetIPRoutePath ¤

Bases: EtherNetIPBaseModel

Ordered EtherNet/IP route path supported by the native backend.

hops class-attribute instance-attribute ¤

hops: list[EtherNetIPBackplaneHop] = Field(
    default_factory=list,
    description="Ordered route hops. Only backplane hops are supported by the native backend today.",
)

model_config class-attribute instance-attribute ¤

model_config = ConfigDict(extra='forbid')

EtherNetIPBackplaneHop ¤

Bases: EtherNetIPBaseModel

One supported EtherNet/IP backplane route hop.

CIP port 1 is implied for backplane hops; the config only exposes the slot.

type class-attribute instance-attribute ¤

type: Literal['backplane'] = Field(description='Route hop kind.')

slot instance-attribute ¤

slot: RouteSlot

model_config class-attribute instance-attribute ¤

model_config = ConfigDict(extra='forbid')

TagDef ¤

Bases: EtherNetIPBaseModel

Definition of one EtherNet/IP tag exposed as an instrument channel.

alias class-attribute instance-attribute ¤

alias: str = Field(
    description="Unique local Nominal channel alias for this tag"
)

tag_name class-attribute instance-attribute ¤

tag_name: str = Field(description='PLC tag name to read/write')

description class-attribute instance-attribute ¤

description: str | None = None

data_type class-attribute instance-attribute ¤

data_type: DataType = Field(
    description="PLC scalar data type for this tag"
)

poll class-attribute instance-attribute ¤

poll: bool = Field(
    default=True, description="Include this tag in background polling"
)

write_min class-attribute instance-attribute ¤

write_min: float | int | None = None

write_max class-attribute instance-attribute ¤

write_max: float | int | None = None

expected_plc_kind_name property ¤

expected_plc_kind_name: str

Return the native PlcKind member name expected for this tag.

model_config class-attribute instance-attribute ¤

model_config = ConfigDict(extra='forbid')

validate_write_value ¤

validate_write_value(value: WriteValue) -> None

Validate a user-provided write value against this tag definition.

validate_integer_raw_value ¤

validate_integer_raw_value(
    raw_value: WriteValue, data_type: str | None = None
) -> int

Validate and coerce a raw integer value for this tag's PLC type.

validate_streamable_read ¤

validate_streamable_read(actual_kind: object) -> None

Validate that this tag can be published as measurement data.