jeka/mega_hacs
1
0
Fork 0
mirror of https://github.com/andvikt/mega_hacs.git synced 2026-02-02 23:47:47 +05:00
Code Issues Packages Projects Releases Wiki Activity

fix dimmer attr

Browse Source
This commit is contained in:
Викторов Андрей Германович
2023-10-15 19:23:46 +03:00
parent 39c23a62cf
commit bd98fa216d
1 changed files with 168 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

285
custom_components/mega/entities.py
View File

@@ -12,8 +12,21 @@ from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.helpers.restore_state import RestoreEntity from homeassistant.helpers.restore_state import RestoreEntity
from . import hub as h from . import hub as h
from .const import DOMAIN, CONF_CUSTOM, CONF_INVERT, EVENT_BINARY_SENSOR, LONG, \ from .const import (
LONG_RELEASE, RELEASE, PRESS, SINGLE_CLICK, DOUBLE_CLICK, EVENT_BINARY, CONF_SMOOTH, CONF_RANGE DOMAIN,
CONF_CUSTOM,
CONF_INVERT,
EVENT_BINARY_SENSOR,
LONG,
LONG_RELEASE,
RELEASE,
PRESS,
SINGLE_CLICK,
DOUBLE_CLICK,
EVENT_BINARY,
CONF_SMOOTH,
CONF_RANGE,
)
_events_on = False _events_on = False
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@@ -22,7 +35,7 @@ _LOGGER = logging.getLogger(__name__)
async def _set_events_on(): async def _set_events_on():
global _events_on, _task_set_ev_on global _events_on, _task_set_ev_on
await asyncio.sleep(10) await asyncio.sleep(10)
_LOGGER.debug('events on') _LOGGER.debug("events on")
_events_on = True _events_on = True
@@ -31,6 +44,7 @@ def set_events_off():
_events_on = False _events_on = False
_task_set_ev_on = None _task_set_ev_on = None
_task_set_ev_on = None _task_set_ev_on = None
@@ -40,20 +54,21 @@ class BaseMegaEntity(CoordinatorEntity, RestoreEntity):
Also provides some basic entity information: unique_id, name, availiability Also provides some basic entity information: unique_id, name, availiability
All base entities are polled in order to be online or offline All base entities are polled in order to be online or offline
""" """
def __init__( def __init__(
self, self,
mega: 'h.MegaD', mega: "h.MegaD",
port: typing.Union[int, str, typing.List[int]], port: typing.Union[int, str, typing.List[int]],
config_entry: ConfigEntry = None, config_entry: ConfigEntry = None,
id_suffix=None, id_suffix=None,
name=None, name=None,
unique_id=None, unique_id=None,
http_cmd='get', http_cmd="get",
addr: str=None, addr: str = None,
index=None, index=None,
customize=None, customize=None,
smooth=None, smooth=None,
**kwargs, **kwargs,
): ):
self._smooth = smooth self._smooth = smooth
self.http_cmd = http_cmd self.http_cmd = http_cmd
@@ -65,11 +80,19 @@ class BaseMegaEntity(CoordinatorEntity, RestoreEntity):
self._mega_id = mega.id self._mega_id = mega.id
self._lg = None self._lg = None
if not isinstance(port, list): if not isinstance(port, list):
self._unique_id = unique_id or f"mega_{mega.id}_{port}" + \ self._unique_id = unique_id or f"mega_{mega.id}_{port}" + (
(f"_{id_suffix}" if id_suffix else "") f"_{id_suffix}" if id_suffix else ""
_pt = port if not mega.new_naming else f'{port:02}' if isinstance(port, int) else port )
self._name = name or f"{mega.id}_{_pt}" + \ _pt = (
(f"_{id_suffix}" if id_suffix else "") port
if not mega.new_naming
else f"{port:02}"
if isinstance(port, int)
else port
)
self._name = name or f"{mega.id}_{_pt}" + (
f"_{id_suffix}" if id_suffix else ""
)
self._customize: dict = None self._customize: dict = None
else: else:
assert id_suffix is not None assert id_suffix is not None
@@ -83,7 +106,7 @@ class BaseMegaEntity(CoordinatorEntity, RestoreEntity):
self.addr = addr self.addr = addr
self.id_suffix = id_suffix self.id_suffix = id_suffix
self._can_smooth_hard = None self._can_smooth_hard = None
if self.http_cmd == 'ds2413': if self.http_cmd == "ds2413":
self.mega.ds2413_ports |= {self.port} self.mega.ds2413_ports |= {self.port}
super().__init__(coordinator=mega.updater) super().__init__(coordinator=mega.updater)
@@ -92,12 +115,12 @@ class BaseMegaEntity(CoordinatorEntity, RestoreEntity):
return False return False
def get_attribute(self, name, default=None): def get_attribute(self, name, default=None):
attr = getattr(self, f'_{name}', None) attr = getattr(self, f"_{name}", None)
if attr is None and self._state is not None: if attr is None and self._state is not None:
if name == 'is_on': if name == "is_on":
attr = self._state.state attr = self._state.state
else: else:
attr = self._state.attributes.get(f'{name}', default) attr = self._state.attributes.get(f"{name}", default)
return attr if attr is not None else default return attr if attr is not None else default
@property @property
@@ -118,7 +141,7 @@ class BaseMegaEntity(CoordinatorEntity, RestoreEntity):
@property @property
def enabled(self): def enabled(self):
if '<' in self.name: if "<" in self.name:
return False return False
else: else:
return super().enabled return super().enabled
@@ -130,12 +153,17 @@ class BaseMegaEntity(CoordinatorEntity, RestoreEntity):
if self.hass is None or self.entity_id is None: if self.hass is None or self.entity_id is None:
return {} return {}
if self._customize is None: if self._customize is None:
c_entity_id = self.hass.data.get(DOMAIN, {}).get(CONF_CUSTOM).get('entities', {}).get(self.entity_id, {}) c_entity_id = (
self.hass.data.get(DOMAIN, {})
.get(CONF_CUSTOM)
.get("entities", {})
.get(self.entity_id, {})
)
c = self.hass.data.get(DOMAIN, {}).get(CONF_CUSTOM) or {} c = self.hass.data.get(DOMAIN, {}).get(CONF_CUSTOM) or {}
c = c.get(self._mega_id) or {} c = c.get(self._mega_id) or {}
c = c.get(self.port) or {} c = c.get(self.port) or {}
if self.addr is not None and self.index is not None and isinstance(c, dict): if self.addr is not None and self.index is not None and isinstance(c, dict):
idx = self.addr.lower() + f'_a' if self.index == 0 else '_b' idx = self.addr.lower() + f"_a" if self.index == 0 else "_b"
c = c.get(idx, {}) c = c.get(idx, {})
c.update(c_entity_id) c.update(c_entity_id)
self._customize = c self._customize = c
@@ -147,18 +175,24 @@ class BaseMegaEntity(CoordinatorEntity, RestoreEntity):
if isinstance(self.port, list): if isinstance(self.port, list):
pt_idx = self.id_suffix pt_idx = self.id_suffix
else: else:
_pt = self.port if not self.mega.new_naming else f'{self.port:02}' if isinstance(self.port, int) else self.port _pt = (
if isinstance(_pt, str) and 'e' in _pt: self.port
pt_idx, _ = _pt.split('e') if not self.mega.new_naming
else f"{self.port:02}"
if isinstance(self.port, int)
else self.port
)
if isinstance(_pt, str) and "e" in _pt:
pt_idx, _ = _pt.split("e")
else: else:
pt_idx = _pt pt_idx = _pt
return DeviceInfo( return DeviceInfo(
identifiers={ identifiers={
# Serial numbers are unique identifiers within a specific domain # Serial numbers are unique identifiers within a specific domain
(DOMAIN, f'{self._mega_id}', pt_idx) (DOMAIN, f"{self._mega_id}", pt_idx)
}, },
name=self.name, name=self.name,
manufacturer='ab-log.ru', manufacturer="ab-log.ru",
sw_version=self.mega.fw, sw_version=self.mega.fw,
via_device=(DOMAIN, self._mega_id), via_device=(DOMAIN, self._mega_id),
) )
@@ -176,7 +210,13 @@ class BaseMegaEntity(CoordinatorEntity, RestoreEntity):
c = self.customize.get(CONF_NAME) c = self.customize.get(CONF_NAME)
if not isinstance(c, str): if not isinstance(c, str):
if not isinstance(self.port, list): if not isinstance(self.port, list):
_pt = self.port if not self.mega.new_naming else f'{self.port:02}' if isinstance(self.port, int) else self.port _pt = (
self.port
if not self.mega.new_naming
else f"{self.port:02}"
if isinstance(self.port, int)
else self.port
)
c = self._name or f"{self.mega.id}_p{_pt}" c = self._name or f"{self.mega.id}_p{_pt}"
else: else:
c = self.id_suffix c = self.id_suffix
@@ -192,7 +232,7 @@ class BaseMegaEntity(CoordinatorEntity, RestoreEntity):
self._state = await self.async_get_last_state() self._state = await self.async_get_last_state()
async def get_state(self): async def get_state(self):
self.lg.debug(f'state is %s', self.state) self.lg.debug(f"state is %s", self.state)
self.async_write_ha_state() self.async_write_ha_state()
@@ -212,51 +252,42 @@ class MegaPushEntity(BaseMegaEntity):
if self.hass is None: if self.hass is None:
return return
self.async_write_ha_state() self.async_write_ha_state()
self.lg.debug(f'state after update %s', self.state) self.lg.debug(f"state after update %s", self.state)
if not self.entity_id.startswith('binary_sensor'): if not self.entity_id.startswith("binary_sensor"):
_LOGGER.debug('skip event because not a bnary sens') _LOGGER.debug("skip event because not a bnary sens")
return return
ll: bool = self.mega.last_long.get(self.port, False) ll: bool = self.mega.last_long.get(self.port, False)
if safe_int(value.get('click', 0)) == 1: if safe_int(value.get("click", 0)) == 1:
self.hass.bus.async_fire( self.hass.bus.async_fire(
event_type=EVENT_BINARY, event_type=EVENT_BINARY,
event_data={ event_data={"entity_id": self.entity_id, "type": SINGLE_CLICK},
'entity_id': self.entity_id,
'type': SINGLE_CLICK
}
) )
elif safe_int(value.get('click', 0)) == 2: elif safe_int(value.get("click", 0)) == 2:
self.hass.bus.async_fire( self.hass.bus.async_fire(
event_type=EVENT_BINARY, event_type=EVENT_BINARY,
event_data={ event_data={"entity_id": self.entity_id, "type": DOUBLE_CLICK},
'entity_id': self.entity_id,
'type': DOUBLE_CLICK
}
) )
elif safe_int(value.get('m', 0)) == 2: elif safe_int(value.get("m", 0)) == 2:
self.mega.last_long[self.port] = True self.mega.last_long[self.port] = True
self.hass.bus.async_fire( self.hass.bus.async_fire(
event_type=EVENT_BINARY, event_type=EVENT_BINARY,
event_data={ event_data={"entity_id": self.entity_id, "type": LONG},
'entity_id': self.entity_id,
'type': LONG
}
) )
elif safe_int(value.get('m', 0)) == 1: elif safe_int(value.get("m", 0)) == 1:
self.hass.bus.async_fire( self.hass.bus.async_fire(
event_type=EVENT_BINARY, event_type=EVENT_BINARY,
event_data={ event_data={
'entity_id': self.entity_id, "entity_id": self.entity_id,
'type': LONG_RELEASE if ll else RELEASE, "type": LONG_RELEASE if ll else RELEASE,
} },
) )
elif safe_int(value.get('m', None)) == 0: elif safe_int(value.get("m", None)) == 0:
self.hass.bus.async_fire( self.hass.bus.async_fire(
event_type=EVENT_BINARY, event_type=EVENT_BINARY,
event_data={ event_data={
'entity_id': self.entity_id, "entity_id": self.entity_id,
'type': PRESS, "type": PRESS,
} },
) )
self.mega.last_long[self.port] = False self.mega.last_long[self.port] = False
return return
@@ -266,21 +297,13 @@ class MegaPushEntity(BaseMegaEntity):
class MegaOutPort(MegaPushEntity): class MegaOutPort(MegaPushEntity):
def __init__(self, dimmer=False, dimmer_scale=1, *args, **kwargs):
def __init__( super().__init__(*args, **kwargs)
self,
dimmer=False,
dimmer_scale=1,
*args, **kwargs
):
super().__init__(
*args, **kwargs
)
self._brightness = None self._brightness = None
self._is_on = None self._is_on = None
self.dimmer = dimmer self.dimmer = dimmer
self.dimmer_scale = dimmer_scale self.dimmer_scale = dimmer_scale
self.is_extender = isinstance(self.port, str) and 'e' in self.port self.is_extender = isinstance(self.port, str) and "e" in self.port
self.task: asyncio.Task = None self.task: asyncio.Task = None
self._restore_brightness = None self._restore_brightness = None
self._last_called: float = 0 self._last_called: float = 0
@@ -313,9 +336,14 @@ class MegaOutPort(MegaPushEntity):
return return
val = self.mega.values.get(self.port, {}) val = self.mega.values.get(self.port, {})
if isinstance(val, dict) and len(val) == 0 and self._state is not None: if isinstance(val, dict) and len(val) == 0 and self._state is not None:
ret = safe_int(self._state.attributes.get("brightness"), def_on=self.max_dim, def_off=0, def_val=0) ret = safe_int(
self._state.attributes.get("brightness"),
def_on=self.max_dim,
def_off=0,
def_val=0,
)
ret = self._calc_brightness(ret) ret = self._calc_brightness(ret)
elif isinstance(self.port, str) and 'e' in self.port: elif isinstance(self.port, str) and "e" in self.port:
if isinstance(val, str): if isinstance(val, str):
val = safe_int(val, def_on=self.max_dim, def_off=0, def_val=0) val = safe_int(val, def_on=self.max_dim, def_off=0, def_val=0)
else: else:
@@ -340,46 +368,68 @@ class MegaOutPort(MegaPushEntity):
def is_on(self) -> bool: def is_on(self) -> bool:
val = self.mega.values.get(self.port, {}) val = self.mega.values.get(self.port, {})
if isinstance(val, dict) and len(val) == 0 and self._state is not None: if isinstance(val, dict) and len(val) == 0 and self._state is not None:
return self._state == 'ON' return self._state == "ON"
elif isinstance(self.port, str) and 'e' in self.port and val: elif isinstance(self.port, str) and "e" in self.port and val:
if val is None: if val is None:
return return
if self.dimmer: if hasattr(self, "dimmer") and self.dimmer:
val = safe_int(val) val = safe_int(val)
if val is not None: if val is not None:
return val > 0 if not self.invert else val == 0 return val > 0 if not self.invert else val == 0
else: else:
return val == 'ON' if not self.invert else val == 'OFF' return val == "ON" if not self.invert else val == "OFF"
elif val is not None: elif val is not None:
val = val.get("value") val = val.get("value")
if not isinstance(val, str) and self.index is not None and self.addr is not None: if (
not isinstance(val, str)
and self.index is not None
and self.addr is not None
):
if not isinstance(val, dict): if not isinstance(val, dict):
self.mega.lg.warning(f'{self.entity_id}: {val} is not a dict') self.mega.lg.warning(f"{self.entity_id}: {val} is not a dict")
return return
_val = val.get(self.addr, val.get(self.addr.lower(), val.get(self.addr.upper()))) _val = val.get(
self.addr, val.get(self.addr.lower(), val.get(self.addr.upper()))
)
if not isinstance(_val, str): if not isinstance(_val, str):
self.mega.lg.warning(f'{self.entity_id}: can not get {self.addr} from {val}, recieved {_val}') self.mega.lg.warning(
f"{self.entity_id}: can not get {self.addr} from {val}, recieved {_val}"
)
return return
_val = _val.split('/') _val = _val.split("/")
if len(_val) >= 2: if len(_val) >= 2:
self.mega.lg.debug('%s parsed values: %s[%s]="%s"', self.entity_id, _val, self.index, _val) self.mega.lg.debug(
'%s parsed values: %s[%s]="%s"',
self.entity_id,
_val,
self.index,
_val,
)
val = _val[self.index] val = _val[self.index]
else: else:
self.mega.lg.warning(f'{self.entity_id}: {_val} has wrong length') self.mega.lg.warning(f"{self.entity_id}: {_val} has wrong length")
return return
elif self.index is not None and self.addr is None: elif self.index is not None and self.addr is None:
self.mega.lg.warning(f'{self.entity_id} does not has addr') self.mega.lg.warning(f"{self.entity_id} does not has addr")
return return
self.mega.lg.debug('%s.state = %s', self.entity_id, val) self.mega.lg.debug("%s.state = %s", self.entity_id, val)
if not self.invert: if not self.invert:
return val == 'ON' or str(val) == '1' or (safe_int(val) is not None and safe_int(val) > 0) return (
val == "ON"
or str(val) == "1"
or (safe_int(val) is not None and safe_int(val) > 0)
)
else: else:
return val == 'OFF' or str(val) == '0' or (safe_int(val) is not None and safe_int(val) == 0) return (
val == "OFF"
or str(val) == "0"
or (safe_int(val) is not None and safe_int(val) == 0)
)
@property @property
def cmd_port(self): def cmd_port(self):
if self.index is not None: if self.index is not None:
return f'{self.port}A' if self.index == 0 else f'{self.port}B' return f"{self.port}A" if self.index == 0 else f"{self.port}B"
else: else:
return self.port return self.port
@@ -400,9 +450,7 @@ class MegaOutPort(MegaPushEntity):
if isinstance(self.port, str): if isinstance(self.port, str):
self.mega.values[self.port] = value[0] self.mega.values[self.port] = value[0]
else: else:
self.mega.values[self.port] = { self.mega.values[self.port] = {"value": value[0]}
'value': value[0]
}
if update_state: if update_state:
self.async_write_ha_state() self.async_write_ha_state()
@@ -412,13 +460,15 @@ class MegaOutPort(MegaPushEntity):
tm = (self.smooth.total_seconds() * pct) if transition is None else transition tm = (self.smooth.total_seconds() * pct) if transition is None else transition
if self.task is not None: if self.task is not None:
self.task.cancel() self.task.cancel()
self.task = asyncio.create_task(self.mega.smooth_dim( self.task = asyncio.create_task(
(self.cmd_port, from_, to_), self.mega.smooth_dim(
time=tm, (self.cmd_port, from_, to_),
can_smooth_hardware=self.can_smooth_hardware, time=tm,
max_values=[255 if self.dimmer_scale == 1 else 4095], can_smooth_hardware=self.can_smooth_hardware,
updater=partial(self.update_from_smooth, update_state=update_state), max_values=[255 if self.dimmer_scale == 1 else 4095],
)) updater=partial(self.update_from_smooth, update_state=update_state),
)
)
def _calc_brightness(self, brightness): def _calc_brightness(self, brightness):
if brightness is None: if brightness is None:
@@ -454,9 +504,9 @@ class MegaOutPort(MegaPushEntity):
brightness = self._calc_brightness(brightness) brightness = self._calc_brightness(brightness)
_prev = safe_int(self.brightness) or 0 _prev = safe_int(self.brightness) or 0
self._brightness = brightness self._brightness = brightness
if self.dimmer and brightness == 0: if hasattr(self, "dimmer") and self.dimmer and brightness == 0:
cmd = self.max_dim cmd = self.max_dim
elif self.dimmer: elif hasattr(self, "dimmer") and self.dimmer:
cmd = min((brightness * self.dimmer_scale, self.max_dim)) cmd = min((brightness * self.dimmer_scale, self.max_dim))
if self.smooth_dim or transition: if self.smooth_dim or transition:
self._set_dim_brightness(from_=_prev, to_=cmd, transition=transition) self._set_dim_brightness(from_=_prev, to_=cmd, transition=transition)
@@ -471,7 +521,7 @@ class MegaOutPort(MegaPushEntity):
"cnt": round(transition / (abs(_prev - brightness) / 255)), "cnt": round(transition / (abs(_prev - brightness) / 255)),
} }
if self.addr: if self.addr:
_cmd['addr'] = self.addr _cmd["addr"] = self.addr
if not (self.smooth_dim or transition): if not (self.smooth_dim or transition):
await self.mega.request(**_cmd, priority=-1) await self.mega.request(**_cmd, priority=-1)
if self.index is not None: if self.index is not None:
@@ -480,29 +530,31 @@ class MegaOutPort(MegaPushEntity):
port=self.port, port=self.port,
force_http=True, force_http=True,
conv=False, conv=False,
http_cmd='list', http_cmd="list",
) )
elif isinstance(self.port, str) and 'e' in self.port: elif isinstance(self.port, str) and "e" in self.port:
if not self.dimmer: if not self.dimmer:
self.mega.values[self.port] = 'ON' if not self.invert else 'OFF' self.mega.values[self.port] = "ON" if not self.invert else "OFF"
else: else:
self.mega.values[self.port] = cmd self.mega.values[self.port] = cmd
else: else:
self.mega.values[self.port] = {'value': cmd} self.mega.values[self.port] = {"value": cmd}
await self.get_state() await self.get_state()
async def async_turn_off(self, transition=None, **kwargs) -> None: async def async_turn_off(self, transition=None, **kwargs) -> None:
if (time.time() - self._last_called) < 0.1: if (time.time() - self._last_called) < 0.1:
return return
self._last_called = time.time() self._last_called = time.time()
self._restore_brightness = self._cal_reverse_brightness(safe_int(self._brightness)) self._restore_brightness = self._cal_reverse_brightness(
safe_int(self._brightness)
)
if not self.dimmer: if not self.dimmer:
transition = None transition = None
cmd = "0" if not self.invert else "1" cmd = "0" if not self.invert else "1"
_cmd = {"cmd": f"{self.cmd_port}:{cmd}"} _cmd = {"cmd": f"{self.cmd_port}:{cmd}"}
_prev = safe_int(self.brightness) or 0 _prev = safe_int(self.brightness) or 0
if self.addr: if self.addr:
_cmd['addr'] = self.addr _cmd["addr"] = self.addr
if not (self.smooth_dim or transition): if not (self.smooth_dim or transition):
await self.mega.request(**_cmd, priority=-1) await self.mega.request(**_cmd, priority=-1)
else: else:
@@ -517,12 +569,12 @@ class MegaOutPort(MegaPushEntity):
port=self.port, port=self.port,
force_http=True, force_http=True,
conv=False, conv=False,
http_cmd='list', http_cmd="list",
) )
elif isinstance(self.port, str) and 'e' in self.port: elif isinstance(self.port, str) and "e" in self.port:
self.mega.values[self.port] = 'OFF' if not self.invert else 'ON' self.mega.values[self.port] = "OFF" if not self.invert else "ON"
else: else:
self.mega.values[self.port] = {'value': cmd} self.mega.values[self.port] = {"value": cmd}
await self.get_state() await self.get_state()
async def async_will_remove_from_hass(self) -> None: async def async_will_remove_from_hass(self) -> None:
@@ -530,13 +582,12 @@ class MegaOutPort(MegaPushEntity):
self.task.cancel() self.task.cancel()
def safe_int(v, def_on=1, def_off=0, def_val=None): def safe_int(v, def_on=1, def_off=0, def_val=None):
if v == 'ON': if v == "ON":
return def_on return def_on
elif v == 'OFF': elif v == "OFF":
return def_off return def_off
try: try:
return int(v) return int(v)
except (ValueError, TypeError): except (ValueError, TypeError):
return def_val return def_val