Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
_lights_dim_on = self.args.get('lights_dim_on', '').split(',')
_lights_dim_off = self.args.get('lights_dim_off', '').split(',')
_lights_off = self.args.get('lights_off', '').split(',')
_switch_dim_group = self.args.get('switch_dim_lights_use')
self._lights = {"dim": {"on": _lights_dim_on, "off": _lights_dim_off},
"off": _lights_off,
"state": self.get_state(_switch_dim_group)}
# Listen for ambilight changes to change light dim group:
self.listen_state(self.ch_dim_lights_group, _switch_dim_group)
self._media_player = conf_data.get('media_player')
self._ios_notifier = conf_data.get('notifier').replace('.', '/')
self._target_sensor = conf_data.get('chatid_sensor')
# Listen for Kodi changes:
self._last_play = utils.get_now()
self.listen_state(self.kodi_state, self._media_player)
self.listen_event(self._receive_kodi_result,
EVENT_KODI_CALL_METHOD_RESULT)
# self.log('KodiAssist Initialized with dim_lights_on={}, '
def run_minutely(self, callback, start, **kwargs):
name = self.name
now = utils.get_now()
if start is None:
event = now + datetime.timedelta(minutes=1)
else:
event = now
event = event.replace(second=start.second)
if event < now:
event = event + datetime.timedelta(minutes=1)
handle = self.run_every(callback, event, 60, **kwargs)
return handle
def run_every(self, callback, start, interval, **kwargs):
name = self.name
now = utils.get_now()
if start < now:
raise ValueError("start cannot be in the past")
utils.log(
conf.logger, "DEBUG",
"Registering run_every starting {} in {}s intervals for {}".format(
start, interval, name
)
)
exec_time = start.timestamp()
handle = utils.insert_schedule(name, exec_time, callback, True, None,
interval=interval, **kwargs)
return handle
def run_once(self, callback, start, **kwargs):
name = self.name
now = utils.get_now()
today = now.date()
event = datetime.datetime.combine(today, start)
if event < now:
one_day = datetime.timedelta(days=1)
event = event + one_day
exec_time = event.timestamp()
handle = utils.insert_schedule(
name, exec_time, callback, False, None, **kwargs
)
return handle
def run_at(self, callback, start, **kwargs):
name = self.name
now = utils.get_now()
if start < now:
raise ValueError(
"{}: run_at() Start time must be "
"in the future".format(self.name)
)
exec_time = start.timestamp()
handle = utils.insert_schedule(
name, exec_time, callback, False, None, **kwargs
)
return handle
def kodi_state(self, entity, attribute, old, new, kwargs):
"""Kodi state change main control."""
if new == 'playing':
kodi_attrs = self.get_state(
entity_id=self._media_player, attribute="attributes")
self._is_playing_video = (
'media_content_type' in kodi_attrs
and kodi_attrs['media_content_type'] in TYPE_HA_ITEMS_NOTIFY)
# self.log('KODI ATTRS: {}, is_playing_video={}'
# .format(kodi_attrs, self._is_playing_video))
if self._is_playing_video:
self._ask_for_playing_item()
elif ((new == 'idle') and self._is_playing_video) or (new == 'off'):
self._is_playing_video = False
self._last_play = utils.get_now()
self.log('KODI STOP. old:{}, new:{}, type_lp={}'
.format(old, new, type(self._last_play)), LOG_LEVEL)
# self._item_playing = None
self._adjust_kodi_lights(play=False)
def run_daily(self, callback, start, **kwargs):
name = self.name
now = utils.get_now()
today = now.date()
event = datetime.datetime.combine(today, start)
if event < now:
event = event + datetime.timedelta(days=1)
handle = self.run_every(callback, event, 24 * 60 * 60, **kwargs)
return handle
def _receive_kodi_result(self, event_id, payload_event, *args):
result = payload_event['result']
method = payload_event['input']['method']
if event_id == EVENT_KODI_CALL_METHOD_RESULT \
and method == METHOD_GET_ITEM:
if 'item' in result:
item = result['item']
new_video = (self._item_playing is None
or self._item_playing != item)
self._is_playing_video = item['type'] in TYPE_ITEMS_NOTIFY
self._item_playing = item
delta = utils.get_now() - self._last_play
if (self._is_playing_video and
(new_video or delta > dt.timedelta(minutes=20))):
self._last_play = utils.get_now()
self._adjust_kodi_lights(play=True)
# Notifications
self._notify_ios_message(self._item_playing)
self._notify_telegram_message(self._item_playing)
else:
self.log('RECEIVED BAD KODI RESULT: {}'
.format(result), 'warn')
elif event_id == EVENT_KODI_CALL_METHOD_RESULT \
and method == METHOD_GET_PLAYERS:
self.log('KODI GET_PLAYERS RECEIVED: {}'.format(result))