Add sensor datapoint access with linear interpolation support

SENSOR_DATAPOINT_FEATURE.md

@@ -0,0 +1,108 @@
+# Sensor Datapoint Access Feature
+
+This feature provides a simple GraphQL API for accessing single sensor datapoints before, after, or interpolated at a given datetime.
+
+## GraphQL Query
+
+```graphql
+query GetSensorDatapoint($sensorId: String!, $targetTime: DateTime!, $direction: String = "before") {
+  sensorDatapoint(sensorId: $sensorId, targetTime: $targetTime, direction: $direction) {
+    value
+    timestamp
+    isInterpolated
+    sourceReadings {
+      timestamp
+      value
+    }
+  }
+}
+```
+
+## Parameters
+
+- `sensorId`: ID of the sensor
+- `targetTime`: Target datetime in ISO format
+- `direction`: One of:
+  - `"before"` - Returns the last reading before the target time
+  - `"after"` - Returns the first reading after the target time  
+  - `"interpolate"` - Returns a linearly interpolated value at the target time
+
+## Response
+
+- `value`: The sensor value (float)
+- `timestamp`: Timestamp of the reading (or target time for interpolated values)
+- `isInterpolated`: Boolean indicating if the value was interpolated
+- `sourceReadings`: Array of readings used to calculate the result
+
+## Examples
+
+### Get Last Reading Before a Time
+```graphql
+{
+  sensorDatapoint(
+    sensorId: "sensor-123"
+    targetTime: "2024-01-15T12:00:00Z"
+    direction: "before"
+  ) {
+    value
+    timestamp
+    isInterpolated
+  }
+}
+```
+
+### Get First Reading After a Time
+```graphql
+{
+  sensorDatapoint(
+    sensorId: "sensor-123"
+    targetTime: "2024-01-15T12:00:00Z"
+    direction: "after"
+  ) {
+    value
+    timestamp
+    isInterpolated
+  }
+}
+```
+
+### Get Interpolated Value at Exact Time
+```graphql
+{
+  sensorDatapoint(
+    sensorId: "sensor-123"
+    targetTime: "2024-01-15T12:00:00Z"
+    direction: "interpolate"
+  ) {
+    value
+    timestamp
+    isInterpolated
+    sourceReadings {
+      timestamp
+      value
+    }
+  }
+}
+```
+
+## Linear Interpolation
+
+When using `direction: "interpolate"`, the system finds the closest readings before and after the target time and calculates a linearly interpolated value:
+
+```
+interpolated_value = before_value + factor * (after_value - before_value)
+```
+
+Where `factor` is the time distance ratio:
+```
+factor = (target_time - before_time) / (after_time - before_time)
+```
+
+If only one reading is available (before or after), it returns that reading without interpolation.
+
+## Use Cases
+
+- **Heat Pump Analysis**: Get precise energy consumption values at period boundaries
+- **Historical Analysis**: Find sensor values at specific timestamps
+- **Data Smoothing**: Interpolate missing values in time series data
+- **Threshold Monitoring**: Check if sensor crossed a threshold at a specific time

app/graphql/resolvers.py

@@ -20,7 +20,7 @@
                                CreateSensorReadingInput, CreateSensorTypeInput,
                                Location, Sensor, SensorReading, SensorType,
                                SensorDataStats, SensorDataRange, SensorReadingsAround,
-                               UpdateAlertInput, UpdateLocationInput,
+                               SingleDatapoint, UpdateAlertInput, UpdateLocationInput,
                                UpdateSensorInput, UpdateSensorReadingInput,
                                UpdateSensorStatusInput, UpdateSensorTypeInput)
 
@@ -200,6 +200,117 @@ def sensor_readings_around(
                 after=[SensorReading.from_model(model) for model in after_readings]
             )
 
+    @strawberry.field
+    def sensor_datapoint(
+        self,
+        info: Info,
+        sensor_id: str,
+        target_time: datetime,
+        direction: str = "before",  # "before", "after", or "interpolate"
+    ) -> Optional[SingleDatapoint]:
+        """Get a single sensor datapoint before, after, or interpolated at a target time.
+
+        Args:
+            sensor_id: ID of the sensor
+            target_time: Target datetime
+            direction: "before" for last reading before target_time,
+                      "after" for first reading after target_time,
+                      "interpolate" for linear interpolation between closest readings
+        """
+
+        with get_db_session() as db:
+            if direction == "before":
+                # Get the last reading before the target time
+                reading = (
+                    db.query(SensorReadingModel)
+                    .filter(SensorReadingModel.sensor_id == sensor_id)
+                    .filter(SensorReadingModel.timestamp < target_time)
+                    .order_by(SensorReadingModel.timestamp.desc())
+                    .first()
+                )
+
+                if reading:
+                    return SingleDatapoint(
+                        value=reading.value,
+                        timestamp=reading.timestamp,
+                        is_interpolated=False,
+                        source_readings=[SensorReading.from_model(reading)]
+                    )
+
+            elif direction == "after":
+                # Get the first reading after the target time
+                reading = (
+                    db.query(SensorReadingModel)
+                    .filter(SensorReadingModel.sensor_id == sensor_id)
+                    .filter(SensorReadingModel.timestamp > target_time)
+                    .order_by(SensorReadingModel.timestamp.asc())
+                    .first()
+                )
+
+                if reading:
+                    return SingleDatapoint(
+                        value=reading.value,
+                        timestamp=reading.timestamp,
+                        is_interpolated=False,
+                        source_readings=[SensorReading.from_model(reading)]
+                    )
+
+            elif direction == "interpolate":
+                # Get the closest readings before and after target time
+                before_reading = (
+                    db.query(SensorReadingModel)
+                    .filter(SensorReadingModel.sensor_id == sensor_id)
+                    .filter(SensorReadingModel.timestamp < target_time)
+                    .order_by(SensorReadingModel.timestamp.desc())
+                    .first()
+                )
+
+                after_reading = (
+                    db.query(SensorReadingModel)
+                    .filter(SensorReadingModel.sensor_id == sensor_id)
+                    .filter(SensorReadingModel.timestamp > target_time)
+                    .order_by(SensorReadingModel.timestamp.asc())
+                    .first()
+                )
+
+                if before_reading and after_reading:
+                    # Linear interpolation
+                    t1 = before_reading.timestamp.timestamp()
+                    t2 = after_reading.timestamp.timestamp()
+                    target_t = target_time.timestamp()
+
+                    # Calculate interpolation factor
+                    factor = (target_t - t1) / (t2 - t1)
+                    interpolated_value = before_reading.value + factor * (after_reading.value - before_reading.value)
+
+                    return SingleDatapoint(
+                        value=interpolated_value,
+                        timestamp=target_time,
+                        is_interpolated=True,
+                        source_readings=[
+                            SensorReading.from_model(before_reading),
+                            SensorReading.from_model(after_reading)
+                        ]
+                    )
+                elif before_reading:
+                    # Only before reading available, return it
+                    return SingleDatapoint(
+                        value=before_reading.value,
+                        timestamp=before_reading.timestamp,
+                        is_interpolated=False,
+                        source_readings=[SensorReading.from_model(before_reading)]
+                    )
+                elif after_reading:
+                    # Only after reading available, return it
+                    return SingleDatapoint(
+                        value=after_reading.value,
+                        timestamp=after_reading.timestamp,
+                        is_interpolated=False,
+                        source_readings=[SensorReading.from_model(after_reading)]
+                    )
+
+            return None
+
     @strawberry.field
     def alerts(
         self,

app/graphql/types.py

@@ -408,3 +408,13 @@ class SensorReadingsAround:
 
     before: List[SensorReading]
     after: List[SensorReading]
+
+
+@strawberry.type
+class SingleDatapoint:
+    """A single sensor datapoint with optional interpolation."""
+
+    value: float
+    timestamp: datetime
+    is_interpolated: bool = False
+    source_readings: Optional[List[SensorReading]] = None

frontend/src/graphql/queries.ts

@@ -115,4 +115,18 @@ export const GET_SENSOR_READINGS_AROUND = gql`
       }
     }
   }
+`;
+
+export const GET_SENSOR_DATAPOINT = gql`
+  query GetSensorDatapoint($sensorId: String!, $targetTime: DateTime!, $direction: String = "before") {
+    sensorDatapoint(sensorId: $sensorId, targetTime: $targetTime, direction: $direction) {
+      value
+      timestamp
+      isInterpolated
+      sourceReadings {
+        timestamp
+        value
+      }
+    }
+  }
 `;

tests/conftest.py

@@ -3,6 +3,7 @@
 """
 import os
 import pytest
+from datetime import datetime, timedelta
 from sqlalchemy import create_engine, text
 from sqlalchemy.orm import sessionmaker
 from fastapi.testclient import TestClient
@@ -135,6 +136,30 @@ def sample_sensor_reading(test_db, sample_sensor):
     return reading
 
 
+@pytest.fixture
+def sample_sensor_with_readings(test_db, sample_sensor):
+    """Create a sensor with multiple time-series readings for testing."""
+    base_time = datetime.now()
+    readings = []
+
+    # Create 3 readings with 5-minute intervals and different values
+    for i in range(3):
+        reading = SensorReading(
+            sensor_id=sample_sensor.id,
+            value=20.0 + i * 2.5,  # Values: 20.0, 22.5, 25.0
+            raw_value=20.0 + i * 2.5,
+            timestamp=base_time + timedelta(minutes=i * 5)
+        )
+        test_db.add(reading)
+        readings.append(reading)
+
+    test_db.commit()
+    for reading in readings:
+        test_db.refresh(reading)
+
+    return sample_sensor.id, readings
+
+
 # GraphQL Query Templates
 class GraphQLQueries:
     """Common GraphQL queries for testing."""