transformation_service

django_spire.metric.visual.services.transformation_service

VISUAL_AGGREGATE_CACHE_TTL_SECONDS = 120 module-attribute

VisualTransformationService

Bases: BaseDjangoModelService['Visual']

Source code in django_spire/contrib/constructor/constructor.py

def __init__(self, obj: Any = None):
    self._obj_type_name: str = str(next(iter(self.__class__.__annotations__.values()))).split(
        '.'
    )[-1]

    if obj is None:
        return

    self._obj_mro_type_names = [cls.__name__ for cls in obj.__class__.__mro__]

    if self._obj_type_name not in self._obj_mro_type_names:
        message = f'{self.__class__.__name__} was instantiated with obj type "{obj.__class__.__name__}" and failed as it was expecting "{self._obj_type_name}".'
        raise ConstructorError(message)

    self._obj_type: type[TypeAny] = obj.__class__

    if self._obj_type is None or self._obj_type is ...:
        message = f'{self.__class__.__name__} top class attribute must have an annotated type.'
        raise ConstructorError(message)

    self.obj: TypeAny = obj

    if ABC not in self.__class__.__bases__:
        if not self._obj_is_valid:
            message = f'{self._obj_type_name} failed to validate on {self.__class__.__name__}'
            raise ConstructorError(message)

    self.__post_init__()

obj instance-attribute

date_range

Source code in django_spire/metric/visual/services/transformation_service.py

def date_range(self, value_date: date | None = None) -> tuple[date, date]:
    value_date = value_date or self.obj.date

    interval = self.obj.statistic.interval if self.obj.statistic_id else None

    if not interval:
        return value_date, value_date

    return interval_range(interval, value_date)

current_value

Source code in django_spire/metric/visual/services/transformation_service.py

def current_value(
    self, value_date: date | None = None, *, reference: str | None = None
) -> Decimal:
    if not self.obj.statistic_id:
        return Decimal(0)

    if self._statistic_deleted():
        return Decimal(0)

    key = self._cache_key('value', value_date=value_date, reference=reference)
    cached = cache.get(key)
    if cached is not None:
        return cached

    value_date = value_date or self.obj.date

    values = self._values_for(reference) if reference is not None else self._statistic_values()

    if self._is_percentage():
        window_days = percentage_moving_window_days(self.obj.statistic.interval)
        result = values.moving_window_average(value_date, window_days)
    else:
        start_date, end_date = self.date_range(value_date)
        result = values.date_range(start_date, end_date).total()

    cache.set(key, result, VISUAL_AGGREGATE_CACHE_TTL_SECONDS)
    return result

current_condition

Source code in django_spire/metric/visual/services/transformation_service.py

def current_condition(
    self, value_date: date | None = None, *, value: Decimal | None = None
) -> VisualCondition | None:
    if value is None:
        value = self.current_value(value_date)

    for condition in self.obj.conditions.all():
        if condition.matches(value):
            return condition

    return None

series_datasets

Source code in django_spire/metric/visual/services/transformation_service.py

def series_datasets(self, value_date: date | None = None) -> list[dict]:
    if not self.obj.statistic_id:
        return []

    if self._statistic_deleted():
        return []

    key = self._cache_key('series', value_date=value_date)
    cached = cache.get(key)
    if cached is not None:
        return cached

    value_date = value_date or self.obj.date

    datasets = self._datasets()

    if not datasets:
        result = [
            {
                'label': self.obj.name,
                'points': self._series_points(value_date, self._statistic_values()),
            }
        ]
    else:
        result = [
            {
                'label': str(dataset),
                'points': self._series_points(value_date, self._values_for(dataset.reference)),
            }
            for dataset in datasets
        ]

    cache.set(key, result, VISUAL_AGGREGATE_CACHE_TTL_SECONDS)
    return result

series_data

Source code in django_spire/metric/visual/services/transformation_service.py

def series_data(self, value_date: date | None = None) -> list[dict]:
    datasets = self.series_datasets(value_date)

    if not datasets:
        return []

    return datasets[0]['points']

series_breakdown

Source code in django_spire/metric/visual/services/transformation_service.py

def series_breakdown(self, value_date: date | None = None) -> list[dict]:
    if not self.obj.statistic_id:
        return []

    if self._statistic_deleted():
        return []

    key = self._cache_key('breakdown', value_date=value_date)
    cached = cache.get(key)
    if cached is not None:
        return cached

    start_date, end_date = self.date_range(value_date)

    values = self._all_values()

    if self._is_percentage():
        breakdown = values.breakdown(start_date, end_date, average=True)
    else:
        breakdown = values.breakdown(start_date, end_date)

    labels = {dataset.reference: str(dataset) for dataset in self._datasets()}

    result = [
        {
            'name': next(
                (label for pattern, label in labels.items() if reference_matches(pattern, ref)),
                ref,
            ),
            'value': float(total),
        }
        for ref, total in breakdown
    ]
    cache.set(key, result, VISUAL_AGGREGATE_CACHE_TTL_SECONDS)
    return result

dataset_values

Source code in django_spire/metric/visual/services/transformation_service.py

def dataset_values(self, value_date: date | None = None) -> list[dict]:
    if not self.obj.statistic_id:
        return []

    if self._statistic_deleted():
        return []

    key = self._cache_key('dataset-values', value_date=value_date)
    cached = cache.get(key)
    if cached is not None:
        return cached

    value_date = value_date or self.obj.date

    datasets = self._datasets()

    if not datasets:
        result = [{'label': self.obj.name, 'value': self.current_value(value_date)}]
    else:
        result = [
            {
                'label': str(dataset),
                'value': self.current_value(value_date, reference=dataset.reference),
            }
            for dataset in datasets
        ]

    cache.set(key, result, VISUAL_AGGREGATE_CACHE_TTL_SECONDS)
    return result

gauge_max

Source code in django_spire/metric/visual/services/transformation_service.py

def gauge_max(self) -> int:
    key = self._cache_key('gauge', include_conditions=True)
    cached = cache.get(key)
    if cached is not None:
        return cached

    ceiling = Decimal(0)

    for condition in self.obj.conditions.all():
        upper = condition.target + condition.tolerance
        ceiling = max(ceiling, upper)

    if ceiling <= 0:
        ceiling = self.current_value() * Decimal(2)

    if ceiling <= 0:
        ceiling = Decimal(100)

    result = int(ceiling)
    cache.set(key, result, VISUAL_AGGREGATE_CACHE_TTL_SECONDS)
    return result

chart

Source code in django_spire/metric/visual/services/transformation_service.py

def chart(self) -> Any | None:
    from django_spire.metric.visual.charts import VISUAL_CHART_CLASSES  # noqa: PLC0415

    chart_class = VISUAL_CHART_CLASSES.get(self.obj.kind)

    if chart_class is None:
        return None

    return chart_class(params={'visual_pk': self.obj.pk})

render_context

Source code in django_spire/metric/visual/services/transformation_service.py

def render_context(self) -> dict:
    if self.obj.is_deleted or self._statistic_deleted():
        return self.empty_render_context()

    current_value = self.current_value()
    period_start, period_end = self.date_range()

    return {
        'visual': self.obj,
        'current_value': current_value,
        'current_condition': self.current_condition(value=current_value),
        'chart': self.chart(),
        'period_start': period_start,
        'period_end': period_end,
    }

empty_render_context staticmethod

Source code in django_spire/metric/visual/services/transformation_service.py

@staticmethod
def empty_render_context() -> dict:
    return {'visual': None, 'current_value': None, 'current_condition': None, 'chart': None}

VisualConditionTransformationService

Bases: BaseDjangoModelService['VisualCondition']

Source code in django_spire/contrib/constructor/constructor.py

def __init__(self, obj: Any = None):
    self._obj_type_name: str = str(next(iter(self.__class__.__annotations__.values()))).split(
        '.'
    )[-1]

    if obj is None:
        return

    self._obj_mro_type_names = [cls.__name__ for cls in obj.__class__.__mro__]

    if self._obj_type_name not in self._obj_mro_type_names:
        message = f'{self.__class__.__name__} was instantiated with obj type "{obj.__class__.__name__}" and failed as it was expecting "{self._obj_type_name}".'
        raise ConstructorError(message)

    self._obj_type: type[TypeAny] = obj.__class__

    if self._obj_type is None or self._obj_type is ...:
        message = f'{self.__class__.__name__} top class attribute must have an annotated type.'
        raise ConstructorError(message)

    self.obj: TypeAny = obj

    if ABC not in self.__class__.__bases__:
        if not self._obj_is_valid:
            message = f'{self._obj_type_name} failed to validate on {self.__class__.__name__}'
            raise ConstructorError(message)

    self.__post_init__()

obj instance-attribute

matches

Source code in django_spire/metric/visual/services/transformation_service.py

def matches(self, value: Decimal) -> bool:
    value = Decimal(value)

    comparisons: dict[str, bool] = {
        VisualConditionOperatorChoices.GT: value > self.obj.target,
        VisualConditionOperatorChoices.GTE: value >= self.obj.target,
        VisualConditionOperatorChoices.LT: value < self.obj.target,
        VisualConditionOperatorChoices.LTE: value <= self.obj.target,
        VisualConditionOperatorChoices.EQ: value == self.obj.target,
        VisualConditionOperatorChoices.BETWEEN: abs(value - self.obj.target)
        <= self.obj.tolerance,
    }

    return comparisons.get(self.obj.operator, False)

VisualRegionTransformationService

Bases: BaseDjangoModelService['VisualRegion']

Source code in django_spire/contrib/constructor/constructor.py

def __init__(self, obj: Any = None):
    self._obj_type_name: str = str(next(iter(self.__class__.__annotations__.values()))).split(
        '.'
    )[-1]

    if obj is None:
        return

    self._obj_mro_type_names = [cls.__name__ for cls in obj.__class__.__mro__]

    if self._obj_type_name not in self._obj_mro_type_names:
        message = f'{self.__class__.__name__} was instantiated with obj type "{obj.__class__.__name__}" and failed as it was expecting "{self._obj_type_name}".'
        raise ConstructorError(message)

    self._obj_type: type[TypeAny] = obj.__class__

    if self._obj_type is None or self._obj_type is ...:
        message = f'{self.__class__.__name__} top class attribute must have an annotated type.'
        raise ConstructorError(message)

    self.obj: TypeAny = obj

    if ABC not in self.__class__.__bases__:
        if not self._obj_is_valid:
            message = f'{self._obj_type_name} failed to validate on {self.__class__.__name__}'
            raise ConstructorError(message)

    self.__post_init__()

obj instance-attribute

display_title property

render_context

Source code in django_spire/metric/visual/services/transformation_service.py

def render_context(self) -> dict:
    if not self.obj.visual_id or self.obj.visual.is_deleted:
        context = VisualTransformationService.empty_render_context()
    else:
        context = self.obj.visual.services.transformation.render_context()

    context['display_title'] = self.display_title
    return context