p1_reader.py

#!/usr/bin/python3.11
# -*- coding: utf-8 -*-

import os
import time
from datetime import datetime
import asyncio
import serial
import sqlite3
from concurrent.futures import ThreadPoolExecutor

from shared_store import set_key
from logger_setup import get_logger
import systemd.daemon
from config import (
    ENERGY_DB, ENERGY_JSON, READER_DATA_JSON, LOADBALANCE_CSV, LOG_MAX_SIZE,
    P1_SERIAL_PORT, P1_BAUDRATE, P1_SERIAL_TIMEOUT, P1_EXPECTED_FIELDS,
    P1_POLL_INTERVAL, P1_WATCHDOG_INTERVAL, P1_RETRY_DELAY, P1_STALE_THRESHOLD,
)

logger = get_logger("p1_reader")

# max_workers=2 kan verstopt raken: ser.readline() (continu, via read_loop)
# en loadbalance_sync()/openSerial_sync/closeSerial_sync (periodiek, via
# energy_manager) delen dezelfde pool. Bij structurele bezetting door de
# continue readline-cyclus kan het periodieke werk hierachter vast komen
# te zitten zonder een exception te geven. Meer workers verkleint dit risico.
executor = ThreadPoolExecutor(max_workers=4)
MAX_SIZE = LOG_MAX_SIZE

# Tijdstip (monotone klok) van de laatst succesvol gelezen, complete
# telegram. De watchdog gebruikt dit om te bepalen of het proces nog
# daadwerkelijk WERKT, niet alleen of het nog LEEFT (zie watchdog_task).
last_telegram_at: float | None = None
# Tijdstip van de laatst voltooide energy_manager()-cyclus. Dit is een
# APARTE achtergrondtaak t.o.v. read_loop() (die last_telegram_at bijhoudt)
# - de twee kunnen onafhankelijk van elkaar vastlopen, dus allebei apart
# bewaken is nodig.
last_energy_cycle_at: float | None = None
process_started_at = time.monotonic()  # referentie voor de opstart-gratieperiode

# 1 telegram tegelijk in de pijplijn
telegram_queue: asyncio.Queue = asyncio.Queue(maxsize=1)
loadbalance_lock = asyncio.Lock()

# DSMR OBIS-code mapping. Dit is protocolstandaard, geen instelbare waarde -
# blijft hier hardcoded. Pas P1_EXPECTED_FIELDS in config.py aan als je
# DSMR-versie een ander aantal velden gebruikt (bijv. zonder gasmeter).
items = {
    '1-3:0.2.8': 'version', '0-0:1.0.0': 'timestamp', '0-0:96.1.1': 'EmeterId',
    '1-0:1.8.1': 'verbruik-1', '1-0:1.8.2': 'verbruik-2',
    '1-0:2.8.1': 'levering-1', '1-0:2.8.2': 'levering-2',
    '0-0:96.14.0': 'act_tarief', '1-0:1.7.0': 'act_verbruik', '1-0:2.7.0': 'act_levering',
    '0-0:96.7.21': '#powerFail', '0-0:96.7.9': '#powerFailLong', '1-0:99.97.0': 'powerFailLog',
    '1-0:32.32.0': '#Vdip_L1', '1-0:52.32.0': '#Vdip_L2', '1-0:72.32.0': '#Vdip_L3',
    '1-0:32.36.0': '#Vpiek_L1', '1-0:52.36.0': '#Vpiek_L2', '1-0:72.36.0': '#Vpiek_L3',
    '0-0:96.13.0': 'message',
    '1-0:32.7.0': 'actV_L1', '1-0:52.7.0': 'actV_L2', '1-0:72.7.0': 'actV_L3',
    '1-0:31.7.0': 'actA_L1', '1-0:51.7.0': 'actA_L2', '1-0:71.7.0': 'actA_L3',
    '1-0:21.7.0': 'actP+_L1', '1-0:41.7.0': 'actP+_L2', '1-0:61.7.0': 'actP+_L3',
    '1-0:22.7.0': 'actP-_L1', '1-0:42.7.0': 'actP-_L2', '1-0:62.7.0': 'actP-_L3',
    '0-1:24.1.0': 'type', '0-1:96.1.0': 'GmeterId', '0-1:24.2.1': 'gasverbruik'
}


def portsettings() -> serial.Serial:
    ser = serial.Serial()
    ser.baudrate = P1_BAUDRATE
    ser.bytesize = serial.EIGHTBITS
    ser.parity = serial.PARITY_NONE
    ser.stopbits = serial.STOPBITS_ONE
    ser.xonxoff = 0
    ser.rtscts = 0
    ser.timeout = P1_SERIAL_TIMEOUT
    ser.port = P1_SERIAL_PORT
    return ser


def rotate_log(path) -> None:
    if os.path.exists(path) and os.path.getsize(path) > MAX_SIZE:
        os.replace(path, str(path) + ".1")


async def append_text(path, line: str) -> None:
    def _append():
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("a", encoding="utf-8-sig") as f:
            f.write(line)
    await asyncio.to_thread(_append)


async def write_csv(regel: str) -> None:
    async with loadbalance_lock:
        await append_text(LOADBALANCE_CSV, regel)


def safe_decode(raw: bytes) -> str | None:
    try:
        return raw.decode("utf-8")
    except UnicodeDecodeError:
        try:
            return raw.decode("utf-8", errors="ignore")
        except Exception as e:
            logger.error(f"P1 decode failed: {e}")
            return None


def openSerial_sync(ser: serial.Serial) -> None:
    try:
        ser.open()
    except Exception as e:
       logger.error(f"P1 open fout: {e}")


def closeSerial_sync(ser: serial.Serial) -> None:
    try:
        ser.close()
    except Exception as e:
       logger.error(f"P1 close fout: {e}")


async def readTelegram(ser: serial.Serial, items_map: dict):
    try:
        _ = ser.isOpen()
    except Exception as error:
        logger.error(f"p1 port error {error}")
        closeSerial_sync(ser)
        openSerial_sync(ser)

    while True:
        result = await getData(ser, items_map)
        if result:
            yield result


async def getData(ser: serial.Serial, items_map: dict) -> dict | None:
    p1_result = True
    global last_telegram_at

    while p1_result:
        result: dict = {}
        count = 0
        p1_log = True
        p1_telegram = False

        while p1_log:
            try:
                loop = asyncio.get_running_loop()
                p1_raw = await loop.run_in_executor(executor, ser.readline)
            except Exception as e:
               logger.error(f"P1 leesfout: {e}")
               closeSerial_sync(ser)
               await asyncio.sleep(P1_RETRY_DELAY)
               openSerial_sync(ser)
               continue
            p1_str = safe_decode(p1_raw)
            if p1_str is None:
                return None
            p1_line = p1_str.strip()
            if p1_line[0:1] == "/" or p1_line[0:1] == "":
                p1_telegram = True
            elif p1_line[0:1] == "!":
                p1_telegram = False
                p1_log = False
            elif p1_telegram:
                if p1_line == "":
                    continue
                count += 1
                if "(" not in p1_line:
                    continue
                p1_split = p1_line.split("(", 1)
                key = p1_split[0].strip()
                if key not in items_map:
                    continue
                if len(p1_split) < 2:
                    continue
                value = p1_split[1].rstrip(")\r\n").strip()
                if not value:
                    continue
                if ")(" in value:
                    value_split = value.split(")(", 1)
                    value = (value_split[0], value_split[1])
                result[items_map[key]] = value

        if count == P1_EXPECTED_FIELDS:
            p1_result = False
            last_telegram_at = time.monotonic()
            logger.info("telegram gelezen")
        else:
            await asyncio.sleep(P1_RETRY_DELAY)
    return result


def checkAmperage(result: dict) -> list[float]:
    amps: list[float] = []
    for i in range(1, 4):
        factor = 1
        if float(result[f"actP-_L{i}"].split("*")[0]) > 0:
            factor = -1
        number = result[f"actA_L{i}"].split("*")
        amp = round(float(number[0]) * factor, 1)
        amps.append(amp)
    return amps


def checkLevering(result: dict) -> float:
    levering: list[float] = []
    for i in range(1, 4):
        number = result[f"actP-_L{i}"].split("*")
        fase = round(float(number[0]), 1)
        levering.append(fase)
    return round(sum(levering), 1)


def dateConversion(result):
    if not isinstance(result, dict):
        return result

    if "timestamp" not in result:
        logger.error(f"GEEN timestamp in result: {result}")
        return result

    date = result["timestamp"]
    year = int("20"+date[0:2])
    month = int(date[2:4])
    day = int(date[4:6])
    hour = int(date[6:8])
    minute = int(date[8:10])
    sec = int(date[10:12])
    datum = datetime(year, month, day).strftime('%Y-%m-%d')
    tijd = datetime(year, month, day, hour, minute, sec).strftime('%H:%M:%S')
    result['timestamp'] = (str(datum)+" "+str(tijd))
    return result


def loadbalance_sync(result: dict) -> None:
    try:
        with sqlite3.connect(ENERGY_DB) as db:
            cur = db.cursor()
            cur.execute("SELECT * FROM LoadBalance ORDER BY datum DESC LIMIT 1;")
            data = cur.fetchone()

            datum, tijd = result["timestamp"].split(" ")
            if not data or data[0] != datum:
                cur.execute("INSERT INTO LoadBalance (datum, tijd) VALUES (?, ?)", (datum, tijd))
                cur.execute("SELECT * FROM LoadBalance ORDER BY datum DESC LIMIT 1;")
                data = cur.fetchone()

            updates = {}
            counter = 1
            for item in ("actA_L", "actP+_L", "actP-_L", "actV_L"):
                for i in range(1, 4):
                    number = result[item + str(i)].split("*")
                    value = round(float(number[0]), 1)
                    current = data[counter + 1]

                    if current is None or current < value:
                        updates[f"{item}{i}"] = value
                    counter += 1
            if updates:
                updates["tijd"] = tijd
                set_clause = ", ".join([f'"{col}" = ?' for col in updates.keys()])
                params = list(updates.values()) + [datum]
                cur.execute(f"UPDATE LoadBalance SET {set_clause} WHERE datum = ?", params)
            db.commit()

    except Exception as error:
        logger.error(f"loadbalance error {error}")


async def LoadBalance(result: dict) -> None:
    rotate_log(LOADBALANCE_CSV)
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(executor, loadbalance_sync, result)
    regel = ",".join(
        [result["timestamp"]] + [str(round(float(result[item + str(i)].split("*")[0]), 1))
            for item in ("actA_L", "actP+_L", "actP-_L", "actV_L")
            for i in range(1, 4)
        ] + ["\n"])
    await write_csv(regel)


async def read_loop(ser: serial.Serial) -> None:
    async for result in readTelegram(ser, items):
       try:
          set_key("net", {"verbruik": float(result["act_verbruik"].split("*")[0]),
                           "levering": float(result["act_levering"].split("*")[0])}, ENERGY_JSON)
       except (KeyError, ValueError, IndexError) as e:
          logger.error(f"read_loop: onvolledig telegram, overgeslagen ({e})")
          continue

       while not telegram_queue.empty():
          try:
             telegram_queue.get_nowait()
          except asyncio.QueueEmpty:
             break

       await telegram_queue.put(result)


async def energy_manager() -> None:
    global last_energy_cycle_at
    last_hour = None
    while True:
        result = await telegram_queue.get()
        try:
            amps = checkAmperage(result)
            levering = checkLevering(result)
            result = dateConversion(result)

            ts = result["timestamp"]
            hour = ts[:13]

            if hour != last_hour:
                store_hourly_snapshot(result)
                last_hour = hour

            await LoadBalance(result)
            timestamp = result["timestamp"]
            set_key("p1_reader", {"levering": levering, "amps": amps, "timestamp": timestamp}, path=READER_DATA_JSON)
        except (KeyError, ValueError, IndexError) as e:
            logger.error(f"energy_manager: fout bij verwerken telegram, overgeslagen ({e})")
        # Bijgewerkt ongeacht try/except-uitkomst: dit toont dat de loop
        # nog daadwerkelijk doorloopt (in tegenstelling tot last_telegram_at,
        # dat alleen read_loop()'s voortgang meet - deze twee taken kunnen
        # onafhankelijk van elkaar vastlopen).
        last_energy_cycle_at = time.monotonic()
        await asyncio.sleep(P1_POLL_INTERVAL)


async def watchdog_task() -> None:
    while True:
        # Zolang er nog nooit een telegram/energy-cyclus is geweest (bijv.
        # net opgestart), gebruik het opstart-tijdstip als referentie voor
        # de gratieperiode - anders zou de allereerste watchdog-cyclus al
        # als 'stale' gelden.
        telegram_ref = last_telegram_at if last_telegram_at is not None else process_started_at
        energy_ref = last_energy_cycle_at if last_energy_cycle_at is not None else process_started_at

        telegram_age = time.monotonic() - telegram_ref
        energy_age = time.monotonic() - energy_ref

        if telegram_age < P1_STALE_THRESHOLD and energy_age < P1_STALE_THRESHOLD:
            systemd.daemon.notify("WATCHDOG=1")
        else:
            # Losse meldingen, zodat in de log meteen zichtbaar is WELKE
            # van de twee taken is vastgelopen (ze kunnen onafhankelijk
            # van elkaar hangen - read_loop() en energy_manager() delen
            # geen gemeenschappelijke lock/resource in de normale flow).
            if telegram_age >= P1_STALE_THRESHOLD:
                logger.error(
                    f"watchdog: geen compleet telegram gelezen in de laatste "
                    f"{P1_STALE_THRESHOLD}s (read_loop lijkt vastgelopen)"
                )
            if energy_age >= P1_STALE_THRESHOLD:
                logger.error(
                    f"watchdog: energy_manager-cyclus niet voltooid in de laatste "
                    f"{P1_STALE_THRESHOLD}s (energy_manager lijkt vastgelopen)"
                )
            logger.error("watchdog-notify overgeslagen, systemd zal het proces herstarten (mits WatchdogSec is ingesteld)")
        await asyncio.sleep(P1_WATCHDOG_INTERVAL)


def store_hourly_snapshot(result: dict):
    try:
        db = sqlite3.connect(ENERGY_DB)
        cur = db.cursor()

        ts = result["timestamp"]
        hour = ts[:13]

        cur.execute("""
            INSERT INTO HourlyMeter (hour, verbruik1, verbruik2, levering1, levering2)
            VALUES (?, ?, ?, ?, ?)
        """, (
            hour,
            float(result["verbruik-1"].split("*")[0]),
            float(result["verbruik-2"].split("*")[0]),
            float(result["levering-1"].split("*")[0]),
            float(result["levering-2"].split("*")[0])
        ))

        db.commit()
        db.close()

    except Exception as e:
        logger.error(f"Hourly snapshot error: {e}")


async def main(ser: serial.Serial) -> None:
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(executor, openSerial_sync, ser)
    systemd.daemon.notify("READY=1")

    try:
        task_reader = asyncio.create_task(read_loop(ser))
        task_energy = asyncio.create_task(energy_manager())
        task_watchdog = asyncio.create_task(watchdog_task())

        await asyncio.gather(
            asyncio.shield(task_reader),
            asyncio.shield(task_energy),
            asyncio.shield(task_watchdog),
        )
    finally:
       loop = asyncio.get_running_loop()
       await loop.run_in_executor(executor, closeSerial_sync, ser)


if __name__ == "__main__":
    ser = portsettings()
    asyncio.run(main(ser))