I’m trying to add my own methods to the simulated component. I need the current state for the recalculation.
"""Adds a test sensor."""
from datetime import datetime
import math
from random import Random
import random
import voluptuous as vol
import logging
from homeassistant.components.sensor import PLATFORM_SCHEMA, ENTITY_ID_FORMAT
from homeassistant.const import CONF_NAME
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
import homeassistant.util.dt as dt_util
from homeassistant.util import slugify
"""
source modified https://www.home-assistant.io/integrations/simulated
To enable detailed logging for this component, add the following to your configuration.yaml file
logger:
default: warn
logs:
custom_components.sensor.testsensor: debug
Samples:
----------------------------
## RANDOM NUMBER (DEFAULT)
- platform: testsensor
name: "s03"
unit: "Wh"
minval: 0.00
maxval: -22.50
## SIMULATED CALC NUMBER
- platform: testsensor
name: 'power used'
unit: 'kWh'
mean: -24.50
spread: 8
period: 600
## RANDOM NUMBER ADD
- platform: testsensor
name: "energy_display_meter_ht"
mode: "add"
minval: 0.00
maxval: 8.34
unit: "kWh"
"""
_LOGGER = logging.getLogger(__name__)
""" all attributes """
CONF_MAXVAL = "maxval"
CONF_MINVAL = "minval"
CONF_AMP = "amplitude"
CONF_FWHM = "spread"
CONF_MEAN = "mean"
CONF_PERIOD = "period"
CONF_PHASE = "phase"
CONF_SEED = "seed"
CONF_UNIT = "unit"
CONF_MODE = "mode"
CONF_ICON = "icon"
CONF_RELATIVE_TO_EPOCH = "relative_to_epoch"
""" default settings """
DEFAULT_NAME = "testsensor"
DEFAULT_maxval = 0
DEFAULT_minval = 0
DEFAULT_AMP = 1
DEFAULT_FWHM = 0
DEFAULT_MEAN = 0
DEFAULT_PERIOD = 1000
DEFAULT_PHASE = 0
DEFAULT_SEED = 999
DEFAULT_UNIT = "value"
DEFAULT_MODE = "random"
DEFAULT_ICON = "mdi:chart-line"
DEFAULT_RELATIVE_TO_EPOCH = False
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Optional(CONF_MAXVAL, default=DEFAULT_maxval): vol.Coerce(float),
vol.Optional(CONF_MINVAL, default=DEFAULT_minval): vol.Coerce(float),
vol.Optional(CONF_AMP, default=DEFAULT_AMP): vol.Coerce(float),
vol.Optional(CONF_FWHM, default=DEFAULT_FWHM): vol.Coerce(float),
vol.Optional(CONF_MEAN, default=DEFAULT_MEAN): vol.Coerce(float),
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_PERIOD, default=DEFAULT_PERIOD): cv.positive_int,
vol.Optional(CONF_PHASE, default=DEFAULT_PHASE): vol.Coerce(float),
vol.Optional(CONF_SEED, default=DEFAULT_SEED): cv.positive_int,
vol.Optional(CONF_UNIT, default=DEFAULT_UNIT): cv.string,
vol.Optional(CONF_MODE, default=DEFAULT_MODE): cv.string,
vol.Optional(CONF_ICON, default=DEFAULT_ICON): cv.string,
vol.Optional(
CONF_RELATIVE_TO_EPOCH, default=DEFAULT_RELATIVE_TO_EPOCH
): cv.boolean,
}
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the test sensor."""
settings = {
"name": config.get(CONF_NAME),
"icon": config.get(CONF_ICON),
"unit": config.get(CONF_UNIT),
"mode": config.get(CONF_MODE),
"minval": config.get(CONF_MINVAL),
"maxval": config.get(CONF_MAXVAL),
"amp": config.get(CONF_AMP),
"mean": config.get(CONF_MEAN),
"period": config.get(CONF_PERIOD),
"phase": config.get(CONF_PHASE),
"fwhm": config.get(CONF_FWHM),
"seed": config.get(CONF_SEED),
"relative_to_epoch": config.get(CONF_RELATIVE_TO_EPOCH)
}
sensor = TestSensor(hass, settings)
add_entities([sensor], True)
class TestSensor(Entity):
"""Class for test sensor."""
def __init__(self, hass, settings):
"""Init the class."""
## _LOGGER.warning("Creating new TestSensor component for %s", name)
self._hass = hass
self._name = settings["name"]
self._unit = settings["unit"]
self._mode = settings["mode"]
self._icon = settings["icon"]
self._amp = settings["amp"]
self._mean = settings["mean"]
self._period = settings["period"]
self._phase = settings["phase"] # phase in degrees
self._fwhm = settings["fwhm"]
self._seed = settings["seed"]
self._minval = settings["minval"]
self._maxval = settings["maxval"]
self._random = Random(settings["seed"]) # A local seeded Random
self._start_time = (
datetime(1970, 1, 1, tzinfo=dt_util.UTC)
if settings["relative_to_epoch"]
else dt_util.utcnow()
)
self._relative_to_epoch = settings["relative_to_epoch"]
# how do get the previous value from the history ??
self._id = ENTITY_ID_FORMAT.format(slugify(self._name).lower())
self.currentState = 0.00
self.entityFound = False
self._state = None
def time_delta(self):
"""Return the time delta."""
dt0 = self._start_time
dt1 = dt_util.utcnow()
return dt1 - dt0
def simulated_calc(self):
"""Simulated calc mode"""
mean = self._mean
amp = self._amp
time_delta = self.time_delta().total_seconds() * 1e6 # to milliseconds
period = self._period * 1e6 # to milliseconds
fwhm = self._fwhm / 2
phase = math.radians(self._phase)
if period == 0:
periodic = 0
else:
periodic = amp * (math.sin((2 * math.pi * time_delta / period) + phase))
noise = self._random.gauss(mu=0, sigma=fwhm)
return round(mean + periodic + noise, 3)
def random_calc(self):
"""Random calc mode."""
if(self._minval == 0 and self._maxval == 0):
return self.simulated_calc()
minval = self._minval
maxval = self._maxval
return round(random.uniform(minval, maxval), 3)
def cumulative_calc(self):
"""Cumulative calc mode."""
minval = self._minval
maxval = self._maxval
return round(float(self.currentState) + random.uniform(minval, maxval), 3)
def substract_calc(self):
"""Substract calc mode."""
minval = self._minval
maxval = self._maxval
return round(float(self.currentState) - random.uniform(minval, maxval), 3)
async def async_update(self):
"""Update the sensor state value."""
# TODO: how to get the previous value from the history on start up ??
# self.currentState = await async_method_that_returns_state_value()
entity = self._hass.states.get(self._id)
if entity is None:
_LOGGER.debug("Unable to find entity %s", self._id)
self.entityFound = False
else:
self.entityFound = True
self.currentState = round(float(entity.state), 3)
if self._mode == 'simulated':
self._state = self.simulated_calc()
elif self._mode == 'random':
self._state = self.random_calc()
elif self._mode == 'add':
self._state = self.cumulative_calc()
elif self._mode == 'substract':
self._state = self.substract_calc()
else:
self._state = self.random_calc()
# fallback on startup ??
if(self.currentState == 0):
self.currentState = self._state
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def state(self):
"""Return the state of the sensor."""
return self._state
@property
def icon(self):
"""Icon to use in the frontend, if any."""
return self._icon
@property
def unit_of_measurement(self):
"""Return the unit this state is expressed in."""
return self._unit
@property
def device_state_attributes(self):
"""Return other details about the sensor state as informations."""
now = datetime.now()
return {
"mode": self._mode,
"id": self._id,
"value": self.currentState,
"minval": self._minval,
"maxval": self._maxval,
"amplitude": self._amp,
"mean": self._mean,
"period": self._period,
"phase": self._phase,
"spread": self._fwhm,
"seed": self._seed,
"relative_to_epoch": self._relative_to_epoch,
"seed": self._seed,
"device": "testsensor v0.0.1",
"timstamp": now.strftime("%d/%m/%Y %H:%M:%S"),
"attribution": "Data provided by custom component",
"relative_to_epoch": self._relative_to_epoch,
}