"""Business semantics for bounded historical monitoring trends.""" from datetime import datetime, timedelta, timezone from uuid import UUID from app.monitoring.exceptions import MonitoringScopeNotFound from app.monitoring.models import ( MonitoringAvailableScopes, MonitoringRoomOption, MonitoringScope, MonitoringSiteOption, MonitoringTrendData, MonitoringTrendMeta, MonitoringTrendPoint, MonitoringTrendQuery, MonitoringTrendResponse, MonitoringTrendSeries, ) from app.monitoring.repository import MonitoringRepository RANGE_CONFIGURATION = { "6h": (timedelta(hours=6), timedelta(minutes=5)), "24h": (timedelta(hours=24), timedelta(minutes=15)), "7d": (timedelta(days=7), timedelta(hours=1)), } MAXIMUM_POINTS = 200 class MonitoringService: """Coordinate factual telemetry aggregation without HTTP dependencies.""" def __init__(self, repository: MonitoringRepository): self.repository = repository @staticmethod def _window( query: MonitoringTrendQuery, generated_at: datetime, ) -> tuple[datetime, datetime, timedelta]: duration, bucket_width = RANGE_CONFIGURATION[query.time_range] generated_at = generated_at.astimezone(timezone.utc) bucket_seconds = int(bucket_width.total_seconds()) current_bucket_epoch = ( int(generated_at.timestamp()) // bucket_seconds * bucket_seconds ) current_bucket_start = datetime.fromtimestamp( current_bucket_epoch, tz=timezone.utc ) window_end = current_bucket_start + bucket_width return window_end - duration, window_end, bucket_width def get_trends( self, query: MonitoringTrendQuery, *, include_location_scopes: bool, generated_at: datetime | None = None, ) -> MonitoringTrendResponse: generated_at = generated_at or datetime.now(timezone.utc) window_start, window_end, bucket_width = self._window(query, generated_at) rows = self.repository.read_trends( scope=query.scope, scope_id=query.scope_id, window_start=window_start, window_end=window_end, bucket_width=bucket_width, ) if not rows or not rows[0]["scope_exists"]: raise MonitoringScopeNotFound(f"{query.scope.title()} not found") options = MonitoringAvailableScopes() query_count = 1 if include_location_scopes: option_rows = self.repository.list_location_scope_options() query_count += 1 options = MonitoringAvailableScopes( sites=[ MonitoringSiteOption.model_validate(row) for row in option_rows if row["option_type"] == "site" ], rooms=[ MonitoringRoomOption.model_validate(row) for row in option_rows if row["option_type"] == "room" ], ) generated_at = generated_at.astimezone(timezone.utc) hashrate_points = [] power_points = [] for row in rows: shared = { "bucket_start": row["bucket_start"], "bucket_end": row["bucket_end"], "observation_count": row["observation_count"], "observed_miner_count": row["observed_miner_count"], "scoped_miner_count": row["scoped_miner_count"], "incomplete": row["bucket_end"] > generated_at, } hashrate_points.append( MonitoringTrendPoint( **shared, value=( float(row["hashrate_ths"]) if row["hashrate_ths"] is not None else None ), metric_sample_count=row["hashrate_sample_count"], contributing_miner_count=( row["hashrate_contributing_miner_count"] ), ) ) power_points.append( MonitoringTrendPoint( **shared, value=( float(row["power_w"]) if row["power_w"] is not None else None ), metric_sample_count=row["power_sample_count"], contributing_miner_count=( row["power_contributing_miner_count"] ), ) ) if len(rows) > MAXIMUM_POINTS: raise RuntimeError("Monitoring query exceeded the fixed point bound") attribution = ( "current_placement_cohort" if query.scope in {"site", "room"} else "miner_identity" if query.scope == "miner" else "fleet" ) latest_observed_at = rows[0]["latest_observed_at"] return MonitoringTrendResponse( data=MonitoringTrendData( scope=MonitoringScope( type=query.scope, id=query.scope_id, label=rows[0]["scope_label"], ), series=[ MonitoringTrendSeries( metric="hashrate", unit="TH/s", label="Average recorded hashrate", points=hashrate_points, ), MonitoringTrendSeries( metric="power", unit="W", label="Average observed power", points=power_points, ), ], available_scopes=options, ), meta=MonitoringTrendMeta( generated_at=generated_at, latest_observed_at=latest_observed_at, attribution=attribution, range=query.time_range, bucket_seconds=int(bucket_width.total_seconds()), current_bucket_incomplete=window_end > generated_at, database_query_count=query_count, ), )