Refactor into config file

.gitignore

@@ -1,3 +1,4 @@
 __pycache__/
 orchestrateur/db.sqlite-shm
-venv/
+venv/
+.env

.vscode/settings.json

@@ -9,6 +9,7 @@
         "${workspaceFolder}/shared",
         "${workspaceFolder}/micro_ondes/esp_lora/lib"
     ],
+    "python.terminal.useEnvFile": true,
     "python.defaultInterpreterPath": "${workspaceFolder}/venv/bin/python",
     "r.lsp.promptToInstall": false,
 }

cloud/app.py

@@ -3,8 +3,8 @@ import base64
 import uuid
 from flask import Flask, request, jsonify
 from pymongo import MongoClient
+from tools.aichat import generate
 
-# Import your shared device types
 from shared import deviceTypes
 
 app = Flask(__name__)
@@ -30,7 +30,7 @@ os.makedirs(CAMERA_IMAGE_DIR, exist_ok=True)
 
 @app.route("/")
 def hello_world():
-    return "<p>Hello, World!</p>"
+    return f"<p>{generate(prompt='Say \"Hello, World!\"')}</p>"
 
 
 @app.route("/cooking-params", methods=["POST"])

cloud/tools/aichat.py

@@ -0,0 +1,98 @@
+import os
+import base64
+import json
+import urllib.request
+import urllib.error
+
+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"
+
+
+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}")
+
+
+def generate(
+    model: str = AI_MODEL,
+    prompt: str = "",
+    images: list[str] = None,
+    output_format: str = 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.
+    """
+    if images is None:
+        images = []
+
+    # Transform image file paths to base64 strings
+    encoded_images = []
+    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"))
+        else:
+            # If it's already a base64 string or an invalid path, keep as-is
+            encoded_images.append(img_path)
+
+    body = {
+        "model": model,
+        "prompt": prompt,
+        "images": encoded_images,
+        "think": should_think,
+        "stream": False,
+    }
+
+    if system_message is not None:
+        body["system"] = system_message
+
+    if output_format is not None:
+        try:
+            body["format"] = json.loads(output_format)
+        except json.JSONDecodeError:
+            body["format"] = output_format
+
+    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", "")
+
+
+if __name__ == "__main__":
+    # Example usage:
+    result = generate(
+        prompt="Explain what you see in the image or answer this prompt.",
+        should_think=AI_MODEL_THINK,
+    )
+    print(result)

orchestrateur/sensors/temp_hum.py

@@ -1,33 +1,3 @@
-# from sensors.lib import grove_i2c_temp_hum_mini
-
-# t= grove_i2c_temp_hum_mini.th02()
-
-# def get_temperature():
-#     """Get the temperature in Celsius from the TH02 sensor."""
-#     # try:
-#     return t.getTemperature()
-#     # except Exception as e:
-#     #     print(f"Error reading temperature: {e}")
-#     #     return None
-
-# def get_humidity():
-#     """Get the humidity in percentage from the TH02 sensor."""
-#     # try:
-#     return t.getHumidity()
-#     # except Exception as e:
-#     #     print(f"Error reading humidity: {e}")
-#     #     return None
-
-# import seeed_dht
-
-# sensor = seeed_dht.DHT("11", 4)  # DHT11 sensor on GPIO pin 4
-
-# def get_humidity_and_temperature():
-#     humi, temp = sensor.read()
-#     return humi, temp
-
-
-# import sensors.lib.grovepi as grovepi
 import grovepi
 import math
 from sensors.lock import grove_lock

orchestrateur/sensors/ultrasonicRanger.py

@@ -1,5 +1,6 @@
 import grovepi
 from sensors.lock import grove_lock
+from shared import config
 
 # Connect the Grove Ultrasonic Ranger to digital port D4
 # SIG,NC,VCC,GND
@@ -25,8 +26,8 @@ def get_dish_height():
         # Assuming the ultrasonic sensor is mounted at a fixed height above the dish
         # and pointing downwards, we can calculate the height of the dish.
         # For example, if the sensor is 30 cm above the dish when it's empty:
-        SENSOR_HEIGHT = 30  # cm
-        dish_height = SENSOR_HEIGHT - distance
+          # cm
+        dish_height = config.COOKING_COMPARTMENT_HEIGHT - distance
         return max(dish_height, 0)  # Ensure height is not negative
     else:
         return None

shared/config.py

@@ -12,4 +12,7 @@ MQTT_KEEPALIVE = 30
 USE_TLS = True
 MQTT_QOS = 1
 # Long because messages are stored into the broker and will be sent when the orchestrator is back online.
-MQTT_HELLO_INTERVAL = 30
+MQTT_HELLO_INTERVAL = 30
+
+# Microwave Model
+COOKING_COMPARTMENT_HEIGHT = 30  # cm