chore(app): Added ijson support

.github/workflows/pylint.yml

@@ -26,13 +26,14 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install pylint
-          pip install flask
-          pip install requests
-          pip install vcrpy
-          pip install prometheus_client
-          pip install redis
-          pip install minio
+            pip install pylint
+            pip install flask
+            pip install requests
+            pip install vcrpy
+            pip install prometheus_client
+            pip install redis
+            pip install minio
+            pip install ijson
       - name: Analysing the code with pylint
         run: |
           # Set PYTHONPATH so pylint can find the app module

app/opensense.py

@@ -1,99 +1,140 @@
-'''Module to get entries from OpenSenseMap API and get the average temperature'''
+"""Module to get entries from OpenSenseMap API and get the average temperature"""
 from datetime import datetime, timezone, timedelta
-import re
+import json
+from typing import Iterable, Dict, Tuple, Optional
 import requests
 import redis
+import ijson
+from ijson.common import JSONError as IjsonJSONError
 from app.config import create_redis_client, CACHE_TTL
 
-# Use shared Redis client
 redis_client, REDIS_AVAILABLE = create_redis_client()
 
 _sensor_stats = {"total_sensors": 0, "null_count": 0}
 
 def classify_temperature(average):
-    '''Classify temperature based on ranges using dictionary approach'''
-    # Define temperature ranges and their classifications
+    """Classify temperature based on ranges using dictionary approach"""
     temp_classifications = {
         "cold": (float('-inf'), 10, "Warning: Too cold"),
-        "good": (10, 36, "Good"), 
-        "hot": (36, float('inf'), "Warning: Too hot")
+        "good": (10, 36, "Good"),
+        "hot": (36, float('inf'), "Warning: Too hot"),
     }
-
-    # Find the appropriate classification
     for _, (min_temp, max_temp, status) in temp_classifications.items():
         if min_temp < average <= max_temp:
             return status
+    return "Unknown"
 
-    return "Unknown"  # Default case
+def _iter_sensors_from_stream(stream) -> Iterable[dict]:
+    """Yield sensors from a streaming JSON array of boxes."""
+    yield from ijson.items(stream, 'item.sensors.item')
+
+def _iter_sensors_from_json(boxes: Iterable[dict]) -> Iterable[dict]:
+    """Yield sensors from a loaded list of boxes."""
+    for box in boxes:
+        yield from box.get("sensors", [])
+
+def _compute_stats(sensors: Iterable[dict]) -> Tuple[float, int, Dict[str, int]]:
+    """Compute temperature sum/count and stats from an iterable of sensors."""
+    temp_sum = 0.0
+    temp_count = 0
+    stats = {"total_sensors": 0, "null_count": 0}
+
+    for sensor in sensors:
+        stats["total_sensors"] += 1
+        if sensor.get('unit') != "°C":
+            continue
+
+        last = sensor.get('lastMeasurement')
+        if not last or 'value' not in last:
+            stats["null_count"] += 1
+            continue
 
-def get_temperature():
-    '''Function to get the average temperature from OpenSenseMap API.'''
-    if REDIS_AVAILABLE:
         try:
-            cached_data = redis_client.get("temperature_data")
-            if cached_data:
-                print("Using cached data from Redis.")
-                # Return cached data with default stats (since we don't have fresh stats)
-                default_stats = {"total_sensors": 0, "null_count": 0}
-                return cached_data, default_stats
-        except redis.RedisError as e:
-            print(f"Redis error: {e}. Proceeding without cache.")
-
-    print("Fetching new data from OpenSenseMap API...")
-
-    # Ensuring that data is not older than 1 hour.
-    time_iso = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat().replace("+00:00", "Z")
+            temp_sum += float(last['value'])
+            temp_count += 1
+        except (TypeError, ValueError):
+            stats["null_count"] += 1
 
-    params = {
-        "date": time_iso,
-        "format": "json"
-    }
+    return temp_sum, temp_count, stats
 
-    print('Getting data from OpenSenseMap API...')
+def _empty_stats() -> Dict[str, int]:
+    return {"total_sensors": 0, "null_count": 0}
+
+def _get_cached_temperature() -> Optional[str]:
+    if not REDIS_AVAILABLE:
+        return None
+    try:
+        cached = redis_client.get("temperature_data")
+        if cached:
+            return cached.decode("utf-8") if isinstance(cached, (bytes, bytearray)) else cached
+    except redis.RedisError:
+        return None
+    return None
+
+def _set_cached_temperature(value: str) -> None:
+    if not REDIS_AVAILABLE:
+        return
     try:
-        response = requests.get(
+        redis_client.setex("temperature_data", CACHE_TTL, value)
+    except redis.RedisError:
+        pass
+
+def _build_params() -> Dict[str, str]:
+    time_iso = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat().replace("+00:00", "Z")
+    return {"date": time_iso, "format": "json"}
+
+def _request_boxes(params: Dict[str, str]):
+    try:
+        resp = requests.get(
             "https://api.opensensemap.org/boxes",
             params=params,
-            timeout=(3, 10)
+            stream=True,
+            timeout=(60, 90)
         )
-        print('Data retrieved successfully!')
+        if hasattr(resp, "raise_for_status"):
+            resp.raise_for_status()
+        if hasattr(resp, "raw") and hasattr(resp.raw, "decode_content"):
+            resp.raw.decode_content = True
+        return resp, None
     except requests.Timeout:
-        print("API request timed out")
-        return "Error: API request timed out\n", {"total_sensors": 0, "null_count": 0}
+        return None, "Error: API request timed out\n"
     except requests.RequestException as e:
-        print(f"API request failed: {e}")
-        return f"Error: API request failed - {e}\n", {"total_sensors": 0, "null_count": 0}
+        return None, f"Error: API request failed - {e}\n"
 
-    _sensor_stats["total_sensors"] = sum(
-        1 for line in response.text.splitlines() if re.search(r'^\s*"sensors"\s*:\s*\[', line)
-    )
+def _make_sensor_iter(response) -> Tuple[Optional[Iterable[dict]], Optional[str]]:
+    # Prefer streaming if raw is available; otherwise load JSON once.
+    if hasattr(response, "raw") and getattr(response, "raw", None):
+        return _iter_sensors_from_stream(response.raw), None
+    try:
+        boxes = response.json()
+    except (json.JSONDecodeError, ValueError) as e:
+        return None, f"Error: parse failed - {e}\n"
+    return _iter_sensors_from_json(boxes), None
 
-    res = [d.get('sensors') for d in response.json() if 'sensors' in d]
+def get_temperature():
+    '''Function to get the average temperature from OpenSenseMap API.'''
+    cached = _get_cached_temperature()
+    if cached:
+        return cached, _empty_stats()
 
-    temp_list = []
-    _sensor_stats["null_count"] = 0  # Initialize counter for null measurements
+    params = _build_params()
+    response, err = _request_boxes(params)
+    if err:
+        return err, _empty_stats()
 
-    for sensor_list in res:
-        for measure in sensor_list:
-            if measure.get('unit') == "°C" and 'lastMeasurement' in measure:
-                last_measurement = measure['lastMeasurement']
-                if last_measurement is not None and 'value' in last_measurement:
-                    last_measurement_int = float(last_measurement['value'])
-                    temp_list.append(last_measurement_int)
-                else:
-                    _sensor_stats["null_count"] += 1
+    sensors_iter, err = _make_sensor_iter(response)
+    if err or sensors_iter is None:
+        return err or "Error: parser unavailable\n", _empty_stats()
 
-    average = sum(temp_list) / len(temp_list) if temp_list else 0
+    try:
+        temp_sum, temp_count, stats = _compute_stats(sensors_iter)
+    except IjsonJSONError as e:
+        return f"Error: parse failed - {e}\n", _empty_stats()
 
-    # Use the dictionary-based classification
+    average = (temp_sum / temp_count) if temp_count else 0.0
     status = classify_temperature(average)
     result = f'Average temperature: {average:.2f} °C ({status})\n'
 
-    if REDIS_AVAILABLE:
-        try:
-            redis_client.setex("temperature_data", CACHE_TTL, result)
-            print("Data cached in Redis.")
-        except redis.RedisError as e:
-            print(f"Redis error while caching data: {e}")
-
+    _set_cached_temperature(result)
+    _sensor_stats.update(stats)
     return result, _sensor_stats

helm-chart/templates/deployment.yaml

@@ -34,7 +34,7 @@ spec:
             httpGet:
               path: /readyz
               port: 5000
-            initialDelaySeconds: 30
+            initialDelaySeconds: 10
             timeoutSeconds: 480
             failureThreshold: 3
             periodSeconds: 600

requirements.in

@@ -2,4 +2,5 @@ Flask==3.1.2
 requests==2.32.5
 prometheus-client==0.22.1
 redis==6.4.0
-minio==7.2.16
+minio==7.2.16
+ijson==3.4.0