Aggregated Widget

AggregatedWidget is the base of a widget showing one number of its source, the latest or an aggregate over a window, like Number. See Aggregating.

dachshund.widgets.aggregate.AggregatedWidget dataclass

AggregatedWidget(name, *, source='', title=None, actions=list(), action_args=dict(), persist=False, y='value', aggregate='last', over=None)

Bases: PersistableWidget

A widget showing one number of its source: the latest, or an aggregate over a window.

Subclass it and write render_value(), which turns the number into the payload the renderer draws. See Aggregating.

Example

@dataclass
class Gauge(AggregatedWidget):
    type: ClassVar[str] = "gauge"
    script: ClassVar[Path] = Path(__file__).parent / "gauge.js"

    def render_value(self, value):
        return {"value": value}


Gauge("latency", aggregate="average", over=timedelta(hours=1))

Options

• y (str) - The field read. Defaults to "value".
• aggregate (str) - "last" for the latest number, or "count", "sum", "average", "min", "max" or "distinct" over over. Defaults to "last".
• over (timedelta | str) - The window: a timedelta back from now, or the current "day", "week" or "month" in the dashboard's timezone. Defaults to None, everything stored. Not for "last".
• persist (bool) - Keeps the latest number across restarts, for "last" only: windows are rebuilt from stored events. Needs the dashboard's storage. Defaults to False.

y class-attribute instance-attribute

y = 'value'

aggregate class-attribute instance-attribute

aggregate = 'last'

over class-attribute instance-attribute

over = None

kept class-attribute instance-attribute

kept = field(default=None, init=False, repr=False, compare=False)

window

window()

aggregate of y over over, None for "last".

Source code in dachshund/widgets/aggregate.py

def window(self) -> Window | None:
    """`aggregate` of `y` over `over`, `None` for `"last"`."""
    if self.aggregate == "last":
        return None

    return Window(fields=(self.y,), aggregate=self.aggregate, over=self.over)

render_value

render_value(value)

The payload drawn for value, None before there's one to show.

Source code in dachshund/widgets/aggregate.py

def render_value(self, value: float | None) -> dict[str, Any]:
    """The payload drawn for `value`, `None` before there's one to show."""
    raise NotImplementedError(f"{type(self).__name__} must write render_value().")

render

render(events)

The latest number of y, or None before anything arrives.

A persisting widget reads its own kept reading instead, since that covers more than whatever events are still around to re-read.

Source code in dachshund/widgets/aggregate.py

def render(self, events: Sequence[Event]) -> dict[str, Any]:
    """The latest number of `y`, or `None` before anything arrives.

    A persisting widget reads its own kept reading instead, since that
    covers more than whatever events are still around to re-read.
    """
    if self.persist:
        value = self.kept["value"] if self.kept else None
    else:
        value = self.value(events[-1], self.y) if events else None

    return self.render_value(value)

render_buckets

render_buckets(buckets)

The aggregate over the window's buckets.

Source code in dachshund/widgets/aggregate.py

def render_buckets(self, buckets: Mapping[str, Sequence[Bucket]]) -> dict[str, Any]:
    """The aggregate over the window's buckets."""
    return self.render_value(aggregated(merged(buckets[self.y]), self.aggregate))

accumulate

accumulate(event)

Keep the latest reading of y.

Source code in dachshund/widgets/aggregate.py

def accumulate(self, event: Event) -> None:
    """Keep the latest reading of `y`."""
    value = self.value(event, self.y)

    if value is None:
        return

    self.kept = {"value": value}

snapshot

snapshot()

The kept reading.

Source code in dachshund/widgets/aggregate.py

def snapshot(self) -> dict[str, Any]:
    """The kept reading."""
    return self.kept or {}

restore

restore(data)

Pick up where the last persisted snapshot left off.

Source code in dachshund/widgets/aggregate.py

def restore(self, data: dict[str, Any]) -> None:
    """Pick up where the last persisted snapshot left off."""
    self.kept = data