DWIS Python client

Python 3.10+ implementation of the supported C# Blackboard client API, using asyncua 2.1.x. DTOs and OPC UA identifiers follow C# client 2.2.52 and DWIS.API.DTO / DWIS.OPCUA.Schemas 2.2.33. Install the distribution named asyncua; opcua-asyncio is the upstream repository name.

Installation and verification

From this directory:

python -m pip install ".[test]" build
python -m pytest
python -m build

Offline tests include synthetic fixtures exported by the real C# DTO serializer. The workflow .github/workflows/python-publish.yml tests Python 3.10, 3.12 and 3.14, validates wheel/source-distribution metadata and retains reports/packages for 14 days. Main-branch pushes, pull requests and manual dispatch only verify/build. Pushing a python-v<version> tag publishes the tested Python 3.12 distributions to PyPI after every matrix job passes. The tag must match the package version in pyproject.toml. Publication uses PyPI Trusted Publishing for D-WIS/DWIS-Blackboard-Clients, workflow python-publish.yml, GitHub environment pypi; no API key is stored. For example, release version 0.1.0 with git tag python-v0.1.0 then git push origin python-v0.1.0. Each subsequent release needs a new package version and matching tag.

Opt in to local Blackboard tests:

python -m pytest --blackboard-url opc.tcp://localhost:48030
python -m pytest --blackboard-url opc.tcp://localhost:48030 --blackboard-writes

The second command also creates uniquely named PythonClientInterop<UUID> manifests and writes only their variables. These test manifests and server-side registrations remain on the Blackboard because the C# API does not implement deletion/unregistration. Use a disposable Blackboard for repeated write testing. Client sessions and monitored items are closed after each test. No server is contacted by the default test command.

Async client

import asyncio
from dwis_client import DwisClientConfiguration, OpcUaDwisClient

async def main():
    config = DwisClientConfiguration(server_address="opc.tcp://localhost:48030")
    async with OpcUaDwisClient(config) as client:
        query = await client.get_query_result("SELECT ?s WHERE { ?s ?p ?o } LIMIT 1")
        for row in query.results:
            print([(item.namespace, item.identifier) if item else None for item in row.items])

asyncio.run(main())

Construction is lazy; async with or await client.connect() establishes a connection. Always call close() or use the context manager. BLACKBOARD_URL overrides the configured address. Configuration accepts both Python field names and the C# ServerAddress, UseWebAPI, and WebAPIUrl spelling; it saves using C# spelling. The Web API fields are preserved without claiming an HTTP implementation.

The default transport uses an anonymous OPC UA session with SecurityPolicy.None. To configure credentials, certificates or another security policy, supply an adapter factory returning UaClientAdapter(endpoint, client_factory=...). The inner factory returns your configured asyncua.Client. Default requests wait at most 15 seconds; the default client advertises a 16 MiB transport-message ceiling. Custom factories control their own transport settings.

Synchronous client

from dwis_client import SyncOpcUaDwisClient

with SyncOpcUaDwisClient() as client:
    result = client.get_query_result("SELECT ?s WHERE { ?s ?p ?o } LIMIT 1")
    print(result.variables_header)

The facade owns one persistent event loop on a background thread, preserving the session across calls. It replaces the scaffold's asyncio.run()-per-operation helper. Do not call synchronous facade methods from its notification callback thread; use the async API there. Closing the facade also stops its loop thread.

C# API correspondence

read_any_variables(nodes) returns a list or None on failure. To use the C# out-parameter pattern, pass an output list: read_any_variables(nodes, out_values) returns bool and preserves the list on failure. Empty input returns None/false. Namespace methods return an index/URI or None; namespace indices are unsigned 0–65535. DwisNodeId.parse() preserves numeric, string, GUID and opaque identifier kinds, including strings containing semicolons or equals signs.

Typed contracts and subscriptions

Manifest inputs, injection mappings, query rows/diffs, resources and nested acquisition results/diffs are dataclasses. Use from_json() / to_json() or from_dict() / to_dict(). C# wire names, including ID/URI acronyms and ITemResults, are preserved. AcquisitionFile serializes as a bare array. Query row cells can be None for unbound SPARQL variables. The unfinished port's generic DTO fields have been replaced; ProvidedVariable now describes manifest input, while InjectionMapping describes injected IDs.

from dwis_client import SubscriptionRequest

await client.subscribe("group context", on_changes, [SubscriptionRequest(node_id, "per-node context")])

on_changes(context, changes) receives values, source/server timestamps and per-node user data. It can be synchronous or async. Callbacks execute in order on the client event loop; exceptions are logged without stopping subsequent callbacks. Keep synchronous callbacks short. Data subscriptions report an initial read as well as later notifications.

Query registration returns its initial diff separately and monitors the diff's QueryResultID variable rather than its registration object. Null initial variable values are ignored. Acquisition subscriptions monitor Server events filtered by SourceNode and decode the Message field, wrapping a single AcquisitionFileDiff into an AcquisitionDiff.

try_reconnect() restores data/query/acquisition monitored items against the same server. It assumes the previously returned registrations still exist; register again after a server restart that drops them. Changing the configured endpoint discards cached mappings and subscription bindings to prevent cross-server ID reuse. Inject and register again on the new Blackboard.

The default transport uses sixty-second request and health-check deadlines and subscription keepalives, with five seconds of extra allowance for the outer operation deadline. Set UaClientAdapter(..., request_timeout=...) to adjust request and health-check deadlines; a custom client factory can configure OPC UA security. Semantic import processing on a busy Blackboard can delay query and acquisition notifications beyond ordinary value writes.

Writes inspect every OPC UA status and return false for rejected writes. Unknown provided-variable IDs prevent the entire request from being sent. Multi-node writes are not transactional: some nodes can change even if another write is rejected. Source timestamps must be timezone-aware datetimes; no server timestamp is fabricated. ua.Variant values may be supplied when a particular OPC UA numeric width is needed.

High-level service/transport failures raise DwisClientError; malformed DTO payloads raise ValueError. Cancellation is propagated, and failed monitored-item creation is cleaned up.

Acquired signals

from dwis_client import AcquiredSignals

signals = await AcquiredSignals.create_with_subscription(acquisition_file, client)
for signal in signals["pressure"]:
    signal.callback = lambda value, timestamp: print(value, timestamp)

The helper selects the lowest-index criterion, deduplicates signal IDs and subscribes to newly discovered signals. Call update_from_acquisition(file, client, convert_to_si=True) to add new results. The default keeps canonical SI values. With convert_to_si=False, it looks up unit factors and converts displayed floating-point values as (value - A) / B. Conversion never writes back to the server. Source time takes precedence over server time. History is bounded by the constructor's max_log_capacity; zero disables logging. get_logged_data(start) returns a snapshot of retained samples at or after the requested time.