697 lines
26 KiB
Python
697 lines
26 KiB
Python
""" local API to manage system, engines and areas """
|
|
|
|
import re, sys, os
|
|
import logging as log
|
|
import asyncio
|
|
|
|
from homeassistant.helpers.entity import DeviceInfo
|
|
from ..const import DOMAIN
|
|
|
|
from . import const
|
|
from .operations import Operations, ModbusConnexionError
|
|
|
|
_LOGGER = log.getLogger(__name__)
|
|
|
|
class Area:
|
|
''' koolnova Area class '''
|
|
|
|
def __init__(self,
|
|
name:str = "",
|
|
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.OFF,
|
|
real_temp:float = 0,
|
|
order_temp:float = 0
|
|
) -> None:
|
|
''' Class constructor '''
|
|
self._name = name
|
|
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 name(self) -> str:
|
|
''' Get area name '''
|
|
return self._name
|
|
|
|
@name.setter
|
|
def name(self, name:str) -> None:
|
|
''' Set area name '''
|
|
if not isinstance(name, str):
|
|
raise AssertionError('Input variable must be a string')
|
|
self._name = name
|
|
|
|
@property
|
|
def id_zone(self) -> int:
|
|
''' Get area 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('Area(Name: {}, Id:{}, State:{}, Register:{}, Fan:{}, Clim:{}, Real Temp:{}, Order Temp:{})'.format(
|
|
self._name,
|
|
self._id,
|
|
self._state,
|
|
self._register,
|
|
self._fan_mode,
|
|
self._clim_mode,
|
|
self._real_temp,
|
|
self._order_temp))
|
|
|
|
class Engine:
|
|
''' koolnova Engine class '''
|
|
|
|
def __init__(self,
|
|
engine_id:int = 0,
|
|
throughput:int = 0,
|
|
state:const.FlowEngine = const.FlowEngine.AUTO,
|
|
order_temp:float = 0
|
|
) -> None:
|
|
''' Constructor class '''
|
|
self._engine_id = engine_id
|
|
self._throughput = throughput
|
|
self._state = state
|
|
self._order_temp = order_temp
|
|
|
|
@property
|
|
def engine_id(self) -> int:
|
|
''' Get Engine ID '''
|
|
return self._engine_id
|
|
|
|
@engine_id.setter
|
|
def engine_id(self, val:int) -> None:
|
|
''' Set Engine ID '''
|
|
if not isinstance(val, int):
|
|
raise AssertionError('Input variable must be Int')
|
|
if val > const.NUM_OF_ENGINES:
|
|
raise NumUnitError('Engine ID must be lower than {}'.format(const.NUM_OF_ENGINES))
|
|
self._engine_id = val
|
|
|
|
@property
|
|
def throughput(self) -> int:
|
|
''' Get throughput Engine '''
|
|
return self._throughput
|
|
|
|
@throughput.setter
|
|
def throughput(self, val:int) -> None:
|
|
''' Set throughput 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('throughput engine value ({}) must be between {} and {}'.format(val,
|
|
const.FLOW_ENGINE_VAL_MIN,
|
|
const.FLOW_ENGINE_VAL_MAX))
|
|
self._throughput = val
|
|
|
|
@property
|
|
def state(self) -> const.FlowEngine:
|
|
''' Get Engine state '''
|
|
return self._state
|
|
|
|
@state.setter
|
|
def state(self, val:const.FlowEngine) -> None:
|
|
''' Set Engine state '''
|
|
if not isinstance(val, const.FlowEngine):
|
|
raise AssertionError('Input variable must be Enum FlowEngine')
|
|
self._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 order temp 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._order_temp = val
|
|
|
|
def __repr__(self) -> str:
|
|
''' repr method '''
|
|
return repr('Unit(Id:{}, Throughput:{}, State:{}, Order Temp:{})'.format(self._engine_id,
|
|
self._throughput,
|
|
self._state,
|
|
self._order_temp))
|
|
|
|
class Koolnova:
|
|
''' koolnova Device class '''
|
|
|
|
def __init__(self,
|
|
name:str = "",
|
|
port:str = "",
|
|
addr:int = const.DEFAULT_ADDR,
|
|
baudrate:int = const.DEFAULT_BAUDRATE,
|
|
parity:str = const.DEFAULT_PARITY,
|
|
bytesize:int = const.DEFAULT_BYTESIZE,
|
|
stopbits:int = const.DEFAULT_STOPBITS,
|
|
timeout:int = 1) -> None:
|
|
''' Class constructor '''
|
|
self._client = Operations(port=port,
|
|
addr=addr,
|
|
baudrate=baudrate,
|
|
parity=parity,
|
|
bytesize=bytesize,
|
|
stopbits=stopbits,
|
|
timeout=timeout)
|
|
self._name = name
|
|
self._global_mode = const.GlobalMode.COLD
|
|
self._efficiency = const.Efficiency.LOWER_EFF
|
|
self._sys_state = const.SysState.SYS_STATE_OFF
|
|
self._engines = []
|
|
self._areas = []
|
|
|
|
def _area_defined(self,
|
|
id_search:int = 0,
|
|
) -> (bool, int):
|
|
""" test if area id is defined """
|
|
_areas_found = [idx for idx, x in enumerate(self._areas) if x.id_zone == id_search]
|
|
_idx = 0
|
|
if not _areas_found:
|
|
_LOGGER.error("Area id ({}) not defined".format(id_search))
|
|
return False, _idx
|
|
elif len(_areas_found) > 1:
|
|
_LOGGER.error("Multiple Area with same id ({})".format(id_search))
|
|
return False, _idx
|
|
else:
|
|
_idx = _areas_found[0]
|
|
_LOGGER.debug("idx found: {}".format(_idx))
|
|
return True, _idx
|
|
|
|
async def async_update(self) -> bool:
|
|
''' update values from modbus '''
|
|
_LOGGER.debug("Retreive system status ...")
|
|
ret, self._sys_state = await self._client.async_system_status()
|
|
if not ret:
|
|
_LOGGER.error("Error retreiving system status")
|
|
self._sys_state = const.SysState.SYS_STATE_OFF
|
|
|
|
_LOGGER.debug("Retreive global mode ...")
|
|
ret, self._global_mode = await self._client.async_global_mode()
|
|
if not ret:
|
|
_LOGGER.error("Error retreiving global mode")
|
|
self._global_mode = const.GlobalMode.COLD
|
|
|
|
_LOGGER.debug("Retreive efficiency ...")
|
|
ret, self._efficiency = await self._client.async_efficiency()
|
|
if not ret:
|
|
_LOGGER.error("Error retreiving efficiency")
|
|
self._efficiency = const.Efficiency.LOWER_EFF
|
|
|
|
await asyncio.sleep(0.1)
|
|
|
|
_LOGGER.debug("Retreive engines ...")
|
|
for idx in range(1, const.NUM_OF_ENGINES + 1):
|
|
engine = Engine(engine_id = idx)
|
|
ret, engine.throughput = await self._client.async_engine_throughput(engine_id = idx)
|
|
ret, engine.state = await self._client.async_engine_state(engine_id = idx)
|
|
ret, engine.order_temp = await self._client.async_engine_order_temp(engine_id = idx)
|
|
self._engines.append(engine)
|
|
await asyncio.sleep(0.1)
|
|
return True
|
|
|
|
async def async_connect(self) -> bool:
|
|
''' connect to the modbus serial server '''
|
|
ret = True
|
|
await self._client.async_connect()
|
|
if not self.connected():
|
|
ret = False
|
|
raise ClientNotConnectedError("Client Modbus connexion error")
|
|
|
|
#_LOGGER.info("Update system values")
|
|
#ret = await self.update()
|
|
|
|
return ret
|
|
|
|
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 async_discover_areas(self) -> None:
|
|
''' Set all registered areas for system '''
|
|
if not self._client.connected:
|
|
raise ModbusConnexionError('Client Modbus not connected')
|
|
zones_lst = await self._client.async_discover_registered_areas()
|
|
for zone in zones_lst:
|
|
self._areas.append(Area(name = zone['name'],
|
|
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 async_add_manual_registered_area(self,
|
|
name:str = "",
|
|
id_zone:int = 0) -> bool:
|
|
''' Add manual area to koolnova system '''
|
|
if not self._client.connected:
|
|
raise ModbusConnexionError('Client Modbus not connected')
|
|
|
|
ret, zone_dict = await self._client.async_area_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._areas:
|
|
if id_zone == zone.id_zone:
|
|
_LOGGER.error('Zone registered with ID: {} is already saved')
|
|
return False
|
|
|
|
self._areas.append(Area(name = name,
|
|
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("Areas registered: {}".format(self._areas))
|
|
return True
|
|
|
|
@property
|
|
def areas(self) -> list:
|
|
''' get areas '''
|
|
return self._areas
|
|
|
|
def get_area(self, zone_id:int = 0) -> Area:
|
|
''' get specific area '''
|
|
return self._areas[zone_id - 1]
|
|
|
|
async def async_update_area(self, zone_id:int = 0) -> bool:
|
|
""" update specific area from zone_id """
|
|
ret, infos = await self._client.async_area_registered(zone_id = zone_id)
|
|
if not ret:
|
|
_LOGGER.error("Error retreiving area ({}) values".format(zone_id))
|
|
return ret, None
|
|
for idx, area in enumerate(self._areas):
|
|
if area.id_zone == zone_id:
|
|
# update areas list values from modbus response
|
|
self._areas[idx].state = infos['state']
|
|
self._areas[idx].register = infos['register']
|
|
self._areas[idx].fan_mode = infos['fan']
|
|
self._areas[idx].clim_mode = infos['clim']
|
|
self._areas[idx].real_temp = infos['real_temp']
|
|
self._areas[idx].order_temp = infos['order_temp']
|
|
break
|
|
return ret, self._areas[zone_id - 1]
|
|
|
|
async def async_update_all_areas(self) -> list:
|
|
""" update all areas registered and all engines values """
|
|
##### Areas
|
|
_ret, _vals = await self._client.async_areas_registered()
|
|
if not _ret:
|
|
_LOGGER.error("Error retreiving areas values")
|
|
return None
|
|
else:
|
|
for k,v in _vals.items():
|
|
for _idx, _area in enumerate(self._areas):
|
|
if k == _area.id_zone:
|
|
# update areas list values from modbus response
|
|
self._areas[_idx].state = v['state']
|
|
self._areas[_idx].register = v['register']
|
|
self._areas[_idx].fan_mode = v['fan']
|
|
self._areas[_idx].clim_mode = v['clim']
|
|
self._areas[_idx].real_temp = v['real_temp']
|
|
self._areas[_idx].order_temp = v['order_temp']
|
|
|
|
##### Engines
|
|
for _idx in range(1, const.NUM_OF_ENGINES + 1):
|
|
ret, self._engines[_idx - 1].throughput = await self._client.async_engine_throughput(engine_id = _idx)
|
|
ret, self._engines[_idx - 1].state = await self._client.async_engine_state(engine_id = _idx)
|
|
ret, self._engines[_idx - 1].order_temp = await self._client.async_engine_order_temp(engine_id = _idx)
|
|
|
|
##### Global mode
|
|
ret, self._global_mode = await self._client.async_global_mode()
|
|
if not ret:
|
|
_LOGGER.error("Error retreiving global mode")
|
|
self._global_mode = const.GlobalMode.COLD
|
|
|
|
##### Efficiency
|
|
ret, self._efficiency = await self._client.async_efficiency()
|
|
if not ret:
|
|
_LOGGER.error("Error retreiving efficiency")
|
|
self._efficiency = const.Efficiency.LOWER_EFF
|
|
|
|
##### Sys state
|
|
ret, self._sys_state = await self._client.async_system_status()
|
|
if not ret:
|
|
_LOGGER.error("Error retreiving system status")
|
|
self._sys_state = const.SysState.SYS_STATE_OFF
|
|
|
|
return {"areas": self._areas,
|
|
"engines": self._engines,
|
|
"glob": self._global_mode,
|
|
"eff": self._efficiency,
|
|
"sys": self._sys_state}
|
|
|
|
@property
|
|
def engines(self) -> list:
|
|
''' get engines '''
|
|
return self._engines
|
|
|
|
@property
|
|
def device_info(self) -> DeviceInfo:
|
|
""" Return a device description for device registry """
|
|
return {
|
|
"name": self._name,
|
|
"manufacturer": "Koolnova",
|
|
"identifiers": {(DOMAIN, "deadbeef")},
|
|
}
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
''' Get name '''
|
|
return self._name
|
|
|
|
async def async_set_engine_state(self,
|
|
val:const.FlowEngine,
|
|
engine_id: int,
|
|
) -> None:
|
|
''' set engine flow from id '''
|
|
_LOGGER.debug("set engine (id:{}) flow : {}".format(engine_id, val))
|
|
if not isinstance(val, const.FlowEngine):
|
|
raise AssertionError('Input variable must be Enum FlowEngine')
|
|
ret = await self._client.async_set_engine_state(engine_id, val)
|
|
if not ret:
|
|
_LOGGER.error("[GLOBAL] Error writing {} to modbus".format(val))
|
|
raise UpdateValueError('Error writing to modbus updated value')
|
|
self._engines[engine_id - 1].state = val
|
|
|
|
@property
|
|
def global_mode(self) -> const.GlobalMode:
|
|
''' Get Global Mode '''
|
|
return self._global_mode
|
|
|
|
async def async_set_global_mode(self,
|
|
val:const.GlobalMode,
|
|
) -> None:
|
|
''' Set Global Mode '''
|
|
_LOGGER.debug("set global mode : {}".format(val))
|
|
if not isinstance(val, const.GlobalMode):
|
|
raise AssertionError('Input variable must be Enum GlobalMode')
|
|
ret = await self._client.async_set_global_mode(val)
|
|
if not ret:
|
|
_LOGGER.error("[GLOBAL] Error writing {} to modbus".format(val))
|
|
raise UpdateValueError('Error writing to modbus updated value')
|
|
self._global_mode = val
|
|
|
|
@property
|
|
def efficiency(self) -> const.Efficiency:
|
|
''' Get Efficiency '''
|
|
return self._efficiency
|
|
|
|
async def async_set_efficiency(self,
|
|
val:const.Efficiency,
|
|
) -> None:
|
|
''' Set Efficiency '''
|
|
_LOGGER.debug("set efficiency : {}".format(val))
|
|
if not isinstance(val, const.Efficiency):
|
|
raise AssertionError('Input variable must be Enum Efficiency')
|
|
ret = await self._client.async_set_efficiency(val)
|
|
if not ret:
|
|
_LOGGER.error("[EFF] Error writing {} to modbus".format(val))
|
|
raise UpdateValueError('Error writing to modbus updated value')
|
|
self._efficiency = val
|
|
|
|
@property
|
|
def sys_state(self) -> const.SysState:
|
|
''' Get System State '''
|
|
return self._sys_state
|
|
|
|
async def async_set_sys_state(self,
|
|
val:const.SysState,
|
|
) -> None:
|
|
''' Set System State '''
|
|
if not isinstance(val, const.SysState):
|
|
raise AssertionError('Input variable must be Enum SysState')
|
|
_LOGGER.debug("set system state : {}".format(val))
|
|
ret = await self._client.async_set_system_status(val)
|
|
if not ret:
|
|
_LOGGER.error("[SYS_STATE] Error writing {} to modbus".format(val))
|
|
raise UpdateValueError('Error writing to modbus updated value')
|
|
self._sys_state = val
|
|
|
|
async def async_get_area_temp(self,
|
|
zone_id:int,
|
|
) -> float:
|
|
""" get current temp of specific Area """
|
|
_ret, _idx = self._area_defined(id_search = zone_id)
|
|
if not _ret:
|
|
_LOGGER.error("Area not defined ...")
|
|
return False
|
|
|
|
ret, temp = await self._client.async_area_temp(id_zone = zone_id)
|
|
if not ret:
|
|
_LOGGER.error("Error reading temp for area with ID: {}".format(zone_id))
|
|
return False
|
|
self._areas[_idx].real_temp = temp
|
|
return temp
|
|
|
|
async def async_set_area_target_temp(self,
|
|
zone_id:int,
|
|
temp:float,
|
|
) -> bool:
|
|
""" set target temp of specific area """
|
|
_ret, _idx = self._area_defined(id_search = zone_id)
|
|
if not _ret:
|
|
_LOGGER.error("Area not defined ...")
|
|
return False
|
|
|
|
ret = await self._client.async_set_area_target_temp(zone_id = zone_id, val = temp)
|
|
if not ret:
|
|
_LOGGER.error("Error writing target temp for area with ID: {}".format(zone_id))
|
|
return False
|
|
self._areas[_idx].order_temp = temp
|
|
return True
|
|
|
|
async def async_get_area_target_temp(self,
|
|
zone_id:int,
|
|
) -> float:
|
|
""" get target temp of specific area """
|
|
_ret, _idx = self._area_defined(id_search = zone_id)
|
|
if not _ret:
|
|
_LOGGER.error("Area not defined ...")
|
|
return False
|
|
|
|
ret, temp = await self._client.async_area_target_temp(id_zone = zone_id)
|
|
if not ret:
|
|
_LOGGER.error("Error reading target temp for area with ID: {}".format(zone_id))
|
|
return 0.0
|
|
self._areas[_idx].order_temp = temp
|
|
return temp
|
|
|
|
async def async_set_area_clim_mode(self,
|
|
zone_id:int,
|
|
mode:const.ZoneClimMode,
|
|
) -> bool:
|
|
""" set climate mode for specific area """
|
|
_ret, _idx = self._area_defined(id_search = zone_id)
|
|
if not _ret:
|
|
_LOGGER.error("Area not defined ...")
|
|
return False
|
|
|
|
if mode == const.ZoneClimMode.OFF:
|
|
_LOGGER.debug("Set area state to OFF")
|
|
ret = await self._client.async_set_area_state(id_zone = zone_id, val = const.ZoneState.STATE_OFF)
|
|
if not ret:
|
|
_LOGGER.error("Error writing area state for area with ID: {}".format(zone_id))
|
|
return False
|
|
self._areas[_idx].state = const.ZoneState.STATE_OFF
|
|
else:
|
|
if self._areas[_idx].state == const.ZoneState.STATE_OFF:
|
|
_LOGGER.debug("Set area state to ON")
|
|
# update area state
|
|
ret = await self._client.async_set_area_state(id_zone = zone_id, val = const.ZoneState.STATE_ON)
|
|
if not ret:
|
|
_LOGGER.error("Error writing area state for area with ID: {}".format(zone_id))
|
|
return False
|
|
_LOGGER.debug("clim mode ? {}".format(mode))
|
|
# update clim mode
|
|
ret = await self._client.async_set_area_clim_mode(id_zone = zone_id, val = mode)
|
|
if not ret:
|
|
_LOGGER.error("Error writing climate mode for area with ID: {}".format(zone_id))
|
|
return False
|
|
self._areas[_idx].clim_mode = mode
|
|
return True
|
|
|
|
async def async_set_area_fan_mode(self,
|
|
zone_id:int,
|
|
mode:const.ZoneFanMode,
|
|
) -> bool:
|
|
""" set fan mode for specific area """
|
|
# test if area id is defined
|
|
_ret, _idx = self._area_defined(id_search = zone_id)
|
|
if not _ret:
|
|
_LOGGER.error("Area not defined ...")
|
|
return False
|
|
|
|
if self._areas[_idx].state == const.ZoneState.STATE_OFF:
|
|
_LOGGER.warning("Area state is off, cannot change fan speed ...")
|
|
return False
|
|
else:
|
|
_LOGGER.debug("fan mode ? {}".format(mode))
|
|
# writing new value to modbus
|
|
ret = await self._client.async_set_area_fan_mode(id_zone = zone_id, val = mode)
|
|
if not ret:
|
|
_LOGGER.error("Error writing fan mode for area with ID: {}".format(zone_id))
|
|
return False
|
|
# update fan mode in list for specific area
|
|
self._areas[_idx].fan_mode = mode
|
|
return True
|
|
|
|
def __repr__(self) -> str:
|
|
''' repr method '''
|
|
return repr('System(Global Mode:{}, Efficiency:{}, State:{})'.format(
|
|
self._global_mode,
|
|
self._efficiency,
|
|
self._sys_state))
|
|
|
|
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
|
|
|
|
class UpdateValueError(Exception):
|
|
''' user defined exception '''
|
|
|
|
def __init__(self,
|
|
msg:str = "") -> None:
|
|
''' Class Constructor '''
|
|
self._msg = msg
|
|
|
|
def __str__(self):
|
|
''' print the message '''
|
|
return self._msg |