import os import json import pickle import re import tempfile from flask import Flask, request, jsonify from flask_cors import CORS from dotenv import load_dotenv from faster_whisper import WhisperModel import anthropic from supabase import create_client load_dotenv() app = Flask(__name__) CORS(app) supabase = create_client(os.environ["SUPABASE_URL"], os.environ["SUPABASE_SERVICE_KEY"]) print("Loading Whisper model (base)…") whisper_model = WhisperModel("base", device="cpu", compute_type="int8") print("Whisper ready.") claude = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"]) SYSTEM_PROMPT = """You are a concise diagnostic assistant for HP Metal Jet S100 industrial 3D metal printers. You receive real-time sensor data and answer operator questions in 1-3 short sentences. Be direct and technical. Always respond in English.""" @app.route("/api/transcribe", methods=["POST"]) def transcribe(): if "audio" not in request.files: return jsonify({"error": "no audio file"}), 400 audio = request.files["audio"] with tempfile.NamedTemporaryFile(suffix=".webm", delete=False) as tmp: audio.save(tmp.name) tmp_path = tmp.name try: segments, _ = whisper_model.transcribe(tmp_path, language="en", task="transcribe") text = " ".join(seg.text for seg in segments).strip() print(f"\n[TRANSCRIPTION] {text}\n") return jsonify({"text": text}) finally: os.unlink(tmp_path) #END STUFF _SQL_SYSTEM = """You are a read-only PostgreSQL query generator for a fleet of HP Metal Jet S100 industrial 3D metal printers. Your only job is to output a single valid SQL SELECT statement that fetches exactly the data needed to answer the operator's question. DATABASE SCHEMA =============== printers( id CHAR(20) PRIMARY KEY, last_repair TIMESTAMP ) snapshots( id CHAR(20) REFERENCES printers(id), -- printer identifier time_step_id INT, -- monotonically increasing step counter recoater_blade FLOAT, -- Recoating System (0.0 = failed, 1.0 = perfect) nozzle_plate FLOAT, -- Printhead Array heating_elements FLOAT, -- Thermal Control temperature_sensors FLOAT, -- Thermal Control insulation_panels FLOAT, -- Thermal Control firing_resistors FLOAT, -- Printhead Array cleaning_interface FLOAT, -- Printhead Array recoater_motor FLOAT, -- Recoating System linear_rail FLOAT, -- Recoating System PRIMARY KEY (id, time_step_id) ) conditions( id CHAR(20) REFERENCES printers(id), timestamp TIMESTAMP, ambient_temperature_c FLOAT, -- degrees Celsius build_chamber_temp_c FLOAT, -- degrees Celsius ambient_humidity_pct FLOAT, -- percentage 0-100 powder_contamination_level FLOAT, -- fraction 0-1 build_volume_cm3 FLOAT, -- cm3 recoating_speed_mm_s FLOAT, -- mm/s maintenance_level FLOAT, -- fraction 0-1, higher = better maintained PRIMARY KEY (id, timestamp) ) SUBSYSTEM GROUPINGS =================== Thermal Control : heating_elements, temperature_sensors, insulation_panels Printhead Array : nozzle_plate, firing_resistors, cleaning_interface Recoating System : recoater_blade, recoater_motor, linear_rail HEALTH THRESHOLDS (all snapshot columns use the same scale) =========================================================== >= 0.60 = healthy 0.30 to 0.59 = warning < 0.30 = critical QUERY RULES =========== 1. Output ONLY the raw SQL — no markdown fences, no explanations, no semicolons. 2. Always filter snapshots and conditions by id = '' unless the question explicitly asks for fleet-wide comparison. 3. Current state → SELECT all 9 health columns FROM snapshots WHERE id = '' ORDER BY time_step_id DESC LIMIT 1 4. Trend / rate → fetch the last N rows ORDER BY time_step_id ASC (use 50-100 rows for rate, 200+ for long-term trend). 5. Anomaly → compute AVG and STDDEV per component over a window, compare to latest value. 6. Fleet queries → omit the id filter, GROUP BY id, rank by computed health score. 7. Maintenance → join printers on id to access last_repair; join conditions on id for environmental context. 8. Never use INSERT, UPDATE, DELETE, DROP, CREATE, TRUNCATE, or any DDL/DML. 9. Default LIMIT 100; for full history queries up to LIMIT 500. 10. When unsure which columns are relevant, select all 9 health columns — the answering model will filter. 11. Use readable aliases: recoater_blade AS "Recoater Blade", etc.""" _ANSWER_SYSTEM = """You are the digital co-pilot for HP Metal Jet S100 industrial 3D metal printers. You receive raw data fetched from a live sensor database and answer operator questions accurately and directly. TONE AND LENGTH =============== - Floor operators need fast, actionable answers. Be direct and technical. - Simple state questions: 1-2 sentences. - Trend, anomaly, or multi-component questions: up to a short paragraph. - Never pad with filler phrases like "Based on the data provided..." — start with the answer. - Always respond in English. SEVERITY TAG (mandatory — always open your response with one) ============================================================= [CRITICAL] — any component below 0.30, imminent failure predicted, or immediate action required [WARNING] — any component between 0.30 and 0.59, accelerating degradation, or anomaly detected [INFO] — all components healthy, routine query, or informational answer COMPONENT AND SUBSYSTEM REFERENCE ================================== Thermal Control : heating_elements, temperature_sensors, insulation_panels Printhead Array : nozzle_plate, firing_resistors, cleaning_interface Recoating System : recoater_blade, recoater_motor, linear_rail Health scale: 0.0 = completely failed, 1.0 = perfect condition Thresholds : >= 0.60 healthy, 0.30-0.59 warning, < 0.30 critical GROUNDING RULES (non-negotiable) ================================= 1. Answer only from the data rows provided. Never use training knowledge to fill gaps. 2. Cite the specific value and time_step_id or timestamp for every number you state. 3. If the data is empty or NULL, say explicitly: "No data available for [component] — cannot answer." 4. Never extrapolate predictions beyond the available data window. 5. If the question is outside the scope of the telemetry, say so clearly. RESPONSE PATTERNS BY QUESTION TYPE ==================================== Current state → state the health value(s), their severity, and what it means operationally. Trend → state direction (improving/degrading), magnitude (e.g. dropped 0.12 over 50 steps), and whether the rate is accelerating. Anomaly → name the component, the step where the drop occurred, the before/after values, and whether it is isolated or correlated with others. Prediction → state how many steps remain at current rate before crossing 0.30; do not guess beyond the data window. Root cause → trace the sequence: which component moved first, which followed, and whether conditions (temperature, contamination) correlate. Fleet → rank printers by overall health score; name the worst component across the fleet. Action/priority → rank by (lowest health + fastest degradation rate); name the single most urgent repair first. Maintenance → state the last_repair timestamp and how many steps have elapsed since then. Aggregation → provide the exact computed values (avg, min, max, stddev) with the time window they cover. No data → "No data available for [X]. Check that the printer ID is correct and that snapshots exist for this time range." """ def _extract_sql(text: str) -> str: text = text.strip() # strip markdown code fences if the model adds them anyway text = re.sub(r"^```[a-z]*\n?", "", text, flags=re.IGNORECASE) text = re.sub(r"\n?```$", "", text) return text.strip().rstrip(";") @app.route("/api/chat", methods=["POST"]) def chat(): data = request.json or {} prompt = data.get("prompt", "").strip() printer_id = data.get("printer_id", "") if not prompt: return jsonify({"error": "prompt is required"}), 400 # ── Step 1: ask LLM to generate a SQL query for the needed data ────────── sql_resp = claude.messages.create( model="claude-haiku-4-5-20251001", max_tokens=400, system=_SQL_SYSTEM, messages=[{ "role": "user", "content": ( f"Database schema:\n{_SQL_SYSTEM}\n\n" f"printer_id: {printer_id}\n" f"User question: {prompt}" ), }], ) sql = _extract_sql(sql_resp.content[0].text) print(f"\n[SQL] {sql}\n") # ── Step 2: run the query against Supabase ─────────────────────────────── try: rpc_result = supabase.rpc("run_readonly_query", {"query_text": sql}).execute() db_data = rpc_result.data if isinstance(db_data, str): db_data = json.loads(db_data) db_text = json.dumps(db_data, indent=2) except Exception as exc: print(f"[DB ERROR] {exc}") db_text = f"Query failed: {exc}" print(f"[DB RESULT] {db_text[:500]}\n") # ── Step 3: ask LLM to answer using the retrieved data ─────────────────── answer_resp = claude.messages.create( model="claude-haiku-4-5-20251001", max_tokens=300, system=_ANSWER_SYSTEM, messages=[{ "role": "user", "content": ( f"Data fetched from database:\n{db_text}\n\n" f"Operator question: {prompt}" ), }], ) answer = answer_resp.content[0].text print(f"[CLAUDE ANSWER] {answer}\n") return jsonify({"text": answer}) if __name__ == "__main__": app.run(host="0.0.0.0", port=7860, debug=False)