added scheduler code

This commit is contained in:
2024-12-11 17:41:52 +01:00
parent aeabdf4c3f
commit 7e36084b0f
8 changed files with 69 additions and 23 deletions
+33 -3
View File
@@ -1,6 +1,36 @@
from app_class import AppClass
from config import Config
from trigger import Trigger
import asyncio
class TriggerScheduler:
def __init__(self):
pass
class TriggerScheduler(AppClass):
def __init__(self, _config: Config) -> None:
super().__init__(_config)
self.tasks = []
async def trigger(self, _trigger: Trigger) -> None:
"""Run a trigger with a specific name at a given interval."""
while True:
self.log.info("Trigger %s is running", _trigger.name)
_trigger.run()
await asyncio.sleep(_trigger.config.interval_in_seconds)
def start_trigger(self, _trigger: Trigger) -> None:
"""Start a new trigger."""
task = asyncio.create_task(self.trigger(_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:
await asyncio.sleep(1) # Keep the main function alive
except KeyboardInterrupt:
await self.stop_triggers()