Software

import pandas as pd
from sklearn.preprocessing import MinMaxScaler
import matplotlib.pyplot as plt
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix, accuracy_score
from sklearn.cluster import KMeans
from sklearn.ensemble import IsolationForest
import joblib

#loads the csv
df = pd.read_csv("mock_rover_sustainability_data.csv", parse_dates=["timestamp"])

#checks if data is corrupted
df.info()
df.describe()

#separates columns
non_numeric_cols = ["timestamp"]
numeric_cols = df.select_dtypes(include="number").columns



#removes extreme outliers
def remove_outliers_iqr(df, columns, k=1.5):
    filtered_df = df.copy()
    for col in columns:
        Q1 = filtered_df[col].quantile(0.25)
        Q3 = filtered_df[col].quantile(0.75)
        IQR = Q3 - Q1

        lower = Q1 - k * IQR
        upper = Q3 + k * IQR

        filtered_df = filtered_df[
            (filtered_df[col] >= lower) &
            (filtered_df[col] <= upper)
        ]
    return filtered_df

#apply removing extreme outliers to numeric features only (not time stamps)
df_clean = remove_outliers_iqr(df, numeric_cols)

print(f"Rows before: {len(df)}, after: {len(df_clean)}")



#normalzies features using minmax scaling
#(making every value between 0 and 1)
scaler = MinMaxScaler()

df_scaled = df_clean.copy()
df_scaled[numeric_cols] = scaler.fit_transform(df_clean[numeric_cols])

#implementing isolationforesting

iso = IsolationForest(
    n_estimators=200,
    contamination=0.05,
    random_state=42
)

df_scaled["anomaly_score"] = iso.fit_predict(
    df_scaled[numeric_cols]
)

# Convert {-1, 1} → {1 = anomaly, 0 = normal}
df_scaled["anomaly_score"] = (df_scaled["anomaly_score"] == -1).astype(int)


#implementing kmeans

kmeans = KMeans(
    n_clusters=2,
    random_state=42,
    n_init=10
)

df_scaled["cluster_label"] = kmeans.fit_predict(
    df_scaled[numeric_cols]
)

#tests if scaling worked (all cols are between 0 and 1)
assert np.all(df_scaled[numeric_cols].min() >= -1e-9)
assert np.all(df_scaled[numeric_cols].max() <= 1 + 1e-9)

print("Scaling Test Passed!!")


#health rules (subject to change)
def label_health(row):
    if row["surface_temp_max_c"] > 0.8:
        return "Heat stress"
    elif row["soil_ph"] < 0.4 or row["soil_ph"] > 0.7:
        return "pH issue"
    else:
        return "Healthy"

df_scaled["health_status"] = df_scaled.apply(label_health, axis=1)

#adding extra "noise" to take into account imperfect sensors
df_scaled["surface_temp_max_c"] += np.random.normal(0, 0.05, len(df_scaled))



#begin random foresting
target_col = "health_status"


#define x -> features and y -> columns
X = df_scaled[
    list(numeric_cols) +
    ["cluster_label", "anomaly_score"]
]

y = df_scaled["health_status"]

#Train test split
X_train, X_test, y_train, y_test = train_test_split(
    X, y,
    test_size=0.2,
    random_state=42,
    stratify=y
)

#training the random forest
rf = RandomForestClassifier(
    n_estimators=200, #num of trees
    max_depth=None,   #trees can fully grow, won't get cut off
    min_samples_split=2,
    random_state=42,
    n_jobs=-1       #uses all CPU cores
)



rf.fit(X_train, y_train)

#evaluation
y_pred = rf.predict(X_test)

#how accurate the predictions were
print("Accuracy:", accuracy_score(y_test, y_pred))

#precision = true pos/(true pos+ false pos) (accuracy of positive predictions)
#recall = true pos/(true pos + false neg) (ability to find all positives)
#f1 score = mean of precision and recall
#support -> actual numbe of real occurances (true positives, true negatives)

print("\nClassification Report: \n", classification_report(y_test, y_pred))

#confusion matrices show the false postives(bottom left), false negatives(top right),
# true positives(top left), true negatives(bottom right)
print("\n Confusion Matrix: \n ", confusion_matrix(y_test, y_pred))



#pickles -> convert python objects to bytes so we can load it later
#trains model saves it to a pickle so we can load it later for inference
#Joblib is a library built on top of pickle where it actually compresses it and loads it
joblib.dump(rf, "rf_model.pkl")
joblib.dump(scaler, "scaler.pkl")
joblib.dump(kmeans, "kmeans.pkl")
joblib.dump(iso, "iso.pkl")
joblib.dump(list(numeric_cols), "numeric_cols.pkl")

from fastapi import APIRouter, HTTPException
from api.routers.sensors import latest_sensor_data
from api.routers.predictions import latest_predictions

router = APIRouter()

def format_geojson(sensor: dict, predictions: dict) -> dict:
    if not sensor:
        return {}
    return {
        "type": "FeatureCollection",
        "features": [
            {
                "type": "Feature",
                "geometry": {
                    "type": "Point",
                    "coordinates": [
                        sensor.get("longitude"),
                        sensor.get("latitude")
                    ]
                },
                "properties": {
                    "timestamp": sensor.get("timestamp"),
                    "ambient_temp_c": sensor.get("ambient_temp_c"),
                    "humidity_pct": sensor.get("humidity_pct"),
                    "soil_moisture_pct": sensor.get("soil_moisture_pct"),
                    "soil_ph": sensor.get("soil_ph"),
                    "battery_pct": sensor.get("battery_pct"),
                    "kmeans_cluster": predictions.get("kmeans", {}).get("cluster"),
                    "is_outlier": predictions.get("isolation_forest", {}).get("is_outlier"),
                    "rf_prediction": predictions.get("random_forest", {}).get("prediction"),
                }
            }
        ]
    }

@router.get("/esri/geojson")
def get_geojson():
    if not latest_sensor_data:
        raise HTTPException(status_code=404, detail="No data available yet")
    return format_geojson(latest_sensor_data, latest_predictions)

import sys
sys.path.append("/app")

from fastapi import FastAPI
from contextlib import asynccontextmanager
from messaging.mqtt_client import start_mqtt, stop_mqtt
from api.routers import sensors, predictions
from api import esri

@asynccontextmanager
async def lifespan(app: FastAPI):
    start_mqtt()
    yield
    stop_mqtt()

app = FastAPI(title="Solara Edge Server", lifespan=lifespan)

app.include_router(sensors.router)
app.include_router(predictions.router)
app.include_router(esri.router)

@app.get("/")
def root():
    return {"service": "solara-edge-server", "status": "running"}

@app.get("/health")
def health():
    return {"status": "ok"}

import time, yaml, json, rclpy
from sensors import environment, light, thermal
import paho.mqtt.client as mqtt
from sensor_msgs.msg import NavSatFix, Imu
from rclpy.node import Node
from datetime import datetime

# LOAD CONFIG
with open("config.yaml", "r") as f:
    config = yaml.safe_load(f)
broker   = config["broker"]
topics   = config["topics"]
INTERVAL = config["collector"]["poll_interval"]
QoS = 1

# MQTT CALLBACKS
def on_connect(client, userdata, flags, reason_code, properties):
    print(f"[MQTT] CONNACK rc={reason_code}")

def on_publish(client, userdata, mid):
    print(f"[MQTT] PUBACK received for PacketId={mid}")

def on_disconnect(client, userdata, rc):
    if rc != 0:
        print(f"[MQTT] Unexpected disconnect (rc={rc}). Will auto-reconnect.")

# COLLECTOR NODE
class CollectorNode(Node):
    def __init__(self, mqtt_client):
        super().__init__('collector')
        self.mqtt_client = mqtt_client

        # Cache for latest GPS fix - None until first message arrives
        self.latest_gps = None

        # Cache for latest IMU data - None until first message arrives
        # self.latest_imu = None

        # Subscribe to GPS topic published by gps_driver.py
        self.gps_sub = self.create_subscription(
            NavSatFix,
            "/gps/fix",
            self.gps_callback,
            10
        )

        # Timer replaces time.sleep() - fires collect_and_publish() every INTERVAL seconds
        self.timer = self.create_timer(INTERVAL, self.collect_and_publish)

        self.get_logger().info(f"Collector started, publishing every {INTERVAL}s")

    def gps_callback(self, msg: NavSatFix):
        self.latest_gps = {
            "latitude":  msg.latitude,
            "longitude": msg.longitude,
            "altitude":  msg.altitude,
        }

    def collect_and_publish(self):
        timestamp = datetime.now(pytz.timezone('America/Los_Angeles')).strftime("%Y-%m-%dT%H:%M:%S PST")
        env_data     = environment.get_data()
        light_data   = light.get_data()
        thermal_data = thermal.get_data()

        gps_data = self.latest_gps if self.latest_gps is not None else {
            "latitude":  None,
            "longitude": None,
            "altitude":  None,
        }

        # ML topic: environment + light
        self.mqtt_client.publish(
            topics["ML"],
            json.dumps({
                "timestamp":   timestamp,
                "environment": env_data,
                "light":       light_data,
            }),
            qos=QoS
        )

        # ESRI topic: thermal + gps + imu
        self.mqtt_client.publish(
            topics["ESRI"],
            json.dumps({
                "timestamp": timestamp,
                "thermal":   thermal_data,
                "gps":       gps_data,
            }),
            qos=QoS
        )

        self.get_logger().info(f"Published at {timestamp}. GPS: {gps_data}")


# MAIN
def main():
    client = mqtt.Client(
        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
        client_id=broker["client_id"],
        clean_session=broker["clean_session"]
    )
    client.on_connect    = on_connect
    client.on_publish    = on_publish
    client.on_disconnect = on_disconnect
    
    client.will_set(topics["status"], json.dumps({"status": "offline"}), qos=QoS, retain=True)
    client.connect(host=broker["host"], port=broker["port"], keepalive=broker["keepalive"])
    client.loop_start()
    client.publish(topics["status"], json.dumps({"status": "online"}), qos=QoS, retain=True)
    
    rclpy.init()
    node = CollectorNode(mqtt_client=client)
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        print("[COLLECTOR] Shutting down.")
    finally:
        node.destroy_node()
        rclpy.shutdown()
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    main()
            

import paho.mqtt.client as mqtt
import json, yaml, os
from datetime import datetime

with open("config.yaml", "r") as f:
    config = yaml.safe_load(f)

broker   = config["broker"]
topics   = config["topics"]
QoS      = config["subscriber"]["qos"]
DATA_DIR = config["subscriber"]["data_dir"]
os.makedirs(DATA_DIR, exist_ok=True)

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"[SUBSCRIBER] Connected rc={reason_code}")
    client.subscribe(topics["ML"],     qos=QoS)
    client.subscribe(topics["ESRI"],   qos=QoS)
    client.subscribe(topics["status"], qos=QoS)

    def on_message(client, userdata, msg):
        try:
            payload = json.loads(msg.payload.decode())
        except json.JSONDecodeError as e:
            print(f"[SUBSCRIBER] Bad JSON on {msg.topic}: {e}")
            return
        timestamp  = datetime.utcnow().strftime("%Y-%m-%dT%H-%M-%SZ")
        topic_slug = msg.topic.replace("/", "_")
        filepath   = os.path.join(DATA_DIR, f"{topic_slug}_{timestamp}.json")
        with open(filepath, "w") as f:
            json.dump(payload, f, indent=2)
        print(f"[SUBSCRIBER] Saved --> {filepath}")

    client = mqtt.Client(
        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
        client_id=broker["client_id"],
        clean_session=broker["clean_session"]
    )
    client.on_connect = on_connect
    client.on_message = on_message

    client.connect(broker["host"], broker["port"], broker["keepalive"])
    client.loop_forever()

-- 1. Create the Raw Sensor Data table
          CREATE TABLE raw_sensor_data (
              id INT PRIMARY KEY IDENTITY(1,1),
              sensor_id VARCHAR(50) NOT NULL,
              json_payload NVARCHAR(MAX) NOT NULL,
              timestamp DATETIME2 DEFAULT SYSUTCDATETIME()
          );
          
          -- 2. Create the Predictions table
          CREATE TABLE predictions (
              prediction_id INT PRIMARY KEY IDENTITY(1,1),
              raw_data_id INT NOT NULL,
              prediction_value FLOAT NOT NULL,
              model_version VARCHAR(20),
              created_at DATETIME2 DEFAULT SYSUTCDATETIME(),
          
              CONSTRAINT FK_RawData FOREIGN KEY (raw_data_id)
              REFERENCES raw_sensor_data(id) ON DELETE CASCADE
          );
          
          -- 3. Define Indexes
          CREATE INDEX IX_SensorData_Timestamp ON raw_sensor_data(timestamp);
          CREATE INDEX IX_Predictions_RawDataID ON predictions(raw_data_id);
          
          GO
          -- Data retention: cleanup of data older than 30 days
          CREATE PROCEDURE sp_CleanupOldData
          AS
          BEGIN
              DELETE FROM raw_sensor_data
              WHERE timestamp < DATEADD(day, -30, GETDATE());
          END;
          GO

-- 1. Simulate the Aggregator Node inserting data
          INSERT INTO raw_sensor_data (sensor_id, json_payload)
          VALUES ('ROVER_01', '{
              "ambient_temp_c": 23.69,
              "humidity_pct": 50.78,
              "soil_moisture_pct": 39.90,
              "battery_pct": 99.73,
              "location": {"lat": 33.97, "long": -117.32}
          }');
          
          -- 2. Simulate the ML Module creating a prediction
          INSERT INTO predictions (raw_data_id, prediction_value, model_version)
          VALUES (SCOPE_IDENTITY(), 0.95, 'v1.0-beta');
          
          -- 3. Join the tables to see the full result
          SELECT
              r.timestamp,
              r.sensor_id,
              r.json_payload,
              p.prediction_value
          FROM raw_sensor_data r
          JOIN predictions p ON r.id = p.raw_data_id;