Source code for shyft.dashboard.time_series.dt_selector

from typing import List, Tuple
from enum import Enum

from shyft.time_series import Calendar, time

from shyft.dashboard.base.ports import Receiver, Sender, States
from shyft.dashboard.base.selector_model import SelectorModelBase
from shyft.dashboard.base.selector_presenter import SelectorPresenter


[docs] class CalendarDtStr(Enum): Year = int(Calendar.YEAR) Quarter = int(Calendar.QUARTER) Month = int(Calendar.MONTH) Week = int(Calendar.WEEK) Day = int(Calendar.DAY) Hour = int(Calendar.HOUR) Minute = int(Calendar.MINUTE) Second = int(Calendar.SECOND)
[docs] def calendar_unit_to_str(dt: Tuple[int, time]) -> str: if isinstance(dt, int): enum = CalendarDtStr(dt) elif isinstance(dt, time): enum = CalendarDtStr(int(dt)) else: raise TypeError() return str(enum.name)
[docs] def dt_to_str(dt: Tuple[int, time]) -> str: """ Convert fix dt to human-readable string (w/o calendar semantics)""" counts = [] dts_options = [e.value for e in CalendarDtStr] tres = int(dt) for option in dts_options: n = tres // option tres -= option * n counts.append(n) ses = ['s' if n > 1 else '' for n in counts] return ' '.join([f'{n} {calendar_unit_to_str(t)}{s}' for n, t, s in zip(counts, dts_options, ses) if n != 0])
[docs] def tdiff_to_str(cal: Calendar, t1: int, t2: int, simple: bool = True) -> str: """ Convert time difference as human-readable string using calendar semantics, like '1 Year 3 Months'""" if t2 < t1: t1, t2 = t2, t1 counts = [] dts = [e.value for e in CalendarDtStr] if simple: dts.remove(int(cal.QUARTER)) tres = t1 for dt in dts: if tres < t2: n = cal.diff_units(tres, t2, dt) tres = cal.add(tres, dt, n) else: n = 0 counts.append(n) ses = ['s' if n > 1 else '' for n in counts] return ' '.join([f'{n} {calendar_unit_to_str(t)}{s}' for n, t, s in zip(counts, dts, ses) if n != 0])
[docs] class DeltaTSelector(SelectorModelBase):
[docs] def __init__(self, presenter: SelectorPresenter, logger=None) -> None: """ dt selctor model used with TsViewer Parameters ---------- presenter: SelectorPresenter instance to use logger: optional logger instance """ super(DeltaTSelector, self).__init__(presenter=presenter, logger=logger) self.presenter.default = 'Auto' self.receive_selection_options = Receiver(parent=self, name='receive dt options', func=self._receive_selection_options, signal_type=List[int]) self.send_dt = Sender(parent=self, name='send dt', signal_type=int) self.state = States.DEACTIVE
[docs] def on_change_selected(self, selection_list: List[str]) -> None: if self.state == States.DEACTIVE or not selection_list or not selection_list[0]: return dt = self._convert_selected_dt(selection_list[0]) self.send_dt(dt)
@staticmethod def _convert_selected_dt(selected_dt: str) -> int: if 'Auto' in selected_dt: selected_dt = selected_dt.split('Auto: ')[-1] return int(selected_dt) def _receive_selection_options(self, dt_list: List[int]): if not isinstance(dt_list, List): self.presenter.set_selector_options(callback=False) return # generate options dt_list = sorted(dt_list) options = [(str(int(dt)), dt_to_str(dt)) for dt in dt_list] if options: self.presenter.default = ('Auto: {}'.format(options[0][0]), 'Auto: {}'.format(options[0][1])) else: self.presenter.default = 'Auto' # selection curr_select = None if len(dt_list) != 0: if not self.presenter.selected_values or not self.presenter.selected_values[0]: curr_select = dt_list[0] else: selected_value = self.presenter.selected_values[0] # if Auto if 'Auto' in selected_value: if options: curr_select = 'Auto: {}'.format(options[0][0]) else: curr_select = None # if not Auto else: selected_value = int(selected_value) if selected_value in dt_list: curr_select = str(selected_value) else: if dt_list[0] > selected_value: curr_select = str(int(dt_list[0])) elif dt_list[-1] < selected_value: curr_select = str(int(dt_list[-1])) self.presenter.set_selector_options(options, sort=False, selected_value=curr_select, callback=False) if curr_select: self.send_dt(self._convert_selected_dt(curr_select)) def _receive_state(self, state: States) -> None: if state == self.state: return self.state = state if state == States.ACTIVE: self.presenter.state_ports.receive_state(state) # Not sending active state since this only done if we can send data to the next widget elif state == States.DEACTIVE: self.presenter.default = ('', 'Auto') self.presenter.state_ports.receive_state(state) self.state_port.send_state(state) else: self.logger.error(f"ERROR: {self} - not handel for received state {state} implemented") self.state_port.send_state(state)