Source code for jvconnected.interfaces.base

import asyncio

from pydispatch import Dispatcher, Property

[docs]class Interface(Dispatcher): """Base interface class Subclasses must override the :meth:`open` and :meth:`close` methods. In order to operate with the :class:`~jvconnected.engine.Engine`, the class should be added to the :attr:`~jvconnected.interfaces.registry` Properties: running (bool): Run state Attributes: loop: The :class:`asyncio.BaseEventLoop` associated with the instance """ running = Property(False) def __init__(self, *args, **kwargs): self._engine = None self.loop = asyncio.get_event_loop() @property def engine(self) -> 'jvconnected.engine.Engine': """Instance of :class:`jvconnected.engine.Engine` """ return self._engine
[docs] async def set_engine(self, engine: 'jvconnected.engine.Engine'): """Attach the interface to a running instance of :class:`jvconnected.engine.Engine` This will be called automatically by the engine if the class is in the :attr:`jvconnected.interfaces.registry`. If the engine is running, the interface will start (using the :meth:`open` method). Otherwise it will automatically start when the engine does. """ if engine is self.engine: return assert self.engine is None self._engine = engine if engine.running: await self.open() engine.bind_async( self.loop, running=self.on_engine_running, )
[docs] async def open(self): """Open all communication methods """ raise NotImplementedError
[docs] async def close(self): """Stop communication """ raise NotImplementedError
async def on_engine_running(self, instance, value, **kwargs): if instance is not self.engine: return if value: if not self.running: await self.open() else: await self.close()