Add Python version of the edge driver library

lib/py-edge-driver/.gitignore

@@ -0,0 +1,3 @@
+dist/
+__pycache__
+.pytest_cache

lib/py-edge-driver/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) University of Sheffield AMRC 2025
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

lib/py-edge-driver/README.md

@@ -0,0 +1,5 @@
+# py-edge-driver
+
+## A Python library for creating Factory+ Edge Agent drivers.
+
+This is an attempt at a straight port of the [ACS JavaScript edge driver library](https://github.com/AMRC-FactoryPlus/amrc-connectivity-stack/tree/main/lib/js-edge-driver). Refer to it for concepts and more comprehensive docs.

lib/py-edge-driver/examples/test_driver.py

@@ -0,0 +1,76 @@
+import time
+import logging
+import math
+from amrc.factoryplus.edge_driver import PolledDriver, BufferX, Handler
+import asyncio
+
+funcs = {
+    'const': lambda p, a: lambda t: a,
+    'sin': lambda p, a: lambda t: a * math.sin(2 * math.pi * (t / p)),
+    'saw': lambda p, a: lambda t: (a / p) * (t % p),
+}
+
+packing = {
+    'bd': [8, lambda v: BufferX.fromDoubleBE(v)],
+    'ld': [8, lambda v: BufferX.fromDoubleLE(v)],
+    'bf': [4, lambda v: BufferX.fromFloatBE(v)],
+    'lf': [4, lambda v: BufferX.fromFloatLE(v)],
+}
+
+
+class TestHandler(Handler):
+    @classmethod
+    def create(cls, driver, conf):
+        return TestHandler()
+
+    def close(self, callback = None):
+        return None
+
+    async def connect(self):
+        return "UP"
+
+    def parseAddr(self, addr):
+        parts = addr.split(":")
+        if len(parts) != 4:
+            return None
+
+        func = funcs.get(parts[0])
+        if not func:
+            return None
+
+        try:
+            period = float(parts[1])
+            amplitude = float(parts[2])
+        except ValueError:
+            return None
+
+        pack = packing.get(parts[3])
+        if not pack:
+            return None
+
+        return {
+            'func': func(period, amplitude),
+            'size': pack[0],
+            'pack': pack[1],
+        }
+
+    async def poll(self, spec):
+        val = spec['func'](time.perf_counter() * 1000)  # Convert to milliseconds
+        buf = spec['pack'](val)
+
+        return buf
+
+driver = PolledDriver(handler=TestHandler, edge_username="my_connection", edge_mqtt="mqtt://localhost", edge_password="my_password")
+
+
+logging.basicConfig(level=logging.DEBUG)
+
+async def main():
+    try:
+        await driver.run()
+        await asyncio.Event().wait()
+    except KeyboardInterrupt:
+        logging.info("Shutting down.")
+
+asyncio.run(main())
+

lib/py-edge-driver/pyproject.toml

@@ -0,0 +1,33 @@
+[project]
+name = "amrc.factoryplus.edge_driver"
+version = "0.1.0"
+description = "An Edge Agent driver library for Factory+."
+authors = [{ name = "KavanPrice", email = "[email protected]" }]
+license = "MIT"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = ["paho-mqtt (>=2.1.0,<3.0.0)", "asyncio (>=3.4.3,<4.0.0)"]
+
+
+[build-system]
+requires = ["poetry-core>=2.0.0,<3.0.0"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.poetry]
+packages = [
+    { include = "amrc", from = "src" }
+]
+
+[tool.poetry.group.dev.dependencies]
+pytest = "^8.3.5"
+pytest-mock = "^3.14.0"
+pytest-asyncio = "^0.26.0"
+
+[tool.poetry.group.test.dependencies]
+pytest = "^8.3.5"
+pytest-mock = "^3.14.0"
+pytest-asyncio = "^0.26.0"
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+asyncio_default_fixture_loop_scope = "function"

lib/py-edge-driver/src/amrc/factoryplus/edge_driver/__init__.py

@@ -0,0 +1,12 @@
+# Copyright (c) University of Sheffield AMRC 2025.
+
+"""
+AMRC Connectivity Stack - Python Edge Driver Library
+
+This library provides a Python interface for writing edge device drivers for the ACS Edge Agent.
+"""
+
+from .async_driver import AsyncDriver
+from .polled_driver import PolledDriver
+from .handler import Handler
+from . import bufferx as BufferX

lib/py-edge-driver/src/amrc/factoryplus/edge_driver/async_driver.py

@@ -0,0 +1,38 @@
+# Copyright (c) University of Sheffield AMRC 2025.
+
+import logging
+
+from typing import Dict, Union, Type
+
+from .driver import Driver
+
+from .handler import Handler
+
+
+class AsyncDriver(Driver):
+    def __init__(self, **opts):
+        """
+        Initialize the Async Driver with the provided options.
+
+        Args:
+            handler: Handler class for device-specific logic
+            edge_username: Username to connect to edge MQTT
+            edge_mqtt: Edge MQTT host
+            edge_password: Password to connect to edge MQTT
+            reconnect_delay: Delay in reconnecting to the southbound device
+        """
+        super().__init__(**opts)
+
+    async def data(self, spec, buf):
+        self.log.debug(f"DATA {buf}")
+        dtopic = self.topics.get(spec) if self.topics else None
+        if not dtopic:
+            return
+
+        mtopic = self.topic("data", dtopic)
+        return self.publish_async(mtopic, buf)
+
+    def publish_async(
+        self, topic: str, payload: Union[str, bytes, bytearray, int, float, None]
+    ):
+        return self.mqtt.publish(topic, payload)