95182d2196
* mock the aiomqtt library and increse coverage * test inv response for a mb scan request * improve test coverage
31 lines
843 B
Python
31 lines
843 B
Python
import logging
|
|
import json
|
|
from mqtt import Mqtt
|
|
from aiocron import crontab
|
|
from infos import ClrAtMidnight
|
|
|
|
logger_mqtt = logging.getLogger('mqtt')
|
|
|
|
|
|
class Schedule:
|
|
mqtt = None
|
|
count = 0
|
|
|
|
@classmethod
|
|
def start(cls) -> None: # pragma: no cover
|
|
'''Start the scheduler and schedule the tasks (cron jobs)'''
|
|
logging.debug("Scheduler init")
|
|
cls.mqtt = Mqtt(None)
|
|
|
|
crontab('0 0 * * *', func=cls.atmidnight, start=True)
|
|
|
|
@classmethod
|
|
async def atmidnight(cls) -> None: # pragma: no cover
|
|
'''Clear daily counters at midnight'''
|
|
logging.info("Clear daily counters at midnight")
|
|
|
|
for key, data in ClrAtMidnight.elm():
|
|
logger_mqtt.debug(f'{key}: {data}')
|
|
data_json = json.dumps(data)
|
|
await cls.mqtt.publish(f"{key}", data_json)
|