server._server.Server

class documentation

class Server(AsyncTaskMixin, LifecycleMixin, ReprStrMixin, BaseServer[UaObject]):

Constructor: Server(config)

View In Hierarchy

Undocumented

Async Method __aenter__ Enter the asynchronous context manager; no setup required.
Async Method __aexit__ Exit the asynchronous context manager and shut down the server.
Method __del__ Undocumented
Method __init__ Undocumented
Async Method add_machine Add given machine to the server. Will initialize the machine if not yet initialized.
Method config.setter Set the server-wide configuration. Only allowed before server initialization.
Async Method ensure_enum Return OPC UA Node representation of given enum type. Creates OPC UA representation automatically.
Method register_enum Use an existing enumeration OPC UA data type for enum_type instead of creating one.
Async Method start Start the internal OPC UA server.
Async Method ua_get_dictionary_entry Get or add a new OPC UA dictionary entry and return the corresponding OPC UA Node.
Async Method ua_import_nodeset Import the OPC UA XML nodeset specified by the given uri and return the namespace index.
Async Method ua_import_xml Import XML nodeset from given path. Returns the namespace index on success.
Method ua_register_nodeset_xml Register the given nodeset XML file living in package. Overrides existing nodesets based on URI.
Property access_control Undocumented
Property config Return the server-wide configuration.
Property logger Return the logger instance of the server. Read-only property.
Property machines Read-only. Return all machines added to the server.
Property running Whether the server is running.
Property state Read-only. Return the state of the server.
Property stopped Whether the server is stopped.
Property ua_address_space Read-only. Return the internal OPC UA address space instance.
Property ua_server Read-only. Return the internal OPC UA server instance.
Async Method _init Undocumented
Async Method _life_cycle Undocumented
Async Method _register_server Undocumented
Method _repr_items Undocumented
Async Method _setup_ua_server Undocumented
Async Method _shutdown Undocumented
Async Method _ua_add_empty_namespace Add a new empty OPC UA namespace with the given uri to the server.
Async Method _update_namespace_map Undocumented
Async Method _watchdog_loop Monitor how busy the event loop is and warn if it is probably too busy.
Instance Variable _access_control Undocumented
Instance Variable _config Undocumented
Instance Variable _dictionary_entries Maps Semantic ID (=Browse name) to live OPC UA node IDs of the corresponding Dictionary entries.
Instance Variable _enum_lock Lock used when creating new OPC UA enums.
Instance Variable _logger Undocumented
Instance Variable _machines Undocumented
Instance Variable _state Undocumented
Instance Variable _ua_address_space Undocumented
Instance Variable _ua_enums Maps created custom enums to OPC UA type nodes.
Instance Variable _ua_machines Undocumented
Instance Variable _ua_nodesets_xml Maps nodeset XML URIs to information how to load the corresponding XML file.
Instance Variable _ua_server Undocumented
async def __aenter__(self) -> Self:

Enter the asynchronous context manager; no setup required.

async def __aexit__(self, exc_type, exc, tb):

Exit the asynchronous context manager and shut down the server.

def __del__(self):

Undocumented

def __init__(self, *, config: str | Path | ServerConfiguration | None = None):

Undocumented

async def add_machine(self, machine: _MachineType) -> _MachineType:

Add given machine to the server. Will initialize the machine if not yet initialized.

def config(self, config: ServerConfiguration):

Set the server-wide configuration. Only allowed before server initialization.

async def ensure_enum(self, enum_type: type[IntEnum]) -> ua.NodeId:

Return OPC UA Node representation of given enum type. Creates OPC UA representation automatically.

def register_enum(self, enum_type: type[IntEnum], *, ua_data_type: ua.NodeId, overwrite: bool = False):

Use an existing enumeration OPC UA data type for enum_type instead of creating one.

async def start(self, blocking: bool = True):

Start the internal OPC UA server.

Server must be initialized beforehand, and must be in state STOPPED.

:param blocking If True, block until server shutdown or KeyboardInterrupt is caught

async def ua_get_dictionary_entry(self, dictionary_entry: DictionaryEntry) -> Node:

Get or add a new OPC UA dictionary entry and return the corresponding OPC UA Node.

async def ua_import_nodeset(self, uri: str | type[NamespaceProvider]) -> int:

Import the OPC UA XML nodeset specified by the given uri and return the namespace index.

Automatically loads all required nodesets beforehand.

Raises
KeyErrorIf given URI is unknown.
async def ua_import_xml(self, *, path: Path | None = None, xml_string: str | None = None) -> int:

Import XML nodeset from given path. Returns the namespace index on success.

def ua_register_nodeset_xml(self, *, namespace: type[NamespaceProvider], package: str):

Register the given nodeset XML file living in package. Overrides existing nodesets based on URI.

@property
access_control: AccessControl =

Undocumented

@property
config: ServerConfiguration =

Return the server-wide configuration.

@property
logger: structlog.stdlib.BoundLogger =

Return the logger instance of the server. Read-only property.

@property
machines: list[BaseMachine] =

Read-only. Return all machines added to the server.

@property
running: bool =

Whether the server is running.

@property
state: ServerState =

Read-only. Return the state of the server.

@property
stopped: bool =

Whether the server is stopped.

@property
ua_address_space: AddressSpace =

Read-only. Return the internal OPC UA address space instance.

@property
ua_server: UaServer =

Read-only. Return the internal OPC UA server instance.

async def _init(self):

Undocumented

@lifecycle
async def _life_cycle(self) -> AsyncGenerator[None]:

Undocumented

async def _register_server(self, retry_seconds: float = 10.0):

Undocumented

@override
def _repr_items(self) -> Iterator[tuple[str, object]]:

Undocumented

async def _setup_ua_server(self):

Undocumented

async def _shutdown(self):

Undocumented

async def _ua_add_empty_namespace(self, uri: str) -> ua.Int16:

Add a new empty OPC UA namespace with the given uri to the server.

Returns
ua.Int16The namespace index for the newly created namespace.
async def _update_namespace_map(self):

Undocumented

async def _watchdog_loop(self, *, interval: float = 0.2, warning_threshold: float = 0.1):

Monitor how busy the event loop is and warn if it is probably too busy.

_access_control: AccessControl =

Undocumented

_config: ServerConfiguration =

Undocumented

_dictionary_entries: dict[str, ua.NodeId] =

Maps Semantic ID (=Browse name) to live OPC UA node IDs of the corresponding Dictionary entries.

_enum_lock: asyncio.Lock =

Lock used when creating new OPC UA enums.

_logger =

Undocumented

_machines: list[BaseMachine] =

Undocumented

_state =

Undocumented

_ua_address_space: AddressSpace =

Undocumented

_ua_enums: dict[type[IntEnum], ua.NodeId] =

Maps created custom enums to OPC UA type nodes.

_ua_machines: Node =

Undocumented

_ua_nodesets_xml: dict[str, NodesetReference] =

Maps nodeset XML URIs to information how to load the corresponding XML file.

_ua_server: UaServer =

Undocumented