elstat/elstat/worker.py

81 lines
2.4 KiB
Python
Raw Normal View History

import time
2018-07-07 23:51:43 +00:00
import asyncio
import logging
from .consts import ADAPTERS
class ServiceWorker:
def __init__(self, manager, name, service):
self.name = name
self.manager = manager
self.service = service
self.adapter = ADAPTERS[service['adapter']]
self.log = logging.getLogger(f'elstat.service.{name}')
self.last_poll = None
2018-07-07 23:51:43 +00:00
self._start()
async def work(self):
return await self.adapter.query(self, self.service['adapter_args'])
async def process_work(self, result: dict):
"""Process given adapter result and insert into
the database."""
columns = self.adapter.spec['db']
conn = self.manager.conn
timestamp = int(time.time() * 1000)
args_str = ','.join(['?'] * (len(result) + 1))
query = f"""
INSERT INTO {self.name} ({','.join(columns)})
VALUES ({args_str})
"""
args = []
for col in columns[1:]:
val = result[col]
args.append(val)
conn.execute(query, (timestamp, ) + tuple(args))
conn.commit()
2018-07-07 23:51:43 +00:00
await self._dispatch_work(columns, timestamp, result)
2018-07-11 06:42:31 +00:00
async def _dispatch_work(self, columns, timestamp: int, result: tuple):
2018-07-11 06:42:31 +00:00
prechan = columns[1:]
chans = [f'{chan}:{self.name}' for chan in prechan]
for col, chan in zip(prechan, chans):
self.manager.publish(chan, [timestamp, result[col]])
2018-07-11 06:42:31 +00:00
2018-07-07 23:51:43 +00:00
async def _work_loop(self):
try:
while True:
self.last_poll = time.monotonic()
res = await self.work()
self.manager.state[self.name] = res
await self.process_work(res)
2018-07-07 23:51:43 +00:00
await asyncio.sleep(self.service['poll'])
except asyncio.CancelledError:
self.log.info('cancelled, stopping')
2018-07-07 23:51:43 +00:00
except Exception:
self.log.exception('fail on work loop, retrying')
try:
self.manager.state[self.name]['status'] = False
self.manager.publish(f'status:{self.name}', False)
except KeyError:
pass
await self._work_loop()
2018-07-07 23:51:43 +00:00
def stop(self):
"""Stop the current worker."""
self._work_task.cancel()
2018-07-07 23:51:43 +00:00
def _start(self):
self.log.info(f'starting work loop for {self.name}')
self._work_task = self.manager.loop.create_task(self._work_loop())