[#340] Merged alarm backend into the alarm plugin.

Closes: #340
This commit is contained in:
Fabio Manganiello 2023-12-06 01:31:50 +01:00
parent 2670d40094
commit 5ad1a62293
Signed by untrusted user: blacklight
GPG key ID: D90FBA7F76362774
7 changed files with 508 additions and 397 deletions

View file

@ -7,7 +7,6 @@ Backends
:caption: Backends: :caption: Backends:
platypush/backend/adafruit.io.rst platypush/backend/adafruit.io.rst
platypush/backend/alarm.rst
platypush/backend/button.flic.rst platypush/backend/button.flic.rst
platypush/backend/camera.pi.rst platypush/backend/camera.pi.rst
platypush/backend/chat.telegram.rst platypush/backend/chat.telegram.rst

View file

@ -1,5 +0,0 @@
``alarm``
===========================
.. automodule:: platypush.backend.alarm
:members:

View file

@ -1,351 +0,0 @@
import datetime
import enum
import os
import time
import threading
from typing import Optional, Union, Dict, Any, List
import croniter
from dateutil.tz import gettz
from platypush.backend import Backend
from platypush.context import get_bus, get_plugin
from platypush.message.event.alarm import (
AlarmStartedEvent,
AlarmDismissedEvent,
AlarmSnoozedEvent,
)
from platypush.plugins.media import MediaPlugin, PlayerState
from platypush.procedure import Procedure
class AlarmState(enum.IntEnum):
WAITING = 1
RUNNING = 2
DISMISSED = 3
SNOOZED = 4
SHUTDOWN = 5
class Alarm:
_alarms_count = 0
_id_lock = threading.RLock()
def __init__(
self,
when: str,
actions: Optional[list] = None,
name: Optional[str] = None,
audio_file: Optional[str] = None,
audio_plugin: Optional[str] = None,
audio_volume: Optional[Union[int, float]] = None,
snooze_interval: float = 300.0,
enabled: bool = True,
):
with self._id_lock:
self._alarms_count += 1
self.id = self._alarms_count
self.when = when
self.name = name or 'Alarm_{}'.format(self.id)
self.audio_file = None
if audio_file:
self.audio_file = os.path.abspath(os.path.expanduser(audio_file))
assert os.path.isfile(self.audio_file), 'No such audio file: {}'.format(
self.audio_file
)
self.audio_plugin = audio_plugin
self.audio_volume = audio_volume
self.snooze_interval = snooze_interval
self.state: Optional[AlarmState] = None
self.timer: Optional[threading.Timer] = None
self.actions = Procedure.build(
name=name, _async=False, requests=actions or [], id=self.id
)
self._enabled = enabled
self._runtime_snooze_interval = snooze_interval
def get_next(self) -> float:
now = datetime.datetime.now().replace(
tzinfo=gettz()
) # lgtm [py/call-to-non-callable]
try:
cron = croniter.croniter(self.when, now)
return cron.get_next()
except (AttributeError, croniter.CroniterBadCronError):
try:
timestamp = datetime.datetime.fromisoformat(self.when).replace(
tzinfo=gettz()
) # lgtm [py/call-to-non-callable]
except (TypeError, ValueError):
timestamp = datetime.datetime.now().replace(
tzinfo=gettz()
) + datetime.timedelta( # lgtm [py/call-to-non-callable]
seconds=int(self.when)
)
return timestamp.timestamp() if timestamp >= now else None
def is_enabled(self):
return self._enabled
def disable(self):
self._enabled = False
def enable(self):
self._enabled = True
def dismiss(self):
self.state = AlarmState.DISMISSED
self.stop_audio()
get_bus().post(AlarmDismissedEvent(name=self.name))
def snooze(self, interval: Optional[float] = None):
self._runtime_snooze_interval = interval or self.snooze_interval
self.state = AlarmState.SNOOZED
self.stop_audio()
get_bus().post(
AlarmSnoozedEvent(name=self.name, interval=self._runtime_snooze_interval)
)
def start(self):
if self.timer:
self.timer.cancel()
if self.get_next() is None:
return
interval = self.get_next() - time.time()
self.timer = threading.Timer(interval, self.callback())
self.timer.start()
self.state = AlarmState.WAITING
def stop(self):
self.state = AlarmState.SHUTDOWN
if self.timer:
self.timer.cancel()
self.timer = None
def _get_audio_plugin(self) -> MediaPlugin:
return get_plugin(self.audio_plugin)
def play_audio(self):
def thread():
self._get_audio_plugin().play(self.audio_file)
if self.audio_volume is not None:
self._get_audio_plugin().set_volume(self.audio_volume)
self.state = AlarmState.RUNNING
audio_thread = threading.Thread(target=thread)
audio_thread.start()
def stop_audio(self):
self._get_audio_plugin().stop()
def callback(self):
def _callback():
while True:
if self.state == AlarmState.SHUTDOWN:
break
if self.is_enabled():
get_bus().post(AlarmStartedEvent(name=self.name))
if self.audio_plugin and self.audio_file:
self.play_audio()
self.actions.execute()
time.sleep(10)
sleep_time = None
if self.state == AlarmState.RUNNING:
while True:
state = self._get_audio_plugin().status().output.get('state')
if state == PlayerState.STOP.value:
if self.state == AlarmState.SNOOZED:
sleep_time = self._runtime_snooze_interval
else:
self.state = AlarmState.WAITING
break
else:
time.sleep(10)
if self.state == AlarmState.SNOOZED:
sleep_time = self._runtime_snooze_interval
elif self.get_next() is None:
self.state = AlarmState.SHUTDOWN
break
if not sleep_time:
sleep_time = (
self.get_next() - time.time() if self.get_next() else 10
)
time.sleep(sleep_time)
return _callback
def to_dict(self):
return {
'name': self.name,
'id': self.id,
'when': self.when,
'next_run': self.get_next(),
'enabled': self.is_enabled(),
'state': self.state.name,
}
class AlarmBackend(Backend):
"""
Backend to handle user-configured alarms.
"""
def __init__(
self,
alarms: Optional[Union[list, Dict[str, Any]]] = None,
audio_plugin: str = 'media.mplayer',
*args,
**kwargs
):
"""
:param alarms: List or name->value dict with the configured alarms. Example:
.. code-block:: yaml
morning_alarm:
when: '0 7 * * 1-5' # Cron expression format: run every weekday at 7 AM
audio_file: ~/path/your_ringtone.mp3
audio_plugin: media.mplayer
audio_volume: 10 # 10%
snooze_interval: 300 # 5 minutes snooze
actions:
- action: tts.say
args:
text: Good morning
- action: light.hue.bri
args:
value: 1
- action: light.hue.bri
args:
value: 140
transitiontime: 150
one_shot_alarm:
when: '2020-02-18T07:00:00.000000' # One-shot execution, with timestamp in ISO format
audio_file: ~/path/your_ringtone.mp3
actions:
- action: light.hue.on
:param audio_plugin: Media plugin (instance of :class:`platypush.plugins.media.MediaPlugin`) that will be
used to play the alarm audio (default: ``media.mplayer``).
"""
super().__init__(*args, **kwargs)
alarms = alarms or []
if isinstance(alarms, dict):
alarms = [{'name': name, **alarm} for name, alarm in alarms.items()]
self.audio_plugin = audio_plugin
alarms = [
Alarm(**{'audio_plugin': self.audio_plugin, **alarm}) for alarm in alarms
]
self.alarms: Dict[str, Alarm] = {alarm.name: alarm for alarm in alarms}
def add_alarm(
self,
when: str,
actions: list,
name: Optional[str] = None,
audio_file: Optional[str] = None,
audio_volume: Optional[Union[int, float]] = None,
enabled: bool = True,
) -> Alarm:
alarm = Alarm(
when=when,
actions=actions,
name=name,
enabled=enabled,
audio_file=audio_file,
audio_plugin=self.audio_plugin,
audio_volume=audio_volume,
)
if alarm.name in self.alarms:
self.logger.info('Overwriting existing alarm {}'.format(alarm.name))
self.alarms[alarm.name].stop()
self.alarms[alarm.name] = alarm
self.alarms[alarm.name].start()
return self.alarms[alarm.name]
def _get_alarm(self, name) -> Alarm:
assert name in self.alarms, 'Alarm {} does not exist'.format(name)
return self.alarms[name]
def enable_alarm(self, name: str):
self._get_alarm(name).enable()
def disable_alarm(self, name: str):
self._get_alarm(name).disable()
def dismiss_alarm(self):
alarm = self.get_running_alarm()
if not alarm:
self.logger.info('No alarm is running')
return
alarm.dismiss()
def snooze_alarm(self, interval: Optional[str] = None):
alarm = self.get_running_alarm()
if not alarm:
self.logger.info('No alarm is running')
return
alarm.snooze(interval=interval)
def get_alarms(self) -> List[Alarm]:
return sorted(
self.alarms.values(),
key=lambda alarm: alarm.get_next(),
)
def get_running_alarm(self) -> Optional[Alarm]:
running_alarms = [
alarm for alarm in self.alarms.values() if alarm.state == AlarmState.RUNNING
]
return running_alarms[0] if running_alarms else None
def __enter__(self):
for alarm in self.alarms.values():
alarm.stop()
alarm.start()
self.logger.info(
'Initialized alarm backend with {} alarms'.format(len(self.alarms))
)
def __exit__(self, *_, **__):
for alarm in self.alarms.values():
alarm.stop()
self.logger.info('Alarm backend terminated')
def loop(self):
for name, alarm in self.alarms.copy().items():
if not alarm.timer or (
not alarm.timer.is_alive() and alarm.state == AlarmState.SHUTDOWN
):
del self.alarms[name]
time.sleep(10)
# vim:sw=4:ts=4:et:

View file

@ -1,10 +0,0 @@
manifest:
events:
platypush.message.event.alarm.AlarmDismissedEvent: when an alarm is dismissed.
platypush.message.event.alarm.AlarmSnoozedEvent: when an alarm is snoozed.
platypush.message.event.alarm.AlarmStartedEvent: when an alarm starts.
platypush.message.event.alarm.AlarmTimeoutEvent: when an alarm times out.
install:
pip: []
package: platypush.backend.alarm
type: backend

View file

@ -1,45 +1,236 @@
import sys
from typing import Optional, Dict, Any, List, Union from typing import Optional, Dict, Any, List, Union
from platypush.context import get_plugin
from platypush.backend.alarm import AlarmBackend from platypush.plugins import RunnablePlugin, action
from platypush.context import get_backend from platypush.plugins.media import MediaPlugin
from platypush.plugins import Plugin, action from platypush.utils import get_plugin_name_by_class
from platypush.utils.media import get_default_media_plugin
from ._model import Alarm, AlarmState
class AlarmPlugin(Plugin): class AlarmPlugin(RunnablePlugin):
""" """
Alarm/timer plugin. Alarm/timer plugin.
Requires: It requires at least one enabled ``media`` plugin to be configured if you
want to play audio resources.
- The :class:`platypush.backend.alarm.AlarmBackend` backend configured and enabled. Example configuration:
.. code-block:: yaml
alarm:
# Media plugin that will be used to play the alarm audio.
# If not specified, the first available configured media plugin
# will be used.
media_plugin: media.vlc
alarms:
morning_alarm:
# Cron expression format: run every weekday at 7 AM
when: '0 7 * * 1-5'
media: ~/path/your_ringtone.mp3
audio_volume: 10 # 10%
snooze_interval: 300 # 5 minutes snooze
actions:
- action: tts.say
args:
text: Good morning
- action: light.hue.bri
args:
value: 1
- action: light.hue.bri
args:
value: 140
transitiontime: 150
one_shot_alarm:
# One-shot execution, with timestamp in ISO format
when: '2020-02-18T07:00:00.000000'
media: ~/path/your_ringtone.mp3
actions:
- action: light.hue.on
timer:
# This alarm will execute the specified number of seconds
# after being initialized (5 minutes after the plugin has
# been initialized in this case)
when: 300
media: ~/path/your_ringtone.mp3
actions:
- action: light.hue.on
""" """
@staticmethod def __init__(
def _get_backend() -> AlarmBackend: self,
return get_backend('alarm') alarms: Optional[Union[list, Dict[str, Any]]] = None,
media_plugin: Optional[str] = None,
poll_interval: Optional[float] = 5.0,
**kwargs,
):
"""
:param alarms: List or name->value dict with the configured alarms. Example:
:param media_plugin: Media plugin (instance of
:class:`platypush.plugins.media.MediaPlugin`) that will be used to
play the alarm audio. It needs to be a supported local media
plugin, e.g. ``media.mplayer``, ``media.vlc``, ``media.mpv``,
``media.gstreamer`` etc. If not specified, the first available
configured local media plugin will be used. This only applies to
alarms that are configured to play an audio resource.
"""
super().__init__(poll_interval=poll_interval, **kwargs)
alarms = alarms or []
if isinstance(alarms, dict):
alarms = [{'name': name, **alarm} for name, alarm in alarms.items()]
if kwargs.get('audio_plugin'):
self.logger.warning(
'The audio_plugin parameter is deprecated. Use media_plugin instead'
)
media_plugin = media_plugin or kwargs.get('audio_plugin')
try:
plugin: Optional[MediaPlugin] = (
get_plugin(media_plugin) if media_plugin else get_default_media_plugin()
)
assert plugin, 'No media/audio plugin configured'
self.media_plugin = get_plugin_name_by_class(plugin.__class__)
except AssertionError:
self.media_plugin = None
self.logger.warning(
'No media plugin configured. Alarms that require audio playback will not work'
)
alarms = [
Alarm(
stop_event=self._should_stop,
**{'media_plugin': self.media_plugin, **alarm},
)
for alarm in alarms
]
self.alarms: Dict[str, Alarm] = {alarm.name: alarm for alarm in alarms}
def _get_alarms(self) -> List[Alarm]:
return sorted(
self.alarms.values(),
key=lambda alarm: alarm.get_next() or sys.maxsize,
)
def _get_alarm(self, name: str) -> Alarm:
assert name in self.alarms, f'The alarm {name} does not exist'
return self.alarms[name]
def _get_current_alarm(self) -> Optional[Alarm]:
return next(
iter(
alarm
for alarm in self.alarms.values()
if alarm.state == AlarmState.RUNNING
),
None,
)
def _enable(self, name: str):
self._get_alarm(name).enable()
def _disable(self, name: str):
self._get_alarm(name).disable()
def _add(
self,
when: Union[str, int, float],
actions: list,
name: Optional[str] = None,
media: Optional[str] = None,
audio_file: Optional[str] = None,
audio_volume: Optional[Union[int, float]] = None,
enabled: bool = True,
) -> Alarm:
alarm = Alarm(
when=when,
actions=actions,
name=name,
enabled=enabled,
media=media or audio_file,
media_plugin=self.media_plugin,
audio_volume=audio_volume,
stop_event=self._should_stop,
)
if alarm.name in self.alarms:
self.logger.info('Overwriting existing alarm: %s', alarm.name)
self.alarms[alarm.name].stop()
self.alarms[alarm.name] = alarm
self.alarms[alarm.name].start()
return self.alarms[alarm.name]
def _dismiss(self):
alarm = self._get_current_alarm()
if not alarm:
self.logger.info('No alarm is running')
return
alarm.dismiss()
def _snooze(self, interval: Optional[float] = None):
alarm = self._get_current_alarm()
if not alarm:
self.logger.info('No alarm is running')
return
alarm.snooze(interval=interval)
@action @action
def add(self, when: str, actions: Optional[list] = None, name: Optional[str] = None, def add(
audio_file: Optional[str] = None, audio_volume: Optional[Union[int, float]] = None, self,
enabled: bool = True) -> str: when: Union[str, int, float],
actions: Optional[list] = None,
name: Optional[str] = None,
media: Optional[str] = None,
audio_file: Optional[str] = None,
audio_volume: Optional[Union[int, float]] = None,
enabled: bool = True,
) -> str:
""" """
Add a new alarm. NOTE: alarms that aren't configured in the :class:`platypush.backend.alarm.AlarmBackend` Add a new alarm. NOTE: alarms that aren't statically defined in the
will only run in the current session. If you want an alarm to be permanently stored, you should configure plugin configuration will only run in the current session. If you want
it in the alarm backend configuration. You may want to add an alarm dynamically if it's a one-time alarm instead an alarm to be permanently stored, you should configure it in the alarm
backend configuration. You may want to add an alarm dynamically if it's
a one-time alarm instead.
:param when: When the alarm should be executed. It can be either a cron expression (for recurrent alarms), or :param when: When the alarm should be executed. It can be either a cron
a datetime string in ISO format (for one-shot alarms/timers), or an integer representing the number of expression (for recurrent alarms), or a datetime string in ISO
seconds before the alarm goes on (e.g. 300 for 5 minutes). format (for one-shot alarms/timers), or an integer/float
representing the number of seconds before the alarm goes on (e.g.
300 for 5 minutes).
:param actions: List of actions to be executed. :param actions: List of actions to be executed.
:param name: Alarm name. :param name: Alarm name.
:param audio_file: Path of the audio file to be played. :param media: Path of the audio file to be played.
:param audio_volume: Volume of the audio. :param audio_volume: Volume of the audio.
:param enabled: Whether the new alarm should be enabled (default: True). :param enabled: Whether the new alarm should be enabled (default: True).
:return: The alarm name. :return: The alarm name.
""" """
alarm = self._get_backend().add_alarm(when=when, audio_file=audio_file, actions=actions or [], if audio_file:
name=name, enabled=enabled, audio_volume=audio_volume) self.logger.warning(
'The audio_file parameter is deprecated. Use media instead'
)
alarm = self._add(
when=when,
media=media,
audio_file=audio_file,
actions=actions or [],
name=name,
enabled=enabled,
audio_volume=audio_volume,
)
return alarm.name return alarm.name
@action @action
@ -49,7 +240,7 @@ class AlarmPlugin(Plugin):
:param name: Alarm name. :param name: Alarm name.
""" """
self._get_backend().enable_alarm(name) self._enable(name)
@action @action
def disable(self, name: str): def disable(self, name: str):
@ -59,14 +250,14 @@ class AlarmPlugin(Plugin):
:param name: Alarm name. :param name: Alarm name.
""" """
self._get_backend().disable_alarm(name) self._disable(name)
@action @action
def dismiss(self): def dismiss(self):
""" """
Dismiss the alarm that is currently running. Dismiss the alarm that is currently running.
""" """
self._get_backend().dismiss_alarm() self._dismiss()
@action @action
def snooze(self, interval: Optional[float] = 300.0): def snooze(self, interval: Optional[float] = 300.0):
@ -76,16 +267,57 @@ class AlarmPlugin(Plugin):
:param interval: Snooze seconds before playing the alarm again (default: 300). :param interval: Snooze seconds before playing the alarm again (default: 300).
""" """
self._get_backend().snooze_alarm(interval=interval) self._snooze(interval=interval)
@action @action
def get_alarms(self) -> List[Dict[str, Any]]: def get_alarms(self) -> List[Dict[str, Any]]:
""" """
Get the list of configured alarms. Deprecated alias for :meth:`.status`.
:return: List of the alarms, sorted by next scheduled run.
""" """
return [alarm.to_dict() for alarm in self._get_backend().get_alarms()] self.logger.warning('get_alarms() is deprecated. Use status() instead')
return self.status() # type: ignore
@action
def status(self) -> List[Dict[str, Any]]:
"""
Get the list of configured alarms and their status.
:return: List of the alarms, sorted by next scheduled run. Example:
.. code-block:: json
[
{
"name": "Morning alarm",
"id": 1,
"when": "0 8 * * 1-5",
"next_run": "2023-12-06T08:00:00.000000",
"enabled": true,
"state": "RUNNING"
}
]
"""
return [alarm.to_dict() for alarm in self._get_alarms()]
def main(self):
for alarm in self.alarms.values():
alarm.start()
while not self.should_stop():
for name, alarm in self.alarms.copy().items():
if not alarm.timer or (
not alarm.timer.is_alive() and alarm.state == AlarmState.SHUTDOWN
):
del self.alarms[name]
self.wait_stop(self.poll_interval)
def stop(self):
for alarm in self.alarms.values():
alarm.stop()
super().stop()
# vim:sw=4:ts=4:et: # vim:sw=4:ts=4:et:

View file

@ -0,0 +1,242 @@
import datetime
import enum
import os
import time
import threading
from typing import Optional, Union
import croniter
from platypush.context import get_bus, get_plugin
from platypush.message.event.alarm import (
AlarmStartedEvent,
AlarmDismissedEvent,
AlarmSnoozedEvent,
)
from platypush.plugins.media import MediaPlugin, PlayerState
from platypush.procedure import Procedure
class AlarmState(enum.IntEnum):
"""
Alarm states.
"""
WAITING = 1
RUNNING = 2
DISMISSED = 3
SNOOZED = 4
SHUTDOWN = 5
UNKNOWN = -1
class Alarm:
"""
Alarm model and controller.
"""
_alarms_count = 0
_id_lock = threading.RLock()
def __init__(
self,
when: Union[str, int, float],
actions: Optional[list] = None,
name: Optional[str] = None,
media: Optional[str] = None,
media_plugin: Optional[str] = None,
audio_volume: Optional[Union[int, float]] = None,
snooze_interval: float = 300,
poll_interval: float = 5,
enabled: bool = True,
stop_event: Optional[threading.Event] = None,
):
with self._id_lock:
self._alarms_count += 1
self.id = self._alarms_count
self.when = when
self.name = name or f'Alarm_{self.id}'
self.media = self._get_media_resource(media)
self.media_plugin = media_plugin
self.audio_volume = audio_volume
self.snooze_interval = snooze_interval
self.state = AlarmState.UNKNOWN
self.timer: Optional[threading.Timer] = None
self.actions = Procedure.build(
name=name, _async=False, requests=actions or [], id=self.id
)
self._enabled = enabled
self._runtime_snooze_interval = snooze_interval
self.stop_event = stop_event or threading.Event()
self.poll_interval = poll_interval
@staticmethod
def _get_media_resource(media: Optional[str]) -> Optional[str]:
if not media:
return None
if media.startswith('file://'):
media = media[len('file://') :]
media_path = os.path.abspath(os.path.expanduser(media))
if os.path.isfile(media_path):
media = media_path
return media
def get_next(self) -> Optional[float]:
now = time.time()
t = 0
try:
# If when is a number, interpret it as number of seconds into the future
delta = float(self.when)
t = now + delta
self.when = datetime.datetime.fromtimestamp(t).isoformat()
except (TypeError, ValueError):
assert isinstance(self.when, str), f'Invalid alarm time {self.when}'
try:
# If when is a cron expression, get the next run time
t = croniter.croniter(self.when, now).get_next()
except (AttributeError, croniter.CroniterBadCronError):
try:
# If when is an ISO-8601 timestamp, parse it
t = datetime.datetime.fromisoformat(self.when).timestamp()
except Exception as e:
raise AssertionError(f'Invalid alarm time {self.when}: {e}') from e
return t if t >= now else None
def is_enabled(self):
return self._enabled
def disable(self):
self._enabled = False
def enable(self):
self._enabled = True
def dismiss(self):
self.state = AlarmState.DISMISSED
self.stop_audio()
get_bus().post(AlarmDismissedEvent(name=self.name))
def snooze(self, interval: Optional[float] = None):
self._runtime_snooze_interval = interval or self.snooze_interval
self.state = AlarmState.SNOOZED
self.stop_audio()
get_bus().post(
AlarmSnoozedEvent(name=self.name, interval=self._runtime_snooze_interval)
)
def start(self):
if self.timer:
self.timer.cancel()
if self.get_next() is None:
return
next_run = self.get_next()
if next_run is None:
return
interval = next_run - time.time()
self.timer = threading.Timer(interval, self.alarm_callback)
self.timer.start()
self.state = AlarmState.WAITING
def stop(self):
self.state = AlarmState.SHUTDOWN
if self.timer:
self.timer.cancel()
self.timer = None
def _get_media_plugin(self) -> MediaPlugin:
plugin = get_plugin(self.media_plugin)
assert plugin and isinstance(plugin, MediaPlugin), (
f'Invalid audio plugin {self.media_plugin}'
if plugin
else f'Missing audio plugin {self.media_plugin}'
)
return plugin
def play_audio(self):
def thread():
self._get_media_plugin().play(self.media)
if self.audio_volume is not None:
self._get_media_plugin().set_volume(self.audio_volume)
self.state = AlarmState.RUNNING
audio_thread = threading.Thread(target=thread)
audio_thread.start()
def stop_audio(self):
self._get_media_plugin().stop()
def alarm_callback(self):
while not self.should_stop():
if self.is_enabled():
get_bus().post(AlarmStartedEvent(name=self.name))
if self.media_plugin and self.media:
self.play_audio()
self.actions.execute()
self.wait_stop(self.poll_interval)
sleep_time = None
if self.state == AlarmState.RUNNING:
while not self.should_stop():
plugin_status = self._get_media_plugin().status().output
if not isinstance(plugin_status, dict):
self.wait_stop(self.poll_interval)
continue
state = plugin_status.get('state')
if state == PlayerState.STOP.value:
if self.state == AlarmState.SNOOZED:
sleep_time = self._runtime_snooze_interval
else:
self.state = AlarmState.WAITING
break
self.wait_stop(self.poll_interval)
if self.state == AlarmState.SNOOZED:
sleep_time = self._runtime_snooze_interval
elif self.get_next() is None:
self.state = AlarmState.SHUTDOWN
break
if not sleep_time:
next_run = self.get_next()
sleep_time = (
next_run - time.time()
if next_run is not None
else self.poll_interval
)
self.stop_event.wait(sleep_time)
def wait_stop(self, timeout: Optional[float] = None):
self.stop_event.wait(timeout)
def should_stop(self):
return self.stop_event.is_set() or self.state == AlarmState.SHUTDOWN
def to_dict(self):
return {
'name': self.name,
'id': self.id,
'when': self.when,
'next_run': self.get_next(),
'enabled': self.is_enabled(),
'state': self.state.name,
}
# vim:sw=4:ts=4:et:

View file

@ -1,5 +1,9 @@
manifest: manifest:
events: {} events:
- platypush.message.event.alarm.AlarmDismissedEvent
- platypush.message.event.alarm.AlarmSnoozedEvent
- platypush.message.event.alarm.AlarmStartedEvent
- platypush.message.event.alarm.AlarmTimeoutEvent
install: install:
pip: [] pip: []
package: platypush.plugins.alarm package: platypush.plugins.alarm