from typing import Dict from .abstract_plugin import AbstractCommand, AbstractPlugin, AbstractCommandCompiler import logging import time class WaitCommand(AbstractCommand): def __init__(self, logger: logging.Logger, secs: float): if type(secs) not in [float, int]: raise ValueError("Secs must be float or int") if secs <= 0: raise ValueError("Secs must be a positive integer") self._secs = secs self._logger = logger def execute(self): self._logger.debug(f"Sleeping for {self._secs} seconds") time.sleep(self._secs) self._logger.debug(f"Slept for {self._secs} seconds") def describe(self) -> dict: return { "command": "wait", "params": { "secs": self._secs } } class WaitCompiler(AbstractCommandCompiler): def __init__(self, logger: logging.Logger): self._logger = logger def compile(self, secs: float) -> AbstractCommand: return WaitCommand(self._logger, secs) class WaitPlugin(AbstractPlugin): plugin_name = "wait" def __init__(self): self._logger = logging.getLogger("plugin").getChild("wait") def load_compilers(self) -> Dict[str, AbstractCommandCompiler]: return { "wait": WaitCompiler(self._logger) } def close(self): pass