HelioSense

import os
import time
import argparse
from datetime import datetime
 
import cv2
import numpy as np
 
try:
    from tflite_runtime.interpreter import Interpreter
except ImportError:
    # Fallback if only full TensorFlow is installed on the board.
    from tensorflow.lite.python.interpreter import Interpreter
 
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
MODEL_PATH = "models/dust_classifier_unoq.tflite"
CLASS_NAMES = ["CLEAN", "DUSTY"]     # must match training label order
IMG_SIZE = (224, 224)                # must match training input size
CONFIDENCE_THRESHOLD = 0.5           # sigmoid threshold: >0.5 -> DUSTY
SNAPSHOT_DIR = "snapshots"
CAMERA_INDEX = 0                     # 0 = default camera device
INFER_EVERY_N_FRAMES = 1             # raise this to reduce CPU load
 
 
def load_interpreter(model_path: str) -> Interpreter:
    if not os.path.isfile(model_path):
        raise FileNotFoundError(
            f"Model not found at '{model_path}'. Train and convert it first "
            f"with train_vgg16_and_convert.py, then copy the .tflite file here."
        )
    interpreter = Interpreter(model_path=model_path)
    interpreter.allocate_tensors()
    return interpreter
 
 
def preprocess(frame_bgr: np.ndarray) -> np.ndarray:
    img = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)
    img = cv2.resize(img, IMG_SIZE)
    img = img.astype(np.float32) / 255.0
    return np.expand_dims(img, axis=0)
 
 
def classify(interpreter: Interpreter, input_details, output_details, frame_bgr: np.ndarray):
    input_tensor = preprocess(frame_bgr)
    interpreter.set_tensor(input_details[0]["index"], input_tensor)
    interpreter.invoke()
    prob_dusty = float(interpreter.get_tensor(output_details[0]["index"]).ravel()[0])
 
    label = CLASS_NAMES[1] if prob_dusty > CONFIDENCE_THRESHOLD else CLASS_NAMES[0]
    confidence = prob_dusty if label == "DUSTY" else 1.0 - prob_dusty
    return label, confidence
 
 
def annotate(frame, label, confidence, fps):
    color = (0, 0, 255) if label == "DUSTY" else (0, 200, 0)
    text = f"{label} ({confidence * 100:.1f}%)"
    cv2.rectangle(frame, (0, 0), (frame.shape[1], 40), (0, 0, 0), -1)
    cv2.putText(frame, text, (10, 28), cv2.FONT_HERSHEY_SIMPLEX, 0.8, color, 2)
    cv2.putText(frame, f"FPS: {fps:.1f}", (frame.shape[1] - 130, 28),
                cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 1)
    return frame
 
 
def save_snapshot(frame, label):
    os.makedirs(SNAPSHOT_DIR, exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    path = os.path.join(SNAPSHOT_DIR, f"{label.lower()}_{timestamp}.jpg")
    cv2.imwrite(path, frame)
    return path
 
 
def main():
    parser = argparse.ArgumentParser(description="HelioSense live dust classifier (UNO Q)")
    parser.add_argument("--model", default=MODEL_PATH)
    parser.add_argument("--camera", type=int, default=CAMERA_INDEX)
    parser.add_argument("--save-every-n-detections", type=int, default=30,
                         help="Save a snapshot every N processed frames (0 = every frame)")
    parser.add_argument("--headless", action="store_true",
                         help="Run without cv2.imshow (use over SSH with no display)")
    args = parser.parse_args()
 
    interpreter = load_interpreter(args.model)
    input_details = interpreter.get_input_details()
    output_details = interpreter.get_output_details()
 
    cap = cv2.VideoCapture(args.camera)
    if not cap.isOpened():
        raise RuntimeError(
            f"Could not open camera index {args.camera}. Check `ls /dev/video*` "
            f"on the UNO Q, or pass a GStreamer pipeline for a CSI camera."
        )
 
    frame_count = 0
    prev_time = time.time()
    fps = 0.0
 
    print("HelioSense live inference started. Press 'q' to quit (or Ctrl+C headless).")
 
    try:
        while True:
            ok, frame = cap.read()
            if not ok:
                print("Warning: failed to read frame, retrying...")
                time.sleep(0.1)
                continue
 
            frame_count += 1
            if frame_count % INFER_EVERY_N_FRAMES == 0:
                label, confidence = classify(interpreter, input_details, output_details, frame)
 
                now = time.time()
                fps = 1.0 / max(now - prev_time, 1e-6)
                prev_time = now
 
                annotated = annotate(frame.copy(), label, confidence, fps)
 
                if args.save_every_n_detections == 0 or frame_count % max(args.save_every_n_detections, 1) == 0:
                    save_snapshot(annotated, label)
 
                print(f"[{datetime.now().isoformat(timespec='seconds')}] "
                      f"{label} conf={confidence:.2f} fps={fps:.1f}")
 
                if not args.headless:
                    cv2.imshow("HelioSense - Dust Classifier", annotated)
                    if cv2.waitKey(1) & 0xFF == ord("q"):
                        break
 
    except KeyboardInterrupt:
        print("Stopped by user.")
    finally:
        cap.release()
        if not args.headless:
            cv2.destroyAllWindows()
 
 
if __name__ == "__main__":
    main()

import os
import numpy as np
import tensorflow as tf
from tensorflow.keras.applications import VGG16
from tensorflow.keras.layers import Dense, Flatten, Dropout, GlobalAveragePooling2D
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.preprocessing.image import ImageDataGenerator
from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint, ReduceLROnPlateau
from sklearn.metrics import classification_report, confusion_matrix
import matplotlib.pyplot as plt
import seaborn as sns
 
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
DATASET_DIR = "dataset"           # expects DATASET_DIR/Clean and DATASET_DIR/Dusty
IMG_SIZE = (224, 224)             # VGG16 native input size
BATCH_SIZE = 32
EPOCHS_HEAD = 15                  # train just the new classifier head
EPOCHS_FINE_TUNE = 10             # then fine-tune top VGG16 blocks
VAL_SPLIT = 0.2
CLASS_NAMES = ["Clean", "Dusty"]  # index 0 = Clean, index 1 = Dusty
 
OUT_H5 = "dust_classifier_vgg16.h5"
OUT_TFLITE = "dust_classifier_unoq.tflite"
OUT_REPORT = "training_report.txt"
 
 
def build_generators():
    """Image augmentation + train/validation split."""
    train_datagen = ImageDataGenerator(
        rescale=1.0 / 255.0,
        rotation_range=20,
        width_shift_range=0.15,
        height_shift_range=0.15,
        shear_range=0.1,
        zoom_range=0.15,
        brightness_range=(0.7, 1.3),   # important: dust detection is lighting-sensitive
        horizontal_flip=True,
        validation_split=VAL_SPLIT,
    )
 
    train_gen = train_datagen.flow_from_directory(
        DATASET_DIR,
        target_size=IMG_SIZE,
        batch_size=BATCH_SIZE,
        class_mode="binary",
        classes=CLASS_NAMES,
        subset="training",
        shuffle=True,
        seed=42,
    )
 
    val_gen = train_datagen.flow_from_directory(
        DATASET_DIR,
        target_size=IMG_SIZE,
        batch_size=BATCH_SIZE,
        class_mode="binary",
        classes=CLASS_NAMES,
        subset="validation",
        shuffle=False,
        seed=42,
    )
 
    return train_gen, val_gen
 
 
def build_model():
    """VGG16 backbone (ImageNet weights) + custom binary classification head."""
    base_model = VGG16(weights="imagenet", include_top=False, input_shape=(*IMG_SIZE, 3))
    base_model.trainable = False  # freeze for initial head training
 
    x = base_model.output
    x = GlobalAveragePooling2D()(x)
    x = Dense(256, activation="relu")(x)
    x = Dropout(0.4)(x)
    x = Dense(64, activation="relu")(x)
    x = Dropout(0.3)(x)
    output = Dense(1, activation="sigmoid")(x)  # binary: 0=Clean, 1=Dusty
 
    model = Model(inputs=base_model.input, outputs=output)
    model.compile(
        optimizer=Adam(learning_rate=1e-4),
        loss="binary_crossentropy",
        metrics=["accuracy", tf.keras.metrics.Precision(name="precision"),
                 tf.keras.metrics.Recall(name="recall")],
    )
    return model, base_model
 
 
def fine_tune(model, base_model):
    """Unfreeze the last VGG16 conv block and continue training at a lower LR."""
    base_model.trainable = True
    for layer in base_model.layers[:-4]:
        layer.trainable = False
 
    model.compile(
        optimizer=Adam(learning_rate=1e-5),
        loss="binary_crossentropy",
        metrics=["accuracy", tf.keras.metrics.Precision(name="precision"),
                 tf.keras.metrics.Recall(name="recall")],
    )
    return model
 
 
def evaluate_and_report(model, val_gen):
    val_gen.reset()
    y_true = val_gen.classes
    y_pred_prob = model.predict(val_gen, verbose=1).ravel()
    y_pred = (y_pred_prob > 0.5).astype(int)
 
    report = classification_report(y_true, y_pred, target_names=CLASS_NAMES)
    cm = confusion_matrix(y_true, y_pred)
 
    print(report)
    print("Confusion matrix:\n", cm)
 
    with open(OUT_REPORT, "w") as f:
        f.write("HelioSense Dust Classifier - Evaluation Report\n")
        f.write("=" * 50 + "\n\n")
        f.write(report)
        f.write("\nConfusion Matrix:\n")
        f.write(str(cm))
 
    plt.figure(figsize=(5, 4))
    sns.heatmap(cm, annot=True, fmt="d", cmap="Blues",
                xticklabels=CLASS_NAMES, yticklabels=CLASS_NAMES)
    plt.xlabel("Predicted")
    plt.ylabel("Actual")
    plt.title("Confusion Matrix")
    plt.tight_layout()
    plt.savefig("confusion_matrix.png")
    print("Saved confusion_matrix.png and", OUT_REPORT)
 
 
def representative_dataset_gen(val_gen, num_samples=100):
    """Feeds real sample images to the converter for full-integer quantization."""
    count = 0
    for images, _ in val_gen:
        for img in images:
            yield [np.expand_dims(img, axis=0).astype(np.float32)]
            count += 1
            if count >= num_samples:
                return
 
 
def convert_to_tflite(model, val_gen):
    """
    Convert to TFLite with dynamic-range quantization (int8 weights, float
    activations). This is the recommended balance of size/speed/accuracy for
    the UNO Q's CPU-based TFLite interpreter. A representative dataset is
    used so the converter can calibrate quantization ranges properly.
    """
    converter = tf.lite.TFLiteConverter.from_keras_model(model)
    converter.optimizations = [tf.lite.Optimize.DEFAULT]
    converter.representative_dataset = lambda: representative_dataset_gen(val_gen)
    # Keep float32 input/output for simplicity of the inference script;
    # internal weights still get quantized by Optimize.DEFAULT.
    tflite_model = converter.convert()
 
    with open(OUT_TFLITE, "wb") as f:
        f.write(tflite_model)
 
    size_kb = os.path.getsize(OUT_TFLITE) / 1024
    print(f"Saved {OUT_TFLITE} ({size_kb:.1f} KB) -- copy this file to the UNO Q.")
 
 
def main():
    if not os.path.isdir(DATASET_DIR):
        raise FileNotFoundError(
            f"Dataset directory '{DATASET_DIR}' not found. Expected structure:\n"
            f"  {DATASET_DIR}/Clean/*.jpg\n  {DATASET_DIR}/Dusty/*.jpg"
        )
 
    train_gen, val_gen = build_generators()
    model, base_model = build_model()
 
    callbacks = [
        EarlyStopping(monitor="val_loss", patience=4, restore_best_weights=True),
        ReduceLROnPlateau(monitor="val_loss", factor=0.5, patience=2, min_lr=1e-7),
        ModelCheckpoint("best_head.h5", monitor="val_accuracy", save_best_only=True),
    ]
 
    print("Stage 1: training classifier head (VGG16 frozen)...")
    model.fit(train_gen, validation_data=val_gen, epochs=EPOCHS_HEAD, callbacks=callbacks)
 
    print("Stage 2: fine-tuning top VGG16 block...")
    model = fine_tune(model, base_model)
    model.fit(train_gen, validation_data=val_gen, epochs=EPOCHS_FINE_TUNE, callbacks=callbacks)
 
    model.save(OUT_H5)
    print(f"Saved Keras model: {OUT_H5}")
 
    evaluate_and_report(model, val_gen)
    convert_to_tflite(model, val_gen)
 
 
if __name__ == "__main__":
    main()

/*
  HelioSense - Dual-Axis Solar Tracker
  Target: Arduino UNO Q -- MCU side (STM32U585, runs as an Arduino sketch under Zephyr)

  This file runs on the UNO Q's MICROCONTROLLER (real-time side), NOT on the Linux MPU.
  It is uploaded exactly like a normal Arduino sketch (Arduino IDE / Arduino CLI / App Lab).

  Why this is UNO Q-safe:
  - Uses the standard Arduino Servo.h API (PWM handled by the core, no manual timer
    register access). No TCCR1A/OCR1A/ICR1 etc. -- those are AVR-specific and do not
    exist on the STM32U585, so touching them would not compile and would be wrong anyway.
  - Uses analogRead() / standard Arduino API only, which the UNO Q core implements
    for the MCU's ADC pins.
  - Optionally reports data to the Linux side (MPU) over the Bridge/RPC link
    (Arduino_RouterBridge), so the Python dust-classifier process can log or react
    to tracker state. This is optional -- comment out if you don't need it.
*/

#include <Servo.h>

// If you want the MCU to expose data to the MPU (Python) side, uncomment:
// #include <Arduino_RouterBridge.h>
// #define USE_BRIDGE

// ---------- Pin configuration ----------
// Adjust these analog pins to match your wiring on the UNO Q MCU header.
const uint8_t LDR_TOP_LEFT     = A0;
const uint8_t LDR_TOP_RIGHT    = A1;
const uint8_t LDR_BOTTOM_LEFT  = A2;
const uint8_t LDR_BOTTOM_RIGHT = A3;

const uint8_t SERVO_PAN_PIN  = 9;   // horizontal axis
const uint8_t SERVO_TILT_PIN = 10;  // vertical axis

// ---------- Tuning parameters ----------
const int   LIGHT_TOLERANCE   = 40;    // dead-band: ignore differences smaller than this
const uint8_t SERVO_STEP_DEG  = 1;     // how many degrees to move per correction
const uint16_t LOOP_DELAY_MS  = 60;    // control loop period

const uint8_t PAN_MIN_DEG  = 0;
const uint8_t PAN_MAX_DEG  = 180;
const uint8_t TILT_MIN_DEG = 0;
const uint8_t TILT_MAX_DEG = 180;

Servo panServo;
Servo tiltServo;

int panAngle  = 90;   // start centered
int tiltAngle = 90;

struct LdrReadings {
  int topLeft;
  int topRight;
  int bottomLeft;
  int bottomRight;
};

LdrReadings readLdrs() {
  LdrReadings r;
  r.topLeft     = analogRead(LDR_TOP_LEFT);
  r.topRight    = analogRead(LDR_TOP_RIGHT);
  r.bottomLeft  = analogRead(LDR_BOTTOM_LEFT);
  r.bottomRight = analogRead(LDR_BOTTOM_RIGHT);
  return r;
}

int clampAngle(int angle, int minDeg, int maxDeg) {
  if (angle < minDeg) return minDeg;
  if (angle > maxDeg) return maxDeg;
  return angle;
}

// Moves pan/tilt servos to chase the brighter side, same logic as the
// classic 4-LDR tracker: compare averaged top/bottom and left/right halves.
void updateTracking(const LdrReadings &r) {
  int avgTop    = (r.topLeft + r.topRight) / 2;
  int avgBottom = (r.bottomLeft + r.bottomRight) / 2;
  int avgLeft   = (r.topLeft + r.bottomLeft) / 2;
  int avgRight  = (r.topRight + r.bottomRight) / 2;

  int vertDiff = avgTop - avgBottom;
  int horizDiff = avgLeft - avgRight;

  if (abs(vertDiff) > LIGHT_TOLERANCE) {
    tiltAngle += (vertDiff > 0) ? SERVO_STEP_DEG : -SERVO_STEP_DEG;
    tiltAngle = clampAngle(tiltAngle, TILT_MIN_DEG, TILT_MAX_DEG);
    tiltServo.write(tiltAngle);
  }

  if (abs(horizDiff) > LIGHT_TOLERANCE) {
    panAngle += (horizDiff > 0) ? -SERVO_STEP_DEG : SERVO_STEP_DEG;
    panAngle = clampAngle(panAngle, PAN_MIN_DEG, PAN_MAX_DEG);
    panServo.write(panAngle);
  }
}

void printDebug(const LdrReadings &r) {
  Serial.print("TL="); Serial.print(r.topLeft);
  Serial.print(" TR="); Serial.print(r.topRight);
  Serial.print(" BL="); Serial.print(r.bottomLeft);
  Serial.print(" BR="); Serial.print(r.bottomRight);
  Serial.print(" | pan="); Serial.print(panAngle);
  Serial.print(" tilt="); Serial.println(tiltAngle);
}

#ifdef USE_BRIDGE
// Exposed to the Python (MPU) side via Bridge.call("get_tracker_state")
String get_tracker_state() {
  LdrReadings r = readLdrs();
  String s = String(r.topLeft) + "," + String(r.topRight) + "," +
             String(r.bottomLeft) + "," + String(r.bottomRight) + "," +
             String(panAngle) + "," + String(tiltAngle);
  return s;
}
#endif

void setup() {
  Serial.begin(115200);

  panServo.attach(SERVO_PAN_PIN);
  tiltServo.attach(SERVO_TILT_PIN);
  panServo.write(panAngle);
  tiltServo.write(tiltAngle);

#ifdef USE_BRIDGE
  Bridge.begin();
  if (!Bridge.provide("get_tracker_state", get_tracker_state)) {
    Serial.println("Bridge: failed to register get_tracker_state");
  }
#endif

  Serial.println("HelioSense MCU tracker ready.");
}

void loop() {
  LdrReadings readings = readLdrs();
  updateTracking(readings);
  printDebug(readings);

#ifdef USE_BRIDGE
  Bridge.update();
#endif

  delay(LOOP_DELAY_MS);
}