utils/history.py

"""In-memory metric history buffer for time-series data."""

from collections import deque
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Any, Optional
import threading


@dataclass
class HistoryPoint:
    """A single point in metric history."""
    timestamp: datetime
    value: float
    labels: Dict[str, str] = field(default_factory=dict)


class MetricHistory:
    """
    Thread-safe in-memory buffer for metric history.

    Maintains a rolling window of metric values for charting.
    """

    def __init__(self, max_length: int = 300):
        """
        Initialize history buffer.

        Args:
            max_length: Maximum number of points to retain
        """
        self.max_length = max_length
        self._data: Dict[str, deque] = {}
        self._lock = threading.Lock()

    def add(self, metric_name: str, value: float, labels: Optional[Dict[str, str]] = None) -> None:
        """
        Add a data point to the history.

        Args:
            metric_name: Name of the metric
            value: Metric value
            labels: Optional labels for the metric
        """
        point = HistoryPoint(
            timestamp=datetime.now(),
            value=value,
            labels=labels or {}
        )

        # Create key including labels for differentiation
        key = self._make_key(metric_name, labels)

        with self._lock:
            if key not in self._data:
                self._data[key] = deque(maxlen=self.max_length)
            self._data[key].append(point)

    def get(
        self,
        metric_name: str,
        labels: Optional[Dict[str, str]] = None,
        limit: Optional[int] = None
    ) -> List[HistoryPoint]:
        """
        Get history for a metric.

        Args:
            metric_name: Name of the metric
            labels: Optional label filter
            limit: Maximum number of points to return

        Returns:
            List of history points
        """
        key = self._make_key(metric_name, labels)

        with self._lock:
            if key not in self._data:
                return []

            points = list(self._data[key])
            if limit:
                points = points[-limit:]
            return points

    def get_latest(
        self,
        metric_name: str,
        labels: Optional[Dict[str, str]] = None
    ) -> Optional[HistoryPoint]:
        """Get the most recent value for a metric."""
        points = self.get(metric_name, labels, limit=1)
        return points[-1] if points else None

    def get_all_series(self, metric_name: str) -> Dict[str, List[HistoryPoint]]:
        """
        Get all label combinations for a metric.

        Args:
            metric_name: Base metric name

        Returns:
            Dictionary mapping label strings to history lists
        """
        result = {}
        prefix = f"{metric_name}:"

        with self._lock:
            for key, points in self._data.items():
                if key == metric_name or key.startswith(prefix):
                    result[key] = list(points)

        return result

    def to_dataframe(self, metric_name: str, labels: Optional[Dict[str, str]] = None):
        """
        Convert history to pandas DataFrame.

        Args:
            metric_name: Name of the metric
            labels: Optional label filter

        Returns:
            pandas DataFrame with time and value columns
        """
        import pandas as pd

        points = self.get(metric_name, labels)

        if not points:
            return pd.DataFrame(columns=["time", "value"])

        return pd.DataFrame([
            {"time": p.timestamp, "value": p.value, **p.labels}
            for p in points
        ])

    def clear(self, metric_name: Optional[str] = None) -> None:
        """
        Clear history.

        Args:
            metric_name: If provided, clear only this metric; otherwise clear all
        """
        with self._lock:
            if metric_name:
                keys_to_remove = [
                    k for k in self._data.keys()
                    if k == metric_name or k.startswith(f"{metric_name}:")
                ]
                for key in keys_to_remove:
                    del self._data[key]
            else:
                self._data.clear()

    def _make_key(self, metric_name: str, labels: Optional[Dict[str, str]]) -> str:
        """Create a unique key from metric name and labels."""
        if not labels:
            return metric_name

        label_str = ",".join(f"{k}={v}" for k, v in sorted(labels.items()))
        return f"{metric_name}:{label_str}"