import os, io, csv from datetime import datetime from collections import deque from typing import Optional import threading import numpy as np from PIL import Image import torch import torch.nn as nn from fastapi import FastAPI, UploadFile, File from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, HTMLResponse from pydantic import BaseModel import torchvision from torchvision import transforms as T from torchvision.transforms.functional import InterpolationMode # basic config torch.set_num_threads(1) _INFER_LOCK = threading.Lock() DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu") MODEL_PATH = os.getenv("MODEL_PATH", "/app/spoilage_model.pth") MAX_POINTS = int(os.getenv("MAX_POINTS", "240")) FRESHNESS_NAMES = ["Fresh", "Spoiled"] # preprocessing IMG_TX = T.Compose([ T.Resize((224, 224), interpolation=InterpolationMode.BICUBIC), T.ToTensor() ]) # FastAPI app = FastAPI(title="Fruit Freshness & Gas Detector") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"] ) LAST = {"vision": None, "vision_updated": None, "gas": None, "gas_updated": None} HISTORY = deque(maxlen=MAX_POINTS) # model class Model(nn.Module): def __init__(self): super().__init__() self.alpha = 0.7 try: self.base = torchvision.models.resnet18(weights=None) except TypeError: self.base = torchvision.models.resnet18(pretrained=False) for m in self.base.modules(): if hasattr(m, "inplace"): m.inplace = False for p in list(self.base.parameters())[:-15]: p.requires_grad = False self.base.fc = nn.Sequential() self.block1 = nn.Sequential( nn.Linear(512, 256), nn.ReLU(), nn.Dropout(0.2), nn.Linear(256, 128), ) self.block2 = nn.Sequential( nn.Linear(128, 128), nn.ReLU(), nn.Dropout(0.1), nn.Linear(128, 9) ) self.block3 = nn.Sequential( nn.Linear(128, 32), nn.ReLU(), nn.Dropout(0.1), nn.Linear(32, 2) ) def forward(self, x): x = self.base(x) x = self.block1(x) y1 = self.block2(x) y2 = self.block3(x) return y1, y2 _model = None def load_model(): global _model if _model is not None: return _model # TorchScript try: m = torch.jit.load(MODEL_PATH, map_location=DEVICE) m.eval().to(DEVICE) _model = m return _model except Exception: pass # full module or state_dict obj = torch.load(MODEL_PATH, map_location=DEVICE) if isinstance(obj, nn.Module): _model = obj.eval().to(DEVICE) return _model if isinstance(obj, dict): m = Model().to(DEVICE) m.load_state_dict(obj, strict=True) m.eval() _model = m return _model raise RuntimeError("Unsupported checkpoint format at MODEL_PATH") def predict_pil(pil: Image.Image): model = load_model() x = IMG_TX(pil).unsqueeze(0).to(DEVICE) with _INFER_LOCK, torch.inference_mode(): out = model(x) if isinstance(out, (tuple, list)) and len(out) >= 2: y2 = out[1] # freshness head else: y2 = out probs_t = torch.softmax(y2, dim=1)[0].tolist() idx = int(np.argmax(probs_t)) label = FRESHNESS_NAMES[idx] conf = float(probs_t[idx]) * 100.0 raw = {FRESHNESS_NAMES[i]: float(p) for i, p in enumerate(probs_t)} return {"label": label, "confidence": round(conf, 1), "raw": {"probs": raw}} # Vision @app.post("/predict") async def predict(image: UploadFile = File(...)): try: data = await image.read() pil = Image.open(io.BytesIO(data)).convert("RGB") except Exception as e: return JSONResponse({"error": "invalid_image", "detail": f"Could not read image ({e})"}, status_code=400) try: out = predict_pil(pil) LAST["vision"] = out LAST["vision_updated"] = datetime.utcnow().isoformat() return JSONResponse(out) except Exception as e: return JSONResponse({"error": "inference_failed", "detail": str(e)}, status_code=500) # Gas class GasReading(BaseModel): vrl: Optional[float] = None adc: Optional[int] = None adc_max: Optional[int] = 4095 vref: Optional[float] = 3.3 rl: Optional[float] = 10000.0 rs: Optional[float] = None r0: Optional[float] = None def _ppm_from_ratio(ratio: float, a: float, b: float) -> float: if ratio is None or ratio <= 0: return 0.0 return max(0.0, a * (ratio ** b)) @app.post("/gas") def gas(g: GasReading): VREF = float(g.vref or 3.3) RL = float(g.rl or 10000.0) used_adc = None adc_max = int(g.adc_max or 4095) if g.vrl is None and g.adc is not None: used_adc = int(g.adc) g.vrl = (used_adc / adc_max) * VREF if g.vrl is None and g.rs is None: return JSONResponse({"error": "need vrl, adc, or rs"}, status_code=400) rs = float(g.rs) if g.rs is not None else ((VREF - g.vrl) * RL) / max(0.001, g.vrl) r0 = float(g.r0) if g.r0 is not None else rs ratio = rs / max(1e-6, r0) data = { "vrl": round(g.vrl, 3), "rs": round(rs, 1), "r0": round(r0, 1), "ratio": round(ratio, 3), "ppm": { "co2": round(_ppm_from_ratio(ratio, 116.6021, -2.7690), 1), "nh3": round(_ppm_from_ratio(ratio, 102.6940, -2.4880), 1), "benzene": round(_ppm_from_ratio(ratio, 76.63, -2.1680), 1), "alcohol": round(_ppm_from_ratio(ratio, 77.255, -3.18), 1), }, "raw": {"adc": used_adc, "adc_max": adc_max, "vref": VREF, "rl": RL, "r0": r0} } LAST["gas"] = data LAST["gas_updated"] = datetime.utcnow().isoformat() # Compute combined decision (vision + gas thresholds) summary = _summarize(LAST) decision = summary["decision"] # Save in history (ppm + decision) HISTORY.append({ "time": datetime.utcnow().isoformat(), "ppm": data["ppm"], "decision": decision }) return {"ok": True, "data": data, "decision": decision} @app.get("/history") def history(): return {"history": list(HISTORY)} @app.get("/export.csv") def export_csv(): buf = io.StringIO() w = csv.writer(buf) w.writerow(["timestamp_utc", "co2_ppm", "nh3_ppm", "benzene_ppm", "alcohol_eq", "decision"]) for r in HISTORY: ppm = r["ppm"] w.writerow([ r["time"], ppm.get("co2"), ppm.get("nh3"), ppm.get("benzene"), ppm.get("alcohol"), r.get("decision") ]) return HTMLResponse( content=buf.getvalue(), media_type="text/csv", headers={"Content-Disposition": 'attachment; filename="gas_history.csv"'} ) # Summary / Health def _summarize(last: dict) -> dict: """ Combine the latest vision prediction and gas ppm into a single, simple decision. - Vision only votes 'rotten' if label says spoiled/rotten AND confidence >= VISION_MIN_CONF. - Any high gas flag can mark the sample as spoiled. """ # thresholds VISION_MIN_CONF = 60.0 # % CO2_HI = 2000.0 # ppm NH3_HI = 15.0 # ppm BENZ_HI = 5.0 # ppm ALC_HI = 10.0 # eq pred = last.get("vision") or {} gas = (last.get("gas") or {}).get("ppm", {}) or {} co2 = gas.get("co2") nh3 = gas.get("nh3") benz = gas.get("benzene") alco = gas.get("alcohol") # gas flags co2_hi = (co2 is not None) and (co2 >= CO2_HI) nh3_hi = (nh3 is not None) and (nh3 >= NH3_HI) voc_hi = ((benz or 0) >= BENZ_HI) or ((alco or 0) >= ALC_HI) # vision vote (label + confidence) label = str(pred.get("label") or "") conf = float(pred.get("confidence") or 0.0) looks_rotten = ("spoiled" in label.lower() or "rotten" in label.lower()) and (conf >= VISION_MIN_CONF) spoiled = bool(looks_rotten or co2_hi or nh3_hi or voc_hi) return { "vision": pred, "gas_ppm": {"co2": co2, "nh3": nh3, "benzene": benz, "alcohol": alco}, "gas_flags": {"co2_high": co2_hi, "nh3_high": nh3_hi, "voc_high": voc_hi}, "decision": "SPOILED" if spoiled else "FRESH", "meta": { "max_points": MAX_POINTS, "thresholds": { "vision_min_conf": VISION_MIN_CONF, "co2_hi": CO2_HI, "nh3_hi": NH3_HI, "benz_hi": BENZ_HI, "alcohol_hi": ALC_HI } } } @app.get("/summary") def summary(): return _summarize(LAST) @app.get("/healthz") def healthz(): return {"ok": True, "time": datetime.utcnow().isoformat()} # UI @app.get("/", response_class=HTMLResponse) def welcome(): return """ Welcome • Smart Freshness Checker
Smart Freshness Checker

Welcome 👋

This simple tool helps you check whether your fruit and veggies are still fresh. Take a quick photo and add air-reading values from a small plug-in sensor. The app looks for early signs of spoilage and gives you a clear “Fresh” or “Spoiled” result.

How it works:
1) Open the app and upload a photo, or use the camera.
2) If you have a gas sensor, send readings to improve the result.
3) See the final prediction and a simple chart of recent readings.

🚀 Launch App ⬇️ Download Data CSV
""" @app.get("/home", response_class=HTMLResponse) def home(): return """ Fruit Freshness & Gas Detector

🍎 Fruit Freshness & Gas Detector

Upload, predict, and view gas-based decision

1) Upload or Capture Fruit Image idle

preview

2) Gas Sensor Reading idle

ADC Vref RL(Ω) R0(Ω)

3) Final Decision


  


  

    
4) Gas Chart (last points)

    

      
      
No readings yet

    

  







"""