lxa_iobus: node: add LxaRemoteNode that uses the HTTP API

This allows using lxa_iobus as a library and talking to a node that
is connected to an IOBus server as if it was connected locally.
This can come in handy for writing command line tools for node types
that do not yet have a fully-fledged API.

The features of a node are discoverd by enumerating the available
protocols via SDO-over-HTTP.

Signed-off-by: Leonard Göhrs <l.goehrs@pengutronix.de>

lxa_iobus/network.py

@@ -21,7 +21,7 @@
     gen_lss_switch_mode_global_message,
     parse_sdo_message,
 )
-from lxa_iobus.node import LxaNode
+from lxa_iobus.node.bus_node import LxaBusNode
 
 logger = logging.getLogger("lxa-iobus.network")
 
@@ -51,7 +51,7 @@ def __init__(self, loop, interface, bustype="socketcan", bitrate=100000, lss_add
 
         self.tx_error = False
 
-        self.isp_node = LxaNode(
+        self.isp_node = LxaBusNode(
             lxa_network=self,
             lss_address=[0, 0, 0, 0],
             node_id=125,
@@ -491,7 +491,7 @@ async def lss_fast_scan(self):
                 # in self.recv().
                 # Otherwise the node would show up as half initialized in the list
                 # of nodes for a moment.
-                self._node_in_setup = LxaNode(
+                self._node_in_setup = LxaBusNode(
                     lxa_network=self,
                     lss_address=lss,
                     node_id=node_id,
@@ -597,7 +597,7 @@ async def setup_single_node(self):
         # The lss_address is a bogus address, because we never discovered the nodes address.
         # This means the node will show up as Unknown type and with default input/output
         # names, even if it is an IOBus device.
-        self.nodes[1] = LxaNode(
+        self.nodes[1] = LxaBusNode(
             lxa_network=self,
             lss_address=[0, 0, 0, 0],
             node_id=1,

lxa_iobus/node/base_node.py

@@ -0,0 +1,93 @@
+import contextlib
+import json
+import logging
+
+from .object_directory import ObjectDirectory
+from .products import find_product
+
+logger = logging.getLogger("lxa_iobus.base_node")
+
+
+class LxaBaseNode(object):
+    def __init__(self, lss_address):
+        self.lss_address = lss_address
+        self.product = find_product(lss_address)
+        self.name = self.product.name()
+        self.address = ".".join(["{:08x}".format(i) for i in lss_address])
+
+        self.locator_state = False
+
+    async def setup_object_directory(self):
+        self.od = await ObjectDirectory.scan(
+            self,
+            self.product.ADC_NAMES,
+            self.product.INPUT_NAMES,
+            self.product.OUTPUT_NAMES,
+        )
+
+    async def ping(self):
+        try:
+            if "locator" in self.od:
+                self.locator_state = await self.od.locator.active()
+            else:
+                # The device does not advertise having an IOBus locator.
+                # Try a CANopen standard endpoint instead
+                await self.od.manufacturer_device_name.name()
+
+            return True
+
+        except TimeoutError:
+            return False
+
+    async def set_locator_state(self, state):
+        if state:
+            await self.od.locator.enable()
+        else:
+            await self.od.locator.disable()
+
+        self.locator_state = state
+
+    async def invoke_isp(self):
+        # The node will enter the bootloader immediately,
+        # so we will not receive a response.
+        with contextlib.suppress(TimeoutError):
+            await self.od.bootloader.enter()
+
+    async def info(self):
+        device_name = await self.od.manufacturer_device_name.name()
+        hardware_version = await self.od.manufacturer_hardware_version.version()
+        software_version = await self.od.manufacturer_software_version.version()
+
+        # check for updates
+        update_name = ""
+
+        bundled_firmware_version = self.product.FIRMWARE_VERSION
+        bundled_firmware_file = self.product.FIRMWARE_FILE
+
+        if (bundled_firmware_version is not None) and (bundled_firmware_file is not None):
+            raw_version = software_version.split(" ")[1]
+            version_tuple = tuple([int(i) for i in raw_version.split(".")])
+
+            if version_tuple < bundled_firmware_version:
+                update_name = bundled_firmware_file
+
+        info = {
+            "device_name": device_name,
+            "address": self.address,
+            "hardware_version": hardware_version,
+            "software_version": software_version,
+            "update_name": update_name,
+        }
+
+        if "version_info" in self.od:
+            info["protocol_version"] = await self.od.version_info.protocol()
+            info["board_version"] = await self.od.version_info.board()
+            info["serial_string"] = await self.od.version_info.serial()
+            info["vendor_name"] = await self.od.version_info.vendor_name()
+            info["notes"] = await self.od.version_info.notes()
+
+            # If the json is not valid we just leave it as string instead
+            with contextlib.suppress(json.decoder.JSONDecodeError):
+                info["notes"] = json.loads(info["notes"])
+
+        return info

lxa_iobus/node.py → lxa_iobus/node/bus_node.py

@@ -1,7 +1,5 @@
 import asyncio
 import concurrent
-import contextlib
-import json
 import logging
 import struct
 
@@ -15,40 +13,26 @@
     gen_sdo_segment_upload,
 )
 
-from .object_directory import ObjectDirectory
-from .products import find_product
+from .base_node import LxaBaseNode
 
 DEFAULT_TIMEOUT = 1
 
-logger = logging.getLogger("lxa_iobus.node")
+logger = logging.getLogger("lxa_iobus.bus_node")
 
 
-class LxaNode:
+class LxaBusNode(LxaBaseNode):
     def __init__(self, lxa_network, lss_address, node_id):
+        super().__init__(lss_address)
+
         self.lxa_network = lxa_network
-        self.lss_address = lss_address
         self.node_id = node_id
 
         self._pending_message = None
         self._lock = asyncio.Lock()
 
-        self.address = ".".join(["{:08x}".format(i) for i in self.lss_address])
-        self.product = find_product(lss_address)
-        self.name = self.product.name()
-
-        self.locator_state = False
-
     def __repr__(self):
         return f"<LxaBusNode(address={self.address}, node_id={self.node_id})>"
 
-    async def setup_object_directory(self):
-        self.od = await ObjectDirectory.scan(
-            self,
-            self.product.ADC_NAMES,
-            self.product.INPUT_NAMES,
-            self.product.OUTPUT_NAMES,
-        )
-
     def set_sdo_response(self, message):
         if self._pending_message and not self._pending_message.done() and not self._pending_message.cancelled():
             self._pending_message.set_result(message)
@@ -297,70 +281,3 @@ async def sdo_write(self, index, sub_index, data, timeout=DEFAULT_TIMEOUT):
 
             # Maybe the complete flag is not correctly set
             raise Exception("Something went wrong with segmented download")
-
-    async def ping(self):
-        try:
-            if "locator" in self.od:
-                self.locator_state = await self.od.locator.active()
-            else:
-                # The device does not advertise having an IOBus locator.
-                # Try a CANopen standard endpoint instead
-                await self.od.manufacturer_device_name.name()
-
-            return True
-
-        except TimeoutError:
-            return False
-
-    async def info(self):
-        device_name = await self.od.manufacturer_device_name.name()
-        hardware_version = await self.od.manufacturer_hardware_version.version()
-        software_version = await self.od.manufacturer_software_version.version()
-
-        # check for updates
-        update_name = ""
-
-        bundled_firmware_version = self.product.FIRMWARE_VERSION
-        bundled_firmware_file = self.product.FIRMWARE_FILE
-
-        if (bundled_firmware_version is not None) and (bundled_firmware_file is not None):
-            raw_version = software_version.split(" ")[1]
-            version_tuple = tuple([int(i) for i in raw_version.split(".")])
-
-            if version_tuple < bundled_firmware_version:
-                update_name = bundled_firmware_file
-
-        info = {
-            "device_name": device_name,
-            "address": self.address,
-            "hardware_version": hardware_version,
-            "software_version": software_version,
-            "update_name": update_name,
-        }
-
-        if "version_info" in self.od:
-            info["protocol_version"] = await self.od.version_info.protocol()
-            info["board_version"] = await self.od.version_info.board()
-            info["serial_string"] = await self.od.version_info.serial()
-            info["vendor_name"] = await self.od.version_info.vendor_name()
-            info["notes"] = await self.od.version_info.notes()
-
-            # If the json is not valid we just leave it as string instead
-            with contextlib.suppress(json.decoder.JSONDecodeError):
-                info["notes"] = json.loads(info["notes"])
-
-        return info
-
-    async def set_locator_state(self, state):
-        if state:
-            await self.od.locator.enable()
-        else:
-            await self.od.locator.disable()
-
-        self.locator_state = state
-
-    async def invoke_isp(self):
-        # The node will enter the bootloader immediately,
-        # so we will not receive a response and waiting for it will timeout.
-        with contextlib.suppress(TimeoutError):
-            await self.od.bootloader.enter()

lxa_iobus/object_directory.py → lxa_iobus/node/object_directory.py

@@ -19,19 +19,19 @@
 Instead the provide an `async def new()` classmethod that should be used
 instead of `__init__()`.
 
-To use just the ADC feature on an LxaNode:
+To use just the ADC feature on an LxaRemoteNode:
 
-    node = LxaNode(...)
+    node = await LxaRemoteNode.new("http://localhost:8080", "<node name>")
     adc = await Adc.new(node)
 
     print("number of ADC channels:", await adc.channel_count())
 
     for name, value in await adc.read_all():
         print(f"Channel {name}: {value}")
 
-To automatically enumerate all features of an LxaNode:
+To automatically enumerate all features of an LxaRemoteNode:
 
-    node = LxaNode(...)
+    node = await LxaRemoteNode.new("http://localhost:8080", "<node name>")
     od = await ObjectDirectory.scan(node)
     adc = od.adc
 
@@ -271,7 +271,7 @@ class ProcessDataObject(object):
         >>>        self.add_sub("example_value", SubIndex.u32(0))
         >>>        self.add_sub_array("example_array", [SubIndex.u32(1), SubIndex.u32(2)])
         >>>
-        >>> node = LxaNode(...)
+        >>> node = await LxaRemoteNode.new("http://localhost:8080", "<node name>")
         >>> ex = ExampleFeature(node)
         >>> await ex.set_example_value(1)
         >>> await ex.example_value()

lxa_iobus/node/remote_node.py

@@ -0,0 +1,63 @@
+import logging
+
+from aiohttp import ClientResponseError, ClientSession
+
+from lxa_iobus.canopen import SdoAbort
+
+from .base_node import LxaBaseNode
+
+logger = logging.getLogger("lxa_iobus.remote_node")
+
+
+class LxaRemoteNode(LxaBaseNode):
+    @classmethod
+    async def new(cls, base_url, node_name):
+        session = ClientSession(raise_for_status=True)
+
+        response = await session.get(f"{base_url}/nodes/{node_name}/")
+        body = await response.json()
+        address = body["result"]["info"]["address"]
+        lss_address = list(int(a) for a in address.split("."))
+
+        this = cls(session, base_url, node_name, lss_address)
+
+        await this.setup_object_directory()
+
+        return this
+
+    def __repr__(self):
+        return f"<LxaRemoteNode(address={self.address}, base_url={self.base_url})>"
+
+    def __init__(self, session, base_url, node_name, lss_address):
+        super().__init__(lss_address)
+
+        self.session = session
+        self.base_url = base_url
+        self.node_name = node_name
+
+    def _sdo_url(self, index, sub_index):
+        return f"{self.base_url}/api/v2/node/{self.node_name}/raw_sdo/0x{index:04x}/{sub_index}"
+
+    async def sdo_read(self, index, sub_index, _timeout=None):
+        try:
+            response = await self.session.get(self._sdo_url(index, sub_index))
+            return await response.read()
+        except ClientResponseError as e:
+            logger.warn(f"sdo_read() failed for node {self.name}: {e}")
+
+            # We do not have all the information we need for a proper SdoAbort,
+            # but no node id and error id "General error" should be good enough.
+            raise SdoAbort(0, index, sub_index, 0x08000000) from e
+
+    async def sdo_write(self, index, sub_index, data, _timeout=None):
+        try:
+            await self.session.post(self._sdo_url(index, sub_index), data=data)
+        except ClientResponseError as e:
+            logger.warn(f"sdo_write() failed for node {self.name}: {e}")
+
+            # We do not have all the information we need for a proper SdoAbort,
+            # but no node id and error id "General error" should be good enough.
+            raise SdoAbort(0, index, sub_index, 0x08000000) from e
+
+    async def close(self):
+        await self.session.close()

pyproject.toml

@@ -30,6 +30,7 @@ packages = [
   "lxa_iobus.lpc11xxcanisp",
   "lxa_iobus.lpc11xxcanisp.firmware",
   "lxa_iobus.lpc11xxcanisp.loader",
+  "lxa_iobus.node",
   "lxa_iobus.server",
 ]
 include-package-data = true