declarative-opcua-server

declarative-opcua-server creates an opinionated synchronous OPC UA server from three flat dictionaries of annotated Python functions. It is intended for small adapters that need a predictable Status, Parameters, and Methods address space without building nodes manually.

Installation

python -m pip install --upgrade pip
python -m pip install declarative-opcua-server

Python 3.8.3 and later are supported. The distribution selects a compatible asyncua release for the active Python version and caps the final cryptography line that supports Python 3.8. Upgrade the old pip bundled with Python 3.8.3 before installing.

Example

import time
import typing

import declarative_opcua_server

state = {"height": 10.0}


def read_height() -> float:
    return state["height"]


def write_height(height: float) -> None:
    state["height"] = height


def load_program(program: str) -> str:
    return "Loaded " + program


def list_programs() -> typing.List[str]:
    return ["Main.urp", "Production/PickPart.urp"]


server = declarative_opcua_server.create_server(
    status_interface={"ActualHeight": read_height},
    parameter_interface={"TargetHeight": write_height},
    method_interface={"LoadProgram": load_program, "ListPrograms": list_programs},
    endpoint="opc.tcp://127.0.0.1:4840/",
    namespace="urn:example:robot",
    root_object="Robot",
)

with server:
    while True:
        time.sleep(1.0)

create_server() returns a plain, unstarted asyncua.sync.Server. Callers retain the normal start(), stop(), and context-manager lifecycle.

Refreshing methods

refresh_method(provider) can be used directly as one value in the method dictionary. The provider returns the complete dynamic portion of the interface. Invoking the exposed refresh method calls it again, validates the result, adds and removes nodes, and returns the sorted dynamic method names:

import typing

dynamic_methods: typing.Dict[str, typing.Callable[..., typing.Any]] = {}


def run_main() -> None:
    print("run Main.urp")


def provide_methods() -> typing.Mapping[str, typing.Callable[..., typing.Any]]:
    return dynamic_methods


dynamic_methods["StartProgram_Main"] = run_main
server = declarative_opcua_server.create_server(
    status_interface={},
    parameter_interface={},
    method_interface={"RefreshMethods": declarative_opcua_server.refresh_method(provide_methods)},
)

Method callbacks run on a worker thread, so RefreshMethods() completes only after the replacement is visible to subsequent OPC UA browsing. A provider mapping is the complete dynamic interface, not a patch. Fixed methods remain in the original dictionary. If a refreshed callback keeps the same name and signature, its existing node and NodeId are retained while the callable behind it is updated.

update_method_interface() remains available for callers that already own their own refresh lifecycle. It applies one validated complete replacement for the whole Methods folder.

Address space

The example creates:

Objects/
    Robot/
        Status/
            ActualHeight
        Parameters/
            TargetHeight
        Methods/
            LoadProgram(program) -> String
            ListPrograms() -> String[]

The selected dictionary defines each callable's role:

• A status getter accepts no required arguments and declares a return type. It becomes a polled read-only variable.
• A parameter setter accepts one required annotated argument and returns no value. It becomes a writable variable whose accepted writes invoke the setter.
• A method exposes required annotated arguments as OPC UA inputs and an annotated return as an optional output.

Defaulted status, parameter, and method arguments are treated as bound application configuration rather than OPC UA inputs. This makes configured callables useful without adding wrapper functions.

Supported annotations

T must be one of the supported scalar annotations. Unsupported or unresolved signatures fail during server creation.

Scope and security

The package intentionally does not provide arbitrary folders, custom node classes, stable NodeId configuration, events, application schemas, or protocol adapters. Applications requiring a general OPC UA framework should use asyncua directly.

The current server defaults to anonymous access and NoSecurity. It is suitable for controlled development and isolated industrial networks; certificate and authentication configuration should be added before use on an untrusted network.

Development

From this package directory:

python -m pip install -e ".[dev]"
python -m pytest tests
python -m mypy
python -m build
python -m twine check dist/*

Tests use a real asyncua client to verify browsing, status polling, parameter writes, typed method calls, and lifecycle behavior.

Release history is recorded in the changelog. The gateway integration is validated independently in ur_dashboard_to_opcua_gateway.

License

This project is licensed under the MIT License. See LICENSE.