copied cosmostat web dashboard for init dashboard work

api/sensors.py

@@ -0,0 +1,207 @@
+from __future__ import annotations
+
+import re, redis
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional, Union
+from config import *
+
+# --------------------------------------------------------------------------- #
+#  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
+        for sensor in comment_lookup:
+            if sensor['mac'] == self.name:
+                self.comment = sensor['comment']
+
+    # ----------------------------------------------------------------------- #
+    #  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))