modularization, structure reorganization, code cleanup, yaml prompts, logging

agent/agent.py

@@ -1,5 +1,7 @@
 import json
 import re
+import os
+import yaml
 from langchain_google_genai import GoogleGenerativeAI
 
 class MaintenanceAgent:
@@ -9,27 +11,21 @@ class MaintenanceAgent:
             temperature=temperature,
             google_api_key=api_key
         )
+        self.prompts = self._load_prompts()
+
+    def _load_prompts(self) -> dict:
+        """Load prompts from YAML file."""
+        # Get the directory where this file is located
+        current_dir = os.path.dirname(os.path.abspath(__file__))
+        prompts_file = os.path.join(current_dir, 'prompts.yaml')
+        
+        with open(prompts_file, 'r') as f:
+            return yaml.safe_load(f)
 
     def _build_prompt(self, phase2_output: dict) -> str:
         """Build the maintenance analysis prompt."""
-        return f"""
-You are a maintenance decision AI.
-You must reason ONLY from the provided JSON.
-Do NOT invent data.
-
-INPUT:
-{json.dumps(phase2_output, indent=2)}
-
-MANDATORY: Return output strictly in JSON format only. Do not include any markdown, code blocks, or extra text.
-
-OUTPUT FORMAT:
-{{
-  "diagnosis": "...",
-  "urgency": "Low | Medium | High",
-  "recommended_action": "...",
-  "justification": ["...", "..."]
-}}
-"""
+        user_template = self.prompts['maintenance']['user_template']
+        return user_template.format(phase2_output=json.dumps(phase2_output, indent=2))
 
     def _parse_response(self, response: str) -> dict:
         """Parse LLM response, handling various JSON formats."""

agent/prompts.yaml

@@ -0,0 +1,19 @@
+maintenance:
+  system: |
+    You are a maintenance decision AI.
+    You must reason ONLY from the provided JSON.
+    Do NOT invent data.
+  
+  user_template: |
+    INPUT:
+    {phase2_output}
+    
+    MANDATORY: Return output strictly in JSON format only. Do not include any markdown, code blocks, or extra text.
+    
+    OUTPUT FORMAT:
+    {{
+      "diagnosis": "...",
+      "urgency": "Low | Medium | High",
+      "recommended_action": "...",
+      "justification": ["...", "..."]
+    }}

app.py

@@ -1,89 +1,21 @@
 from fastapi import FastAPI, HTTPException
-from pydantic import BaseModel
-import pandas as pd
-import os
 import logging
-from dotenv import load_dotenv
-from ml.inference import MLEngine
-from agent.agent import MaintenanceAgent
-
-load_dotenv()
+from src.config import Config
+from src.models import SensorData, AnalysisResponse
+from src.services import AnalysisService
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-app = FastAPI(title="Solar PV Predictive Maintenance API", version="1.0.0")
-
-# Load ML models once on startup for production performance
-ml_engine = MLEngine()
-
-# ============ Helper Functions ============
-
-def validate_sensor_data(vdc1: list, idc1: list) -> None:
-    """Validate sensor data consistency. Raises HTTPException on error."""
-    if len(vdc1) != len(idc1):
-        raise HTTPException(status_code=400, detail="Voltage and current lists must have the same length")
-    if len(vdc1) < 3:
-        raise HTTPException(status_code=400, detail="Need at least 3 data points")
-
-def prepare_dataframe(vdc1: list, idc1: list) -> pd.DataFrame:
-    """Prepare sensor data for ML inference by padding to 100 points."""
-    return pd.DataFrame({
-        "vdc1": (vdc1 * (100 // len(vdc1) + 1))[:100],
-        "idc1": (idc1 * (100 // len(idc1) + 1))[:100]
-    })
-
-def get_agent_output(api_key: str, ml_output: dict) -> dict:
-    """Get agent analysis if API key is provided, otherwise return no-key message."""
-    if not api_key:
-        return {
-            "diagnosis": "No API key provided - LLM features disabled",
-            "urgency": "Unknown",
-            "recommended_action": "Provide Google API key in request for AI diagnosis",
-            "justification": ["Google API key required for maintenance reasoning"]
-        }
-    
-    try:
-        agent = MaintenanceAgent(
-            api_key=api_key,
-            model_name="gemini-2.5-flash-lite",
-            temperature=0.0
-        )
-        return agent.run(ml_output)
-    except Exception as e:
-        logger.warning(f"Agent initialization failed: {e}")
-        return {
-            "diagnosis": "Agent initialization failed",
-            "urgency": "Unknown",
-            "recommended_action": "Check your Google API key",
-            "justification": [f"Error: {str(e)}"]
-        }
-
-class SensorData(BaseModel):
-    vdc1: list[float]
-    idc1: list[float]
-    api_key: str = None  # Optional Google API key for LLM features
-
-class AnalysisResponse(BaseModel):
-    ml_output: dict
-    agent_output: dict
+config = Config()
+app = FastAPI(title=config.APP_TITLE, version=config.APP_VERSION)
+service = AnalysisService(config)
 
 @app.post("/analyze", response_model=AnalysisResponse)
 async def analyze_sensor_data(data: SensorData):
     try:
         logger.info(f"Processing request with {len(data.vdc1)} voltage and {len(data.idc1)} current data points")
-        
-        # Validate input
-        validate_sensor_data(data.vdc1, data.idc1)
-        
-        # Prepare data and run ML inference
-        raw_df = prepare_dataframe(data.vdc1, data.idc1)
-        ml_output = ml_engine.predict_from_raw(raw_df)
-        
-        # Get agent analysis
-        agent_output = get_agent_output(data.api_key, ml_output)
-        
+        ml_output, agent_output = service.analyze(data.vdc1, data.idc1, data.api_key)
         return AnalysisResponse(ml_output=ml_output, agent_output=agent_output)
-    
     except HTTPException:
         raise
     except Exception as e:
@@ -92,8 +24,8 @@ async def analyze_sensor_data(data: SensorData):
 
 @app.get("/")
 async def root():
-    return {"message": "Solar PV Predictive Maintenance API", "endpoint": "/analyze (POST)"}
+    return {"message": "Solar PV Predictive Maintenance", "endpoint": "/analyze (POST)"}
 
 if __name__ == "__main__":
     import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=7860)
+    uvicorn.run(app, host=config.HOST, port=config.PORT)

data/phase2_output.json

@@ -1,7 +0,0 @@
-{
-  "asset_id": "PV_INVERTER_001",
-  "failure_probability": 0.0,
-  "expected_ttf_days": 10338.5,
-  "expected_rul_days": 10942.0,
-  "confidence": 1.0
-}

main.py

@@ -1,9 +1,9 @@
-import pandas as pd
 import os
+import pandas as pd
+import numpy as np
 from dotenv import load_dotenv
 from ml.inference import MLEngine
 from agent.agent import MaintenanceAgent
-import numpy as np
 
 load_dotenv()
 
@@ -12,14 +12,12 @@ raw_df = pd.DataFrame({
     "idc1": np.random.normal(10.0, 0.2, 200)
 })
 
-
 engine = MLEngine()
 phase2_output = engine.predict_from_raw(raw_df)
 
 print("\n=== ML OUTPUT ===")
 print(phase2_output)
 
-# ---- LLM AGENT ----
 agent = MaintenanceAgent(
     api_key=os.getenv("GOOGLE_API_KEY"),
     model_name="gemini-2.5-flash-lite",

ml/features.py

@@ -3,25 +3,15 @@ import pandas as pd
 
 def build_features(df, window):
     df = df.copy()
-
     df["pdc1"] = df["vdc1"] * df["idc1"]
-
     df["vdc_mean"] = df["vdc1"].rolling(window).mean()
-    df["vdc_std"]  = df["vdc1"].rolling(window).std()
-
+    df["vdc_std"] = df["vdc1"].rolling(window).std()
     df["pdc_mean"] = df["pdc1"].rolling(window).mean()
-    df["pdc_std"]  = df["pdc1"].rolling(window).std()
-
+    df["pdc_std"] = df["pdc1"].rolling(window).std()
     df["pdc_delta"] = df["pdc1"].diff()
-
     df["pdc_slope"] = df["pdc1"].rolling(window).apply(
-        lambda x: np.polyfit(range(len(x)), x, 1)[0],
-        raw=False
+        lambda x: np.polyfit(range(len(x)), x, 1)[0], raw=False
     )
-
     df["efficiency"] = df["pdc1"] / (df["vdc1"] * df["idc1"] + 1e-6)
-    df["efficiency_norm"] = (
-        df["efficiency"] / df["efficiency"].rolling(window).mean()
-    )
-
+    df["efficiency_norm"] = df["efficiency"] / df["efficiency"].rolling(window).mean()
     return df

ml/inference.py

@@ -4,41 +4,43 @@ import joblib
 import torch
 import numpy as np
 import pandas as pd
+import logging
 from sklearn.preprocessing import StandardScaler
 from sklearn.ensemble import IsolationForest
 from safetensors.torch import load_file
 
 from ml.features import build_features
 from ml.lstm_model import LSTMAutoencoder
+from src.config import MLConfig
+
+logger = logging.getLogger(__name__)
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 ARTIFACTS_DIR = os.path.join(BASE_DIR, "artifacts")
 
+
 class MLEngine:
     def __init__(self):
-        # Load configuration
-        self._load_config()
-        # Load all models
+        logger.info("Initializing ML Engine...")
+        self._load_ml_config()
         self._load_scaler()
         self._load_isolation_forest()
         self._load_xgboost_models()
         self._load_lstm_model()
+        logger.info("ML Engine initialized successfully")
 
-    def _load_config(self):
-        """Load ML configuration from JSON."""
-        with open(os.path.join(ARTIFACTS_DIR, "ml_config.json")) as f:
-            self.config = json.load(f)
-
-        self.feature_cols = self.config["feature_cols"]
-        self.window = self.config["window"]
-        self.seq_len = self.config["seq_len"]
-        self.design_life_days = self.config["design_life_days"]
+    def _load_ml_config(self):
+        """Load ML configuration from config."""
+        config = MLConfig.load()
+        self.feature_cols = config["feature_cols"]
+        self.window = config["window"]
+        self.seq_len = config["seq_len"]
+        self.design_life_days = config["design_life_days"]
 
     def _load_scaler(self):
         """Load and reconstruct StandardScaler from JSON."""
         with open(os.path.join(ARTIFACTS_DIR, "scaler.json"), "r") as f:
             params = json.load(f)
-        
         self.scaler = StandardScaler()
         self.scaler.mean_ = np.array(params["mean"])
         self.scaler.scale_ = np.array(params["scale"])
@@ -58,10 +60,8 @@ class MLEngine:
     def _load_xgboost_models(self):
         """Load XGBoost models from JSON artifacts."""
         import xgboost as xgb
-        
         self.ttf_model = xgb.XGBRegressor()
         self.ttf_model.load_model(os.path.join(ARTIFACTS_DIR, "xgb_ttf.json"))
-        
         self.fail_model = xgb.XGBClassifier()
         self.fail_model.load_model(os.path.join(ARTIFACTS_DIR, "xgb_fail.json"))
 
@@ -77,57 +77,33 @@ class MLEngine:
 
     def _compute_anomalies(self, df_scaled: pd.DataFrame) -> tuple:
         """Compute anomaly scores from LSTM and IsolationForest.
-        
-        Returns:
-            (anomaly_lstm, health) tuple
+        Returns: (anomaly_lstm, health) tuple
         """
-        # --- Isolation Forest anomaly ---
         df_scaled["anomaly_iforest"] = -self.iso.decision_function(df_scaled)
-
-        # --- LSTM anomaly ---
         X = df_scaled[self.feature_cols].values
         X_seq = np.array([X[-self.seq_len:]])
-
         with torch.no_grad():
             recon = self.lstm(torch.tensor(X_seq, dtype=torch.float32))
-
         anomaly_lstm = float(((recon - torch.tensor(X_seq)) ** 2).mean())
-
-        # --- Health (0–1) ---
-        # Normalize anomaly_lstm (assuming max error ~1e6 from training)
         anomaly_norm = min(anomaly_lstm / 1e6, 1.0)
         health = max(0.0, 1.0 - anomaly_norm)
-
         return anomaly_lstm, health
 
     def _make_predictions(self, df_scaled: pd.DataFrame, anomaly_lstm: float, health: float) -> dict:
         """Make TTF and failure probability predictions.
-        
-        Returns:
-            Dictionary with ttf, failure_prob, and rul predictions
+        Returns: Dictionary with ttf, failure_prob, and rul predictions
         """
         latest_features = df_scaled[self.feature_cols].iloc[[-1]].copy()
         latest_features["anomaly_lstm"] = anomaly_lstm
         latest_features["health_index"] = health
-
         expected_ttf_days = float(
             self.ttf_model.predict(latest_features, validate_features=False)[0]
         )
-
         failure_probability = float(
             self.fail_model.predict_proba(latest_features, validate_features=False)[0][1]
         )
-
-        # --- RUL ---
         expected_rul_days = float(health * self.design_life_days)
-
-        # --- Confidence ---
-        confidence = round(
-            0.5 * abs(failure_probability - 0.5) * 2
-            + 0.5 * health,
-            2
-        )
-
+        confidence = round(0.5 * abs(failure_probability - 0.5) * 2 + 0.5 * health, 2)
         return {
             "ttf_days": expected_ttf_days,
             "failure_prob": failure_probability,
@@ -136,30 +112,21 @@ class MLEngine:
         }
 
     def predict_from_raw(self, raw_df: pd.DataFrame):
-        # --- Feature engineering ---
+        logger.info("ML analysis start")
         df = build_features(raw_df, self.window)
         df = df[self.feature_cols].dropna()
-
         if len(df) < self.seq_len:
             raise ValueError("Not enough data for LSTM sequence")
-
-        # --- Scaling ---
         df_scaled = pd.DataFrame(
-            self.scaler.transform(df),
-            columns=self.feature_cols,
-            index=df.index
+            self.scaler.transform(df), columns=self.feature_cols, index=df.index
         )
-
-        # --- Compute anomalies ---
         anomaly_lstm, health = self._compute_anomalies(df_scaled)
-
-        # --- Make predictions ---
         predictions = self._make_predictions(df_scaled, anomaly_lstm, health)
-
+        logger.info("ML analysis end")
         return {
             "asset_id": "PV_INVERTER_001",
             "failure_probability": round(predictions["failure_prob"], 2),
             "expected_ttf_days": round(predictions["ttf_days"], 1),
             "expected_rul_days": round(predictions["rul_days"], 1),
             "confidence": predictions["confidence"]
-        }
+        }

requirements.txt

@@ -8,4 +8,5 @@ scikit-learn
 xgboost
 fastapi
 uvicorn
-safetensors
+safetensors
+pyyaml

src/config.py

@@ -0,0 +1,36 @@
+import json
+import os
+from typing import Dict, Any
+
+
+class AppConfig:
+    """Application and API configuration."""
+
+    # LLM Configuration
+    MODEL_NAME: str = "gemini-2.5-flash-lite"
+    TEMPERATURE: float = 0.0
+
+    # Server Configuration
+    HOST: str = "0.0.0.0"
+    PORT: int = 7860
+
+    # Application Metadata
+    APP_TITLE: str = "Solar PV Predictive Maintenance API"
+    APP_VERSION: str = "1.0.0"
+
+
+class MLConfig:
+    """ML model configuration from ml/artifacts/ml_config.json."""
+
+    @staticmethod
+    def load() -> Dict[str, Any]:
+        """Load and return ML configuration."""
+        config_path = os.path.join(
+            os.path.dirname(__file__), "..", "ml", "artifacts", "ml_config.json"
+        )
+        with open(config_path) as f:
+            return json.load(f)
+
+
+# For backwards compatibility
+Config = AppConfig

src/models.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel
+
+class SensorData(BaseModel):
+    vdc1: list[float]
+    idc1: list[float]
+    api_key: str = None  # Optional Google API key for LLM features
+
+class AnalysisResponse(BaseModel):
+    ml_output: dict
+    agent_output: dict

src/services.py

@@ -0,0 +1,54 @@
+import logging
+from ml.inference import MLEngine
+from agent.agent import MaintenanceAgent
+from src.utils import validate_sensor_data, prepare_dataframe
+
+logger = logging.getLogger(__name__)
+
+class AnalysisService:
+    """Service class for handling sensor data analysis logic."""
+    
+    def __init__(self, config):
+        self.config = config
+        self.ml_engine = MLEngine()
+
+    def analyze(self, vdc1: list, idc1: list, api_key: str) -> tuple:
+        """Analyze sensor data and return ML and agent outputs."""
+        logger.info(f"Complete analysis start - processing {len(vdc1)} data points")
+        validate_sensor_data(vdc1, idc1)
+        raw_df = prepare_dataframe(vdc1, idc1)
+        ml_output = self.ml_engine.predict_from_raw(raw_df)
+        agent_output = self.get_agent_output(api_key, ml_output)
+        
+        logger.info("Complete analysis end")
+        return ml_output, agent_output
+    
+    def get_agent_output(self, api_key: str, ml_output: dict) -> dict:
+        """Get agent analysis if API key is provided, otherwise return no-key message."""
+        if not api_key:
+            logger.info("No API key provided - skipping agent analysis")
+            return {
+                "diagnosis": "No API key provided - LLM features disabled",
+                "urgency": "Unknown",
+                "recommended_action": "Provide Google API key in request for AI diagnosis",
+                "justification": ["Google API key required for maintenance reasoning"]
+            }
+        
+        try:
+            logger.info("Agent analysis start")
+            agent = MaintenanceAgent(
+                api_key=api_key,
+                model_name=self.config.MODEL_NAME,
+                temperature=self.config.TEMPERATURE
+            )
+            result = agent.run(ml_output)
+            logger.info("Agent analysis end")
+            return result
+        except Exception as e:
+            logger.warning(f"Agent initialization failed: {e}")
+            return {
+                "diagnosis": "Agent initialization failed",
+                "urgency": "Unknown",
+                "recommended_action": "Check your Google API key",
+                "justification": [f"Error: {str(e)}"]
+            }

src/utils.py

@@ -0,0 +1,16 @@
+from fastapi import HTTPException
+import pandas as pd
+
+def validate_sensor_data(vdc1: list, idc1: list) -> None:
+    """Validate sensor data consistency. Raises HTTPException on error."""
+    if len(vdc1) != len(idc1):
+        raise HTTPException(status_code=400, detail="Voltage and current lists must have the same length")
+    if len(vdc1) < 3:
+        raise HTTPException(status_code=400, detail="Need at least 3 data points")
+
+def prepare_dataframe(vdc1: list, idc1: list) -> pd.DataFrame:
+    """Prepare sensor data for ML inference by padding to 100 points."""
+    return pd.DataFrame({
+        "vdc1": (vdc1 * (100 // len(vdc1) + 1))[:100],
+        "idc1": (idc1 * (100 // len(idc1) + 1))[:100]
+    })