ethernetip2

A Python implementation of the EtherNet/IP protocol for industrial automation and control systems.

🎯 Overview

This library provides a clean, Pythonic interface to communicate with industrial devices using the EtherNet/IP (Ethernet Industrial Protocol) standard. It supports both explicit messaging (request/response) and implicit messaging (I/O connections) commonly used in industrial automation.

What is EtherNet/IP?

EtherNet/IP is an industrial network protocol that adapts the Common Industrial Protocol (CIP) to standard Ethernet. It's widely used in factory automation, process control, and building automation systems.

✨ Features

• 🔌 Explicit Messaging - Request/response communication for configuration and data access
• ⚡ Implicit I/O - High-speed cyclic data exchange for real-time control
• 🎛️ Adapter / Target Mode - Act as an EtherNet/IP device: serve Identity, respond to Forward Open, and run Class 1 cyclic I/O
• 🔍 Device Discovery - Network scanning and device identification
• 🧵 Thread-Safe - Built-in locking mechanisms for concurrent operations
• ⚙️ Configurable - Centralized configuration for timeouts, buffer sizes, and I/O parameters
• 📝 Well-Documented - Clear API with comprehensive docstrings

📦 Installation

Release Installation

pip install ethernetip2

From Source

git clone https://github.com/krisl/python-ethernetip.git
cd python-ethernetip
uv sync

Dependencies

• dpkt - Packet creation and parsing

pip install dpkt

🚀 Quick Start

Device Discovery

Scan your network for EtherNet/IP devices:

import ethernetip

# Create EtherNet/IP instance
enip = ethernetip.EtherNetIP()

# Create explicit connection
conn = enip.explicit_conn("192.168.1.100")

# Scan network
devices = conn.scanNetwork("192.168.1.255", timeout=5)

for device in devices:
    print(f"Found: {device.product_name.decode()}")

Read Device Identity

import ethernetip

# Connect to device
enip = ethernetip.EtherNetIP("192.168.1.100")
conn = enip.explicit_conn()

# Register session
conn.registerSession()

# Read device attributes (Identity Object, Instance 1)
for attr in range(1, 8):
    status, data = conn.getAttrSingle(
        ethernetip.CIP_OBJ_IDENTITY,
        instance=1,
        attribute=attr
    )
    if status == 0:
        print(f"Attribute {attr}: {data}")

I/O Connection (Real-time Data Exchange)

import ethernetip
import time

# Setup connection
enip = ethernetip.EtherNetIP("192.168.1.100")
conn = enip.explicit_conn()
conn.registerSession()

# Register I/O assemblies (1 byte input, 1 byte output)
input_data = enip.registerAssembly(
    ethernetip.EtherNetIP.ENIP_IO_TYPE_INPUT,
    size=1,
    instance=101,
    conn=conn
)
output_data = enip.registerAssembly(
    ethernetip.EtherNetIP.ENIP_IO_TYPE_OUTPUT,
    size=1,
    instance=100,
    conn=conn
)

# Start I/O communication
enip.startIO()
conn.sendFwdOpenReq(
    inputinst=101,
    outputinst=100,
    configinst=1
)
conn.produce()

# Control outputs
try:
    while True:
        # Set output bit 0
        output_data[0] = True
        time.sleep(0.5)

        # Read input bit 0
        print(f"Input bit 0: {input_data[0]}")

        output_data[0] = False
        time.sleep(0.5)
except KeyboardInterrupt:
    pass

# Cleanup
conn.stopProduce()
conn.sendFwdCloseReq(101, 100, 1)
enip.stopIO()

Adapter / Target Mode

Besides acting as a scanner (originator), the library can act as an EtherNet/IP adapter (target/device): it listens for scanner connections, serves the Identity object, responds to Forward Open / Forward Close, and runs Class 1 cyclic I/O.

import time
import ethernetip
from ethernetip.adapter import EtherNetIPAdapter, IO_TYPE_CONSUME, IO_TYPE_PRODUCE
from ethernetip.cip_objects import IdentityObject

identity = IdentityObject(vendor_id=0x1234, device_type=0x000C,
                          product_code=0x0001, product_name="my-device")
adapter = EtherNetIPAdapter(ip="0.0.0.0", identity=identity)

# O->T (consumed): bytes the scanner writes to us
adapter.registerAssembly(150, 4, IO_TYPE_CONSUME,
                         on_update=lambda data: print("got output:", data.hex()))
# T->O (produced): bytes we serve to the scanner
inp = adapter.registerAssembly(100, 4, IO_TYPE_PRODUCE)

adapter.start()
try:
    while True:
        time.sleep(1.0)
        inp.data[0] ^= 0x01      # toggle a bit the scanner will see
finally:
    adapter.stop()

Assembly direction follows the device-vendor convention:

• IO_TYPE_PRODUCE — produced by the adapter (Target → Originator "input").
• IO_TYPE_CONSUME — consumed by the adapter (Originator → Target "output").

A runnable version is in examples/adapter_example.py.

📚 API Overview

Main Classes

EtherNetIP

Main class for managing EtherNet/IP communications.

enip = ethernetip.EtherNetIP(ip="192.168.1.100")

Methods:

• explicit_conn(ipaddr) - Create an explicit messaging connection
• registerAssembly(iotype, size, inst, conn) - Register I/O assembly
• startIO() / stopIO() - Start/stop I/O communication
• listIDUDP(ipaddr, timeout) - Send ListIdentity request via UDP

EtherNetIPExpConnection

Explicit connection for request/response messaging and I/O.

Session Management:

• registerSession() - Establish session
• unregisterSession() - Close session

Device Information:

• listID() - Get device identity
• listServices() - Get available services
• scanNetwork(broadcast, timeout) - Scan for devices

Attribute Services:

• getAttrSingle(class, instance, attribute) - Read single attribute
• setAttrSingle(class, instance, attribute, data) - Write single attribute
• getAttrAll(class, instance) - Read all attributes

I/O Connection:

• sendFwdOpenReq(...) - Open I/O connection
• sendFwdCloseReq(...) - Close I/O connection
• produce() - Start producing output data
• stopProduce() - Stop producing output data

CIP Object IDs

Common CIP object class IDs are provided as constants:

ethernetip.CIP_OBJ_IDENTITY        # 0x01 - Device identity
ethernetip.CIP_OBJ_MESSAGE_ROUTER  # 0x02 - Message router
ethernetip.CIP_OBJ_ASSEMBLY        # 0x04 - Assembly object
ethernetip.CIP_OBJ_CONNECTION      # 0x05 - Connection object
ethernetip.CIP_OBJ_TCPIP           # 0xF5 - TCP/IP interface
ethernetip.CIP_OBJ_ETHERNET_LINK   # 0xF6 - Ethernet link
# ... and more

⚙️ Configuration

Global Configuration

Customize default behavior using the global config object:

from ethernetip import config

# Adjust timeouts
config.DEFAULT_TIMEOUT = 15  # seconds
config.IO_SOCKET_SELECT_TIMEOUT = 3  # seconds

# Adjust buffer sizes
config.RECV_BUFFER_SIZE = 2048  # bytes
config.RECV_BUFFER_SIZE_LARGE = 8192  # bytes

# Adjust I/O timing
config.UDP_IO_RPI_DEFAULT = 50  # ms (faster I/O updates)
config.UDP_IO_MIN_RPI = 5  # ms

Custom Configuration Class

For application-specific settings:

from ethernetip import EtherNetIPConfig

class ProductionConfig(EtherNetIPConfig):
    """Optimized for production environment"""
    DEFAULT_TIMEOUT = 30
    UDP_IO_RPI_DEFAULT = 100
    RECV_BUFFER_SIZE = 4096

# Apply configuration
import ethernetip
ethernetip.config = ProductionConfig()

Available Configuration Options

Parameter	Default	Description
DEFAULT_TIMEOUT	10	Default socket timeout (seconds)
IO_SOCKET_SELECT_TIMEOUT	2	I/O socket select timeout (seconds)
RECV_BUFFER_SIZE	1024	Standard receive buffer size (bytes)
RECV_BUFFER_SIZE_LARGE	4096	Large receive buffer size (bytes)
UDP_IO_RPI_DEFAULT	100	Default I/O update rate (milliseconds)
UDP_IO_MIN_RPI	8	Minimum I/O update rate (milliseconds)
FWD_OPEN_RPI_MULTIPLIER	1000	RPI conversion multiplier (ms to µs)
TICK_TIME	250	Timeout ticks for unconnected send

📖 Examples

Check out the examples/ directory for more detailed examples:

• example1.py - Complete workflow: discovery, configuration, I/O
• dlrscanner.py - Network scanning example

🛠️ Development

This project uses uv for environment management and nox for test/lint/docs.

uv sync        # install the project into the venv
uv run pytest  # run tests
make nox       # run full nox suite

Code Style

This project follows PEP 8 with some modifications (see pyproject.toml):

Building Documentation

make docs

🤝 Contributing

Contributions are welcome! Here's how you can help:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

Guidelines

• Follow the existing code style
• Add tests for new features
• Update documentation as needed
• Ensure all tests pass before submitting

📝 License

This project is licensed under the MIT License - see the LICENSE file for details.

🔗 Resources

• EtherNet/IP Specification: ODVA Website
• CIP Common Services: ODVA CIP Documentation

Version History

See CHANGELOG.md for complete version history.

---

Made with ❤️ for Industrial Automation

Original project by Sebastian Block — https://codeberg.org/paperwork/python-ethernetip