Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
123 changes: 123 additions & 0 deletions custom_components/SMioplus/binary_sensor.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,123 @@
DEFAULT_ICONS = {
"on": "mdi:numeric-1",
"off": "mdi:numeric-0",
}

import logging
import time
import types
import inspect
from inspect import signature

from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.helpers.entity import generate_entity_id
from homeassistant.helpers.event import async_track_time_interval
from datetime import timedelta

from . import (
DOMAIN, CONF_STACK, CONF_TYPE, CONF_CHAN, CONF_NAME,
CONF_UPDATE_INTERVAL,
SM_MAP, SM_API
)
SM_MAP = SM_MAP["binary_sensor"]

_LOGGER = logging.getLogger(__name__)

async def async_setup_platform(hass, config, add_devices, discovery_info=None):
# We want this platform to be setup via discovery
if discovery_info == None:
return
add_devices([BinarySensor(
hass=hass,
name=discovery_info.get(CONF_NAME, ""),
stack=discovery_info.get(CONF_STACK, 0),
type=discovery_info.get(CONF_TYPE),
chan=discovery_info.get(CONF_CHAN),
update_interval=discovery_info.get(CONF_UPDATE_INTERVAL) or 30,
)])

class BinarySensor(BinarySensorEntity):
def __init__(self, name, stack, type, chan, hass, update_interval):
generated_name = DOMAIN + str(stack) + "_" + type + "_" + str(chan)
self._unique_id = generate_entity_id("binary_sensor.{}", generated_name, hass=hass)
self._name = name or generated_name
self._stack = int(stack)
self._type = type
self._chan = int(chan)
self._update_interval = float(update_interval)
self._short_timeout = .05
self._icons = DEFAULT_ICONS | SM_MAP[self._type].get("icon", {})
self._icon = self._icons["off"]
self._uom = SM_MAP[self._type].get("uom", "")
self._value = 0
self._remove_hooks = []
self.__SM__init()
### __CUSTOM_SETUP__ START
### __CUSTOM_SETUP__ END

def __SM__init(self):
com = SM_MAP[self._type]["com"]
self._SM = SM_API
if inspect.isclass(self._SM):
self._SM = self._SM(self._stack)
self._SM_get = getattr(self._SM, com["get"])
if len(signature(self._SM_get).parameters) == 0:
def _aux2_SM_get(self, _):
return getattr(self, com["get"])()
self._SM_get = types.MethodType(_aux2_SM_get, self._SM)
else:
_SM_get = getattr(self._SM, com["get"])
if len(signature(_SM_get).parameters) == 1:
def _aux3_SM_get(_, *args):
return _SM_get(self._stack, *args)
self._SM_get = _aux3_SM_get
else:
def _aux_SM_get(*args):
return _SM_get(self._stack, *args)
self._SM_get = _aux_SM_get

async def async_added_to_hass(self):
new_hook = async_track_time_interval(
self.hass, self.async_update_ha_state, timedelta(seconds=self._update_interval)
)
self._remove_hooks.append(new_hook)

async def async_will_remove_from_hass(self):
for remove_hook in self._remove_hooks:
remove_hook()

@property
def should_poll(self):
return False

def update(self):
time.sleep(self._short_timeout)
try:
self._value = self._SM_get(self._chan)
except Exception as ex:
_LOGGER.error(DOMAIN + " %s update() failed, %e, %s, %s", self._type, ex, str(self._stack), str(self._chan))
return
if self._value != 0:
self._icon = self._icons["on"]
else:
self._icon = self._icons["off"]

@property
def unique_id(self):
return self._unique_id

@property
def name(self):
return self._name

@property
def icon(self):
return self._icon

@property
def native_unit_of_measurement(self):
return self._uom

@property
def is_on(self):
return self._value
7 changes: 3 additions & 4 deletions custom_components/SMioplus/data.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,17 +14,16 @@
},
}
},
"sensor": {
"binary_sensor": {
"opto": {
"chan_no": 8,
"uom": "",
"com": {
"get": "getOptoCh",
},
"icon": {
"on": "mdi:numeric-1",
}
},
},
"sensor": {
"opto_cnt": {
"chan_no": 8,
"uom": "",
Expand Down