mirror of
https://github.com/basnijholt/adaptive-lighting.git
synced 2026-09-27 20:34:20 +02:00
Merge pull request #107 from basnijholt/styling
Rewrite and simplify code
This commit is contained in:
commit
7289957ce8
4 changed files with 579 additions and 546 deletions
463
custom_components/circadian_lighting/__init__.py
Normal file → Executable file
463
custom_components/circadian_lighting/__init__.py
Normal file → Executable file
|
|
@ -28,289 +28,304 @@ Technical notes: I had to make a lot of assumptions when writing this app
|
|||
"""
|
||||
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
|
||||
import astral
|
||||
import voluptuous as vol
|
||||
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.components.light import (
|
||||
VALID_TRANSITION, ATTR_TRANSITION)
|
||||
from homeassistant.components.light import ATTR_TRANSITION, VALID_TRANSITION
|
||||
from homeassistant.const import (
|
||||
CONF_LATITUDE, CONF_LONGITUDE, CONF_ELEVATION,
|
||||
SUN_EVENT_SUNRISE, SUN_EVENT_SUNSET)
|
||||
from homeassistant.util import Throttle
|
||||
CONF_ELEVATION,
|
||||
CONF_LATITUDE,
|
||||
CONF_LONGITUDE,
|
||||
SUN_EVENT_SUNRISE,
|
||||
SUN_EVENT_SUNSET,
|
||||
)
|
||||
from homeassistant.helpers.discovery import load_platform
|
||||
from homeassistant.helpers.dispatcher import dispatcher_send
|
||||
from homeassistant.helpers.event import track_sunrise, track_sunset, track_time_change
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
||||
from homeassistant.helpers.event import (
|
||||
async_track_sunrise,
|
||||
async_track_sunset,
|
||||
async_track_time_change,
|
||||
async_track_time_interval,
|
||||
)
|
||||
from homeassistant.util.color import (
|
||||
color_temperature_to_rgb, color_RGB_to_xy,
|
||||
color_xy_to_hs)
|
||||
from homeassistant.util.dt import now as dt_now, get_time_zone
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
VERSION = '1.0.13'
|
||||
color_RGB_to_xy,
|
||||
color_temperature_to_rgb,
|
||||
color_xy_to_hs,
|
||||
)
|
||||
from homeassistant.util.dt import get_time_zone
|
||||
from homeassistant.util.dt import now as dt_now
|
||||
from timezonefinder import TimezoneFinder
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DOMAIN = 'circadian_lighting'
|
||||
CIRCADIAN_LIGHTING_PLATFORMS = ['sensor', 'switch']
|
||||
CIRCADIAN_LIGHTING_UPDATE_TOPIC = '{0}_update'.format(DOMAIN)
|
||||
DATA_CIRCADIAN_LIGHTING = 'data_cl'
|
||||
DOMAIN = "circadian_lighting"
|
||||
CIRCADIAN_LIGHTING_UPDATE_TOPIC = f"{DOMAIN}_update"
|
||||
SUN_EVENT_NOON = "solar_noon"
|
||||
SUN_EVENT_MIDNIGHT = "solar_midnight"
|
||||
|
||||
CONF_MIN_CT = 'min_colortemp'
|
||||
DEFAULT_MIN_CT = 2500
|
||||
CONF_MAX_CT = 'max_colortemp'
|
||||
DEFAULT_MAX_CT = 5500
|
||||
CONF_SUNRISE_OFFSET = 'sunrise_offset'
|
||||
CONF_SUNSET_OFFSET = 'sunset_offset'
|
||||
CONF_SUNRISE_TIME = 'sunrise_time'
|
||||
CONF_SUNSET_TIME = 'sunset_time'
|
||||
CONF_INTERVAL = 'interval'
|
||||
DEFAULT_INTERVAL = 300
|
||||
CONF_MIN_CT, DEFAULT_MIN_CT = "min_colortemp", 2500
|
||||
CONF_MAX_CT, DEFAULT_MAX_CT = "max_colortemp", 5500
|
||||
CONF_INTERVAL, DEFAULT_INTERVAL = "interval", 300
|
||||
CONF_SUNRISE_OFFSET = "sunrise_offset"
|
||||
CONF_SUNSET_OFFSET = "sunset_offset"
|
||||
CONF_SUNRISE_TIME = "sunrise_time"
|
||||
CONF_SUNSET_TIME = "sunset_time"
|
||||
DEFAULT_TRANSITION = 60
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema({
|
||||
DOMAIN: vol.Schema({
|
||||
vol.Optional(CONF_MIN_CT, default=DEFAULT_MIN_CT):
|
||||
vol.All(vol.Coerce(int), vol.Range(min=1000, max=10000)),
|
||||
vol.Optional(CONF_MAX_CT, default=DEFAULT_MAX_CT):
|
||||
vol.All(vol.Coerce(int), vol.Range(min=1000, max=10000)),
|
||||
vol.Optional(CONF_SUNRISE_OFFSET): cv.time_period_str,
|
||||
vol.Optional(CONF_SUNSET_OFFSET): cv.time_period_str,
|
||||
vol.Optional(CONF_SUNRISE_TIME): cv.time,
|
||||
vol.Optional(CONF_SUNSET_TIME): cv.time,
|
||||
vol.Optional(CONF_LATITUDE): cv.latitude,
|
||||
vol.Optional(CONF_LONGITUDE): cv.longitude,
|
||||
vol.Optional(CONF_ELEVATION): float,
|
||||
vol.Optional(CONF_INTERVAL, default=DEFAULT_INTERVAL): cv.positive_int,
|
||||
vol.Optional(ATTR_TRANSITION, default=DEFAULT_TRANSITION): VALID_TRANSITION
|
||||
}),
|
||||
}, extra=vol.ALLOW_EXTRA)
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
DOMAIN: vol.Schema(
|
||||
{
|
||||
vol.Optional(CONF_MIN_CT, default=DEFAULT_MIN_CT): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=1000, max=10000)
|
||||
),
|
||||
vol.Optional(CONF_MAX_CT, default=DEFAULT_MAX_CT): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=1000, max=10000)
|
||||
),
|
||||
vol.Optional(CONF_SUNRISE_OFFSET): cv.time_period_str,
|
||||
vol.Optional(CONF_SUNSET_OFFSET): cv.time_period_str,
|
||||
vol.Optional(CONF_SUNRISE_TIME): cv.time,
|
||||
vol.Optional(CONF_SUNSET_TIME): cv.time,
|
||||
vol.Optional(CONF_LATITUDE): cv.latitude,
|
||||
vol.Optional(CONF_LONGITUDE): cv.longitude,
|
||||
vol.Optional(CONF_ELEVATION): float,
|
||||
vol.Optional(CONF_INTERVAL, default=DEFAULT_INTERVAL): cv.time_period,
|
||||
vol.Optional(
|
||||
ATTR_TRANSITION, default=DEFAULT_TRANSITION
|
||||
): VALID_TRANSITION,
|
||||
}
|
||||
),
|
||||
},
|
||||
extra=vol.ALLOW_EXTRA,
|
||||
)
|
||||
|
||||
|
||||
def setup(hass, config):
|
||||
"""Set up the Circadian Lighting component."""
|
||||
"""Set up the Circadian Lighting platform."""
|
||||
conf = config[DOMAIN]
|
||||
min_colortemp = conf.get(CONF_MIN_CT)
|
||||
max_colortemp = conf.get(CONF_MAX_CT)
|
||||
sunrise_offset = conf.get(CONF_SUNRISE_OFFSET)
|
||||
sunset_offset = conf.get(CONF_SUNSET_OFFSET)
|
||||
sunrise_time = conf.get(CONF_SUNRISE_TIME)
|
||||
sunset_time = conf.get(CONF_SUNSET_TIME)
|
||||
|
||||
latitude = conf.get(CONF_LATITUDE, hass.config.latitude)
|
||||
longitude = conf.get(CONF_LONGITUDE, hass.config.longitude)
|
||||
elevation = conf.get(CONF_ELEVATION, hass.config.elevation)
|
||||
|
||||
load_platform(hass, 'sensor', DOMAIN, {}, config)
|
||||
|
||||
interval = conf.get(CONF_INTERVAL)
|
||||
transition = conf.get(ATTR_TRANSITION)
|
||||
|
||||
cl = CircadianLighting(hass, min_colortemp, max_colortemp,
|
||||
sunrise_offset, sunset_offset, sunrise_time, sunset_time,
|
||||
latitude, longitude, elevation,
|
||||
interval, transition)
|
||||
|
||||
hass.data[DATA_CIRCADIAN_LIGHTING] = cl
|
||||
hass.data[DOMAIN] = CircadianLighting(
|
||||
hass,
|
||||
min_colortemp=conf.get(CONF_MIN_CT),
|
||||
max_colortemp=conf.get(CONF_MAX_CT),
|
||||
sunrise_offset=conf.get(CONF_SUNRISE_OFFSET),
|
||||
sunset_offset=conf.get(CONF_SUNSET_OFFSET),
|
||||
sunrise_time=conf.get(CONF_SUNRISE_TIME),
|
||||
sunset_time=conf.get(CONF_SUNSET_TIME),
|
||||
latitude=conf.get(CONF_LATITUDE, hass.config.latitude),
|
||||
longitude=conf.get(CONF_LONGITUDE, hass.config.longitude),
|
||||
elevation=conf.get(CONF_ELEVATION, hass.config.elevation),
|
||||
interval=conf.get(CONF_INTERVAL),
|
||||
transition=conf.get(ATTR_TRANSITION),
|
||||
)
|
||||
load_platform(hass, "sensor", DOMAIN, {}, config)
|
||||
|
||||
return True
|
||||
|
||||
class CircadianLighting(object):
|
||||
|
||||
class CircadianLighting:
|
||||
"""Calculate universal Circadian values."""
|
||||
|
||||
def __init__(self, hass, min_colortemp, max_colortemp,
|
||||
sunrise_offset, sunset_offset, sunrise_time, sunset_time,
|
||||
latitude, longitude, elevation,
|
||||
interval, transition):
|
||||
def __init__(
|
||||
self,
|
||||
hass,
|
||||
min_colortemp,
|
||||
max_colortemp,
|
||||
sunrise_offset,
|
||||
sunset_offset,
|
||||
sunrise_time,
|
||||
sunset_time,
|
||||
latitude,
|
||||
longitude,
|
||||
elevation,
|
||||
interval,
|
||||
transition,
|
||||
):
|
||||
self.hass = hass
|
||||
self.data = {}
|
||||
self.data['min_colortemp'] = min_colortemp
|
||||
self.data['max_colortemp'] = max_colortemp
|
||||
self.data['sunrise_offset'] = sunrise_offset
|
||||
self.data['sunset_offset'] = sunset_offset
|
||||
self.data['sunrise_time'] = sunrise_time
|
||||
self.data['sunset_time'] = sunset_time
|
||||
self.data['latitude'] = latitude
|
||||
self.data['longitude'] = longitude
|
||||
self.data['elevation'] = elevation
|
||||
self.data['interval'] = interval
|
||||
self.data['transition'] = transition
|
||||
self.data['timezone'] = self.get_timezone()
|
||||
self.data['percent'] = self.calc_percent()
|
||||
self.data['colortemp'] = self.calc_colortemp()
|
||||
self.data['rgb_color'] = self.calc_rgb()
|
||||
self.data['xy_color'] = self.calc_xy()
|
||||
self.data['hs_color'] = self.calc_hs()
|
||||
self._min_colortemp = min_colortemp
|
||||
self._max_colortemp = max_colortemp
|
||||
self._sunrise_offset = sunrise_offset
|
||||
self._sunset_offset = sunset_offset
|
||||
self._manual_sunset = sunset_time
|
||||
self._manual_sunrise = sunrise_time
|
||||
self._latitude = latitude
|
||||
self._longitude = longitude
|
||||
self._elevation = elevation
|
||||
self._transition = transition
|
||||
self._timezone = self.get_timezone()
|
||||
|
||||
self.update = Throttle(timedelta(seconds=interval))(self._update)
|
||||
self._percent = self.calc_percent()
|
||||
self._colortemp = self.calc_colortemp()
|
||||
self._rgb_color = self.calc_rgb()
|
||||
self._xy_color = self.calc_xy()
|
||||
self._hs_color = self.calc_hs()
|
||||
|
||||
if self.data['sunrise_time'] is not None:
|
||||
track_time_change(self.hass, self._update, hour=int(self.data['sunrise_time'].strftime("%H")), minute=int(self.data['sunrise_time'].strftime("%M")), second=int(self.data['sunrise_time'].strftime("%S")))
|
||||
if self._manual_sunrise is not None:
|
||||
async_track_time_change(
|
||||
self.hass,
|
||||
self.update,
|
||||
hour=self._manual_sunrise.hour,
|
||||
minute=self._manual_sunrise.minute,
|
||||
second=self._manual_sunrise.second,
|
||||
)
|
||||
else:
|
||||
track_sunrise(self.hass, self._update, self.data['sunrise_offset'])
|
||||
if self.data['sunset_time'] is not None:
|
||||
track_time_change(self.hass, self._update, hour=int(self.data['sunset_time'].strftime("%H")), minute=int(self.data['sunset_time'].strftime("%M")), second=int(self.data['sunset_time'].strftime("%S")))
|
||||
async_track_sunrise(self.hass, self.update, self._sunrise_offset)
|
||||
|
||||
if self._manual_sunset is not None:
|
||||
async_track_time_change(
|
||||
self.hass,
|
||||
self.update,
|
||||
hour=self._manual_sunset.hour,
|
||||
minute=self._manual_sunset.minute,
|
||||
second=self._manual_sunset.second,
|
||||
)
|
||||
else:
|
||||
track_sunset(self.hass, self._update, self.data['sunset_offset'])
|
||||
async_track_sunset(self.hass, self.update, self._sunset_offset)
|
||||
|
||||
async_track_time_interval(self.hass, self.update, interval)
|
||||
|
||||
def get_timezone(self):
|
||||
from timezonefinder import TimezoneFinder
|
||||
tf = TimezoneFinder()
|
||||
timezone_string = tf.timezone_at(lng=self.data['longitude'], lat=self.data['latitude'])
|
||||
timezone = get_time_zone(timezone_string)
|
||||
_LOGGER.debug("Timezone: " + str(timezone))
|
||||
return timezone
|
||||
|
||||
def get_sunrise_sunset(self, date = None):
|
||||
if self.data['sunrise_time'] is not None and self.data['sunset_time'] is not None:
|
||||
if date is None:
|
||||
date = dt_now(self.data['timezone'])
|
||||
sunrise = date.replace(hour=int(self.data['sunrise_time'].strftime("%H")), minute=int(self.data['sunrise_time'].strftime("%M")), second=int(self.data['sunrise_time'].strftime("%S")), microsecond=int(self.data['sunrise_time'].strftime("%f")))
|
||||
sunset = date.replace(hour=int(self.data['sunset_time'].strftime("%H")), minute=int(self.data['sunset_time'].strftime("%M")), second=int(self.data['sunset_time'].strftime("%S")), microsecond=int(self.data['sunset_time'].strftime("%f")))
|
||||
solar_noon = sunrise + (sunset - sunrise)/2
|
||||
solar_midnight = sunset + ((sunrise + timedelta(days=1)) - sunset)/2
|
||||
timezone_string = tf.timezone_at(lng=self._longitude, lat=self._latitude)
|
||||
return get_time_zone(timezone_string)
|
||||
|
||||
def _replace_time(self, date, key):
|
||||
other_date = self._manual_sunrise if key == "sunrise" else self._manual_sunset
|
||||
return date.replace(
|
||||
hour=other_date.hour,
|
||||
minute=other_date.minute,
|
||||
second=other_date.second,
|
||||
microsecond=other_date.microsecond,
|
||||
)
|
||||
|
||||
def get_sunrise_sunset(self, date):
|
||||
if self._manual_sunrise is not None and self._manual_sunset is not None:
|
||||
sunrise = self._replace_time(date, "sunrise")
|
||||
sunset = self._replace_time(date, "sunset")
|
||||
solar_noon = sunrise + (sunset - sunrise) / 2
|
||||
solar_midnight = sunset + ((sunrise + timedelta(days=1)) - sunset) / 2
|
||||
else:
|
||||
import astral
|
||||
location = astral.Location()
|
||||
location.name = 'name'
|
||||
location.region = 'region'
|
||||
location.latitude = self.data['latitude']
|
||||
location.longitude = self.data['longitude']
|
||||
location.elevation = self.data['elevation']
|
||||
_LOGGER.debug("Astral location: " + str(location))
|
||||
if self.data['sunrise_time'] is not None:
|
||||
if date is None:
|
||||
date = dt_now(self.data['timezone'])
|
||||
sunrise = date.replace(hour=int(self.data['sunrise_time'].strftime("%H")), minute=int(self.data['sunrise_time'].strftime("%M")), second=int(self.data['sunrise_time'].strftime("%S")), microsecond=int(self.data['sunrise_time'].strftime("%f")))
|
||||
location.name = "name"
|
||||
location.region = "region"
|
||||
location.latitude = self._latitude
|
||||
location.longitude = self._longitude
|
||||
location.elevation = self._elevation
|
||||
|
||||
if self._manual_sunrise is not None:
|
||||
sunrise = self._replace_time(date, "sunrise")
|
||||
else:
|
||||
sunrise = location.sunrise(date)
|
||||
if self.data['sunset_time'] is not None:
|
||||
if date is None:
|
||||
date = dt_now(self.data['timezone'])
|
||||
sunset = date.replace(hour=int(self.data['sunset_time'].strftime("%H")), minute=int(self.data['sunset_time'].strftime("%M")), second=int(self.data['sunset_time'].strftime("%S")), microsecond=int(self.data['sunset_time'].strftime("%f")))
|
||||
|
||||
if self._manual_sunset is not None:
|
||||
sunset = self._replace_time(date, "sunset")
|
||||
else:
|
||||
sunset = location.sunset(date)
|
||||
|
||||
solar_noon = location.solar_noon(date)
|
||||
solar_midnight = location.solar_midnight(date)
|
||||
if self.data['sunrise_offset'] is not None:
|
||||
sunrise = sunrise + self.data['sunrise_offset']
|
||||
if self.data['sunset_offset'] is not None:
|
||||
sunset = sunset + self.data['sunset_offset']
|
||||
|
||||
if self._sunrise_offset is not None:
|
||||
sunrise = sunrise + self._sunrise_offset
|
||||
if self._sunset_offset is not None:
|
||||
sunset = sunset + self._sunset_offset
|
||||
|
||||
datetimes = {
|
||||
SUN_EVENT_SUNRISE: sunrise,
|
||||
SUN_EVENT_SUNSET: sunset,
|
||||
SUN_EVENT_NOON: solar_noon,
|
||||
SUN_EVENT_MIDNIGHT: solar_midnight,
|
||||
}
|
||||
|
||||
return {
|
||||
SUN_EVENT_SUNRISE: sunrise.astimezone(self.data['timezone']),
|
||||
SUN_EVENT_SUNSET: sunset.astimezone(self.data['timezone']),
|
||||
'solar_noon': solar_noon.astimezone(self.data['timezone']),
|
||||
'solar_midnight': solar_midnight.astimezone(self.data['timezone'])
|
||||
k: dt.astimezone(self._timezone).timestamp() for k, dt in datetimes.items()
|
||||
}
|
||||
|
||||
def calc_percent(self):
|
||||
now = dt_now(self.data['timezone'])
|
||||
_LOGGER.debug("now: " + str(now))
|
||||
now = dt_now(self._timezone)
|
||||
now_ts = now.timestamp()
|
||||
|
||||
today_sun_times = self.get_sunrise_sunset(now)
|
||||
_LOGGER.debug("today_sun_times: " + str(today_sun_times))
|
||||
|
||||
# Convert everything to epoch timestamps for easy calculation
|
||||
now_seconds = now.timestamp()
|
||||
sunrise_seconds = today_sun_times[SUN_EVENT_SUNRISE].timestamp()
|
||||
sunset_seconds = today_sun_times[SUN_EVENT_SUNSET].timestamp()
|
||||
solar_noon_seconds = today_sun_times['solar_noon'].timestamp()
|
||||
solar_midnight_seconds = today_sun_times['solar_midnight'].timestamp()
|
||||
|
||||
if now < today_sun_times[SUN_EVENT_SUNRISE]: # It's before sunrise (after midnight)
|
||||
# Because it's before sunrise (and after midnight) sunset must have happend yesterday
|
||||
yesterday_sun_times = self.get_sunrise_sunset(now - timedelta(days=1))
|
||||
_LOGGER.debug("yesterday_sun_times: " + str(yesterday_sun_times))
|
||||
sunset_seconds = yesterday_sun_times[SUN_EVENT_SUNSET].timestamp()
|
||||
if today_sun_times['solar_midnight'] > today_sun_times[SUN_EVENT_SUNSET] and yesterday_sun_times['solar_midnight'] > yesterday_sun_times[SUN_EVENT_SUNSET]:
|
||||
today = self.get_sunrise_sunset(now)
|
||||
if now_ts < today[SUN_EVENT_SUNRISE]:
|
||||
# It's before sunrise (after midnight), because it's before
|
||||
# sunrise (and after midnight) sunset must have happend yesterday.
|
||||
yesterday = self.get_sunrise_sunset(now - timedelta(days=1))
|
||||
today[SUN_EVENT_SUNSET] = yesterday[SUN_EVENT_SUNSET]
|
||||
if (
|
||||
today[SUN_EVENT_MIDNIGHT] > today[SUN_EVENT_SUNSET]
|
||||
and yesterday[SUN_EVENT_MIDNIGHT] > yesterday[SUN_EVENT_SUNSET]
|
||||
):
|
||||
# Solar midnight is after sunset so use yesterdays's time
|
||||
solar_midnight_seconds = yesterday_sun_times['solar_midnight'].timestamp()
|
||||
elif now > today_sun_times[SUN_EVENT_SUNSET]: # It's after sunset (before midnight)
|
||||
# Because it's after sunset (and before midnight) sunrise should happen tomorrow
|
||||
tomorrow_sun_times = self.get_sunrise_sunset(now + timedelta(days=1))
|
||||
_LOGGER.debug("tomorrow_sun_times: " + str(tomorrow_sun_times))
|
||||
sunrise_seconds = tomorrow_sun_times[SUN_EVENT_SUNRISE].timestamp()
|
||||
if today_sun_times['solar_midnight'] < today_sun_times[SUN_EVENT_SUNRISE] and tomorrow_sun_times['solar_midnight'] < tomorrow_sun_times[SUN_EVENT_SUNRISE]:
|
||||
today[SUN_EVENT_MIDNIGHT] = yesterday[SUN_EVENT_MIDNIGHT]
|
||||
elif now_ts > today[SUN_EVENT_SUNSET]:
|
||||
# It's after sunset (before midnight), because it's after sunset
|
||||
# (and before midnight) sunrise should happen tomorrow.
|
||||
tomorrow = self.get_sunrise_sunset(now + timedelta(days=1))
|
||||
today[SUN_EVENT_SUNRISE] = tomorrow[SUN_EVENT_SUNRISE]
|
||||
if (
|
||||
today[SUN_EVENT_MIDNIGHT] < today[SUN_EVENT_SUNRISE]
|
||||
and tomorrow[SUN_EVENT_MIDNIGHT] < tomorrow[SUN_EVENT_SUNRISE]
|
||||
):
|
||||
# Solar midnight is before sunrise so use tomorrow's time
|
||||
solar_midnight_seconds = tomorrow_sun_times['solar_midnight'].timestamp()
|
||||
today[SUN_EVENT_MIDNIGHT] = tomorrow[SUN_EVENT_MIDNIGHT]
|
||||
|
||||
_LOGGER.debug("now_seconds: " + str(now_seconds))
|
||||
_LOGGER.debug("sunrise_seconds: " + str(sunrise_seconds))
|
||||
_LOGGER.debug("sunset_seconds: " + str(sunset_seconds))
|
||||
_LOGGER.debug("solar_midnight_seconds: " + str(solar_midnight_seconds))
|
||||
_LOGGER.debug("solar_noon_seconds: " + str(solar_noon_seconds))
|
||||
# Figure out where we are in time so we know which half of the
|
||||
# parabola to calculate. We're generating a different
|
||||
# sunset-sunrise parabola for before and after solar midnight.
|
||||
# because it might not be half way between sunrise and sunset.
|
||||
# We're also generating a different parabola for sunrise-sunset.
|
||||
|
||||
# Figure out where we are in time so we know which half of the parabola to calculate
|
||||
# We're generating a different sunset-sunrise parabola for before and after solar midnight
|
||||
# because it might not be half way between sunrise and sunset
|
||||
# We're also (obviously) generating a different parabola for sunrise-sunset
|
||||
|
||||
# sunrise-sunset parabola
|
||||
if now_seconds > sunrise_seconds and now_seconds < sunset_seconds:
|
||||
h = solar_noon_seconds
|
||||
# sunrise -> sunset parabola
|
||||
if today[SUN_EVENT_SUNRISE] < now_ts < today[SUN_EVENT_SUNSET]:
|
||||
h = today[SUN_EVENT_NOON]
|
||||
k = 100
|
||||
# parabola before solar_noon
|
||||
if now_seconds < solar_noon_seconds:
|
||||
x = sunrise_seconds
|
||||
# parabola after solar_noon
|
||||
else:
|
||||
x = sunset_seconds
|
||||
y = 0
|
||||
# parabola before solar_noon else after solar_noon
|
||||
x = (
|
||||
today[SUN_EVENT_SUNRISE]
|
||||
if now_ts < today[SUN_EVENT_NOON]
|
||||
else today[SUN_EVENT_SUNSET]
|
||||
)
|
||||
|
||||
# sunset_sunrise parabola
|
||||
elif now_seconds > sunset_seconds and now_seconds < sunrise_seconds:
|
||||
h = solar_midnight_seconds
|
||||
# sunset -> sunrise parabola
|
||||
elif today[SUN_EVENT_SUNSET] < now_ts < today[SUN_EVENT_SUNRISE]:
|
||||
h = today[SUN_EVENT_MIDNIGHT]
|
||||
k = -100
|
||||
# parabola before solar_midnight
|
||||
if now_seconds < solar_midnight_seconds:
|
||||
x = sunset_seconds
|
||||
# parabola after solar_midnight
|
||||
else:
|
||||
x = sunrise_seconds
|
||||
y = 0
|
||||
|
||||
a = (y-k)/(h-x)**2
|
||||
percentage = a*(now_seconds-h)**2+k
|
||||
|
||||
_LOGGER.debug("h: " + str(h))
|
||||
_LOGGER.debug("k: " + str(k))
|
||||
_LOGGER.debug("x: " + str(x))
|
||||
_LOGGER.debug("y: " + str(y))
|
||||
_LOGGER.debug("a: " + str(a))
|
||||
_LOGGER.debug("percentage: " + str(percentage))
|
||||
# parabola before solar_midnight else after solar_midnight
|
||||
x = (
|
||||
today[SUN_EVENT_SUNSET]
|
||||
if now_ts < today[SUN_EVENT_MIDNIGHT]
|
||||
else today[SUN_EVENT_SUNRISE]
|
||||
)
|
||||
|
||||
y = 0
|
||||
a = (y - k) / (h - x) ** 2
|
||||
percentage = a * (now_ts - h) ** 2 + k
|
||||
return percentage
|
||||
|
||||
def calc_colortemp(self):
|
||||
if self.data['percent'] > 0:
|
||||
return ((self.data['max_colortemp'] - self.data['min_colortemp']) * (self.data['percent'] / 100)) + self.data['min_colortemp']
|
||||
if self._percent > 0:
|
||||
delta = self._max_colortemp - self._min_colortemp
|
||||
percent = self._percent / 100
|
||||
return (delta * percent) + self._min_colortemp
|
||||
else:
|
||||
return self.data['min_colortemp']
|
||||
return self._min_colortemp
|
||||
|
||||
def calc_rgb(self):
|
||||
return color_temperature_to_rgb(self.data['colortemp'])
|
||||
return color_temperature_to_rgb(self._colortemp)
|
||||
|
||||
def calc_xy(self):
|
||||
rgb = self.calc_rgb()
|
||||
iR = rgb[0]
|
||||
iG = rgb[1]
|
||||
iB = rgb[2]
|
||||
|
||||
return color_RGB_to_xy(iR, iG, iB)
|
||||
return color_RGB_to_xy(*self.calc_rgb())
|
||||
|
||||
def calc_hs(self):
|
||||
xy = self.calc_xy()
|
||||
vX = xy[0]
|
||||
vY = xy[1]
|
||||
return color_xy_to_hs(*self.calc_xy())
|
||||
|
||||
return color_xy_to_hs(vX, vY)
|
||||
|
||||
def _update(self, *args, **kwargs):
|
||||
async def update(self, _=None):
|
||||
"""Update Circadian Values."""
|
||||
self.data['percent'] = self.calc_percent()
|
||||
self.data['colortemp'] = self.calc_colortemp()
|
||||
self.data['rgb_color'] = self.calc_rgb()
|
||||
self.data['xy_color'] = self.calc_xy()
|
||||
self.data['hs_color'] = self.calc_hs()
|
||||
dispatcher_send(self.hass, CIRCADIAN_LIGHTING_UPDATE_TOPIC)
|
||||
_LOGGER.debug("Circadian Lighting Component Updated")
|
||||
self._percent = self.calc_percent()
|
||||
self._colortemp = self.calc_colortemp()
|
||||
self._rgb_color = self.calc_rgb()
|
||||
self._xy_color = self.calc_xy()
|
||||
self._hs_color = self.calc_hs()
|
||||
async_dispatcher_send(self.hass, CIRCADIAN_LIGHTING_UPDATE_TOPIC)
|
||||
|
|
|
|||
|
|
@ -4,5 +4,5 @@
|
|||
"documentation": "https://github.com/claytonjn/hass-circadian_lighting",
|
||||
"dependencies": [],
|
||||
"codeowners": ["@claytonjn"],
|
||||
"requirements": ["timezonefinder==4.2.0"]
|
||||
"requirements": ["timezonefinder==4.2.0", "astral==1.10.1"]
|
||||
}
|
||||
|
|
|
|||
83
custom_components/circadian_lighting/sensor.py
Normal file → Executable file
83
custom_components/circadian_lighting/sensor.py
Normal file → Executable file
|
|
@ -2,56 +2,43 @@
|
|||
Circadian Lighting Sensor for Home-Assistant.
|
||||
"""
|
||||
|
||||
DEPENDENCIES = ['circadian_lighting']
|
||||
|
||||
import logging
|
||||
|
||||
from custom_components.circadian_lighting import DOMAIN, CIRCADIAN_LIGHTING_UPDATE_TOPIC, DATA_CIRCADIAN_LIGHTING
|
||||
|
||||
from homeassistant.helpers.dispatcher import dispatcher_connect
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.entity import Entity
|
||||
|
||||
import datetime
|
||||
from . import CIRCADIAN_LIGHTING_UPDATE_TOPIC, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
ICON = "mdi:theme-light-dark"
|
||||
|
||||
ICON = 'mdi:theme-light-dark'
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Set up the Circadian Lighting sensor."""
|
||||
cl = hass.data.get(DATA_CIRCADIAN_LIGHTING)
|
||||
if cl:
|
||||
cs = CircadianSensor(hass, cl)
|
||||
add_devices([cs])
|
||||
circadian_lighting = hass.data.get(DOMAIN)
|
||||
if circadian_lighting is not None:
|
||||
sensor = CircadianSensor(hass, circadian_lighting)
|
||||
add_devices([sensor], True)
|
||||
|
||||
def update(call=None):
|
||||
"""Update component."""
|
||||
cl._update()
|
||||
circadian_lighting.update()
|
||||
|
||||
service_name = "values_update"
|
||||
hass.services.register(DOMAIN, service_name, update)
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
class CircadianSensor(Entity):
|
||||
"""Representation of a Circadian Lighting sensor."""
|
||||
|
||||
def __init__(self, hass, cl):
|
||||
def __init__(self, hass, circadian_lighting):
|
||||
"""Initialize the Circadian Lighting sensor."""
|
||||
self._cl = cl
|
||||
self._name = 'Circadian Values'
|
||||
self._entity_id = 'sensor.circadian_values'
|
||||
self._state = self._cl.data['percent']
|
||||
self._unit_of_measurement = '%'
|
||||
self._circadian_lighting = circadian_lighting
|
||||
self._name = "Circadian Values"
|
||||
self._entity_id = "sensor.circadian_values"
|
||||
self._unit_of_measurement = "%"
|
||||
self._icon = ICON
|
||||
self._hs_color = self._cl.data['hs_color']
|
||||
self._attributes = {}
|
||||
self._attributes['colortemp'] = self._cl.data['colortemp']
|
||||
self._attributes['rgb_color'] = self._cl.data['rgb_color']
|
||||
self._attributes['xy_color'] = self._cl.data['xy_color']
|
||||
|
||||
"""Register callbacks."""
|
||||
dispatcher_connect(hass, CIRCADIAN_LIGHTING_UPDATE_TOPIC, self.update_sensor)
|
||||
|
||||
@property
|
||||
def entity_id(self):
|
||||
|
|
@ -66,7 +53,7 @@ class CircadianSensor(Entity):
|
|||
@property
|
||||
def state(self):
|
||||
"""Return the state of the sensor."""
|
||||
return self._state
|
||||
return self._circadian_lighting._percent
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
|
|
@ -80,25 +67,31 @@ class CircadianSensor(Entity):
|
|||
|
||||
@property
|
||||
def hs_color(self):
|
||||
return self._hs_color
|
||||
return self._circadian_lighting._hs_color
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the attributes of the sensor."""
|
||||
return self._attributes
|
||||
return {
|
||||
"colortemp": self._circadian_lighting._colortemp,
|
||||
"rgb_color": self._circadian_lighting._rgb_color,
|
||||
"xy_color": self._circadian_lighting._xy_color,
|
||||
}
|
||||
|
||||
def update(self):
|
||||
"""Fetch new state data for the sensor.
|
||||
@property
|
||||
def should_poll(self) -> bool:
|
||||
"""Disable polling."""
|
||||
return False
|
||||
|
||||
This is the only method that should fetch new data for Home Assistant.
|
||||
"""
|
||||
self._cl.update()
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Connect dispatcher to signal from CircadianLighting object."""
|
||||
self.async_on_remove(
|
||||
async_dispatcher_connect(
|
||||
self.hass, CIRCADIAN_LIGHTING_UPDATE_TOPIC, self._update_callback
|
||||
)
|
||||
)
|
||||
|
||||
def update_sensor(self):
|
||||
if self._cl.data is not None:
|
||||
self._state = self._cl.data['percent']
|
||||
self._hs_color = self._cl.data['hs_color']
|
||||
self._attributes['colortemp'] = self._cl.data['colortemp']
|
||||
self._attributes['rgb_color'] = self._cl.data['rgb_color']
|
||||
self._attributes['xy_color'] = self._cl.data['xy_color']
|
||||
_LOGGER.debug("Circadian Lighting Sensor Updated")
|
||||
@callback
|
||||
def _update_callback(self) -> None:
|
||||
"""Triggers update of properties."""
|
||||
self.async_schedule_update_ha_state(force_refresh=False)
|
||||
|
|
|
|||
527
custom_components/circadian_lighting/switch.py
Normal file → Executable file
527
custom_components/circadian_lighting/switch.py
Normal file → Executable file
|
|
@ -2,133 +2,194 @@
|
|||
Circadian Lighting Switch for Home-Assistant.
|
||||
"""
|
||||
|
||||
DEPENDENCIES = ['circadian_lighting', 'light']
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
from custom_components.circadian_lighting import DOMAIN, CIRCADIAN_LIGHTING_UPDATE_TOPIC, DATA_CIRCADIAN_LIGHTING
|
||||
from itertools import repeat
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.dispatcher import dispatcher_connect
|
||||
from homeassistant.helpers.event import track_state_change
|
||||
from homeassistant.helpers.restore_state import RestoreEntity
|
||||
from homeassistant.components.light import (
|
||||
is_on, ATTR_BRIGHTNESS, ATTR_COLOR_TEMP, ATTR_RGB_COLOR, ATTR_TRANSITION,
|
||||
VALID_TRANSITION, ATTR_WHITE_VALUE, ATTR_XY_COLOR, DOMAIN as LIGHT_DOMAIN)
|
||||
|
||||
try:
|
||||
from homeassistant.components.switch import SwitchEntity
|
||||
except ImportError:
|
||||
from homeassistant.components.switch import SwitchDevice as SwitchEntity
|
||||
|
||||
ATTR_BRIGHTNESS,
|
||||
ATTR_COLOR_TEMP,
|
||||
ATTR_RGB_COLOR,
|
||||
ATTR_TRANSITION,
|
||||
ATTR_WHITE_VALUE,
|
||||
ATTR_XY_COLOR,
|
||||
)
|
||||
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
|
||||
from homeassistant.components.light import VALID_TRANSITION, is_on
|
||||
from homeassistant.components.switch import SwitchEntity
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID, CONF_NAME, CONF_PLATFORM, STATE_ON,
|
||||
SERVICE_TURN_ON)
|
||||
ATTR_ENTITY_ID,
|
||||
CONF_NAME,
|
||||
CONF_PLATFORM,
|
||||
SERVICE_TURN_ON,
|
||||
STATE_ON,
|
||||
)
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.event import async_track_state_change
|
||||
from homeassistant.helpers.restore_state import RestoreEntity
|
||||
from homeassistant.util import slugify
|
||||
from homeassistant.util.color import (
|
||||
color_RGB_to_xy, color_temperature_kelvin_to_mired,
|
||||
color_temperature_to_rgb, color_xy_to_hs)
|
||||
color_RGB_to_xy,
|
||||
color_temperature_kelvin_to_mired,
|
||||
color_temperature_to_rgb,
|
||||
color_xy_to_hs,
|
||||
)
|
||||
|
||||
from . import CIRCADIAN_LIGHTING_UPDATE_TOPIC, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ICON = 'mdi:theme-light-dark'
|
||||
ICON = "mdi:theme-light-dark"
|
||||
|
||||
CONF_LIGHTS_CT = 'lights_ct'
|
||||
CONF_LIGHTS_RGB = 'lights_rgb'
|
||||
CONF_LIGHTS_XY = 'lights_xy'
|
||||
CONF_LIGHTS_BRIGHT = 'lights_brightness'
|
||||
CONF_DISABLE_BRIGHTNESS_ADJUST = 'disable_brightness_adjust'
|
||||
CONF_MIN_BRIGHT = 'min_brightness'
|
||||
DEFAULT_MIN_BRIGHT = 1
|
||||
CONF_MAX_BRIGHT = 'max_brightness'
|
||||
DEFAULT_MAX_BRIGHT = 100
|
||||
CONF_SLEEP_ENTITY = 'sleep_entity'
|
||||
CONF_SLEEP_STATE = 'sleep_state'
|
||||
CONF_SLEEP_CT = 'sleep_colortemp'
|
||||
CONF_SLEEP_BRIGHT = 'sleep_brightness'
|
||||
CONF_DISABLE_ENTITY = 'disable_entity'
|
||||
CONF_DISABLE_STATE = 'disable_state'
|
||||
CONF_INITIAL_TRANSITION = 'initial_transition'
|
||||
DEFAULT_INITIAL_TRANSITION = 1
|
||||
CONF_LIGHTS_CT = "lights_ct"
|
||||
CONF_LIGHTS_RGB = "lights_rgb"
|
||||
CONF_LIGHTS_XY = "lights_xy"
|
||||
CONF_LIGHTS_BRIGHT = "lights_brightness"
|
||||
CONF_DISABLE_BRIGHTNESS_ADJUST = "disable_brightness_adjust"
|
||||
CONF_MIN_BRIGHT, DEFAULT_MIN_BRIGHT = "min_brightness", 1
|
||||
CONF_MAX_BRIGHT, DEFAULT_MAX_BRIGHT = "max_brightness", 100
|
||||
CONF_SLEEP_ENTITY = "sleep_entity"
|
||||
CONF_SLEEP_STATE = "sleep_state"
|
||||
CONF_SLEEP_CT, DEFAULT_SLEEP_CT = "sleep_colortemp", 1000
|
||||
CONF_SLEEP_BRIGHT, DEFAULT_SLEEP_BRIGHT = "sleep_brightness", 1
|
||||
CONF_DISABLE_ENTITY = "disable_entity"
|
||||
CONF_DISABLE_STATE = "disable_state"
|
||||
CONF_INITIAL_TRANSITION, DEFAULT_INITIAL_TRANSITION = "initial_transition", 1
|
||||
CONF_ONLY_ONCE = "only_once"
|
||||
|
||||
PLATFORM_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_PLATFORM): "circadian_lighting",
|
||||
vol.Optional(CONF_NAME, default="Circadian Lighting"): cv.string,
|
||||
vol.Optional(CONF_LIGHTS_CT): cv.entity_ids,
|
||||
vol.Optional(CONF_LIGHTS_RGB): cv.entity_ids,
|
||||
vol.Optional(CONF_LIGHTS_XY): cv.entity_ids,
|
||||
vol.Optional(CONF_LIGHTS_BRIGHT): cv.entity_ids,
|
||||
vol.Optional(CONF_DISABLE_BRIGHTNESS_ADJUST, default=False): cv.boolean,
|
||||
vol.Optional(CONF_MIN_BRIGHT, default=DEFAULT_MIN_BRIGHT): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=1, max=100)
|
||||
),
|
||||
vol.Optional(CONF_MAX_BRIGHT, default=DEFAULT_MAX_BRIGHT): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=1, max=100)
|
||||
),
|
||||
vol.Optional(CONF_SLEEP_ENTITY): cv.entity_id,
|
||||
vol.Optional(CONF_SLEEP_STATE): vol.All(cv.ensure_list, [cv.string]),
|
||||
vol.Optional(CONF_SLEEP_CT, default=DEFAULT_SLEEP_CT): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=1000, max=10000)
|
||||
),
|
||||
vol.Optional(CONF_SLEEP_BRIGHT, default=DEFAULT_SLEEP_BRIGHT): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=1, max=100)
|
||||
),
|
||||
vol.Optional(CONF_DISABLE_ENTITY): cv.entity_id,
|
||||
vol.Optional(CONF_DISABLE_STATE): vol.All(cv.ensure_list, [cv.string]),
|
||||
vol.Optional(
|
||||
CONF_INITIAL_TRANSITION, default=DEFAULT_INITIAL_TRANSITION
|
||||
): VALID_TRANSITION,
|
||||
vol.Optional(CONF_ONLY_ONCE, default=False): cv.boolean,
|
||||
}
|
||||
)
|
||||
|
||||
PLATFORM_SCHEMA = vol.Schema({
|
||||
vol.Required(CONF_PLATFORM): 'circadian_lighting',
|
||||
vol.Optional(CONF_NAME, default="Circadian Lighting"): cv.string,
|
||||
vol.Optional(CONF_LIGHTS_CT): cv.entity_ids,
|
||||
vol.Optional(CONF_LIGHTS_RGB): cv.entity_ids,
|
||||
vol.Optional(CONF_LIGHTS_XY): cv.entity_ids,
|
||||
vol.Optional(CONF_LIGHTS_BRIGHT): cv.entity_ids,
|
||||
vol.Optional(CONF_DISABLE_BRIGHTNESS_ADJUST, default=False): cv.boolean,
|
||||
vol.Optional(CONF_MIN_BRIGHT, default=DEFAULT_MIN_BRIGHT):
|
||||
vol.All(vol.Coerce(int), vol.Range(min=1, max=100)),
|
||||
vol.Optional(CONF_MAX_BRIGHT, default=DEFAULT_MAX_BRIGHT):
|
||||
vol.All(vol.Coerce(int), vol.Range(min=1, max=100)),
|
||||
vol.Optional(CONF_SLEEP_ENTITY): cv.entity_id,
|
||||
vol.Optional(CONF_SLEEP_STATE): cv.string,
|
||||
vol.Optional(CONF_SLEEP_CT):
|
||||
vol.All(vol.Coerce(int), vol.Range(min=1000, max=10000)),
|
||||
vol.Optional(CONF_SLEEP_BRIGHT):
|
||||
vol.All(vol.Coerce(int), vol.Range(min=1, max=100)),
|
||||
vol.Optional(CONF_DISABLE_ENTITY): cv.entity_id,
|
||||
vol.Optional(CONF_DISABLE_STATE): cv.string,
|
||||
vol.Optional(CONF_INITIAL_TRANSITION, default=DEFAULT_INITIAL_TRANSITION):
|
||||
VALID_TRANSITION
|
||||
})
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Set up the Circadian Lighting switches."""
|
||||
cl = hass.data.get(DATA_CIRCADIAN_LIGHTING)
|
||||
if cl:
|
||||
lights_ct = config.get(CONF_LIGHTS_CT)
|
||||
lights_rgb = config.get(CONF_LIGHTS_RGB)
|
||||
lights_xy = config.get(CONF_LIGHTS_XY)
|
||||
lights_brightness = config.get(CONF_LIGHTS_BRIGHT)
|
||||
disable_brightness_adjust = config.get(CONF_DISABLE_BRIGHTNESS_ADJUST)
|
||||
name = config.get(CONF_NAME)
|
||||
min_brightness = config.get(CONF_MIN_BRIGHT)
|
||||
max_brightness = config.get(CONF_MAX_BRIGHT)
|
||||
sleep_entity = config.get(CONF_SLEEP_ENTITY)
|
||||
sleep_state = config.get(CONF_SLEEP_STATE)
|
||||
sleep_colortemp = config.get(CONF_SLEEP_CT)
|
||||
sleep_brightness = config.get(CONF_SLEEP_BRIGHT)
|
||||
disable_entity = config.get(CONF_DISABLE_ENTITY)
|
||||
disable_state = config.get(CONF_DISABLE_STATE)
|
||||
initial_transition = config.get(CONF_INITIAL_TRANSITION)
|
||||
cs = CircadianSwitch(hass, cl, name, lights_ct, lights_rgb, lights_xy, lights_brightness,
|
||||
disable_brightness_adjust, min_brightness, max_brightness,
|
||||
sleep_entity, sleep_state, sleep_colortemp, sleep_brightness,
|
||||
disable_entity, disable_state, initial_transition)
|
||||
add_devices([cs])
|
||||
circadian_lighting = hass.data.get(DOMAIN)
|
||||
if circadian_lighting is not None:
|
||||
switch = CircadianSwitch(
|
||||
hass,
|
||||
circadian_lighting,
|
||||
name=config.get(CONF_NAME),
|
||||
lights_ct=config.get(CONF_LIGHTS_CT, []),
|
||||
lights_rgb=config.get(CONF_LIGHTS_RGB, []),
|
||||
lights_xy=config.get(CONF_LIGHTS_XY, []),
|
||||
lights_brightness=config.get(CONF_LIGHTS_BRIGHT, []),
|
||||
disable_brightness_adjust=config.get(CONF_DISABLE_BRIGHTNESS_ADJUST),
|
||||
min_brightness=config.get(CONF_MIN_BRIGHT),
|
||||
max_brightness=config.get(CONF_MAX_BRIGHT),
|
||||
sleep_entity=config.get(CONF_SLEEP_ENTITY),
|
||||
sleep_state=config.get(CONF_SLEEP_STATE),
|
||||
sleep_colortemp=config.get(CONF_SLEEP_CT),
|
||||
sleep_brightness=config.get(CONF_SLEEP_BRIGHT),
|
||||
disable_entity=config.get(CONF_DISABLE_ENTITY),
|
||||
disable_state=config.get(CONF_DISABLE_STATE),
|
||||
initial_transition=config.get(CONF_INITIAL_TRANSITION),
|
||||
only_once=config.get(CONF_ONLY_ONCE),
|
||||
)
|
||||
add_devices([switch])
|
||||
|
||||
def update(call=None):
|
||||
"""Update lights."""
|
||||
cs.update_switch()
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
def _difference_between_states(from_state, to_state):
|
||||
start = "Lights adjusting because "
|
||||
if from_state is None and to_state is None:
|
||||
return start + "Both states None"
|
||||
if from_state is None:
|
||||
return start + f"from_state: None, to_state: {to_state}"
|
||||
if to_state is None:
|
||||
return start + f"from_state: {from_state}, to_state: None"
|
||||
|
||||
changed_attrs = ", ".join(
|
||||
[
|
||||
f"{key}: {val}"
|
||||
for key, val in to_state.attributes.items()
|
||||
if from_state.attributes.get(key) != val
|
||||
]
|
||||
)
|
||||
if from_state.state == to_state.state:
|
||||
return start + (
|
||||
f"{from_state.entity_id} is still {to_state.state} but"
|
||||
f" these attributes changes: {changed_attrs}."
|
||||
)
|
||||
elif changed_attrs != "":
|
||||
return start + (
|
||||
f"{from_state.entity_id} changed from {from_state.state} to"
|
||||
f" {to_state.state} and these attributes changes: {changed_attrs}."
|
||||
)
|
||||
else:
|
||||
return start + (
|
||||
f"{from_state.entity_id} changed from {from_state.state} to"
|
||||
f" {to_state.state} and no attributes changed."
|
||||
)
|
||||
|
||||
|
||||
class CircadianSwitch(SwitchEntity, RestoreEntity):
|
||||
"""Representation of a Circadian Lighting switch."""
|
||||
|
||||
def __init__(self, hass, cl, name, lights_ct, lights_rgb, lights_xy, lights_brightness,
|
||||
disable_brightness_adjust, min_brightness, max_brightness,
|
||||
sleep_entity, sleep_state, sleep_colortemp, sleep_brightness,
|
||||
disable_entity, disable_state, initial_transition):
|
||||
def __init__(
|
||||
self,
|
||||
hass,
|
||||
circadian_lighting,
|
||||
name,
|
||||
lights_ct,
|
||||
lights_rgb,
|
||||
lights_xy,
|
||||
lights_brightness,
|
||||
disable_brightness_adjust,
|
||||
min_brightness,
|
||||
max_brightness,
|
||||
sleep_entity,
|
||||
sleep_state,
|
||||
sleep_colortemp,
|
||||
sleep_brightness,
|
||||
disable_entity,
|
||||
disable_state,
|
||||
initial_transition,
|
||||
only_once,
|
||||
):
|
||||
"""Initialize the Circadian Lighting switch."""
|
||||
self.hass = hass
|
||||
self._cl = cl
|
||||
self._circadian_lighting = circadian_lighting
|
||||
self._name = name
|
||||
self._entity_id = "switch." + slugify("{} {}".format('circadian_lighting', name))
|
||||
self._entity_id = f"switch.circadian_lighting_{slugify(name)}"
|
||||
self._state = None
|
||||
self._icon = ICON
|
||||
self._hs_color = None
|
||||
self._lights_ct = lights_ct
|
||||
self._lights_rgb = lights_rgb
|
||||
self._lights_xy = lights_xy
|
||||
self._lights_brightness = lights_brightness
|
||||
self._brightness = None
|
||||
self._disable_brightness_adjust = disable_brightness_adjust
|
||||
self._min_brightness = min_brightness
|
||||
self._max_brightness = max_brightness
|
||||
|
|
@ -139,27 +200,12 @@ class CircadianSwitch(SwitchEntity, RestoreEntity):
|
|||
self._disable_entity = disable_entity
|
||||
self._disable_state = disable_state
|
||||
self._initial_transition = initial_transition
|
||||
self._attributes = {}
|
||||
self._attributes['hs_color'] = self._hs_color
|
||||
self._attributes['brightness'] = None
|
||||
|
||||
self._lights = []
|
||||
if lights_ct != None:
|
||||
self._lights += lights_ct
|
||||
if lights_rgb != None:
|
||||
self._lights += lights_rgb
|
||||
if lights_xy != None:
|
||||
self._lights += lights_xy
|
||||
if lights_brightness != None:
|
||||
self._lights += lights_brightness
|
||||
|
||||
"""Register callbacks."""
|
||||
dispatcher_connect(hass, CIRCADIAN_LIGHTING_UPDATE_TOPIC, self.update_switch)
|
||||
track_state_change(hass, self._lights, self.light_state_changed)
|
||||
if self._sleep_entity is not None:
|
||||
track_state_change(hass, self._sleep_entity, self.sleep_state_changed)
|
||||
if self._disable_entity is not None:
|
||||
track_state_change(hass, self._disable_entity, self.disable_state_changed)
|
||||
self._only_once = only_once
|
||||
self._lights_types = dict(zip(lights_ct, repeat("ct")))
|
||||
self._lights_types.update(zip(lights_rgb, repeat("rgb")))
|
||||
self._lights_types.update(zip(lights_xy, repeat("xy")))
|
||||
self._lights_types.update(zip(lights_brightness, repeat("brightness")))
|
||||
self._lights = list(self._lights_types.keys())
|
||||
|
||||
@property
|
||||
def entity_id(self):
|
||||
|
|
@ -178,9 +224,31 @@ class CircadianSwitch(SwitchEntity, RestoreEntity):
|
|||
|
||||
async def async_added_to_hass(self):
|
||||
"""Call when entity about to be added to hass."""
|
||||
# If not None, we got an initial value.
|
||||
await super().async_added_to_hass()
|
||||
if self._state is not None:
|
||||
# Add callback
|
||||
self.async_on_remove(
|
||||
async_dispatcher_connect(
|
||||
self.hass, CIRCADIAN_LIGHTING_UPDATE_TOPIC, self._update_switch
|
||||
)
|
||||
)
|
||||
|
||||
# Add listeners
|
||||
async_track_state_change(
|
||||
self.hass, self._lights, self._light_state_changed, to_state="on"
|
||||
)
|
||||
track_kwargs = dict(hass=self.hass, action=self._state_changed)
|
||||
if self._sleep_entity is not None:
|
||||
sleep_kwargs = dict(track_kwargs, entity_ids=self._sleep_entity)
|
||||
async_track_state_change(**sleep_kwargs, to_state=self._sleep_state)
|
||||
async_track_state_change(**sleep_kwargs, from_state=self._sleep_state)
|
||||
|
||||
if self._disable_entity is not None:
|
||||
async_track_state_change(
|
||||
**track_kwargs,
|
||||
entity_ids=self._disable_entity,
|
||||
from_state=self._disable_state,
|
||||
)
|
||||
|
||||
if self._state is not None: # If not None, we got an initial value
|
||||
return
|
||||
|
||||
state = await self.async_get_last_state()
|
||||
|
|
@ -198,165 +266,122 @@ class CircadianSwitch(SwitchEntity, RestoreEntity):
|
|||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the attributes of the switch."""
|
||||
return self._attributes
|
||||
return {"hs_color": self._hs_color, "brightness": self._brightness}
|
||||
|
||||
def turn_on(self, **kwargs):
|
||||
async def turn_on(self, **kwargs):
|
||||
"""Turn on circadian lighting."""
|
||||
self._state = True
|
||||
|
||||
# Make initial update
|
||||
self.update_switch(self._initial_transition)
|
||||
|
||||
self.schedule_update_ha_state()
|
||||
await self._force_update_switch()
|
||||
|
||||
def turn_off(self, **kwargs):
|
||||
"""Turn off circadian lighting."""
|
||||
self._state = False
|
||||
self.schedule_update_ha_state()
|
||||
self._hs_color = None
|
||||
self._attributes['hs_color'] = self._hs_color
|
||||
self._attributes['brightness'] = None
|
||||
self._brightness = None
|
||||
|
||||
def is_sleep(self):
|
||||
return self._sleep_entity is not None and self.hass.states.get(self._sleep_entity).state == self._sleep_state
|
||||
return (
|
||||
self._sleep_entity is not None
|
||||
and self.hass.states.get(self._sleep_entity).state in self._sleep_state
|
||||
)
|
||||
|
||||
def calc_ct(self):
|
||||
if self.is_sleep():
|
||||
_LOGGER.debug(self._name + " in Sleep mode")
|
||||
return color_temperature_kelvin_to_mired(self._sleep_colortemp)
|
||||
else:
|
||||
return color_temperature_kelvin_to_mired(self._cl.data['colortemp'])
|
||||
def _color_temperature(self):
|
||||
return (
|
||||
self._circadian_lighting._colortemp
|
||||
if not self.is_sleep()
|
||||
else self._sleep_colortemp
|
||||
)
|
||||
|
||||
def calc_rgb(self):
|
||||
if self.is_sleep():
|
||||
_LOGGER.debug(self._name + " in Sleep mode")
|
||||
return color_temperature_to_rgb(self._sleep_colortemp)
|
||||
else:
|
||||
return color_temperature_to_rgb(self._cl.data['colortemp'])
|
||||
def _calc_ct(self):
|
||||
return color_temperature_kelvin_to_mired(self._color_temperature())
|
||||
|
||||
def calc_xy(self):
|
||||
return color_RGB_to_xy(*self.calc_rgb())
|
||||
def _calc_rgb(self):
|
||||
return color_temperature_to_rgb(self._color_temperature())
|
||||
|
||||
def calc_hs(self):
|
||||
return color_xy_to_hs(*self.calc_xy())
|
||||
def _calc_xy(self):
|
||||
return color_RGB_to_xy(*self._calc_rgb())
|
||||
|
||||
def calc_brightness(self):
|
||||
if self._disable_brightness_adjust is True:
|
||||
def _calc_hs(self):
|
||||
return color_xy_to_hs(*self._calc_xy())
|
||||
|
||||
def _calc_brightness(self) -> float:
|
||||
if self._disable_brightness_adjust:
|
||||
return None
|
||||
else:
|
||||
if self.is_sleep():
|
||||
_LOGGER.debug(self._name + " in Sleep mode")
|
||||
return self._sleep_brightness
|
||||
else:
|
||||
if self._cl.data['percent'] > 0:
|
||||
return self._max_brightness
|
||||
else:
|
||||
return ((self._max_brightness - self._min_brightness) * ((100+self._cl.data['percent']) / 100)) + self._min_brightness
|
||||
if self.is_sleep():
|
||||
return self._sleep_brightness
|
||||
if self._circadian_lighting._percent > 0:
|
||||
return self._max_brightness
|
||||
delta_brightness = self._max_brightness - self._min_brightness
|
||||
percent = (100 + self._circadian_lighting._percent) / 100
|
||||
return (delta_brightness * percent) + self._min_brightness
|
||||
|
||||
def update_switch(self, transition=None):
|
||||
if self._cl.data is not None:
|
||||
self._hs_color = self.calc_hs()
|
||||
self._attributes['hs_color'] = self._hs_color
|
||||
self._attributes['brightness'] = self.calc_brightness()
|
||||
_LOGGER.debug(self._name + " Switch Updated")
|
||||
async def _update_switch(self, lights=None, transition=None, force=False):
|
||||
if self._only_once and not force:
|
||||
return
|
||||
self._hs_color = self._calc_hs()
|
||||
self._brightness = self._calc_brightness()
|
||||
await self._adjust_lights(lights or self._lights, transition)
|
||||
|
||||
self.adjust_lights(self._lights, transition)
|
||||
async def _force_update_switch(self, lights=None):
|
||||
return await self._update_switch(
|
||||
lights, transition=self._initial_transition, force=True
|
||||
)
|
||||
|
||||
def should_adjust(self):
|
||||
def _is_disabled(self):
|
||||
return (
|
||||
self._disable_entity is not None
|
||||
and self.hass.states.get(self._disable_entity).state in self._disable_state
|
||||
)
|
||||
|
||||
def _should_adjust(self):
|
||||
if self._state is not True:
|
||||
_LOGGER.debug(self._name + " off - not adjusting")
|
||||
return False
|
||||
elif self._cl.data is None:
|
||||
_LOGGER.debug(self._name + " could not retrieve Circadian Lighting data")
|
||||
if self._is_disabled():
|
||||
return False
|
||||
elif self._disable_entity is not None and self.hass.states.get(self._disable_entity).state == self._disable_state:
|
||||
_LOGGER.debug(self._name + " disabled by " + str(self._disable_entity))
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
return True
|
||||
|
||||
def adjust_lights(self, lights, transition=None):
|
||||
if self.should_adjust():
|
||||
if transition == None:
|
||||
transition = self._cl.data['transition']
|
||||
async def _adjust_lights(self, lights, transition):
|
||||
if not self._should_adjust():
|
||||
return
|
||||
|
||||
brightness = int((self._attributes['brightness'] / 100) * 254) if self._attributes['brightness'] is not None else None
|
||||
mired = int(self.calc_ct()) if self._lights_ct is not None else None
|
||||
rgb = tuple(map(int, self.calc_rgb())) if self._lights_rgb is not None else None
|
||||
xy = self.calc_xy() if self._lights_xy is not None else None
|
||||
if transition is None:
|
||||
transition = self._circadian_lighting._transition
|
||||
|
||||
for light in lights:
|
||||
"""Set color of array of ct light if on."""
|
||||
if self._lights_ct is not None and light in self._lights_ct and is_on(self.hass, light):
|
||||
service_data = {ATTR_ENTITY_ID: light}
|
||||
if mired is not None:
|
||||
service_data[ATTR_COLOR_TEMP] = mired
|
||||
if brightness is not None:
|
||||
service_data[ATTR_BRIGHTNESS] = brightness
|
||||
if transition is not None:
|
||||
service_data[ATTR_TRANSITION] = transition
|
||||
self.hass.services.call(
|
||||
LIGHT_DOMAIN, SERVICE_TURN_ON, service_data)
|
||||
_LOGGER.debug(light + " CT Adjusted - color_temp: " + str(mired) + ", brightness: " + str(brightness) + ", transition: " + str(transition))
|
||||
tasks = []
|
||||
for light in lights:
|
||||
if not is_on(self.hass, light):
|
||||
continue
|
||||
|
||||
"""Set color of array of rgb light if on."""
|
||||
if self._lights_rgb is not None and light in self._lights_rgb and is_on(self.hass, light):
|
||||
service_data = {ATTR_ENTITY_ID: light}
|
||||
if rgb is not None:
|
||||
service_data[ATTR_RGB_COLOR] = rgb
|
||||
if brightness is not None:
|
||||
service_data[ATTR_BRIGHTNESS] = brightness
|
||||
if transition is not None:
|
||||
service_data[ATTR_TRANSITION] = transition
|
||||
self.hass.services.call(
|
||||
LIGHT_DOMAIN, SERVICE_TURN_ON, service_data)
|
||||
_LOGGER.debug(light + " RGB Adjusted - rgb_color: " + str(rgb) + ", brightness: " + str(brightness) + ", transition: " + str(transition))
|
||||
service_data = {ATTR_ENTITY_ID: light}
|
||||
if self._brightness is not None:
|
||||
service_data[ATTR_BRIGHTNESS] = int((self._brightness / 100) * 254)
|
||||
if transition is not None:
|
||||
service_data[ATTR_TRANSITION] = transition
|
||||
|
||||
"""Set color of array of xy light if on."""
|
||||
if self._lights_xy is not None and light in self._lights_xy and is_on(self.hass, light):
|
||||
service_data = {ATTR_ENTITY_ID: light}
|
||||
if xy is not None:
|
||||
service_data[ATTR_XY_COLOR] = xy
|
||||
if brightness is not None:
|
||||
service_data[ATTR_BRIGHTNESS] = brightness
|
||||
service_data[ATTR_WHITE_VALUE] = brightness
|
||||
if transition is not None:
|
||||
service_data[ATTR_TRANSITION] = transition
|
||||
self.hass.services.call(
|
||||
LIGHT_DOMAIN, SERVICE_TURN_ON, service_data)
|
||||
_LOGGER.debug(light + " XY Adjusted - xy_color: " + str(xy) + ", brightness: " + str(brightness) + ", transition: " + str(transition) + ", white_value: " + str(brightness))
|
||||
light_type = self._lights_types[light]
|
||||
if light_type == "ct":
|
||||
service_data[ATTR_COLOR_TEMP] = int(self._calc_ct())
|
||||
elif light_type == "rgb":
|
||||
r, g, b = self._calc_rgb()
|
||||
service_data[ATTR_RGB_COLOR] = (int(r), int(g), int(b))
|
||||
elif light_type == "xy":
|
||||
service_data[ATTR_XY_COLOR] = self._calc_xy()
|
||||
if service_data.get(ATTR_BRIGHTNESS, False):
|
||||
service_data[ATTR_WHITE_VALUE] = service_data[ATTR_BRIGHTNESS]
|
||||
|
||||
"""Set color of array of brightness light if on."""
|
||||
if self._lights_brightness is not None and light in self._lights_brightness and is_on(self.hass, light):
|
||||
service_data = {ATTR_ENTITY_ID: light}
|
||||
if brightness is not None:
|
||||
service_data[ATTR_BRIGHTNESS] = brightness
|
||||
if transition is not None:
|
||||
service_data[ATTR_TRANSITION] = transition
|
||||
self.hass.services.call(
|
||||
LIGHT_DOMAIN, SERVICE_TURN_ON, service_data)
|
||||
_LOGGER.debug(light + " Brightness Adjusted - brightness: " + str(brightness) + ", transition: " + str(transition))
|
||||
tasks.append(
|
||||
self.hass.services.async_call(
|
||||
LIGHT_DOMAIN, SERVICE_TURN_ON, service_data
|
||||
)
|
||||
)
|
||||
if tasks:
|
||||
await asyncio.wait(tasks)
|
||||
|
||||
def light_state_changed(self, entity_id, from_state, to_state):
|
||||
try:
|
||||
_LOGGER.debug(entity_id + " change from " + str(from_state) + " to " + str(to_state))
|
||||
if to_state.state == 'on' and from_state.state != 'on':
|
||||
self.adjust_lights([entity_id], self._initial_transition)
|
||||
except:
|
||||
pass
|
||||
async def _light_state_changed(self, entity_id, from_state, to_state):
|
||||
if to_state.state == "on" and from_state.state != "on":
|
||||
_LOGGER.debug(_difference_between_states(from_state, to_state))
|
||||
await self._force_update_switch(lights=[entity_id])
|
||||
|
||||
def sleep_state_changed(self, entity_id, from_state, to_state):
|
||||
try:
|
||||
_LOGGER.debug(entity_id + " change from " + str(from_state) + " to " + str(to_state))
|
||||
if to_state.state == self._sleep_state or from_state.state == self._sleep_state:
|
||||
self.update_switch(self._initial_transition)
|
||||
except:
|
||||
pass
|
||||
|
||||
def disable_state_changed(self, entity_id, from_state, to_state):
|
||||
try:
|
||||
_LOGGER.debug(entity_id + " change from " + str(from_state) + " to " + str(to_state))
|
||||
if from_state.state == self._disable_state:
|
||||
self.update_switch(self._initial_transition)
|
||||
except:
|
||||
pass
|
||||
async def _state_changed(self, entity_id, from_state, to_state):
|
||||
_LOGGER.debug(_difference_between_states(from_state, to_state))
|
||||
await self._force_update_switch()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue