vincent/koolnova-BMS-Integration
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add koolnova modbus api and operations classes

Browse Source
This commit is contained in:
vbenoit
2023-11-01 09:06:26 +01:00
parent d26324306e
commit 67b1181ee7
9 changed files with 837 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/koolnova_bms/koolnova/__init__.py
View File
custom_components/koolnova_bms/koolnova/__pycache__/__init__.cpython-39.pyc
BIN
View File
Binary file not shown.
custom_components/koolnova_bms/koolnova/__pycache__/const.cpython-39.pyc
BIN
View File
Binary file not shown.
custom_components/koolnova_bms/koolnova/__pycache__/device.cpython-39.pyc
BIN
View File
Binary file not shown.
custom_components/koolnova_bms/koolnova/__pycache__/operations.cpython-39.pyc
BIN
View File
Binary file not shown.
custom_components/koolnova_bms/koolnova/app.py
View File
custom_components/koolnova_bms/koolnova/const.py
+115
View File
@@ -0,0 +1,115 @@
# Constants for koolnova Modbus RTU BMS
# d'après la documentation officielle A52102 - Registre contrôle Modbus
from enum import Enum
# La couche physique est Modbus RTU sur RS485 à 9600, avec 8 bits de données, sans parité
# ou même parité et un bit d'arrêt. Par défaut: 9600 8E1
# L'adresse Modbus par défaut est 49
DEFAULT_ADDR = 49
DEFAULT_BAUDRATE = 9600
DEFAULT_PARITY = 'E'
DEFAULT_STOPBITS = 1
DEFAULT_BYTESIZE = 8
# Nombre de machines (AC1, AC2, AC3 et AC4)
NUM_OF_ENGINES = 4
# Nombre de registres par zone
NUM_REG_PER_ZONE = 4
# Nombre max de zone pour un systeme
NB_ZONE_MAX = 16
# Chaque zone climatique est définie par 4 registres et il y a 16 zones possibles,
# donc le climat est défini par 64 registres
REG_START_ZONE = 0
# Commandes Modbus prises en charge sont le Read Holding Register (0x03)
# et le Write Single Register (0x06)
REG_LOCK_ZONE = 0 # 40001, 40005, 40009, etc ...
REG_STATE_AND_FLOW = 1 # 40002, 40006, 40010, etc ...
class ZoneState(Enum):
STATE_OFF = 0
STATE_ON = 1
class ZoneRegister(Enum):
REGISTER_OFF = 0
REGISTER_ON = 1
class ZoneFanMode(Enum):
FAN_OFF = 0
FAN_LOW = 1
FAN_MEDIUM = 2
FAN_HIGH = 3
FAN_AUTO = 4
class ZoneClimMode(Enum):
COLD = 1
HOT = 2
HEATING_FLOOR = 4
REFRESHING_FLOOR = 5
HEATING_FLOOR_2 = 6
# Température de consigne = (data / 2) => delta: 15°C -> 35°C
REG_TEMP_ORDER = 2 # 40003, 40007, 40011, etc ...
# Température réelle = (data / 2) => delta: 0°C -> 50°C
REG_TEMP_REAL = 3 # 40004, 40008, 40012, etc ...
# Temperature maximale de consigne : 35°C
MAX_TEMP_ORDER = 35.0
# Temperature minimale de consigne : 15°C
MIN_TEMP_ORDER = 15.0
# (4 registres: 64 -> 67) Debit des machines (0: arret -> 15: debit maximum)
REG_START_FLOW_ENGINE = 64
NUM_REG_FLOG_ENGINE = 4
FLOW_ENGINE_VAL_MAX = 15
FLOW_ENGINE_VAL_MIN = 0
# (4 registres : 68 -> 71) Température de consigne de la machine.
# Valeur décimale de 30 à 60 = double de la temp de consigne des consignes AC1, AC2, AC3 et AC4
REG_START_ORDER_TEMP = 68
NUM_REG_ORDER_TEMP = 4
# (4 registres : 72 -> 75) Programmation des débit des machines du système
REG_START_FLOW_STATE_ENGINE = 72
NUM_REG_FLOW_STATE_ENGINE = 4
class FlowEngine(Enum):
MANUAL_MIN = 1
MANUAL_MED = 2
MANUAL_HIGH = 3
AUTO = 4
# Communication Modbus
REG_COMM = 76
REG_ADDR_MODBUS = 77 # dispo (1 - 127)
REG_EFFICIENCY = 78
# Point d'equilibre entre efficience/vitesse du système de zone
# chiffre élevé = meilleur efficience
# chiffre bas = temp réglée atteinte au plus tot
class Efficiency(Enum):
LOWER_EFF = 1
LOW_EFF = 2
MED_EFF = 3
HIGH_EFF = 4
HIGHER_EFF = 5
REG_CLIM_ID = 79
REG_SYS_STATE = 80
class SysState(Enum):
SYS_STATE_OFF = 0
SYS_STATE_ON = 1
REG_GLOBAL_MODE = 81
class GlobalMode(Enum):
COLD = 1
HEAT = 2
HEATING_FLOOR = 4
REFRESHING_FLOOR = 5
HEATING_FLOOR_2 = 6
custom_components/koolnova_bms/koolnova/device.py
+414
View File
@@ -0,0 +1,414 @@
#!/usr/bin/env python3
# -*- encoding: utf-8 -*-
# @author: sinseman44 <vincent.benoit@benserv.fr>
# @date: 10/2023
# @brief: System, Unit and Zone classes
###########################################################################
# import external modules
import re, sys, os
import logging as log
import asyncio
###########################################################################
# import internal modules
from . import const
from .operations import Operations, ModbusConnexionError
###########################################################################
# class & methods
logger = log.getLogger('koolnova_bms')
class Koolnova:
''' koolnova Device class '''
def __init__(self,
port:str = "",
timeout:int = 0,
) -> None:
''' Class constructor '''
self._client = Operations(port=port, timeout=timeout)
self._global_mode = const.GlobalMode.COLD
self._efficiency = const.Efficiency.LOWER_EFF
self._sys_state = const.SysState.SYS_STATE_OFF
self._units = []
self._zones = []
async def connect(self) -> bool:
''' connect to the modbus serial server '''
await self._client.connect()
if not self.connected():
raise ClientNotConnectedError("Client Modbus connexion error")
logger.info("Retreive system status ...")
ret, self._sys_state = await self._client.system_status()
if not ret:
logger.error("Error retreiving system status")
self._sys_state = const.SysState.SYS_STATE_OFF
logger.info("Retreive global mode ...")
ret, self._global_mode = await self._client.global_mode()
if not ret:
logger.error("Error retreiving global mode")
self._global_mode = const.GlobalMode.COLD
logger.info("Retreive efficiency ...")
ret, self._efficiency = await self._client.efficiency()
if not ret:
logger.error("Error retreiving efficiency")
self._efficiency = const.Efficiency.LOWER_EFF
await asyncio.sleep(0.5)
logger.info("Retreive units ...")
for idx in range(1, const.NUM_OF_ENGINES + 1):
logger.debug("Unit id: {}".format(idx))
unit = Unit(unit_id = idx)
ret, unit.flow_engine = await self._client.flow_engine(unit_id = idx)
ret, unit.flow_state = await self._client.flow_state_engine(unit_id = idx)
ret, unit.order_temp = await self._client.order_temp_engine(unit_id = idx)
self._units.append(unit)
await asyncio.sleep(0.5)
return True
def connected(self) -> bool:
''' get modbus client status '''
return self._client.connected
def disconnect(self) -> None:
''' close the underlying socket connection '''
self._client.disconnect()
async def discover_zones(self) -> None:
''' Set all registered zones for system '''
if not self._client.connected:
raise ModbusConnexionError('Client Modbus not connected')
zones_lst = await self._client.discover_registered_zones()
for zone in zones_lst:
self._zones.append(Zone(id_zone = zone['id'],
state = zone['state'],
register = zone['register'],
fan_mode = zone['fan'],
clim_mode = zone['clim'],
real_temp = zone['real_temp'],
order_temp = zone['order_temp']
))
return
async def add_manual_registered_zone(self,
id_zone:int = 0) -> bool:
''' Add zone to koolnova System '''
if not self._client.connected:
raise ModbusConnexionError('Client Modbus not connected')
ret, zone_dict = await self._client.zone_registered(zone_id = id_zone)
if not ret:
logger.error("Zone with ID: {} is not registered".format(id_zone))
return False
for zone in self._zones:
if id_zone == zone.id_zone:
logger.error('Zone registered with ID: {} is already saved')
return False
self._zones.append(Zone(id_zone = id_zone,
state = zone_dict['state'],
register = zone_dict['register'],
fan_mode = zone_dict['fan'],
clim_mode = zone_dict['clim'],
real_temp = zone_dict['real_temp'],
order_temp = zone_dict['order_temp']
))
logger.debug("Zones registered: {}".format(self._zones))
return True
def get_zones(self) -> list:
''' get zones '''
return self._zones
def get_zone(self, zone_id:int = 0) -> str:
''' get specific zone '''
return self._zones[zone_id - 1]
def get_units(self) -> list:
''' get units '''
return self._units
@property
def global_mode(self) -> const.GlobalMode:
''' Get Global Mode '''
return self._global_mode
@global_mode.setter
def global_mode(self, val:const.GlobalMode) -> None:
''' Set Global Mode '''
if not isinstance(val, const.GlobalMode):
raise AssertionError('Input variable must be Enum GlobalMode')
self._global_mode = val
@property
def efficiency(self) -> const.Efficiency:
''' Get Efficiency '''
return self._efficiency
@efficiency.setter
def efficiency(self, val:const.Efficiency) -> None:
''' Set Efficiency '''
if not isinstance(val, const.Efficiency):
raise AssertionError('Input variable must be Enum Efficiency')
self._efficiency = val
@property
def sys_state(self) -> const.SysState:
''' Get System State '''
return self.sys_state
@sys_state.setter
def sys_state(self, val:const.SysState) -> None:
''' Set System State '''
if not isinstance(val, const.SysState):
raise AssertionError('Input variable must be Enum SysState')
self._sys_state = val
def __repr__(self) -> str:
''' repr method '''
return repr('System(Global Mode:{}, Efficiency:{}, State:{})'.format( self._global_mode,self._efficiency,self._sys_state))
class Unit:
''' koolnova Unit class '''
def __init__(self,
unit_id:int = 0,
flow_engine:int = 0,
flow_state:const.FlowEngine = const.FlowEngine.AUTO,
order_temp:float = 0
) -> None:
''' Constructor class '''
self._unit_id = unit_id
self._flow_engine = flow_engine
self._flow_state = flow_state
self._order_temp = order_temp
@property
def unit_id(self) -> int:
''' Get Unit ID '''
return self._unit_id
@unit_id.setter
def unit_id(self, val:int) -> None:
''' Set Unit ID '''
if not isinstance(val, int):
raise AssertionError('Input variable must be Int')
if val > const.NUM_OF_ENGINES:
raise NumUnitError('Unit ID must be lower than {}'.format(const.NUM_OF_ENGINES))
self._unit_id = val
@property
def flow_engine(self) -> int:
''' Get Flow Engine '''
return self._flow_engine
@flow_engine.setter
def flow_engine(self, val:int) -> None:
''' Set Flow Engine '''
if not isinstance(val, int):
raise AssertionError('Input variable must be Int')
if val > const.FLOW_ENGINE_VAL_MAX or val < const.FLOW_ENGINE_VAL_MIN:
raise FlowEngineError('Flow Engine value ({}) must be between {} and {}'.format(val, const.FLOW_ENGINE_VAL_MIN, const.FLOW_ENGINE_VAL_MAX))
self._flow_engine = val
@property
def flow_state(self) -> const.FlowEngine:
''' Get Flow State '''
return self._flow_state
@flow_state.setter
def flow_state(self, val:const.FlowEngine) -> None:
''' Set Flow State '''
if not isinstance(val, const.FlowEngine):
raise AssertionError('Input variable must be Enum FlowEngine')
self._flow_state = val
@property
def order_temp(self) -> float:
''' Get Order Temp '''
return self._order_temp
@order_temp.setter
def order_temp(self, val:float = 0.0) -> None:
''' Set Flow Engine '''
if not isinstance(val, float):
raise AssertionError('Input variable must be Int')
if val > 0 and (val > 30.0 or val < 15.0):
raise OrderTempError('Flow Engine value ({}) must be between 15 and 30'.format(val))
self._flow_engine = val
def __repr__(self) -> str:
''' repr method '''
return repr('Unit(Id:{}, Flow Engine:{}, Flow State:{}, Order Temp:{})'.format(self._unit_id,
self._flow_engine,
self._flow_state,
self._order_temp))
class Zone:
''' koolnova Zone class '''
def __init__(self,
id_zone:int = 0,
state:const.ZoneState = const.ZoneState.STATE_OFF,
register:const.ZoneRegister = const.ZoneRegister.REGISTER_OFF,
fan_mode:const.ZoneFanMode = const.ZoneFanMode.FAN_OFF,
clim_mode:const.ZoneClimMode = const.ZoneClimMode.COLD,
real_temp:float = 0,
order_temp:float = 0
) -> None:
''' Class constructor '''
self._id = id_zone
self._state = state
self._register = register
self._fan_mode = fan_mode
self._clim_mode = clim_mode
self._real_temp = real_temp
self._order_temp = order_temp
@property
def id_zone(self) -> int:
''' Get Zone Id '''
return self._id
@property
def state(self) -> const.ZoneState:
''' Get state '''
return self._state
@state.setter
def state(self, val:const.ZoneState) -> None:
''' Set state '''
if not isinstance(val, const.ZoneState):
raise AssertionError('Input variable must be Enum ZoneState')
self._state = val
@property
def register(self) -> const.ZoneRegister:
''' Get register state '''
return self._register
@register.setter
def register(self, val:const.ZoneRegister) -> None:
''' Set register state '''
if not isinstance(val, const.ZoneRegister):
raise AssertionError('Input variable must be Enum ZoneRegister')
self._register = val
@property
def fan_mode(self) -> const.ZoneFanMode:
''' Get Fan Mode '''
return self._fan_mode
@fan_mode.setter
def fan_mode(self, val:const.ZoneFanMode) -> None:
''' Set Fan Mode '''
if not isinstance(val, const.ZoneFanMode):
raise AssertionError('Input variable must be Enum ZoneFanMode')
self._fan_mode = val
@property
def clim_mode(self) -> const.ZoneClimMode:
''' Get Clim Mode '''
return self._clim_mode
@clim_mode.setter
def clim_mode(self, val:const.ZoneClimMode) -> None:
''' Set Clim Mode '''
if not isinstance(val, const.ZoneClimMode):
raise AssertionError('Input variable must be Enum ZoneClimMode')
self._clim_mode = val
@property
def real_temp(self) -> float:
''' Get real temp '''
return self._real_temp
@real_temp.setter
def real_temp(self, val:float) -> None:
''' Set Real Temp '''
if not isinstance(val, float):
raise AssertionError('Input variable must be Float')
self._real_temp = val
@property
def order_temp(self) -> float:
''' Get order temp '''
return self._order_temp
@order_temp.setter
def order_temp(self, val:float) -> None:
''' Set Order Temp '''
if not isinstance(val, float):
raise AssertionError('Input variable must be float')
if val > const.MAX_TEMP_ORDER or val < const.MIN_TEMP_ORDER:
raise OrderTempError('Order temp value must be between {} and {}'.format(const.MIN_TEMP_ORDER, const.MAX_TEMP_ORDER))
self._order_temp = val
def __repr__(self) -> str:
''' repr method '''
return repr('Zone(Id:{}, State:{}, Register:{}, Fan:{}, Clim:{}, Real Temp:{}, Order Temp:{})'.format(self._id,
self._state,
self._register,
self._fan_mode,
self._clim_mode,
self._real_temp,
self._order_temp))
class NumUnitError(Exception):
''' user defined exception '''
def __init__(self,
msg:str = "") -> None:
''' Class Constructor '''
self._msg = msg
def __str__(self):
''' print the message '''
return self._msg
class FlowEngineError(Exception):
''' user defined exception '''
def __init__(self,
msg:str = "") -> None:
''' Class Constructor '''
self._msg = msg
def __str__(self):
''' print the message '''
return self._msg
class OrderTempError(Exception):
''' user defined exception '''
def __init__(self,
msg:str = "") -> None:
''' Class Constructor '''
self._msg = msg
def __str__(self):
''' print the message '''
return self._msg
class ClientNotConnectedError(Exception):
''' user defined exception '''
def __init__(self,
msg:str = "") -> None:
''' Class Constructor '''
self._msg = msg
def __str__(self):
''' print the message '''
return self._msg
custom_components/koolnova_bms/koolnova/operations.py
+308
View File
@@ -0,0 +1,308 @@
#!/usr/bin/env python3
# -*- encoding: utf-8 -*-
# @author: sinseman44 <vincent.benoit@benserv.fr>
# @date: 10/2023
# @brief: Communicate with Koolnova BMS Modbus RTU
###########################################################################
# import external modules
import re, sys, os
import logging as log
import asyncio
from pymodbus import pymodbus_apply_logging_config
from pymodbus.client import AsyncModbusSerialClient as ModbusClient
from pymodbus.exceptions import ModbusException
from pymodbus.pdu import ExceptionResponse
from pymodbus.transaction import ModbusRtuFramer
###########################################################################
# import internal modules
from . import const
###########################################################################
# class & methods
logger = log.getLogger('koolnova_bms')
class Operations:
''' koolnova BMS Modbus operations class '''
def __init__(self, port:str, timeout:int) -> None:
''' Class constructor '''
self._port = port
self._timeout = timeout
self._client = ModbusClient(port=self._port,
baudrate=const.DEFAULT_BAUDRATE,
parity=const.DEFAULT_PARITY,
stopbits=const.DEFAULT_STOPBITS,
bytesize=const.DEFAULT_BYTESIZE,
timeout=self._timeout)
pymodbus_apply_logging_config("DEBUG")
async def __read_register(self, reg:int) -> (int, bool):
''' Read one holding register (code 0x03) '''
ret = True
rr = None
if not self._client.connected:
raise ModbusConnexionError('Client Modbus not connected')
try:
rr = await self._client.read_holding_registers(address=reg, count=1, slave=const.DEFAULT_ADDR)
if rr.isError():
ret = False
except ModbusException as e:
logger.error("{}".format(e))
ret = False
if isinstance(rr, ExceptionResponse):
logger.error("Received modbus exception ({})".format(rr))
ret = False
return rr.registers[0], ret
async def __read_registers(self, start_reg:int, count:int) -> (int, bool):
''' Read holding registers (code 0x03) '''
ret = True
rr = None
if not self._client.connected:
raise ModbusConnexionError('Client Modbus not connected')
try:
rr = await self._client.read_holding_registers(address=start_reg, count=count, slave=const.DEFAULT_ADDR)
if rr.isError():
ret = False
except ModbusException as e:
logger.error("{}".format(e))
ret = False
if isinstance(rr, ExceptionResponse):
logger.error("Received modbus exception ({})".format(rr))
ret = False
return rr.registers, ret
async def __write_register(self, reg:int, val:int) -> bool:
''' Write one register (code 0x06) '''
rq = None
ret = True
if not self._client.connected:
raise ModbusConnexionError('Client Modbus not connected')
try:
rq = await self._client.write_register(address=reg, value=val, slave=const.DEFAULT_ADDR)
if rq.isError():
logger.error("Write error: {}".format(rq))
ret = False
except ModbusException as e:
logger.error("{}".format(e))
ret = False
if isinstance(rq, ExceptionResponse):
logger.error("Received modbus exception ({})".format(rr))
ret = False
return ret
async def connect(self) -> None:
''' connect to the modbus serial server '''
await self._client.connect()
def connected(self) -> bool:
''' get modbus client status '''
return self._client.connected
def disconnect(self) -> None:
''' close the underlying socket connection '''
self._client.close()
async def discover_registered_zones(self) -> list:
''' Discover all zones registered to the system '''
regs, ret = await self.__read_registers(start_reg=const.REG_START_ZONE, count=const.NB_ZONE_MAX * const.NUM_REG_PER_ZONE)
if not ret:
raise ReadRegistersError("Read holding regsiter error")
zones_lst = []
zone_dict = {}
jdx = 1
flag = False
for idx, reg in enumerate(regs):
if idx % const.NUM_REG_PER_ZONE == 0:
zone_dict = {}
if const.ZoneRegister(reg >> 1) == const.ZoneRegister.REGISTER_ON:
zone_dict['id'] = jdx
zone_dict['state'] = const.ZoneState(reg & 0b01)
zone_dict['register'] = const.ZoneRegister(reg >> 1)
flag = True
elif idx % const.NUM_REG_PER_ZONE == 1 and flag:
zone_dict['fan'] = const.ZoneFanMode((reg & 0xF0) >> 4)
zone_dict['clim'] = const.ZoneClimMode(reg & 0x0F)
elif idx % const.NUM_REG_PER_ZONE == 2 and flag:
zone_dict['order_temp'] = reg/2
elif idx % const.NUM_REG_PER_ZONE == 3 and flag:
zone_dict['real_temp'] = reg/2
jdx += 1
flag = False
zones_lst.append(zone_dict)
return zones_lst
async def zone_registered(self, zone_id:int = 0) -> (bool, dict):
''' Get Zone Status from Id '''
if zone_id > const.NB_ZONE_MAX or zone_id == 0:
raise ZoneIdError('Zone Id must be between 1 to 16')
zone_dict = {}
regs, ret = await self.__read_registers(start_reg = const.REG_START_ZONE + (4 * (zone_id - 1)), count = const.NUM_REG_PER_ZONE)
if not ret:
raise ReadRegistersError("Read holding regsiter error")
if const.ZoneRegister(regs[0] >> 1) == const.ZoneRegister.REGISTER_OFF:
return False, {}
zone_dict['state'] = const.ZoneState(regs[0] & 0b01)
zone_dict['register'] = const.ZoneRegister(regs[0] >> 1)
zone_dict['fan'] = const.ZoneFanMode((regs[1] & 0xF0) >> 4)
zone_dict['clim'] = const.ZoneClimMode(regs[1] & 0x0F)
zone_dict['order_temp'] = regs[2]/2
zone_dict['real_temp'] = regs[3]/2
return True, zone_dict
async def system_status(self) -> (bool, const.SysState):
''' Read system status register '''
reg, ret = await self.__read_register(const.REG_SYS_STATE)
if not ret:
logger.error('Error retreive system status')
reg = 0
return ret, const.SysState(reg)
async def global_mode(self) -> (bool, const.GlobalMode):
''' Read global mode '''
reg, ret = await self.__read_register(const.REG_GLOBAL_MODE)
if not ret:
logger.error('Error retreive global mode')
reg = 0
return ret, const.GlobalMode(reg)
async def efficiency(self) -> (bool, const.Efficiency):
''' read efficiency/speed '''
reg, ret = await self.__read_register(const.REG_EFFICIENCY)
if not ret:
logger.error('Error retreive efficiency')
reg = 0
return ret, const.Efficiency(reg)
async def flow_engines(self) -> (bool, list):
''' read flow engines AC1, AC2, AC3, AC4 '''
engines_lst = []
regs, ret = await self.__read_registers(const.REG_START_FLOW_ENGINE, const.NUM_OF_ENGINES)
if ret:
for idx, reg in enumerate(regs):
engines_lst.append(const.FlowEngine(reg))
else:
logger.error('Error retreive flow engines')
return ret, engines_lst
async def flow_engine(self, unit_id:int = 0) -> (bool, int):
''' read flow unit specified by unit id '''
if unit_id < 1 or unit_id > 4:
raise UnitIdError("Unit Id must be between 1 and 4")
reg, ret = await self.__read_register(const.REG_START_FLOW_ENGINE + (unit_id - 1))
if not ret:
logger.error('Error retreive flow engine for id:{}'.format(unit_id))
reg = 0
return ret, reg
async def flow_state_engine(self, unit_id:int = 0) -> (bool, const.FlowEngine):
if unit_id < 1 or unit_id > 4:
raise UnitIdError("Unit Id must be between 1 and 4")
reg, ret = await self.__read_register(const.REG_START_FLOW_STATE_ENGINE + (unit_id - 1))
if not ret:
logger.error('Error retreive flow state for id:{}'.format(unit_id))
reg = 0
return ret, const.FlowEngine(reg)
async def order_temp_engine(self, unit_id:int = 0) -> (bool, float):
if unit_id < 1 or unit_id > 4:
raise UnitIdError("Unit Id must be between 1 and 4")
reg, ret = await self.__read_register(const.REG_START_ORDER_TEMP + (unit_id - 1))
if not ret:
logger.error('Error retreive order temp for id:{}'.format(unit_id))
reg = 0
return ret, reg / 2
async def orders_temp(self) -> (bool, list):
''' read orders temperature AC1, AC2, AC3, AC4 '''
engines_lst = []
regs, ret = await self.__read_registers(const.REG_START_ORDER_TEMP, const.NUM_OF_ENGINES)
if ret:
for idx, reg in enumerate(regs):
engines_lst.append(reg/2)
else:
logger.error('error reading flow engines registers')
return ret, engines_lst
async def set_zone_order_temp(self, zone_id:int = 0, val:float = 0.0) -> bool:
''' Set zone order temp '''
if zone_id > const.NB_ZONE_MAX or zone_id == 0:
raise ZoneIdError('Zone Id must be between 1 to 16')
if val > const.MAX_TEMP_ORDER or val < const.MIN_TEMP_ORDER:
logger.error('Order Temperature must be between {} and {}'.format(const.MIN_TEMP_ORDER, const.MAX_TEMP_ORDER))
return False
ret = await self.__write_register(reg = const.REG_START_ZONE + (4 * (zone_id - 1)) + const.REG_TEMP_ORDER, val = int(val * 2))
if not ret:
logger.error('Error writing zone order temperature')
return ret
@property
def port(self) -> str:
''' Get Port '''
return self._port
@property
def timeout(self) -> int:
''' Get Timeout '''
return self._timeout
class ModbusConnexionError(Exception):
''' user defined exception '''
def __init__(self,
msg:str = "") -> None:
''' Class Constructor '''
self._msg = msg
def __str__(self):
''' print the message '''
return self._msg
class ReadRegistersError(Exception):
''' user defined exception '''
def __init__(self,
msg:str = "") -> None:
''' Class Constructor '''
self._msg = msg
def __str__(self):
''' print the message '''
return self._msg
class ZoneIdError(Exception):
''' user defined exception '''
def __init__(self,
msg:str = "") -> None:
''' Class Constructor '''
self._msg = msg
def __str__(self):
''' print the message '''
return self._msg
class UnitIdError(Exception):
''' user defined exception '''
def __init__(self,
msg:str = "") -> None:
''' Class Constructor '''
self._msg = msg
def __str__(self):
''' print the message '''
return self._msg