68 lines
2.9 KiB
Python
68 lines
2.9 KiB
Python
from app_class import AppClass
|
|
from config import Config
|
|
from trigger import Trigger
|
|
from governor import GovernorControl, Governor, _governor_list
|
|
import asyncio
|
|
from asyncio import Task
|
|
from functools import cached_property
|
|
|
|
|
|
class TriggerScheduler(AppClass):
|
|
def __init__(self, _config: Config) -> None:
|
|
super().__init__(_config)
|
|
self.tasks: list[Task] = []
|
|
self.running_triggers: list[Trigger] = []
|
|
|
|
@cached_property
|
|
def governor_control(self) -> GovernorControl:
|
|
return GovernorControl(self._config)
|
|
|
|
def establish_preferred_governor(self) -> str:
|
|
preferred_governor: Governor = [gov for gov in _governor_list if gov.name == 'powersave'].pop()
|
|
for trigger in self.running_triggers:
|
|
self.log.debug('trigger %s, state: %s, preferred governor: %s', trigger.name, trigger.trigger_state, trigger.governor.name)
|
|
if trigger.active:
|
|
if trigger.governor.priority < preferred_governor.priority:
|
|
preferred_governor = trigger.governor
|
|
self.log.debug('preferred governor: %s', preferred_governor.name)
|
|
return preferred_governor.name
|
|
|
|
async def callback_trigger(self, _trigger: Trigger) -> None:
|
|
"""Run a callback_trigger with a specific name and check its status at a given interval."""
|
|
await _trigger.async_run()
|
|
while True:
|
|
self.log.debug("Checking state of trigger %s, state: %s", _trigger.name, _trigger.trigger_state)
|
|
await asyncio.sleep(_trigger.config.interval_in_seconds)
|
|
|
|
async def run_once_trigger(self, _trigger: Trigger) -> None:
|
|
"""Run a trigger with a specific name at a given interval."""
|
|
while True:
|
|
_trigger.run()
|
|
self.log.debug("Checking state of trigger %s, state: %s", _trigger.name, _trigger.trigger_state)
|
|
await asyncio.sleep(_trigger.config.interval_in_seconds)
|
|
|
|
def start_trigger(self, _trigger: Trigger) -> None:
|
|
"""Start a new trigger."""
|
|
if _trigger.config.type == "callback":
|
|
task = asyncio.create_task(self.callback_trigger(_trigger))
|
|
else:
|
|
task = asyncio.create_task(self.run_once_trigger(_trigger))
|
|
self.running_triggers.append(_trigger)
|
|
self.tasks.append(task)
|
|
|
|
async def stop_triggers(self) -> None:
|
|
"""Stop all triggers."""
|
|
for task in self.tasks:
|
|
task.cancel()
|
|
await asyncio.gather(*self.tasks, return_exceptions=True)
|
|
|
|
async def run(self) -> None:
|
|
"""Run the scheduler and keep it alive until stopped."""
|
|
try:
|
|
while True:
|
|
self.establish_preferred_governor()
|
|
await asyncio.sleep(1) # Keep the main function alive
|
|
except asyncio.exceptions.CancelledError:
|
|
self.log.info("Exiting, stopping all triggers")
|
|
await self.stop_triggers()
|