ble_api/sensors.py

from __future__ import annotations

import re
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Union

# --------------------------------------------------------------------------- #
#  Helper utilities
# --------------------------------------------------------------------------- #

def _current_timestamp() -> int:
    """Return the current UTC timestamp (seconds)."""
    return int(datetime.now(timezone.utc).timestamp())

def _ensure_timestamp(ts: Optional[Union[int, datetime]]) -> int:
    """
    Normalise a timestamp that can be an int (Unix seconds) or a datetime.
    Raises ValueError if the value is not recognised.
    """
    if ts is None:
        return _current_timestamp()
    if isinstance(ts, int):
        return ts
    if isinstance(ts, datetime):
        return int(ts.replace(tzinfo=timezone.utc).timestamp())
    raise ValueError(f"Unsupported timestamp type: {type(ts)}")

# --------------------------------------------------------------------------- #
#  Sensor implementation
# --------------------------------------------------------------------------- #

@dataclass
class Sensor:
    """
    Represents a sensor that records timestamped values under a set of keys.

    Attributes
    ----------
    name : str
        The unique identifier for the sensor itself.
    first_seen : int
        Unix timestamp of when the sensor was first instantiated.
    last_seen : int
        Unix timestamp of the most recent data update (or creation time).
    data : List[Dict[str, Any]]
        A list of data points, each with keys:
            - name   (str)
            - value  (Any)
            - first_seen (int)
            - last_seen (int)
    """
    name: str
    first_seen: int = field(init=False)
    last_seen: int = field(init=False)
    data: List[Dict[str, Any]] = field(default_factory=list)

    # ----------------------------------------------------------------------- #
    #  Validity rules
    # ----------------------------------------------------------------------- #

    # Current rule: key must be a "word" (letters, digits, underscore)
    VALID_NAME_RE = re.compile(r"^[A-Za-z0-9_]+$")

    def _is_valid_key(self, key: str) -> bool:
        """Return True if the key meets current validity criteria."""
        is_valid = False
        if bool(self.VALID_NAME_RE.match(key)):
            is_valid = True
        return is_valid

    # ----------------------------------------------------------------------- #
    #  Life‑cycle hooks
    # ----------------------------------------------------------------------- #

    def __post_init__(self) -> None:
        """Initialise timestamps when the object is created."""
        print(f"Instantiating new sensor object - {self.name}")
        ts = _current_timestamp()
        self.first_seen = ts
        self.last_seen = ts
        self.comment = self.name

    # ----------------------------------------------------------------------- #
    #  Core API
    # ----------------------------------------------------------------------- #

    def add_or_update(self,
                      key: str,
                      value: Any,
                      timestamp: Optional[Union[int, datetime]] = None) -> None:
        """
        Add a new data point or update an existing one.

        Parameters
        ----------
        key : str
            The data key. Must pass :meth:`_is_valid_key`.
        value : Any
            The data value to store.
        timestamp : int | datetime | None
            When the data was recorded. If omitted, current time is used.

        Raises
        ------
        ValueError
            If the key is not valid.
        """
        if not self._is_valid_key(key):
            raise ValueError(f"Invalid key '{key}'. Must contain only "
                             "letters, digits, or underscores.")

        ts = _ensure_timestamp(timestamp)

        # Update sensor-level timestamps
        self.last_seen = max(self.last_seen, ts)

        # Search for existing entry
        for entry in self.data:
            if entry["name"] == key:
                # Update existing entry
                entry["value"] = value
                entry["last_seen"] = ts
                if ts < entry["first_seen"]:
                    entry["first_seen"] = ts
                return

        # If we get here, it's a new key
        new_entry: Dict[str, Any] = {
            "name": key,
            "value": value,
            "first_seen": ts,
            "last_seen": ts
        }
        self.data.append(new_entry)

    def get_value(self, key: str) -> Optional[Any]:
        """Return the most recent value for a key, or None if it doesn't exist."""
        for entry in self.data:
            if entry["name"] == key:
                return entry["value"]
        return None

    def update_comment(self, new_comment: str):
        self.comment = new_comment

    def to_dict(self, simple_return = False) -> Dict[str, List[Dict[str, Any]]]:
        """
        Return the data in the exact structure requested.

        Example
        -------
        >>> sensor.to_dict()
        {
            "data": [
                {"name": "temp", "value": 23, "first_seen": 1710000000, "last_seen": 1710000005},
                {"name": "hum",  "value": 45, "first_seen": 1710000010, "last_seen": 1710000010}
            ]
        }
        """
        if simple_return:
            simple_return = {
                "temp": self.get_value(key="temp"),
                "humidity": self.get_value(key="humidity")
            }
            return simple_return
        
        else:
            return {"data": self.data.copy()}

    # ----------------------------------------------------------------------- #
    #  Convenience & debugging
    # ----------------------------------------------------------------------- #

    def __repr__(self) -> str:
        return (f"{self.__class__.__name__}(name={self.name!r}, "
                f"first_seen={self.first_seen}, last_seen={self.last_seen}, "
                f"data={self.data!r})")

# --------------------------------------------------------------------------- #
#  Example usage
# --------------------------------------------------------------------------- #

if __name__ == "__main__":
    # Instantiate a sensor
    sensor = Sensor(name="weather_station_01")

    # Add some data
    sensor.add_or_update("temperature", 23.5)
    sensor.add_or_update("humidity", 45)

    # Update an existing key
    sensor.add_or_update("temperature", 24.0)

    # Try an invalid key (will raise ValueError)
    try:
        sensor.add_or_update("bad key!", 0)
    except ValueError as e:
        print(e)

    # Print the structured data
    import json
    print(json.dumps(sensor.to_dict(), indent=4))