Añadir protocolos

Contrato Connector, ProtocolDefinition y registro de nuevos adaptadores de comunicación.

3 min de lectura

Las pantallas y variables nunca conocen DB, registros ni NodeIds. Cada protocolo aporta un adaptador y una definición con los campos de conexión y enlace. Studio genera los formularios a partir de esa definición.

Contrato del adaptador

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 es el tipo lógico SCADA (bool, int, float, string). La codificación de transporte la decide el enlace.

Definición del protocolo

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("Área de solo lectura")

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

class MyProtocol:
    definition = ProtocolDefinition(
        "Mi protocolo",
        connection_fields=(
            Field("host", "IP / host", "127.0.0.1"),
            Field("port", "Puerto TCP", 5000, 1, 65535),
        ),
        binding_fields=(
            Field("area", "Área", "holding", choices=(
                ("holding", "Holding", ()), ("input", "Input", ()))),
            Field("index", "Índice", 0, 0, 65535),
        ),
        normalize=normalize, check_binding=check, describe=describe,
    )

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

    def connect(self):
        import my_network_lib  # importar al conectar, no al abrir Studio
        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): ...

Registrar

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

El registro rechaza IDs duplicados y adaptadores sin ProtocolDefinition.

Cómo lo usa el Runtime

Un trabajador por conexión con su propio cliente, cola de escritura y reloj monotónico.
El siguiente ciclo se calcula desde el comienzo del anterior; no hay lecturas solapadas.
Tras un fallo se cierra el cliente y se reintenta a los 2 s.
Las escrituras pendientes de una conexión caída se descartan.

Lista de comprobación

Módulo con definición y adaptador.
Dependencia de red como extra opcional en pyproject.toml.
Pruebas de esquema y pruebas de integración contra un servidor externo.
Si cambia el formato de enlace, sube su version y añade una migración probada.

Ejemplo real: Beckhoff ADS

src/abscada/ads.py es un adaptador completo de unas 250 líneas que sirve de referencia:

AdsClient habla AMS/TCP por socket: cabecera de 32 bytes, Read, Write, ReadWrite y ReadState.
El adaptador ADS cumple el contrato connect/read/write/close. Resuelve símbolos a handles (grupo 0xF003), los guarda en caché y los renueva si el PLC cambia de programa.
DEFINITION declara los campos de conexión (IP, AMS Net ID, puerto ADS) y de enlace (símbolo, tipo PLC, longitud de STRING), con los que Studio genera sus formularios.
ads_simulator.py es el servidor de pruebas en Python puro que usan tests/test_ads.py y 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

Implementado en opcua.py (cliente con polling) y opcua_server.py (servidor del runtime). pki.py gestiona certificados del proyecto y secrets_store.py sus credenciales. Se resuelve nsu= al conectar; las escrituras consultan y conservan en caché el tipo del nodo. Las suscripciones, lecturas por lotes, exploración de nodos y límite de sesiones siguen pendientes. Consulta OPC UA.