Projet de remplacement du "RPiPasserelle" d'Otec.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

variable.py 3.5KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132
  1. from pyheatpump.db import RowClass
  2. from pyheatpump.db import DB
  3. from datetime import date
  4. from .variable_type import VariableType
  5. from pyheatpump.logger import logger_init
  6. logger = logger_init()
  7. class Variable(RowClass):
  8. address: int = None
  9. unit: str = None
  10. last_update: date = None
  11. type: VariableType = None
  12. def __init__(self, **kwargs):
  13. if 'type' in kwargs.keys() and type(kwargs['type']) != VariableType:
  14. kwargs['type'] = VariableType.get(kwargs['type'])
  15. super().__init__(**kwargs)
  16. @property
  17. def type_slabel(self):
  18. return self.type.slabel
  19. def insert(self):
  20. try:
  21. DB.sql(
  22. """INSERT INTO variable
  23. (type, address)
  24. VALUES
  25. (:type_slabel, :address)
  26. """, {
  27. 'type_slabel': self.type_slabel,
  28. 'address': self.address
  29. })
  30. return True
  31. except Exception as e:
  32. print(e)
  33. return False
  34. def exists(self):
  35. try:
  36. return bool(next(DB.sql(
  37. """SELECT 1 FROM variable
  38. WHERE type=:type AND address=:address
  39. """, {
  40. 'type': self.type.slabel,
  41. 'address': self.address
  42. })))
  43. except StopIteration:
  44. return False
  45. @staticmethod
  46. def getall():
  47. return {
  48. row.slabel: Variable.getall_of_type(row)
  49. for _, row in VariableType.getall().items()
  50. }
  51. @staticmethod
  52. def getall_of_type(var_type: VariableType) -> dict:
  53. return {
  54. row['address']: Variable(**dict(row))
  55. for row in DB.sql(
  56. """SELECT * FROM variable
  57. WHERE type = :slabel
  58. AND address >= :start_address
  59. AND address <= :end_address""", {
  60. 'slabel': var_type.slabel,
  61. 'start_address': var_type.start_address,
  62. 'end_address': var_type.end_address
  63. })
  64. }
  65. @staticmethod
  66. def getall_values_of_type(var_type: VariableType) -> dict:
  67. return Variable.getall_values_of_type_since(var_type, 0)
  68. @staticmethod
  69. def getall_values_of_type_since(var_type: VariableType, since: int) -> dict:
  70. """
  71. floatcast = lambda x: round(float(x) / 10, 2)
  72. cast_fct = floatcast if type.slabel == 'A' else lambda x: x
  73. """
  74. if not var_type.cast():
  75. # Should not happen
  76. return {}
  77. params = {
  78. 'slabel': var_type.slabel,
  79. 'start_address': var_type.start_address,
  80. 'end_address': var_type.end_address,
  81. 'since': since
  82. }
  83. return {
  84. row['address']: var_type.cast()(row['value'])
  85. for row in DB.sql(
  86. """SELECT var.address as address, val.value as value
  87. FROM variable var
  88. LEFT JOIN var_value val ON
  89. var.type = val.type
  90. AND var.address = val.address
  91. AND var.last_update = val.time
  92. WHERE
  93. var.type = :slabel
  94. AND var.address >= :start_address
  95. AND var.address <= :end_address
  96. AND var.last_update > :since""",
  97. params)
  98. if 'address' in row.keys() and 'value' in row.keys() and
  99. row['value'] is not None
  100. }
  101. def modbus_update(self):
  102. if self.type == 'A':
  103. pass
  104. elif self.type == 'I':
  105. pass
  106. elif self.type == 'D':
  107. pass