fix wrong bank calculation, add features in read/writereg

Author: heli2src

pylab_ml/common/data.py

@@ -65,10 +65,16 @@ def complement(value, bits):
     bvalue = formatstring.format(value)
     return int(''.join({'0': '1', '1': '0'}[x] for x in bvalue), 2)
 
+def complement2(value, bits):
+    """2's complement representation from a value with number of bits."""
+    if value >= 2**(bits-1):
+        raise ValueError(f"complement2 value must be < {2**(bits-1)}")
+    result = value if value >=0 else complement(-value, bits)+1
+    return result & (2**bits -1)
 
 def crc4(data):             # width
     """
-    Calculate 4-bit crc from a data with widht bits.
+    Calculate 4-bit crc from a data with width bits.
 
     polynomial = X4+X+1
 

pylab_ml/misc/registermaster.py

@@ -21,6 +21,9 @@
 """
 
 import re
+import fnmatch
+import ast
+from typing import Iterable, List, Union
 import sys
 import os
 import pandas as pd
@@ -1303,12 +1306,15 @@ def init(self):
             pass
         object.__setattr__(self, "mapping", {})
         object.__setattr__(self, "_len_slices", None)
+        object.__setattr__(self, "banks", {})
         self.register = self.__dict__.copy()  # create attribute register with all registernames
         for reg in self.__dict__:
             if reg != "register" and not hasattr(self.register[reg], "_Register__has_reset"):
                 self.register.pop(reg, None)
             elif reg != "register":
                 bank = self.register[reg]._bank
+                if bank is not None and (bank not in self.banks or self.banks[bank] < self.register[reg].addr):
+                    self.banks[bank] = self.register[reg].addr
                 if bank is not None and bank != "":
                     self.mapping[bank] = self.register[reg]._len_slices()
         self.mqtt_list = list(self.register) + self.mqtt_all
@@ -1340,6 +1346,11 @@ def _mqtt2json(self, value, attr=None):
             return value
         return "nomqtt"
 
+    def _register_to_list(patterns: Union[str, Iterable[str]]) ->List[str]:
+        if isinstance(patterns, str):
+            return[patterns]
+        return list(patterns)
+
     def close(self):
         pass
 
@@ -1383,19 +1394,68 @@ def set_bank(self, adr):
         -------
           adr.
         """
+        if self.banks:
+            for bank in self.banks.keys():
+                if adr <= self.banks[bank]:
+                    self._bank = bank
+                    break
         if self.mapping != {}:
             mybank = 0
             for bank in self.mapping:
                 mybank = bank if adr >= bank else mybank
-            self._bank = mybank
+            # self._bank = mybank
             self._len_slices = self.mapping[mybank]
-            adr = adr - mybank
+            # adr = adr - mybank   # TODO: cj 7.2.2026 das ist doch falsch, überprüfen mit HATC!!!!
         else:
             self._bank = None
             self._len_slices = None
         return adr
 
+    def _call_from_string(self, adr, callstr):
+        regs = self._register_to_list()
+        call = ast.parse(callstr, mode='eval').body
+        args = [ast.literal_eval(a) for a in call.args]
+        kwargs = {kw.arg:ast.literal_eval(kw.value) for kw in call.keywords}
+        index = 0
+        dat = args[0]
+        result = 0 if type(dat) == list else []
+        for reg in regs:
+            if fnmatch.fnmatchcase(reg._name, adr):
+                function = getattr(getattr(self, reg._name), call.func.id)
+                if type(dat) == list:
+                    args[0] = dat[index]
+                resu = function(*args, **kwargs)
+                if type(dat) == list:
+                    result += resu
+                else:
+                    result.append(resu)
+                index += 1
+        return result
+
     def readreg(self, adr, bank=None, compare=None, onlycheck=True, tolerance=0, mask=None):
+        if type(adr) is str:
+           if len(adr) == 0:
+               mylogger.log_message(LogLevel.Error(), 'readreg: adr is empty')
+               return
+           elif adr[0].isdigit():
+               adr = list(common.arange(adr))
+           else:
+               return self._call_from_string(adr, f"read({compare}, {onlycheck}, {tolerance}, {mask})")
+        if type(adr) is  list:
+            result = [] if compare is None else 0
+            index = 0
+            for a in adr:
+                comp = compare[index] if type(compare) is list else compare
+                resu = self._readreg(a, bank, comp, onlycheck, tolerance, mask)
+                if compare is None:
+                    result.append(resu)
+                else:
+                    result += resu
+                index += 1
+            return result
+        return self._readreg(adr, bank, compare, onlycheck, tolerance, mask)
+
+    def _readreg(self, adr, bank=None, compare=None, onlycheck=True, tolerance=0, mask=None):
         """
         Read Register with selected protokoll.
 
@@ -1417,13 +1477,16 @@ def readreg(self, adr, bank=None, compare=None, onlycheck=True, tolerance=0, mas
             0 = ok
             1 = error
         """
+        hadr = adr
         if bank is not None:
             self._bank = bank
         else:
             adr = self.set_bank(adr)
         if self._bank is not None and self._bank != "":
             self._protocol.writebase(self._bank)
         value = self._protocol.readreg(adr, compare=compare, tolerance=tolerance, mask=mask)
+        mylogger.log_message(LogLevel.Measure(), f"readreg(0x{hadr:02x}) == {hex(value)}")
+        self.publish_set(f"readreg({hadr}:02x)", value)
         if self._len_slices is not None:
             value &= 2**self._len_slices - 1
         if compare is not None:
@@ -1439,6 +1502,24 @@ def readreg(self, adr, bank=None, compare=None, onlycheck=True, tolerance=0, mas
         return value
 
     def writereg(self, adr, dat, bank=None):
+        if type(adr) is str:
+            if len(adr) == 0:
+                mylogger.log_message(LogLevel.Error(), 'writereg: adr is empty')
+                return
+            elif adr[0].isdigit():
+                adr = list(common.arange(adr))
+            else:
+                return self._call_from_string(adr, f"write({dat})")
+        if type(adr) is list:
+            index = 0
+            for a in adr:
+                d = dat[index] if type(dat) is list else dat
+                self._writereg(a, d, bank)
+                index += 1
+            return
+        self._writereg(adr, dat, bank)
+
+    def _writereg(self, adr, dat, bank=None):
         """
         Write Register with selected protokoll.
 
@@ -1456,13 +1537,17 @@ def writereg(self, adr, dat, bank=None):
         None.
 
         """
+        hadr = adr
         if bank is not None:
             self._bank = bank
         else:
             adr = self.set_bank(adr)
         if self._bank is not None and self._bank != "":
             self._protocol.writebase(self._bank)
         self._protocol.writereg(adr, dat)
+        msg = f"writereg(0x{hadr:02x}) := {hex(dat)}"
+        mylogger.log_message(LogLevel.Measure(), msg)
+        self.publish_set(f"writereg({hadr})", dat)
 
     def reset(self):
         self._protocol.reset()