Don't crash if no internet connection

cloud/APIs/aichat.py

@@ -1,102 +1,124 @@
 import os
 import base64
 import json
-import urllib.request
-import urllib.error
+from typing import Any, Union
+from openai import OpenAI
+from pydantic import BaseModel
 
-API_HOST = os.getenv("OPENAI_API_HOST", "https://chat.matthiasg.dev/ollama")
-AI_MODEL = os.getenv("OPENAI_MODEL", "llava:7b-v1.6-mistral-q4_1")
-AI_MODEL_THINK = os.getenv("OPENAI_MODEL_THINK", "True").lower() in ("true", "1", "t")
-OPENAPI_TOKEN = os.getenv("OPENAI_API_TOKEN", None)
-OPENAPI_ENDPOINT = "/api/generate"
+# Default base_url pointing to Lemonade's OpenAI-compatible endpoint
+API_HOST = os.getenv("OPENAI_API_HOST", "https://lemonade.matthiasg.dev/api/v1")
+AI_MODEL = os.getenv("OPENAI_MODEL", "LFM2.5-VL-1.6B-GGUF-Q8_0")
+AI_MODEL_THINK = os.getenv("OPENAI_MODEL_THINK", "False").lower() in ("true", "1", "t")
+OPENAPI_TOKEN = os.getenv("OPENAI_API_TOKEN", "lemonade")
 
 print(f"Using API Host: {API_HOST}")
 print(f"Using API Model: {AI_MODEL}")
 print(f"Using API Model Think: {AI_MODEL_THINK}")
 print(f"Using API Token: {'Yes' if OPENAPI_TOKEN else 'No'} {OPENAPI_TOKEN[:5] + '...' if OPENAPI_TOKEN else ''}")
 
-def call_api(body: dict, endpoint: str = OPENAPI_ENDPOINT) -> str:
-    """Call the API with the given endpoint and body dict."""
-    url = f"{API_HOST}{endpoint}"
-    headers = {
-        "Content-Type": "application/json",
-    }
-    if OPENAPI_TOKEN:
-        headers["Authorization"] = f"Bearer {OPENAPI_TOKEN}"
-
-    json_data = json.dumps(body).encode("utf-8")
-    req = urllib.request.Request(url, data=json_data, headers=headers, method="POST")
-
-    try:
-        with urllib.request.urlopen(req) as response:
-            return response.read().decode("utf-8")
-    except urllib.error.HTTPError as e:
-        error_body = e.read().decode("utf-8")
-        raise Exception(f"Error calling API: HTTP {e.code} - {error_body}")
-    except urllib.error.URLError as e:
-        raise Exception(f"Failed to reach server: {e.reason}")
+# Initialize OpenAI Client for Lemonade
+client = OpenAI(
+    base_url=API_HOST,
+    api_key=OPENAPI_TOKEN if OPENAPI_TOKEN else "lemonade"
+)
 
 
 def generate(
     model: str = AI_MODEL,
     prompt: str = "",
     images: list[str] = None,
-    output_format: str = None,
+    output_format: Union[str, dict, type[BaseModel]] = None,
     system_message: str = None,
     keep_alive: bool = True,
     should_think: bool = AI_MODEL_THINK,
 ) -> str:
     """
-    Generate a response for a given prompt with a provided model via the Ollama/OpenAI API.
-    Handles base64 encoding for local image file paths and structures the request body.
+    Generate a completion response using the OpenAI SDK against Lemonade / OpenAI compatible APIs.
+    Handles multimodal image content, structured outputs, and custom Lemonade parameters.
     """
     if images is None:
         images = []
 
-    # Transform image file paths to base64 strings
-    encoded_images = []
+    messages = []
+    
+    # 1. Add System Message if present
+    if system_message:
+        messages.append({"role": "system", "content": system_message})
+
+    # 2. Build User Content Payload (Text + Multimodal Images)
+    user_content = []
+    if prompt:
+        user_content.append({"type": "text", "text": prompt})
+
     for img_path in images:
         if os.path.isfile(img_path):
             with open(img_path, "rb") as image_file:
-                encoded_images.append(base64.b64encode(image_file.read()).decode("utf-8"))
+                b64_str = base64.b64encode(image_file.read()).decode("utf-8")
         else:
-            # If it's already a base64 string or an invalid path, keep as-is
-            encoded_images.append(img_path)
+            b64_str = img_path
 
-    body = {
-        "model": model,
-        "prompt": prompt,
-        "images": encoded_images,
+        # Ensure base64 string includes Data URI prefix for OpenAI vision format
+        if not b64_str.startswith("data:"):
+            image_url = f"data:image/jpeg;base64,{b64_str}"
+        else:
+            image_url = b64_str
+
+        user_content.append({
+            "type": "image_url",
+            "image_url": {"url": image_url}
+        })
+
+    # Simplify content payload if text-only
+    if len(user_content) == 1 and user_content[0]["type"] == "text":
+        messages.append({"role": "user", "content": prompt})
+    else:
+        messages.append({"role": "user", "content": user_content})
+
+    # 3. Prepare parameters and custom body flags
+    extra_body = {
         "think": should_think,
-        "stream": False,
+    }
+    if not keep_alive:
+        extra_body["keep_alive"] = "0m"
+
+    request_kwargs: dict[str, Any] = {
+        "model": model,
+        "messages": messages,
+        "extra_body": extra_body,
     }
 
-    if system_message is not None:
-        body["system"] = system_message
-
+    # 4. Handle Structured Outputs / JSON mode
     if output_format is not None:
-        try:
-            body["format"] = json.loads(output_format)
-        except json.JSONDecodeError:
-            body["format"] = output_format
+        if isinstance(output_format, type) and issubclass(output_format, BaseModel):
+            # Pydantic Model -> Use structured output parsing API
+            response = client.beta.chat.completions.parse(
+                response_format=output_format,
+                **request_kwargs
+            )
+            return response.choices[0].message.content or ""
+            
+        elif isinstance(output_format, dict):
+            # JSON Schema dictionary
+            request_kwargs["response_format"] = {
+                "type": "json_schema",
+                "json_schema": {
+                    "name": "structured_response",
+                    "strict": True,
+                    "schema": output_format,
+                }
+            }
+        elif isinstance(output_format, str) and output_format.lower() == "json":
+            # Standard JSON Mode
+            request_kwargs["response_format"] = {"type": "json_object"}
 
-    if not keep_alive:
-        body["keep_alive"] = "0m"
-
-    response_text = call_api(body)
-
-    try:
-        decoded_response = json.loads(response_text)
-    except json.JSONDecodeError as e:
-        raise Exception(f"Error decoding JSON response: {e}")
-
-    return decoded_response.get("response", "")
+    # Standard completion request
+    response = client.chat.completions.create(**request_kwargs)
+    return response.choices[0].message.content or ""
 
 
 if __name__ == "__main__":
-    # Example usage:
     result = generate(
-        prompt="Explain what you see in the image or answer this prompt.",
+        prompt="Explain what you see or answer this prompt.",
         should_think=AI_MODEL_THINK,
     )
     print(result)

cloud/app.py

@@ -84,7 +84,11 @@ async def cooking_params():
 
         # 2. Run Vision Safety Check & Cook Planner Concurrently
         try:
-            safety_task = asyncio.to_thread(safety_checker.check_dish_safety, filepath)
+            if getattr(config, "DEBUG", True):
+                print(f"[Debug] Running safety check on predefined image")
+                safety_task = asyncio.to_thread(safety_checker.check_dish_safety, os.path.join(CAMERA_IMAGE_DIR, "dish_0de0ee1dab8949fc8e78796947e24ed3.jpg"))
+            else:
+                safety_task = asyncio.to_thread(safety_checker.check_dish_safety, filepath)
             planner_task = asyncio.to_thread(
                 microwave_cook_planner.generate_plan,
                 image_path=filepath,

cloud/requirements.txt

@@ -3,4 +3,6 @@ pymongo==4.6.1
 gunicorn==21.2.0
 opencv-python-headless
 requests==2.32.3
-paho-mqtt>=1.6,<3
+paho-mqtt>=1.6,<3
+openai>=1.0.0
+pydantic>=2.0.0

cloud/safety_checker.py

@@ -1,27 +1,20 @@
 import json
+from pydantic import BaseModel, Field
 from APIs.aichat import generate
 
-DISH_SAFETY_SCHEMA = {
-    "type": "object",
-    "properties": {
-        "is_safe": {
-            "type": "boolean",
-            "description": "True if no microwave hazards (metal, foil, sealed packaging) are present."
-        },
-        "warning_message": {
-            "type": "string",
-            "description": "Explanation of any hazard found, or an empty string if safe."
-        },
-        "detected_hazards": {
-            "type": "array",
-            "items": {
-                "type": "string"
-            },
-            "description": "List of specific hazard items detected."
-        }
-    },
-    "required": ["is_safe", "warning_message", "detected_hazards"]
-}
+
+# 1. Define the output schema as a Pydantic model
+class DishSafetyResult(BaseModel):
+    is_safe: bool = Field(
+        description="True if no microwave hazards (metal, foil, sealed packaging) are present."
+    )
+    warning_message: str = Field(
+        description="Explanation of any hazard found, or an empty string if safe."
+    )
+    detected_hazards: list[str] = Field(
+        description="List of specific hazard items detected."
+    )
+
 
 def check_dish_safety(image_path: str) -> dict:
     prompt = (
@@ -29,15 +22,28 @@ def check_dish_safety(image_path: str) -> dict:
         "Inspect the area for metal utensils, aluminum foil, metallic dish patterns, or unvented plastic wraps."
     )
 
-    # Pass the schema directly to the generation call
+    # 2. Pass the Pydantic class directly to generate()
     response_raw = generate(
         prompt=prompt,
         images=[image_path],
-        output_format=json.dumps(DISH_SAFETY_SCHEMA)
+        output_format=DishSafetyResult,
+        should_think=False
     )
     print(f"[Debug] Raw response from AI generator: {response_raw}")
 
-    # 1. If response_raw is already a dict, return it directly
+    # 3. Handle response parsing
+    if isinstance(response_raw, str):
+        # Parse and validate the JSON string into the Pydantic model, then return as a dict
+        try:
+            validated_result = DishSafetyResult.model_validate_json(response_raw)
+            return validated_result.model_dump()
+        except Exception:
+            # Fallback to standard json.loads if raw parsing is needed
+            return json.loads(response_raw)
+
+    if isinstance(response_raw, DishSafetyResult):
+        return response_raw.model_dump()
+
     if isinstance(response_raw, dict):
         return response_raw
 

orchestrateur/main.py

@@ -106,9 +106,12 @@ cloud_mqtt_client = get_mqtt_client(
     password="myMicrowaveVerySecret",
     keepalive=config.MQTT_KEEPALIVE,
 )
-cloud_mqtt_client.connect()
-cloud_mqtt_client.subscribe("cmd/all", qos=config.CLOUD_MQTT_QOS)
-cloud_mqtt_client.subscribe(f"cmd/{DEVICE_ID}", qos=config.CLOUD_MQTT_QOS)
+try:
+    cloud_mqtt_client.connect()
+    cloud_mqtt_client.subscribe("cmd/all", qos=config.CLOUD_MQTT_QOS)
+    cloud_mqtt_client.subscribe(f"cmd/{DEVICE_ID}", qos=config.CLOUD_MQTT_QOS)
+except Exception as e:
+    print(f"[MQTT Cloud] Error connecting to cloud MQTT broker: {e}")
 
 if hasattr(cloud_mqtt_client._client, "loop_start"):
     cloud_mqtt_client._client.loop_start()
@@ -264,7 +267,7 @@ async def handle_new_dish(microwave_id, detected_height):
         ir_data_events.pop(microwave_id, None)
 
 async def request_cloud_cooking_plan(microwave_id, sensors_data):
-    """Sends all data to the cloud with up to 3 retries (30s interval)."""
+    """Sends all data to the cloud with up to 3 retries (330s timeout for AI generation)."""
     global cloud_alert
     microwave_states[microwave_id] = MicrowaveState.WAITING_FOR_CLOUD
     URL = "https://smartwave.matthiasg.dev/cooking-params"
@@ -276,7 +279,10 @@ async def request_cloud_cooking_plan(microwave_id, sensors_data):
     print(f"[{microwave_id}] Requesting cooking plan from cloud app...")
 
     max_retries = 3
-    retry_delay_seconds = 30
+    retry_delay_seconds = 10  # Reduced delay since each request already waits up to 5 mins
+
+    # Tuple format: (connect_timeout_seconds, read_timeout_seconds)
+    HTTP_TIMEOUT = (10, 330)
 
     for attempt in range(1, max_retries + 1):
         if microwave_states.get(microwave_id) != MicrowaveState.WAITING_FOR_CLOUD:
@@ -284,30 +290,43 @@ async def request_cloud_cooking_plan(microwave_id, sensors_data):
             return
 
         try:
-            print(f"[{microwave_id}] Connection attempt {attempt}/{max_retries}...")
-            response = await asyncio.to_thread(requests.post, URL, json=sensors_data, timeout=30)
-            # Abort if state changed (e.g. user removed dish while waiting for wifi)
+            print(f"[{microwave_id}] Connection attempt {attempt}/{max_retries} (Waiting up to 5.5 min)...")
+            
+            response = await asyncio.to_thread(
+                requests.post, 
+                URL, 
+                json=sensors_data, 
+                timeout=HTTP_TIMEOUT
+            )
+
+            # Abort if state changed while waiting for cloud AI response
             if microwave_states.get(microwave_id) != MicrowaveState.WAITING_FOR_CLOUD:
                 print(f"[{microwave_id}] Dish removed during API request. Discarding API plan.")
                 return
 
             response.raise_for_status()
-            plan = response.json().get("cook_plan", {})
+
+            # Robust JSON extraction: handles direct dict or nested {"cook_plan": {...}}
+            res_json = response.json()
+            plan = res_json.get("cook_plan", res_json)
+
             c_time = plan.get("cook_time_seconds")
             c_power = plan.get("effective_power_watts")
             c_temp = plan.get("target_temp")
             
             if c_time is None or c_power is None or c_temp is None:
-                print(f"[{microwave_id}] ❌ Invalid plan received: {response.json()}")
+                print(f"[{microwave_id}] ❌ Invalid plan received: {res_json}")
                 microwave_states[microwave_id] = MicrowaveState.DONE
                 return
 
             print(f"[{microwave_id}] Cloud Plan Received! Starting microwave: {c_time}s @ {c_power}W")
             cloud_alert = False  # Reset alert flag on success
             microwave_states[microwave_id] = MicrowaveState.COOKING
+
             if config.DEBUG:
                 c_time = 20  # Set to 20s for debug
                 c_temp = 50  # Set to 50°C for debug
+
             mqtt_client.publish(
                 config.MQTT_TOPIC_COOKING, 
                 payloads.mqtt_cooking_config(microwave_id, c_time, c_power, c_temp), 
@@ -320,7 +339,7 @@ async def request_cloud_cooking_plan(microwave_id, sensors_data):
             
             if attempt < max_retries:
                 print(f"[{microwave_id}] Retrying in {retry_delay_seconds} seconds...")
-                # Interruptible wait loop in case user removes the dish mid-wait
+                # Interruptible wait loop in case state changes mid-wait
                 for _ in range(retry_delay_seconds):
                     if microwave_states.get(microwave_id) != MicrowaveState.WAITING_FOR_CLOUD:
                         print(f"[{microwave_id}] State changed during retry wait. Aborting retries.")
@@ -328,7 +347,7 @@ async def request_cloud_cooking_plan(microwave_id, sensors_data):
                     await asyncio.sleep(1)
 
     # Executed only if all 3 retries failed
-    print(f"[{microwave_id}] All 3 cloud retries failed. Setting global alert flag.")
+    print(f"[{microwave_id}] All cloud retries failed. Setting global alert flag.")
     cloud_alert = True
     microwave_states[microwave_id] = MicrowaveState.DONE