pyHeatpump/pyheatpump/models/variable.py

from pyheatpump.db import RowClass
from pyheatpump.db import DB
from datetime import date

from .variable_type import VariableType

from pyheatpump.logger import logger_init
logger = logger_init()

class Variable(RowClass):
    address: int = None
    unit: str = None
    last_update: date = None
    type: VariableType = None

    def __init__(self, **kwargs):
        if 'type' in kwargs.keys() and type(kwargs['type']) != VariableType:
            kwargs['type'] = VariableType.get(kwargs['type'])

        super().__init__(**kwargs)

    @property
    def type_slabel(self):
        return self.type.slabel

    def insert(self):

        try:
            DB.sql(
            """INSERT INTO variable
            (type, address)
            VALUES
            (:type_slabel, :address)
            """, {
                'type_slabel': self.type_slabel,
                'address': self.address
            })
            return True
        except Exception as e:
            logger.error('Variable.insert: %s', e)
            return False


    def exists(self):
        try:
            return bool(next(DB.sql(
            """SELECT 1 FROM variable
            WHERE type=:type AND address=:address
            """, {
                'type': self.type.slabel,
                'address': self.address
            })))
        except StopIteration:
            return False

    @staticmethod
    def getall():
        return {
            row.slabel: Variable.getall_of_type(row)
            for _, row in VariableType.getall().items()
        }


    @staticmethod
    def getall_of_type(var_type: VariableType) -> dict:
        return {
            row['address']: Variable(**dict(row))
            for row in DB.sql(
            """SELECT * FROM variable
            WHERE type = :slabel
                AND address >= :start_address
                AND address <= :end_address""", {

                'slabel': var_type.slabel,
                'start_address': var_type.start_address,
                'end_address': var_type.end_address
            })
       }


    @staticmethod
    def getall_values_of_type(var_type: VariableType) -> dict:
        return Variable.getall_values_of_type_since(var_type, 0)


    @staticmethod
    def getall_values_of_type_since(var_type: VariableType, since: int) -> dict:
        """
        floatcast = lambda x: round(float(x) / 10, 2)
        cast_fct = floatcast if type.slabel == 'A' else lambda x: x
        """
        if not var_type.cast():
            # Should not happen
            return {}

        params = {
            'slabel': var_type.slabel,
            'start_address': var_type.start_address,
            'end_address': var_type.end_address,
            'since': since
        }

        return {
            row['address']: var_type.cast()(row['value'])

            for row in DB.sql(
            """SELECT var.address as address, val.value as value
            FROM variable var
            LEFT JOIN var_value val ON
                var.type = val.type
                AND var.address = val.address
                AND var.last_update = val.time
            WHERE
                var.type = :slabel
                AND var.address >= :start_address
                AND var.address <= :end_address
                AND var.last_update > :since""",
            params)

            if 'address' in row.keys() and 'value' in row.keys() and
                row['value'] is not None
        }



    def modbus_update(self):
        if self.type == 'A':
            pass
        elif self.type == 'I':
            pass
        elif self.type == 'D':
            pass