58 lines
2.0 KiB
Python
58 lines
2.0 KiB
Python
from app_class import AppClass
|
|
from config import Config, TriggerTuple
|
|
from exceptions import MissingConfig, TriggerImportError
|
|
from governor import Governor, _governor_list
|
|
from functools import cached_property
|
|
|
|
|
|
class Trigger(AppClass):
|
|
def __init__(self, _config: Config) -> None:
|
|
super().__init__(_config)
|
|
self.active: bool = False
|
|
|
|
def __hash__(self):
|
|
return super().__hash__(hash(self.name))
|
|
|
|
@property
|
|
def trigger_state(self) -> str:
|
|
return "active" if self.active else "not active"
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
if self.__module__ == 'trigger':
|
|
raise TriggerImportError(
|
|
"the Trigger class can't used directly, but must be inherited in a trigger specific class"
|
|
)
|
|
_ , _name = self.__module__.split('.')
|
|
return _name
|
|
|
|
@property
|
|
def config(self) -> TriggerTuple:
|
|
_config = self._config.get_trigger_by_name(self.name)
|
|
if _config is None:
|
|
raise MissingConfig(f"Trigger {self.name} hasn't been properly configured")
|
|
return _config
|
|
|
|
@cached_property
|
|
def governor(self) -> Governor:
|
|
_governor = next((gov for gov in _governor_list if self.config.governor == gov.name), None)
|
|
if _governor is None:
|
|
raise ValueError("unknown governor: %s", _governor)
|
|
return _governor
|
|
|
|
def run(self) -> None:
|
|
# this is the main function of the trigger class and has to stay active by using a while True loop
|
|
|
|
while True:
|
|
raise TriggerImportError(
|
|
"the Trigger class can't used directly, but must be inherited in a trigger specific class"
|
|
)
|
|
|
|
async def async_run(self) -> None:
|
|
# this is the main function of the trigger class and can be used to execute a callback function
|
|
|
|
while True:
|
|
raise TriggerImportError(
|
|
"the Trigger class can't used directly, but must be inherited in a trigger specific class"
|
|
)
|