Adding protocols

The Connector contract, ProtocolDefinition and registering new communication adapters.

2 min read

Screens and variables never know about DBs, registers or NodeIds. Each protocol provides an adapter and a definition with its connection and binding fields. Studio generates the forms from that definition.

Adapter contract

class Connector(Protocol):
    def connect(self): ...
    def read(self, address: dict | str, kind: str): ...
    def write(self, address: dict | str, kind: str, value): ...
    def close(self): ...

kind is the logical SCADA type (bool, int, float, string). Transport encoding is decided by the binding.

Protocol definition

from abscada.protocol_definition import Field, ProtocolDefinition

def normalize(address, kind):
    return dict(address)

def check(address, kind, writable):
    if writable and address["area"] == "input":
        raise ValueError("Read-only area")

def describe(address, kind):
    return f"{address['area']}:{address['index']}"

class MyProtocol:
    definition = ProtocolDefinition(
        "My protocol",
        connection_fields=(
            Field("host", "IP / host", "127.0.0.1"),
            Field("port", "TCP port", 5000, 1, 65535),
        ),
        binding_fields=(
            Field("area", "Area", "holding", choices=(
                ("holding", "Holding", ()), ("input", "Input", ()))),
            Field("index", "Index", 0, 0, 65535),
        ),
        normalize=normalize, check_binding=check, describe=describe,
    )

    def __init__(self, config):
        self.config = config

    def connect(self):
        import my_network_lib  # import on connect, not when Studio opens
        self.client = my_network_lib.Client(self.config["host"], self.config["port"])

    def read(self, address, kind): ...
    def write(self, address, kind, value): ...
    def close(self): ...

Register

from abscada.connectors import register
register("my_protocol", MyProtocol)

Registration rejects duplicate IDs and adapters without a ProtocolDefinition.

How Runtime uses it

One worker per connection with its own client, write queue and monotonic clock.
The next cycle is scheduled from the start of the previous one; reads never overlap.
After a failure the client is closed and retried after 2 s.
Pending writes on a down connection are discarded.

Checklist

Module with definition and adapter.
Network dependency as an optional extra in pyproject.toml.
Schema tests and integration tests against an external server.
If the binding format changes, bump its version and add a tested migration.

Real example: Beckhoff ADS

src/abscada/ads.py is a complete adapter of about 250 lines you can use as a reference:

AdsClient speaks AMS/TCP over a socket: 32-byte header, Read, Write, ReadWrite and ReadState.
The ADS adapter fulfils the connect/read/write/close contract. It resolves symbols to handles (group 0xF003), caches them and renews them if the PLC program changes.
DEFINITION declares the connection fields (IP, AMS Net ID, ADS port) and binding fields (symbol, PLC type, STRING length) that Studio turns into forms.
ads_simulator.py is the pure-Python test server used by tests/test_ads.py and examples/beckhoff.

from abscada.connectors import create

plc = create({"protocol": "ads", "host": "192.168.1.50", "ams_net_id": "192.168.1.50.1.1", "ams_port": 851})
plc.connect()
print(plc.read({"symbol": "MAIN.rVelocidad", "encoding": "REAL"}, "float"))
plc.write({"symbol": "MAIN.bMarcha", "encoding": "BOOL"}, "bool", True)
plc.close()

OPC UA

Implemented in opcua.py (polling client) and opcua_server.py (runtime server). pki.py manages project certificates and secrets_store.py manages credentials. Connections resolve nsu=; writes query and cache the node type. Subscriptions, batch reads, node browsing and a server session limit remain pending. See OPC UA.