better api

app.py

@@ -110,7 +110,7 @@ def create_app():
         def _prepare_data(kit_id):
             # Fetch data for the given kit to ensure availability before enabling generation
             try:
-                from utils import get_kit_measurements_df
+                from api_call import get_kit_measurements_df
 
                 kit = 1001
                 if kit_id is not None:
@@ -119,7 +119,7 @@ def create_app():
                     except Exception:
                         kit = 1001
 
-                df = get_kit_measurements_df(kit)
+                df = get_kit_measurements_df(kit, page_size=60, max_pages=2)
                 if df is None or getattr(df, "empty", True):
                     return (
                         f"No data found for kit {kit}. Please try another kit.",

utils.py

@@ -1,149 +1,34 @@
 """
-Utility helpers for Teleagriculture kits API
+Deprecated API shim.
 
-Provides:
-- BASE_URL, HEADERS (with optional Bearer from KIT_API_KEY env)
-- get_kit_info(kit_id)
-- get_kit_measurements_df(kit_id, sensors=None, page_size=100)
+All API-related functions/constants moved to `api_call.py` to keep
+`utils.py` free of network concerns. Import directly from `api_call`.
+
+This module re-exports the public API for backward compatibility.
 """
 from __future__ import annotations
 
-import os
-from typing import Any, Iterable, Optional
-
-import pandas as pd
-import requests
-
-# API configuration
-BASE_URL = os.getenv("KITS_API_BASE", "https://kits.teleagriculture.org/api")
-KIT_API_KEY = os.getenv("KIT_API_KEY")
-
-HEADERS: dict[str, str] = {
-    "Accept": "application/json",
-}
-if KIT_API_KEY:
-    HEADERS["Authorization"] = f"Bearer {KIT_API_KEY}"
-
-
-def get_kit_info(kit_id: int) -> Optional[dict]:
-    """Fetch metadata for a kit (board).
-
-    Returns the JSON 'data' object or None if not found / error.
-    """
-    url = f"{BASE_URL}/kits/{kit_id}"
-    try:
-        r = requests.get(url, headers=HEADERS, timeout=30)
-        if r.status_code == 200:
-            body = r.json()
-            return body.get("data")
-        return None
-    except requests.RequestException:
-        return None
-
-
-def _paginate(
-    url: str,
-    *,
-    params: Optional[dict] = None,
-    headers: Optional[dict] = None,
-    page_size: int = 100,
-    max_pages: int = 500,
-):
-    """Cursor pagination helper yielding lists of items from {'data': [...]} pages.
-
-    Stops when no next_cursor is provided or on any non-200/parse error.
-    """
-    q = dict(params or {})
-    q["page[size]"] = str(page_size)
-    cursor = None
-    pages = 0
-    while pages < max_pages:
-        if cursor:
-            q["page[cursor]"] = cursor
-        try:
-            r = requests.get(url, headers=headers, params=q, timeout=30)
-        except requests.RequestException:
-            break
-        if r.status_code != 200:
-            break
-        try:
-            payload = r.json()
-        except Exception:
-            break
-        data = payload.get("data")
-        meta = payload.get("meta", {})
-        yield data if isinstance(data, list) else []
-        cursor = meta.get("next_cursor")
-        pages += 1
-        if not cursor:
-            break
-
-
-def get_kit_measurements_df(
-    kit_id: int,
-    sensors: Optional[Iterable[str]] = None,
-    *,
-    page_size: int = 100,
-) -> pd.DataFrame:
-    """Fetch all measurements for the given kit across its sensors as a DataFrame.
-
-    - If sensors is None, discover sensors via get_kit_info(kit_id).
-    - Returns columns: kit_id, sensor, timestamp, value, unit, _raw
-      (depending on API, some fields may be None/NaT)
-    """
-    # Determine sensor list
-    if sensors is None:
-        kit = get_kit_info(kit_id)
-        if not kit:
-            return pd.DataFrame(columns=["kit_id", "sensor", "timestamp", "value", "unit", "_raw"])
-        sensor_list = [
-            s.get("name")
-            for s in (kit.get("sensors") or [])
-            if isinstance(s, dict) and s.get("name")
-        ]
-    else:
-        sensor_list = [s for s in sensors if s]
-
-    rows: list[dict[str, Any]] = []
-
-    for sname in sensor_list:
-        endpoint = f"{BASE_URL}/kits/{kit_id}/{sname}/measurements"
-        for page in _paginate(endpoint, headers=HEADERS, page_size=page_size):
-            for item in page:
-                if not isinstance(item, dict):
-                    continue
-                
-                # Some APIs nest details under 'attributes'
-                rec = item.get("attributes", {})
-                rec.update({k: v for k, v in item.items() if k != "attributes"})
-
-                ts = rec.get("timestamp") or rec.get("time") or rec.get("created_at") or rec.get("datetime")
-                val = rec.get("value") or rec.get("reading") or rec.get("measurement") or rec.get("val")
-                unit = rec.get("unit") or rec.get("units")
-                rows.append(
-                    {
-                        "kit_id": kit_id,
-                        "sensor": sname,
-                        "timestamp": ts,
-                        "value": val,
-                        "unit": unit,
-                        "_raw": item,  # preserve original
-                    }
-                )
-
-    df = pd.DataFrame(rows)
-    if not df.empty and "timestamp" in df.columns:
-        try:
-            df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce", utc=True)
-            df = df.sort_values(["sensor", "timestamp"], kind="stable")
-        except Exception:
-            pass
-    return df
-
-
-def fetch_kit_dataframe(kit_id: int) -> pd.DataFrame:
-    """Simplest API: return all measurements for the given kit as a DataFrame.
-
-    Equivalent to get_kit_measurements_df(kit_id) with sensible defaults.
-    """
-    return get_kit_measurements_df(kit_id)
+import warnings
+
+from api_call import (
+    BASE_URL,
+    HEADERS,
+    get_kit_info,
+    get_kit_measurements_df,
+    fetch_kit_dataframe,
+)
+
+warnings.warn(
+    "The API helpers were moved from utils.py to api_call.py. "
+    "Please import from 'api_call' going forward.",
+    DeprecationWarning,
+    stacklevel=2,
+)
+
+__all__ = [
+    "BASE_URL",
+    "HEADERS",
+    "get_kit_info",
+    "get_kit_measurements_df",
+    "fetch_kit_dataframe",
+]

weather_data_visualisation.py

@@ -7,21 +7,53 @@ import pandas as pd
 import matplotlib.pyplot as plt
 from matplotlib.backends.backend_agg import FigureCanvasAgg as FigureCanvas
 from PIL import Image
-
-
-def weather_data_visualisation(save_to_disk: bool = False) -> Image.Image:
-    weatherdata_df = pd.read_csv("data/kit_1001_2025-09-22.csv", index_col=2)
-    weatherdata_df.drop(columns=['kit_id', "unit","_raw"], inplace=True)
-    weatherdata_df.dropna(inplace=True)
-    weatherdata_df.index = pd.to_datetime(weatherdata_df.index)
-    weatherdata_df = weatherdata_df.pivot(columns='sensor', values='value')
+from typing import Optional
+from api_call import get_kit_measurements_df
+
+
+def weather_data_visualisation(
+    kit: int = 1001,
+    save_to_disk: bool = False,
+    df: Optional[pd.DataFrame] = None,
+) -> Image.Image:
+    """Generates a 'Monsoon Mandala' visualization from weather data.
+
+    Args:
+        kit: The kit ID to fetch data for, if df is not provided.
+        save_to_disk: Whether to save the output image to disk.
+        df: An optional DataFrame with pre-loaded weather data. If None, data will be fetched.
+
+    Returns:
+        A PIL.Image object of the generated visualization.
+    """
+    # If no DataFrame is provided, fetch the data using the kit ID
+    if df is None:
+        df = get_kit_measurements_df(kit)
+
+    # If data is still unavailable, return a placeholder or raise an error
+    if df is None or df.empty:
+        raise ValueError(f"No data available for kit {kit}")
+
+    # --- Data cleaning and pivoting ---
+    # Drop columns that are not needed for the visualization (new API df has no _raw)
+    df.drop(columns=['kit_id', "unit"], inplace=True, errors='ignore')
+    df.dropna(inplace=True)
+    
+    # Ensure the index is a datetime object before pivoting
+    if not isinstance(df.index, pd.DatetimeIndex):
+        df['timestamp'] = pd.to_datetime(df['timestamp'])
+        df.set_index('timestamp', inplace=True)
+        
+    df = df.pivot(columns='sensor', values='value')
+
+    # Ensure required columns exist (fill with zeros if missing)
+    for col in ['ftTemp', 'gbHum', 'NH3', 'C3H8', 'CO']:
+        if col not in df.columns:
+            df[col] = 0.0
 
     # ---- Mapping to polar "Monsoon Mandala" ----
     # Angles map to time; radii encode a blended metric; thickness & dot size encode other variables.
-
-    df = weatherdata_df
-
-    theta = np.linspace(0, 2*np.pi, len(df), endpoint=False)
+    theta = np.linspace(0, 2 * np.pi, len(df), endpoint=False)
 
     # Normalize helpers (avoid specifying colors, per instructions).
     def norm(x):
@@ -96,5 +128,10 @@ def weather_data_visualisation(save_to_disk: bool = False) -> Image.Image:
     plt.close(fig)
     return pil_img
 
+
 if __name__ == "__main__":
-    weather_data_visualisation()
+    # Example: Generate visualization for a specific kit and save it
+    img = weather_data_visualisation(kit=1001, save_to_disk=True)
+    if img:
+        print("Visualization generated and saved to 'output/'.")
+        img.show()  # Display the image