Source code for draugr.writers.mixins.scalar_writer_mixin
from abc import ABC, abstractmethod
from itertools import cycle
from warg import Number, passes_kws_to, drop_unused_kws
from draugr.python_utilities import CounterFilter
__all__ = ["ScalarWriterMixin"]
[docs]class ScalarWriterMixin(CounterFilter, ABC):
"""description"""
@abstractmethod
def _scalar(self, tag: str, value: float, step: int):
raise NotImplementedError
[docs] @passes_kws_to(CounterFilter.__init__)
@drop_unused_kws
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._blip_iterators = {}
[docs] def scalar(self, tag: str, value: Number, step_i: int = None) -> None:
"""
:param tag:
:type tag:
:param value:
:type value:
:param step_i:
:type step_i:"""
if step_i:
self._counter[tag] = step_i
else:
self._counter[tag] += 1
if self.filter(tag):
self._scalar(tag, value, self._counter[tag])
[docs] def blip(self, tag: str, step_i: int = None) -> None:
"""
:param tag:
:type tag:
:param step_i:
:type step_i:"""
if tag not in self._blip_iterators:
self._blip_iterators[tag] = iter(cycle(range(2)))
self.scalar(tag, next(self._blip_iterators[tag]), step_i=step_i)