publish cache retained

.woodpecker.yml

@@ -1,29 +1,69 @@
+when:
+  event:
+    - tag
+
+variables:
+  - &NAMESPACE 'homea-ctrl-1'
+
 steps:
   build:
     image: plugins/kaniko
     settings:
-      repo: gitea.hottis.de/wn/pv-controller
       registry:
-        from_secret: container_registry
+        from_secret: local_registry
-      tags: latest,${CI_COMMIT_TAG}
       username:
-        from_secret: container_registry_username
+        from_secret: local_username
       password:
-        from_secret: container_registry_password
+        from_secret: local_password
+      repo: ${FORGE_NAME}/${CI_REPO}
+      auto_tag: true
       dockerfile: Dockerfile
     when:
-      - event: tag
+      ref:
+        exclude:
+        - refs/tags/*-configchange
 
-  deploy:
+  namespace:
-    image: portainer/kubectl-shell:latest
+    image: quay.io/wollud1969/k8s-admin-helper:0.3.4
-    secrets:
+    environment:
-      - source: kube_config
+      KUBE_CONFIG_CONTENT:
-        target: KUBE_CONFIG_CONTENT
+        from_secret: kube_config
+      NAMESPACE: *NAMESPACE
     commands:
-      - export IMAGE_TAG=$CI_COMMIT_TAG
       - printf "$KUBE_CONFIG_CONTENT" > /tmp/kubeconfig
       - export KUBECONFIG=/tmp/kubeconfig
-      - cat ./deployment/install-yml.tmpl | sed -e 's,%IMAGETAG%,'$IMAGE_TAG','g | kubectl apply -f -
+      - kubectl create namespace $NAMESPACE || echo "Namespace $NAMESPACE already exists"
     when:
-      - event: tag
+      ref:
+        exclude:
+        - refs/tags/*-configchange
 
+  configuration:
+    image: quay.io/wollud1969/k8s-admin-helper:0.3.4
+    environment:
+      KUBE_CONFIG_CONTENT:
+        from_secret: kube_config
+      NAMESPACE: *NAMESPACE
+    commands:
+      - printf "$KUBE_CONFIG_CONTENT" > /tmp/kubeconfig
+      - export KUBECONFIG=/tmp/kubeconfig
+      - kubectl create configmap pv-controller-config 
+        --from-file=config.yaml=config/config.yaml 
+        --namespace=$NAMESPACE 
+        --dry-run=client -o yaml | kubectl apply -f - 
+
+  deploy:
+    image: quay.io/wollud1969/k8s-admin-helper:0.3.4
+    environment:
+      KUBE_CONFIG_CONTENT:
+        from_secret: kube_config
+      NAMESPACE: *NAMESPACE
+      IMAGE: "${FORGE_NAME}/${CI_REPO}:${CI_COMMIT_TAG}"
+    commands:
+      - printf "$KUBE_CONFIG_CONTENT" > /tmp/kubeconfig
+      - export KUBECONFIG=/tmp/kubeconfig
+      - cat deployment/install-yml.tmpl | sed "s,%IMAGE%,$IMAGE,g" | kubectl apply -n $NAMESPACE -f -
+    when:
+      ref:
+        exclude:
+        - refs/tags/*-configchange

Dockerfile

@@ -1,36 +1,23 @@
-FROM python:latest
+FROM python:3.14-alpine
 
 LABEL Maintainer="Wolfgang Hottgenroth wolfgang.hottgenroth@icloud.com"
 LABEL ImageName="registry.hottis.de/dockerized/pv-controller"
-LABEL HubImageName="wollud1969/pv-controller"
+LABEL HubImageName="wn/pv-controller"
 
 ARG APP_DIR="/opt/app"
+ARG APP_USER="app"
+ENV CFG_FILE ""
 
-
-ENV MQTT__BROKER ""
-ENV MQTT__PORT "1883"
-ENV MQTT__METERPUBLISHTOPIC "IoT/PV/Values"
-ENV MQTT__METERPUBLISHPERIOD "15"
-ENV MQTT__RELAISSUBSCRIBETOPIC "IoT/PV/Cmd"
-ENV MODBUS__GATEWAY ""
-
-RUN \
-    apt update && \
-    pip3 install loguru && \
-    pip3 install pymodbus && \
-    pip3 install paho-mqtt
-
-RUN \
-    mkdir -p ${APP_DIR} && \
-    useradd -d ${APP_DIR} -u 1000 user
-
-COPY ./src/pv_controller/*.py ${APP_DIR}/
-
-USER 1000:1000
 WORKDIR ${APP_DIR}
 
+COPY ./src/pv_controller/requirements.txt requirements.txt
+COPY ./src/pv_controller/*.py ${APP_DIR}/
+
+RUN addgroup -g 10001 -S ${APP_USER} && \
+    adduser -u 10001 -S ${APP_USER} -G ${APP_USER} && \
+    pip install --no-cache-dir -r requirements.txt
+
+USER ${APP_USER}
 
 CMD ["python", "pvc.py"]
 
-
-

config/config.yaml

@@ -1,30 +1,31 @@
+global:
+  scan_interval: 0.25
+  log_level: INFO
+
 mqtt:
-  broker: 172.16.2.16
+  broker: emqx01-anonymous-cluster-internal.broker.svc.cluster.local
   port: 1883
-  publish_period: 15
 
 modbus:
   gateway: 172.16.2.42
 
-# REGISTERS = [
+
-#   { "slave":2, "addr":0x0048, "type":"input", "attr": "importEnergyActive", "name":"Import active energy", "unit":"kWh", "adaptor": floatAdaptor },
+input:
-#   { "slave":2, "addr":0x004c, "type":"input", "attr": "importEnergyReactive", "name":"Import reactive energy", "unit":"kVAh", "adaptor": floatAdaptor },
+  - name: pv_control
-#   { "slave":2, "addr":0x004a, "type":"input", "attr": "exportEnergyActive", "name":"Export active energy", "unit":"kWh", "adaptor": floatAdaptor },
+    subscribe_topic: IoT/PV/Control
-#   { "slave":2, "addr":0x004e, "type":"input", "attr": "exportEnergyReactive", "name":"Export reactive energy", "unit":"kVAh", "adaptor": floatAdaptor },
+    slave_id: 1
-#   { "slave":2, "addr":0x0012, "type":"input", "attr": "powerApparent", "name":"Apparent Power", "unit":"W", "adaptor": floatAdaptor },
+    address: 0
-#   { "slave":2, "addr":0x000c, "type":"input", "attr": "powerActive", "name":"Active Power", "unit":"W", "adaptor": floatAdaptor },
+    register_type: coil
-#   { "slave":2, "addr":0x0018, "type":"input", "attr": "powerReactive", "name":"Reactive Power", "unit":"W", "adaptor": floatAdaptor },
+  - name: car_control
-#   { "slave":2, "addr":0x0058, "type":"input", "attr": "powerDemandPositive", "name":"PositivePowerDemand", "unit":"W", "adaptor": floatAdaptor },
+    subscribe_topic: IoT/Car/Control
-#   { "slave":2, "addr":0x005c, "type":"input", "attr": "powerDemandReverse", "name":"ReversePowerDemand", "unit":"W", "adaptor": floatAdaptor },
+    slave_id: 5
-#   { "slave":2, "addr":0x001e, "type":"input", "attr": "factor", "name":"Factor", "unit":"-", "adaptor": floatAdaptor },
+    address: 0
-#   { "slave":2, "addr":0x0024, "type":"input", "attr": "angle", "name":"Angle", "unit":"degree", "adaptor": floatAdaptor },
+    register_type: coil
-#   { "slave":2, "addr":0x0000, "type":"input", "attr": "voltage", "name":"Voltage", "unit":"V", "adaptor": floatAdaptor },
-#   { "slave":2, "addr":0x0006, "type":"input", "attr": "current", "name":"Current", "unit":"A", "adaptor": floatAdaptor },
-#   { "slave":1, "addr":0x0001, "type":"holding", "attr": "state", "name":"State", "unit":"-", "adaptor": onOffAdaptor },
-# ]
 
 output:
   - name: pv_meter
+    enabled: true 
+    scan_rate: 15
     publish_topic: IoT/PV/Values
     slave_id: 2
     registers:
@@ -120,22 +121,27 @@ output:
         data_type: float32
         adaptor: floatAdaptor
   - name: pv_control
-    publish_topic: IoT/PV/Control
+    publish_topic: IoT/PV/Control/State
+    scan_rate: 1
+    raw_output: true # use only for output device with only one register, name this register 'output'
     slave_id: 1
     registers:
       - address: 0x0001
-        attribute: state
+        attribute: output
         name: State
         unit: "-"
         register_type: holding
         data_type: int32
         adaptor: onOffAdaptor
   - name: car_control
-    publish_topic: IoT/Car/Control
+    enabled: true
+    publish_topic: IoT/Car/Control/State
+    scan_rate: 1
+    raw_output: true # use only for output device with only one register, name this register 'output'
     slave_id: 5
     registers:
       - address: 0x0001
-        attribute: state
+        attribute: output
         name: State
         unit: "-"
         register_type: holding
@@ -144,26 +150,97 @@ output:
   - name: car_meter
     enabled: true
     publish_topic: IoT/Car/Values
+    scan_rate: 15
     slave_id: 6
     registers:
-      - address: 14
+      - address: 0
         attribute: voltageL1
         name: Voltage L1
         unit: V
-        register_type: holding
+        register_type: input
         data_type: float32
         adaptor: floatAdaptor
-      - address: 16
+      - address: 2
         attribute: voltageL2
         name: Voltage L2
         unit: V
-        register_type: holding
+        register_type: input
         data_type: float32
         adaptor: floatAdaptor
-      - address: 18
+      - address: 4
         attribute: voltageL3
         name: Voltage L3
         unit: V
-        register_type: holding
+        register_type: input
         data_type: float32
         adaptor: floatAdaptor
+      - address: 6
+        attribute: currentL1
+        name: Current L1
+        unit: A
+        register_type: input
+        data_type: float32
+        adaptor: floatAdaptor
+      - address: 8
+        attribute: currentL2
+        name: Current L2
+        unit: A
+        register_type: input
+        data_type: float32
+        adaptor: floatAdaptor
+      - address: 10
+        attribute: currentL3
+        name: Current L3
+        unit: A
+        register_type: input
+        data_type: float32
+        adaptor: floatAdaptor
+      - address: 12
+        attribute: powerL1
+        name: Power L1
+        unit: W
+        register_type: input
+        data_type: float32
+        adaptor: floatAdaptor
+      - address: 14
+        attribute: powerL2
+        name: Power L2
+        unit: W
+        register_type: input
+        data_type: float32
+        adaptor: floatAdaptor
+      - address: 16
+        attribute: powerL3
+        name: Power L3
+        unit: W
+        register_type: input
+        data_type: float32
+        adaptor: floatAdaptor
+      - address: 0x0048
+        attribute: totalImportEnergy
+        name: Total Import Energy
+        unit: kWh
+        register_type: input
+        data_type: float32
+        adaptor: floatAdaptor
+      - address: 0x004a
+        attribute: totalExportEnergy
+        name: Total Export Energy
+        unit: kWh
+        register_type: input
+        data_type: float32
+        adaptor: floatAdaptor
+  - name: car_feedback
+    enabled: true
+    publish_topic: IoT/Car/Feedback/State
+    scan_rate: 1
+    raw_output: true # use only for output device with only one register, name this register 'output'
+    slave_id: 7
+    registers:
+      - address: 0x0010
+        attribute: output
+        name: State
+        unit: "-"
+        register_type: holding
+        data_type: int32
+        adaptor: onOffAdaptor

deployment/install-yml.tmpl

@@ -1,28 +1,12 @@
-apiVersion: v1
-kind: Namespace
-metadata:
-  name: homea
----
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: pv-controller
-  namespace: homea
-data:
-  MQTT__BROKER: "emqx01-anonymous-cluster-internal.broker.svc.cluster.local"
-  MQTT__PORT: "1883"
-  MQTT__METERPUBLISHTOPIC: "IoT/PV/Values"
-  MQTT__METERPUBLISHPERIOD: "15"
-  MQTT__RELAISSUBSCRIBETOPIC: "IoT/PV/Cmd"
-  MODBUS__GATEWAY: "172.16.2.42"
----
 apiVersion: apps/v1
 kind: Deployment
 metadata:
   name: pv-controller
-  namespace: homea
   labels:
     app: pv-controller
+  annotations:
+    reloader.stakater.com/auto: "true"
+    reloader.stakater.com/configmap: "pv-controller-config"
 spec:
   replicas: 1
   selector:
@@ -35,7 +19,15 @@ spec:
     spec:
       containers:
         - name: pv-controller
-          image: gitea.hottis.de/wn/pv-controller:%IMAGETAG%
+          image: %IMAGE%
-          envFrom:
+          env:
-            - configMapRef:
+            - name: CFG_FILE
-                name: pv-controller
+              value: /config/config.yaml
+          volumeMounts:
+            - name: pv-controller-config
+              mountPath: /config/config.yaml
+              subPath: config.yaml
+      volumes:
+        - name: pv-controller-config
+          configMap:
+            name: pv-controller-config

src/pv_controller/FromDevices.py

@@ -9,7 +9,7 @@ def floatAdaptor(i):
     return float(f"{i:0.2f}") if i else 0.0
 
 def onOffAdaptor(i):
-    return i[0] if i else '-1'
+    return 'on' if bool(i) else 'off'
 
 
 
@@ -22,27 +22,40 @@ class FromDevices(AbstractMqttPublisher):
         cnt = 0
         while not self.killBill:
             cnt += 1
-            payload = {}
 
+            for device in self.config.output:
                 try:
+                    payload = {}
                     payload['status'] = "Error"
                     payload['timestamp'] = datetime.datetime.isoformat(datetime.datetime.utcnow())
-                for device in self.config.output:
                     logger.debug(f"{device.name=} {device.publish_topic=}")
                     if not device.enabled:
                         logger.debug(f"  device disabled, skipping")
                         continue
+                    if cnt % device.scan_rate != 0:
+                        logger.debug(f"  not scan_rate yet, skipping")
+                        continue
                     for registers in device.registers:
                         logger.debug(f"  {registers.name=} {registers.address=} {registers.register_type=}")
                         rawValue = self.modbusHandler.readRegister(registers.register_type, device.slave_id, registers.address, registers.data_type)
                         logger.debug(f"    {rawValue=}")
+                        if registers.adaptor == "floatAdaptor":
+                            value = floatAdaptor(rawValue)
+                        elif registers.adaptor == "onOffAdaptor":
+                            value = onOffAdaptor(rawValue)
+                        else:
+                            value = rawValue
+                        logger.debug(f"    {value=}")
+                        payload[registers.attribute] = value
+
                     payload['status'] = "Ok"
+                    payload['cnt'] = cnt
+                
+                    payloadStr = json.dumps(payload) if not device.raw_output else str(payload['output'])
+                    self.publish_with_cache(device.publish_topic, payloadStr)
+                    logger.debug(f"mqtt message sent: {device.publish_topic} -> {payloadStr}")
                 except Exception as e:
                     logger.error(f"Caught exception: {str(e)}")
 
-#            payload['cnt'] = cnt
-#            payloadStr = json.dumps(payload)
-#            self.client.publish(topic, payloadStr)
-#            logger.info(f"mqtt message sent: {topic} -> {payloadStr}")
 
-            self.killEvent.wait(timeout=float(self.config.mqtt.publish_period))
+            self.killEvent.wait(timeout=float(self.config.global_.scan_interval))

src/pv_controller/ModbusBase.py

@@ -31,7 +31,9 @@ READ_REGISTER_FUNCTIONS = {
 
 DATA_TYPES = {
   'int16': ModbusTcpClient.DATATYPE.INT16,
+  'uint16': ModbusTcpClient.DATATYPE.UINT16,
   'int32': ModbusTcpClient.DATATYPE.INT32,
+  'uint32': ModbusTcpClient.DATATYPE.UINT32,
   'float32': ModbusTcpClient.DATATYPE.FLOAT32,
   'float64': ModbusTcpClient.DATATYPE.FLOAT64
 }
@@ -51,7 +53,7 @@ class ModbusHandler:
       dataType = DATA_TYPES[data_type]
       count = dataType.value[1]
 
-      logger.debug(f"{self.client=}, {addr=}, {count=}, {slave=}")
+      logger.debug(f"{addr=}, {count=}, {slave=}")
       res = readFunc(self.client, addr, count=count, device_id=slave)
       if (isinstance(res, pymodbus.pdu.register_message.ReadHoldingRegistersResponse) or
           isinstance(res, pymodbus.pdu.register_message.ReadInputRegistersResponse) or
@@ -66,7 +68,7 @@ class ModbusHandler:
 
 
   def writeCoil(self, slave, addr, value):
-    res = self.client.write_coil(addr, value, slave=slave)
+    res = self.client.write_coil(addr, value, device_id=slave)
     logger.debug(f"write coil result {res}")
     return value
 

src/pv_controller/MqttBase.py

@@ -1,7 +1,7 @@
 import paho.mqtt.client as mqtt
 from loguru import logger
+import uuid
 import threading
-import ssl
 
 
 
@@ -21,7 +21,11 @@ class AbstractMqttPublisher(threading.Thread):
 
         self.config = config
 
-        self.client = mqtt.Client(userdata=self)
+        client_id = f"pv-controller-{uuid.uuid4()}" 
+        logger.info(f"mqtt client id: {client_id}")
+        self.client = mqtt.Client(client_id=client_id, userdata=self)
+
+        self.cache = {}
 
         # consider this flag in the localLoop
         self.killBill = False
@@ -60,3 +64,10 @@ class AbstractMqttPublisher(threading.Thread):
     def onMessage(self, topic, payload):
         logger.warning("mqtt unexpected message received: {} -> {}".format(topic, str(payload)))
 
+    def publish_with_cache(self, topic, payload):
+        if topic in self.cache and self.cache[topic] == payload:
+            logger.debug(f"mqtt message unchanged, not publishing: {topic} -> {payload}")
+            return
+        self.cache[topic] = payload
+        self.client.publish(topic, payload, retain=True)
+

src/pv_controller/ToDevices.py

@@ -13,17 +13,26 @@ class ToDevices(AbstractMqttPublisher):
             sleep(60.0)
 
     def onMessage(self, topic, payload):
-        logger.info("mqtt message received: {} -> {}".format(topic, str(payload)))
+        try:
-        if payload == b'On':
+            logger.debug("mqtt message received: {} -> {}".format(topic, str(payload)))
-            self.modbusHandler.writeCoil(1, 0, 1)
+            for device in self.config.input:
-        elif payload == b'Off':
+                if topic != device.subscribe_topic:
-            self.modbusHandler.writeCoil(1, 0, 0)
+                    continue
-        else:
+                logger.debug(f"{topic=} matches {device.subscribe_topic=}, processing")
-            logger.warning(f"Illegal command {payload} received")
+                if not device.enabled:
-
+                    logger.debug(f"  device disabled, skipping")
+                    continue
+                if device.register_type != 'coil':
+                    raise Exception(f"Unsupported register type {device.register_type} for input device {device.name}")
+                value = payload == b'on'
+                self.modbusHandler.writeCoil(device.slave_id, device.address, value)
+        except Exception as e:
+            logger.error(f"Caught exception in onMessage: {str(e)}")
 
 
     def onConnect(self):
         logger.info("mqtt connected")
-        self.client.subscribe("{}".format(self.config["relaisSubscribeTopic"]))
+        for device in self.config.input:
+            self.client.subscribe(device.subscribe_topic)
+            logger.info(f"subscribed to topic: {device.subscribe_topic}")
         logger.info("subscribed")

src/pv_controller/config.py

@@ -22,16 +22,26 @@ class OutputConfig(BaseModel):
     """Output Configuration for Modbus Devices"""
     name: str
     enabled: bool = Field(default=True)
+    scan_rate: Optional[int] = Field(default=60)
     publish_topic: str
+    raw_output: Optional[bool] = Field(default=False)
     slave_id: int
     registers: List[RegisterConfig]
 
+class InputConfig(BaseModel):
+    """Input Configuration for Modbus Devices (MQTT -> Modbus)"""
+    name: str
+    enabled: bool = Field(default=True)
+    subscribe_topic: str
+    slave_id: int
+    address: int
+    register_type: str
+
 
 class MqttConfig(BaseModel):
     """MQTT Configuration"""
     broker: str
     port: int
-    publish_period: int
 
 
 class ModbusConfig(BaseModel):
@@ -39,10 +49,18 @@ class ModbusConfig(BaseModel):
     gateway: str
 
 
+class GlobalConfig(BaseModel):
+    """Global settings"""
+    scan_interval: float
+    log_level: str
+
+
 class Config(BaseModel):
     """Main Configuration"""
+    global_: GlobalConfig = Field(alias="global")
     mqtt: MqttConfig
     modbus: ModbusConfig
+    input: List[InputConfig]
     output: List[OutputConfig]
 
     @classmethod

src/pv_controller/pvc.py

@@ -5,11 +5,9 @@ from loguru import logger
 from config import Config
 import logging
 import threading
+import sys
 
 
-l = logging.getLogger()
-for h in l.handlers:
-  l.removeHandler(h)
 
 deathBell = threading.Event()
 
@@ -24,11 +22,16 @@ logger.info("pv controller starting")
 
 config = Config.load_from_file()
 
+# configure loguru: only log INFO and above
+logger.remove()
+logger.add(sys.stdout, level=config.global_.log_level)
+
+
 modbusHandler = ModbusHandler(config)
 
-# toDevicesThread = ToDevices(config, modbusHandler)
+toDevicesThread = ToDevices(config, modbusHandler)
-# toDevicesThread.start()
+toDevicesThread.start()
-# logger.info("toDevices started")
+logger.info("toDevices started")
 
 fromDevicesThread = FromDevices(config, modbusHandler)
 fromDevicesThread.start()
@@ -43,11 +46,11 @@ logger.info("pv controller is running")
 deathBell.wait()
 logger.error("pv controller is dying")
 
-# toDevicesThread.stop()
+toDevicesThread.stop()
 fromDevicesThread.stop()
 
-# toDevicesThread.join()
+toDevicesThread.join()
-# logger.error("toDevices joined")
+logger.error("toDevices joined")
 fromDevicesThread.join()
 logger.error("fromDevices joined")