server._server.Server

class documentation

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

Constructor: Server(config_path)

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.
Async Method get_enum Return OPC UA Node representation of given enum type.
Method load_config Load configuration from given path.
Async Method start Start the internal OPC UA server.
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. Read-only property.
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 _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 _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_path: str | Path | 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.

async def get_enum(self, enum_type: type[IntEnum], *, option_set: bool = False) -> ua.NodeId:

Return OPC UA Node representation of given enum type.

def load_config(self, path: str | Path):

Load configuration from given path.

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_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. Read-only property.

@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 _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

_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