Watch remotely written CDIVar(s) by space and address, preserving type information.

Author: Poikilos

examples/example_node_memory_implementation.py

@@ -18,6 +18,7 @@
                           the address and port). Defaults to a hard-coded test
                           address and port.
 '''
+from collections import OrderedDict
 import os
 import socket
 import struct
@@ -28,6 +29,7 @@
 
 # region same code as other examples
 from examples_settings import Settings
+from openlcb.cdivar import CDIVar
 from openlcb.convert import Convert
 from openlcb.localnode import LocalNode
 from openlcb.memoryspace import MemorySpace
@@ -208,8 +210,47 @@ def memoryReadFail(memo):
 # localNodeProcessor = LocalNodeProcessor(canLink, localNode)
 # canLink.registerMessageReceivedListener(localNodeProcessor.process)
 localNodeProcessor = localNode.localNodeProcessor
-localNode.setInt(0, 0, settings['port'], 2, False)
-localNode.setFloat(0, 2, settings['timeout'], 2)
+
+# region simple local configuration
+# localNode.setInt(0, 0, settings['port'], 2, False)
+# localNode.setFloat(0, 2, settings['timeout'], 2)
+# endregion simple local configuration
+
+
+# region observable configuration
+
+
+def valueChangedRemotely(var: CDIVar):
+    # f"REMOTE memory configuration: set value={repr(var.getSerializable())}"
+    value = var.getInt() if (var.className == "int") else var.getFloat()
+    data = var.getData()
+    if data is None:
+        data = bytearray([])
+    print()
+    print(
+        f"[valueChangedRemotely] set {type(value).__name__}"
+        f" value={repr(value)} ({Convert.toHex(data)})"
+        f" at address {var.address} (tag={var.tag})")
+    return
+
+
+vars = OrderedDict()
+vars['port'] = CDIVar("int", space=0, address=0, _size=2,
+                      _default=CDIVar.fromInt(12021, 2))
+port = settings['port']
+assert port is not None
+vars['port'].setInt(port)
+vars['timeout'] = CDIVar("float", space=0, address=2, _size=2,
+                         _default=CDIVar.fromFloat(0.5, 2))
+timeout = settings['timeout']
+assert timeout is not None
+vars['timeout'].setFloat(timeout)
+# assert vars['timeout'].getData() == bytearray([0x38, 0])
+for name, var in vars.items():
+    var.tag = name  # optional application-specific data
+    memoryService.memory.registerWatchVar(var)
+memoryService.memory.registerWriteListener(valueChangedRemotely)
+# endregion observable configuration
 
 
 def displayOtherNodeIds(message: Message) :

openlcb/cdivar.py

@@ -10,6 +10,7 @@
 
 from openlcb import emit_cast, formatted_ex
 from openlcb.eventid import EventID
+from openlcb.memoryspace import MemorySpace
 from openlcb.openlcbaction import OpenLCBAction
 
 logger = getLogger(__name__)
@@ -104,6 +105,9 @@ class CDIVar:
             .signed = True)
         _max (int): Maximum value (only for int/float className)
         _size (int): Size of int/float (not allowed for other className).
+        tag (str): A key that can be used to access the var externally
+            such as if in a dictionary, or any other data useful to
+            the application.
 
     Attributes:
         className (str): An OpenLCB CDI type. Must be a key in
@@ -131,13 +135,28 @@ class CDIVar:
     """
     TYPED_KEYS = ['min', 'max', 'default']
 
-    def __init__(self, className, _min=None, _max=None,
-                 _size=None, _default=None, assert_range=False,
-                 _no_min=False, _no_max=False, _default_data=None,
-                 signed=None):
+    def __init__(self, className,
+                 _min: Union['CDIVar', None] = None,
+                 _max: Union['CDIVar', None] = None,
+                 _size: Union[int, None] = None,
+                 _default: Union['CDIVar', None] = None,
+                 assert_range=False,
+                 _no_min=False, _no_max=False,
+                 _default_data: Union[bytearray, bytes, None] = None,
+                 signed: Union[bool, None] = None,
+                 space: Union[MemorySpace, int, None] = None,
+                 address: Union[int, None] = None,
+                 tag: Union[str, None] = None):
         self.data = None  # type: bytes|None
         self.min = _min  # type: CDIVar|None
         self.max = _max  # type: CDIVar|None
+        self.tag = tag  # type: str|None
+        if space is not None:
+            assert isinstance(space, (MemorySpace, int))
+        self.space = space.value if isinstance(space, MemorySpace) else space
+        if address is not None:
+            assert isinstance(address, int)
+        self.address = address
 
         assert isinstance(className, str), \
             f"Expected {CLASSNAME_TYPES.keys()} got {emit_cast(className)}"
@@ -196,7 +215,7 @@ def __init__(self, className, _min=None, _max=None,
                         logger.error(error)
         elif (className in NUM_TYPES) and not _no_min:
             # self.min = CDIVar(className, _size=_size,
-            #                   _no_min=True, _no_max=True)  # prevent inf recurs
+            #                   _no_min=True, _no_max=True)  # prevent inf rec
             # Set minimum based on size,
             #   as per Configuration Description Information Standard.
             assert _size is not None
@@ -267,9 +286,17 @@ def __init__(self, className, _min=None, _max=None,
                 (f"Expected size in {sizes}"
                  f" for {self.className} but got {self.size}")
         self.floatFormat = None  # type: str|None
-        self.address = None  # type: int|None
         self.element = None  # type: Any|None
-        self.space = None  # type: int|None
+
+    def copy(self, assert_range=False) -> 'CDIVar':
+        default = None
+        if self.default is not None:
+            default = self.default.copy()
+        return CDIVar(self.className, _min=self.min, _max=self.max,
+                      _size=self.size, _default=default,
+                      assert_range=assert_range, _no_min=self._no_min,
+                      _no_max=self._no_max, signed=self.signed,
+                      space=self.space, address=self.address)
 
     @staticmethod
     def cmp_float(left: 'CDIVar', right: Union['CDIVar', float]) -> CompareOp:
@@ -606,15 +633,22 @@ def getDict(self, add_name=True):
         return result
 
     def setInt(self, value: int):
+        if self.className != "int":
+            logger.warning(
+                f"setInt called on {self.className} CDIVar (may be wrong)!")
         self.data = self.intToData(value)
 
     def floatToData(self, value: float) -> bytes:
-        assert self.className == "float", \
-            f"floatToData attempted on non-float: {self.className}"
+        if self.className != "float":
+            logger.warning(
+                f"floatToData on {self.className} CDIVar (may be wrong)!")
         assert isinstance(value, float)
         return struct.pack(self.packFormat(), value)
 
     def setFloat(self, value: float):
+        if self.className != "float":
+            logger.warning(
+                f"setFloat called on {self.className} CDIVar (may be wrong)!")
         self.data = self.floatToData(value)
 
     def stringToData(self, value: str) -> bytes:

openlcb/memorymanager.py

@@ -1,6 +1,6 @@
 from logging import getLogger
 import struct
-from typing import Union
+from typing import Callable, Union
 
 from openlcb.cdivar import SUBTYPE_FORMATS, CDIVar
 from openlcb.memoryspace import MemorySpace
@@ -9,6 +9,8 @@
 
 
 class Segment:
+    """A memory segment that is contiguous or behaves as such.
+    """
     def __init__(self, size=0, readOnly=False):
         assert isinstance(size, int)
         assert size >= 0
@@ -78,7 +80,7 @@ def extend(self, data):
         self._data += data
 
     def setSlice(self, address: int, data: Union[bytearray, bytes],
-                size: Union[int, None] = None, force=True):
+                 size: Union[int, None] = None, force=True):
         assert isinstance(data, (bytearray, bytes))
         assert isinstance(address, int)
         assert address >= 0
@@ -109,16 +111,37 @@ def size(self) -> int:
 
 
 class MemoryManager:
+    """A collection of memory segments.
+    Attributes:
+        watchVars (Dict[int, Dict[int, CDIVar]]): Variables that will be
+            used as the argument to callbacks, as well as set when
+            memory is set, at the address corresponding to the 2nd
+            tier's integer key. The 1st tier of the dict is keyed by a
+            space int.
+            - If not present, and setSlice is used (without the optional
+              callbackVar) no registered callbacks will be called (Not
+              enough information).
+
+    """
+
     def __init__(self):
         self._segments = {}  # type: dict[int, Segment]
+        self._writeListeners = []
+        self.watchVars = {}  # type: dict[int, dict[int, CDIVar]]
 
     def set(self, var: CDIVar):
-        assert isinstance(var, CDIVar)
+        assert issubclass(type(var), CDIVar)
         assert var.space is not None
-        assert var.address
+        assert var.address is not None
         data = var.getData()
         assert data is not None
-        self.setSlice(var.space, var.address, data, size=var.size)
+        watchVar = self.getWatchVar(var.space, var.address)
+        if watchVar is not None:
+            var = watchVar
+            assert var.space is not None
+            assert var.address is not None
+        self.setSlice(var.space, var.address, data, size=var.size,
+                      callbackVar=var)
 
     def getFirst(self, space: Union[MemorySpace, int]):
         """Get first address"""
@@ -196,7 +219,7 @@ def get(self, var: CDIVar) -> CDIVar:
         Returns:
             CDIVar: Same var instance (returned by reference) modified.
         """
-        assert isinstance(var, CDIVar)
+        assert issubclass(type(var), CDIVar)
         assert var.space is not None
         assert var.address
         assert var.size is not None
@@ -206,8 +229,13 @@ def get(self, var: CDIVar) -> CDIVar:
         return var
 
     def setSlice(self, space: Union[MemorySpace, int], address: int,
-                 data: Union[bytes, bytearray], size=None):
-        """Set address in virtual memory space to data"""
+                 data: Union[bytes, bytearray], size=None,
+                 callbackVar: Union[CDIVar, None] = None) -> Segment:
+        """Set address in virtual memory space to data.
+        fireWriteListeners can only be called if
+        callbackVar is known, otherwise type information
+        is not available at this level.
+        """
         assert isinstance(data, (bytearray, bytes))
         if isinstance(space, MemorySpace):
             space = space.value
@@ -223,6 +251,18 @@ def setSlice(self, space: Union[MemorySpace, int], address: int,
             segment = Segment()
             self._segments[space] = segment
         segment.setSlice(address, data, size=size)
+        if callbackVar is None:
+            callbackVar = self.getWatchVar(space, address)
+            if callbackVar is not None:
+                print(f"[setSlice] getWatchVar {callbackVar.className}")
+            else:
+                print(f"[setSlice] getWatchVar (None)")
+        else:
+            print(f"[setSlice] callbackVar {callbackVar.className}")
+        if callbackVar is not None:
+            callbackVar.setData(data)
+            self.fireWriteListeners(callbackVar)
+        return segment
 
     def getSlice(self, space: Union[MemorySpace, int], address: int,
                  size: int, force=False) -> bytearray:
@@ -255,7 +295,14 @@ def setInt(self, space: Union[MemorySpace, int], address: int,
         data = struct.pack(dataFormat, value)
         assert len(data) == size, \
             f"Expected {size} byte(s) for {typeStr}, got {len(data)}"
-        return self.setSlice(space, address, data)
+        var = self.getWatchVar(space, address)
+        if var is None:
+            var = CDIVar("int", _size=size, _no_min=True, _no_max=True,
+                         space=space, address=address)
+        result = self.setSlice(space, address, data, callbackVar=var)
+        if self._writeListeners:
+            self.fireWriteListeners(var)
+        return result
 
     def getInt(self, space: Union[MemorySpace, int], address: int,
                size: int, signed: bool) -> int:
@@ -289,7 +336,14 @@ def setFloat(self, space: Union[MemorySpace, int], address: int,
         data = struct.pack(dataFormat, value)
         assert len(data) == size, \
             f"Expected {size} byte(s) for {typeStr}, got {len(data)}"
-        return self.setSlice(space, address, data)
+        var = self.getWatchVar(space, address)
+        if var is None:
+            var = CDIVar("float", _size=size, _no_min=True, _no_max=True,
+                         space=space, address=address)
+        result = self.setSlice(space, address, data, callbackVar=var)
+        if self._writeListeners:
+            self.fireWriteListeners(var)
+        return result
 
     def getFloat(self, space: Union[MemorySpace, int], address: int,
                  size: int) -> float:
@@ -306,3 +360,80 @@ def getFloat(self, space: Union[MemorySpace, int], address: int,
         assert len(values) == 1, f"Expected 1 {typeStr}, got {len(values)}"
         assert isinstance(values[0], float)
         return values[0]
+
+    def registerWriteListener(self, callback: Callable[[CDIVar], None]):
+        """Register a function to call when a value is written.
+        NOTE: You must also call registerWatchVar or low-level
+        (setSlice) calls will not trigger a callback due to not enough
+        information (callback takes a CDIVar).
+        """
+        self._writeListeners.append(callback)
+
+    def registerWatchVar(self, var: CDIVar):
+        """Register a CDIVar to track writes.
+
+        Arguments:
+            var (CDIVar): Must have var.space
+                and var.address in order to be tracked. This var's value
+                will be edited remotely, and will allow write listeners
+                to fire even when setting memory of a non-number or
+                unspecified type (using setSlice).
+        Raises:
+            AssertionError: space or address is None.
+        """
+        # a.k.a. setWatchVar
+        assert var.space is not None, \
+            'cdivar.space is required in order to listen for change'
+        if isinstance(var.space, MemorySpace):
+            var.space = var.space.value
+        assert isinstance(var.space, int)
+        assert var.address is not None, \
+            'cdivar.address is required in order to listen for change'
+        assert isinstance(var.address, int)
+        if var.space not in self.watchVars:
+            self.watchVars[var.space] = {}
+        if var.address in self.watchVars[var.space]:
+            raise KeyError(
+                f"Address {var.address} of space {var.space}"
+                " is already registered.")
+        self.watchVars[var.space][var.address] = var
+
+    def getWatchVar(self, space: Union[MemorySpace, int], address: int,
+                    default: Union[CDIVar, None] = None):
+        assert space is not None
+        assert address is not None
+        if isinstance(space, MemorySpace):
+            space = space.value
+        assert isinstance(space, int)
+        assert isinstance(address, int)
+        if default is not None:
+            assert issubclass(type(default), CDIVar)
+        spaceVars = self.watchVars.get(space)
+        if spaceVars is None:
+            return None
+        var = spaceVars.get(address, default)
+        if var is not None:
+            # Fix space & address so fireWriteListeners works correctly
+            if var.space is None:
+                logger.warning(
+                    f"Setting var.space={space} using its location")
+                var.space = space
+            elif var.space != space:
+                logger.warning(
+                    f"Setting incorrect var.space {var.space}"
+                    f" to {space} using its location")
+                var.space = space
+            if var.address is None:
+                logger.warning(
+                    f"Setting var.address={address} using its location")
+                var.address = address
+            elif var.address != address:
+                logger.warning(
+                    f"Setting incorrect var.address {var.address}"
+                    f" to {address} using its location")
+                var.address = address
+        return var
+
+    def fireWriteListeners(self, var: CDIVar):
+        for writeListener in self._writeListeners:
+            writeListener(var)