Add OPC UA interface, device, init, and states

lewis_emulators/opcua/__init__.py

@@ -0,0 +1,5 @@
+from ..lewis_versions import LEWIS_LATEST
+from .device import SimulatedFinsPLC
+
+framework_version = LEWIS_LATEST
+__all__ = ["SimulatedFinsPLC"]

lewis_emulators/opcua/device.py

@@ -0,0 +1,24 @@
+from collections import OrderedDict
+
+from lewis.devices import StateMachineDevice
+from lewis.core.statemachine import State
+
+from .states import DefaultState
+
+class SimulatedOPCUA(StateMachineDevice):
+    """Class representing a simulated OPC UA PLC.
+    """
+
+    #Need a dictionary ? of some sort to store PV values and their
+    #corresponding address/name on the OPCUA server. Also need
+    #IP address and other server info, and possibly library to start
+    #the server... 
+
+    def _get_state_handlers(self):
+        return super()._get_state_handlers()
+
+    def _get_initial_state(self):
+        return super()._get_initial_state()
+
+    def _get_transition_handlers(self):
+        return super()._get_transition_handlers()

lewis_emulators/opcua/interfaces/opcua_interface.py

@@ -0,0 +1,54 @@
+from lewis.adapters import opcua
+
+"""
+This is a fake OPC UA server interface which creates three values:
+a temperature float, a name string, and a status boolean.
+"""
+
+class OPCUAInterface:
+    """Interface for my device."""
+
+    def __init__(self):
+        self._temperature = 25.0
+        self._vacuum_status = False
+
+    @property
+    def temperature(self):
+        """Current temperature (read/write)"""
+        return self._temperature
+
+    @temperature.setter
+    def set_temperature(self, value):
+        """Set the temperature."""
+        self.temperature = float(value)
+
+    @property
+    def vacuum_status(self):
+        """Vacuum status (read-only)"""
+        return self._vacuum_status
+
+    def set_vacuum(self, status: bool):
+        """Simulate a change in vacuum status"""
+        self._vacuum_status = bool(status)
+
+    @property
+    def name(self):
+        """Device name (read-only)"""
+        return "TestOPCUADevice"
+
+
+
+    def reset(self):
+        """Reset the device."""
+        self.temperature = 25.0
+        self._vacuum_status = False
+        return True
+
+# Configuration for the adapter
+opcua_adapter = {
+    'options': {
+        'port': 4840,
+        'server_name': 'OPC UA Server',
+        'read_only_properties': ['status']
+    }
+}

lewis_emulators/opcua/states.py

@@ -0,0 +1,5 @@
+from lewis.core.statemachine import State
+
+
+class DefaultState(State):
+    pass

other_emulators/opcua/simple-test.py

@@ -0,0 +1,57 @@
+import asyncio
+import logging
+
+from asyncua import Server, ua
+from asyncua.common.methods import uamethod
+
+
+@uamethod
+def func(parent, value):
+    return value * 2
+
+
+async def main():
+    _logger = logging.getLogger(__name__)
+    # setup our server
+    server = Server()
+    await server.init()
+    server.set_endpoint("opc.tcp://0.0.0.0:4840")
+
+    # set up our own namespace, not really necessary but should as spec
+    uri = "http://isiscomputing-namespace.epics"
+    idx = await server.register_namespace(uri)
+
+    # populating our address space
+    # server.nodes, contains links to very common nodes like objects and root
+    print(type(idx), idx)
+    myobj = await server.nodes.objects.add_object(idx, bname="Pressure_Values")
+    pressure_var = await myobj.add_variable(idx, "Pressure", 6.7)
+    temp_var = await myobj.add_variable(idx, "Temperature", 30.5)
+    # Set Pressure and Temperature to be writable by clients
+    await pressure_var.set_writable()
+    await temp_var.set_writable()
+    await server.nodes.objects.add_method(
+        ua.NodeId("ServerMethod", idx),
+        ua.QualifiedName("ServerMethod", idx),
+        func,
+        [ua.VariantType.Int64],
+        [ua.VariantType.Int64],
+    )
+    _logger.info("Starting server!")
+
+    print(f"Pressure NodeId: {pressure_var.nodeid}")
+    print(f"Temperature NodeId: {temp_var.nodeid}")
+    async with server:
+        while True:
+            await asyncio.sleep(1)
+            new_val = await pressure_var.get_value() + 0.1
+            new_val2 = await temp_var.get_value() + 0.05
+            _logger.info("Set value of %s to %.1f", pressure_var, new_val)
+            _logger.info("Set value of %s to %.1f", temp_var, new_val2)
+            await pressure_var.write_value(new_val)
+            await temp_var.write_value(new_val2)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
+    asyncio.run(main(), debug=False)

requirements.txt

@@ -1 +1,2 @@
 lewis
+asyncua