Source code for shyft.dashboard.time_series.tools.base
import abc
import logging
from typing import Any
from shyft.dashboard.base.hashable import Hashable
from shyft.dashboard.time_series.bindable import BindableToMany
from shyft.dashboard.base.ports import States, StatePorts
[docs]
class BaseTool(Hashable, BindableToMany):
[docs]
def __init__(self, logger=None, *, parent_limit=None):
"""
Parameters
----------
logger:
optional logger
parent_limit:
number of max parents to be bound to
"""
Hashable.__init__(self)
BindableToMany.__init__(self, parent_limit=parent_limit)
self.logger = logger or logging.getLogger()
self._state: States = States.ACTIVE
self.state_port: StatePorts = StatePorts(parent=self, _receive_state=self._receive_state)
[docs]
@abc.abstractmethod
def on_bind(self, *, parent: Any) -> None:
"""
Abstract method which is call on bind to a parent
"""
pass
def _receive_state(self, state: States) -> None:
"""
State port function on receiving state, can be overwritten by inherited class
"""
if self._state == state:
return
if state in [States.ACTIVE, States.DEACTIVE]:
self._state = state
self.state_port.send_state(self._state)