restrucuted whole model

Dockerfile

@@ -2,11 +2,10 @@ FROM python:3.10
 
 WORKDIR /app
 
-COPY requirements.txt .
-RUN pip install --no-cache-dir -r requirements.txt
+COPY . /app
 
-COPY app ./app
+RUN pip install --no-cache-dir -r requirements.txt
 
 EXPOSE 7860
 
-CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]
+CMD ["uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "7860"]

api/main.py

@@ -0,0 +1,62 @@
+from fastapi import FastAPI, UploadFile, File
+import shutil
+import os
+
+from predictor import predict_logs
+from shap_explainer import explain_logs
+
+app = FastAPI(
+    title="RTL Failure Prediction API",
+    description="Predict RTL module failure risk from verification logs",
+    version="1.0"
+)
+
+
+@app.get("/")
+def health():
+    return {"status": "running"}
+
+
+@app.post("/predict_file")
+async def predict_file(file: UploadFile = File(...)):
+
+    path = f"temp_{file.filename}"
+
+    with open(path, "wb") as buffer:
+        shutil.copyfileobj(file.file, buffer)
+
+    result = predict_logs(path)
+
+    os.remove(path)
+
+    return result
+
+
+@app.post("/predict_single")
+def predict_single(log_line: str):
+
+    path = "temp_single.txt"
+
+    with open(path, "w") as f:
+        f.write(log_line)
+
+    result = predict_logs(path)
+
+    os.remove(path)
+
+    return result
+
+
+@app.post("/explain")
+async def explain(file: UploadFile = File(...)):
+
+    path = f"temp_{file.filename}"
+
+    with open(path, "wb") as buffer:
+        shutil.copyfileobj(file.file, buffer)
+
+    result = explain_logs(path)
+
+    os.remove(path)
+
+    return result

api/predictor.py

@@ -0,0 +1,57 @@
+import pandas as pd
+import joblib
+
+from features.log_feature_extraction import run_pipeline
+
+
+MODEL_PATH = "models/failure_model.pkl"
+FEATURE_PATH = "models/feature_columns.pkl"
+
+
+def predict_logs(log_file):
+
+    run_pipeline(log_file, "temp_features.csv")
+
+    df = pd.read_csv("temp_features.csv")
+
+    model = joblib.load(MODEL_PATH)
+    feature_cols = joblib.load(FEATURE_PATH)
+
+    X = df[feature_cols]
+
+    probs = model.predict_proba(X)[:, 1]
+
+    df["failure_probability"] = probs
+
+    module_risk = (
+        df.groupby("module")["failure_probability"]
+        .mean()
+        .sort_values(ascending=False)
+    )
+
+    module_results = []
+
+    for module, prob in module_risk.items():
+
+        if prob > 0.75:
+            risk = "HIGH"
+        elif prob > 0.4:
+            risk = "MEDIUM"
+        else:
+            risk = "LOW"
+
+        module_results.append({
+            "module": module,
+            "failure_probability": float(prob),
+            "risk": risk
+        })
+
+    summary = {
+        "total_logs": int(len(df)),
+        "modules_analyzed": int(df["module"].nunique())
+    }
+
+    return {
+        "summary": summary,
+        "module_risk": module_results
+    }

api/shap_explainer.py

@@ -0,0 +1,40 @@
+import pandas as pd
+import joblib
+import shap
+
+from features.log_feature_extraction import run_pipeline
+
+
+MODEL_PATH = "models/failure_model.pkl"
+FEATURE_PATH = "models/feature_columns.pkl"
+
+
+def explain_logs(log_file):
+
+    run_pipeline(log_file, "temp_features.csv")
+
+    df = pd.read_csv("temp_features.csv")
+
+    model = joblib.load(MODEL_PATH)
+    feature_cols = joblib.load(FEATURE_PATH)
+
+    X = df[feature_cols]
+
+    explainer = shap.TreeExplainer(model.estimator)
+
+    shap_values = explainer.shap_values(X)
+
+    importance = abs(shap_values).mean(axis=0)
+
+    feature_importance = sorted(
+        zip(feature_cols, importance),
+        key=lambda x: x[1],
+        reverse=True
+    )[:10]
+
+    return {
+        "top_features": [
+            {"feature": f, "impact": float(v)}
+            for f, v in feature_importance
+        ]
+    }

app/main.py

@@ -1,28 +0,0 @@
-from fastapi import FastAPI, UploadFile, File
-from fastapi.middleware.cors import CORSMiddleware
-import pandas as pd
-
-app = FastAPI(title="RTL Reliability Engine")
-
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-@app.get("/")
-def root():
-    return {"status": "running"}
-
-
-@app.post("/analyze")
-async def analyze(file: UploadFile = File(...)):
-
-    df = pd.read_csv(file.file, sep="\t")
-
-    return {
-        "rows": len(df),
-        "columns": list(df.columns)
-    }

app/parser.py

@@ -1,32 +0,0 @@
-import re
-import pandas as pd
-
-LOG_PATTERN = re.compile(r"\[(\d+)\]\s+\[(\w+)\]\s+(.*)")
-
-def parse_log_file(file_content: str) -> pd.DataFrame:
-    """
-    Parse RTL simulation logs into structured dataframe
-    """
-
-    records = []
-
-    for line in file_content.splitlines():
-
-        match = LOG_PATTERN.match(line.strip())
-
-        if not match:
-            continue
-
-        sim_time = int(match.group(1))
-        severity = match.group(2).upper()
-        message = match.group(3)
-
-        records.append({
-            "time": sim_time,
-            "severity": severity,
-            "message": message
-        })
-
-    df = pd.DataFrame(records)
-
-    return df

app/reliability_engine.py

@@ -1,61 +0,0 @@
-import numpy as np
-import pandas as pd
-
-SEVERITY_WEIGHTS = {
-    "INFO": 1,
-    "WARNING": 2,
-    "ERROR": 4,
-    "CRITICAL": 8,
-    "PASS": 0,
-    "DRV": 0
-}
-
-
-def compute_metrics(df: pd.DataFrame):
-
-    if df.empty:
-        return {}
-
-    total_logs = len(df)
-
-    severity_counts = df["severity"].value_counts().to_dict()
-
-    info = severity_counts.get("INFO", 0)
-    warning = severity_counts.get("WARNING", 0)
-    error = severity_counts.get("ERROR", 0)
-    critical = severity_counts.get("CRITICAL", 0)
-
-    failures = error + critical
-
-    severity_score = (
-        info * SEVERITY_WEIGHTS["INFO"]
-        + warning * SEVERITY_WEIGHTS["WARNING"]
-        + error * SEVERITY_WEIGHTS["ERROR"]
-        + critical * SEVERITY_WEIGHTS["CRITICAL"]
-    )
-
-    failure_rate = failures / total_logs if total_logs > 0 else 0
-
-    critical_ratio = critical / failures if failures > 0 else 0
-
-    mtbf = None
-    failure_times = df[df["severity"].isin(["ERROR", "CRITICAL"])]["time"]
-
-    if len(failure_times) > 1:
-        mtbf = np.mean(np.diff(failure_times))
-
-    risk_score = (
-        0.5 * severity_score
-        + 0.3 * failure_rate * 100
-        + 0.2 * critical_ratio * 100
-    )
-
-    return {
-        "total_logs": total_logs,
-        "severity_counts": severity_counts,
-        "failure_rate": round(failure_rate, 4),
-        "critical_ratio": round(critical_ratio, 4),
-        "severity_score": severity_score,
-        "mtbf_cycles": mtbf,
-        "risk_score": round(risk_score, 3)
-    }

features/log_feature_extraction.py

@@ -0,0 +1,157 @@
+import pandas as pd
+import numpy as np
+import re
+import joblib
+from sklearn.feature_extraction.text import TfidfVectorizer
+
+
+WINDOW = 10
+
+
+def parse_log_file(log_file):
+
+    records = []
+
+    pattern = re.compile(r"(\d+)ns\s+\[(\w+)\]\s+(\w+)\s+(.*)")
+
+    with open(log_file) as f:
+
+        for line in f:
+
+            m = pattern.match(line.strip())
+
+            if m:
+
+                records.append({
+                    "time": int(m.group(1)),
+                    "severity": m.group(2),
+                    "module": m.group(3),
+                    "message": m.group(4)
+                })
+
+    return pd.DataFrame(records)
+
+
+def severity_flags(df):
+
+    df["error_flag"] = (df["severity"] == "ERROR").astype(int)
+    df["critical_flag"] = (df["severity"] == "CRITICAL").astype(int)
+    df["warning_flag"] = (df["severity"] == "WARNING").astype(int)
+
+    return df
+
+
+def temporal_features(df):
+
+    df = df.sort_values("time")
+
+    df["time_since_last_event"] = df["time"].diff().fillna(0)
+
+    last_error = df["time"].where(df["severity"] == "ERROR")
+    last_critical = df["time"].where(df["severity"] == "CRITICAL")
+
+    df["time_since_last_error"] = df["time"] - last_error.ffill()
+    df["time_since_last_critical"] = df["time"] - last_critical.ffill()
+
+    df["time_since_last_error"] = df["time_since_last_error"].fillna(0)
+    df["time_since_last_critical"] = df["time_since_last_critical"].fillna(0)
+
+    # transform to reduce dominance
+    df["log_time_since_last_error"] = np.log1p(df["time_since_last_error"])
+    df["log_time_since_last_critical"] = np.log1p(df["time_since_last_critical"])
+
+    return df
+
+
+def rolling_features(df):
+
+    df["error_count_last_10"] = df["error_flag"].rolling(WINDOW).sum().shift(1).fillna(0)
+
+    df["critical_count_last_10"] = df["critical_flag"].rolling(WINDOW).sum().shift(1).fillna(0)
+
+    df["warning_count_last_10"] = df["warning_flag"].rolling(WINDOW).sum().shift(1).fillna(0)
+
+    df["failure_rate_recent_window"] = (
+        df["error_count_last_10"] + df["critical_count_last_10"]
+    ) / WINDOW
+
+    # trend features
+    df["rolling_error_rate_20"] = df["error_flag"].rolling(20).mean().shift(1)
+
+    df["rolling_warning_rate_20"] = df["warning_flag"].rolling(20).mean().shift(1)
+
+    df["error_acceleration"] = df["error_flag"].diff().rolling(10).sum()
+
+    return df
+
+
+def module_features(df):
+
+    stats = df.groupby("module").agg(
+        total_logs=("severity", "count"),
+        error_logs=("error_flag", "sum"),
+        critical_logs=("critical_flag", "sum")
+    )
+
+    stats["historical_error_rate"] = stats["error_logs"] / stats["total_logs"]
+
+    stats["historical_critical_ratio"] = stats["critical_logs"] / stats["total_logs"]
+
+    stats["module_failure_density"] = (
+        stats["error_logs"] + stats["critical_logs"]
+    ) / stats["total_logs"]
+
+    df = df.merge(stats, on="module", how="left")
+
+    return df
+
+
+def text_features(df):
+
+    df["clean_message"] = df["message"].str.lower()
+
+    df["message_length"] = df["clean_message"].str.len()
+
+    keywords = ["timeout", "overflow", "stall", "violation"]
+
+    for k in keywords:
+        df[f"kw_{k}"] = df["clean_message"].str.contains(k).astype(int)
+
+    vectorizer = TfidfVectorizer(max_features=300)
+
+    X = vectorizer.fit_transform(df["clean_message"])
+
+    tfidf = pd.DataFrame(
+        X.toarray(),
+        columns=[f"tfidf_{i}" for i in range(X.shape[1])]
+    )
+
+    df = pd.concat([df.reset_index(drop=True), tfidf], axis=1)
+
+    joblib.dump(vectorizer, "tfidf_vectorizer.pkl")
+
+    return df
+
+
+def run_pipeline(input_file, output_file):
+
+    df = parse_log_file(input_file)
+
+    df = severity_flags(df)
+
+    df = temporal_features(df)
+
+    df = rolling_features(df)
+
+    df = module_features(df)
+
+    df = text_features(df)
+
+    df.to_csv(output_file, index=False)
+
+    print("Feature extraction complete")
+
+
+if __name__ == "__main__":
+
+    run_pipeline("C:/Codes/SanDisk/rtl_logs_with_severity.txt", "data/features.csv")

models/failure_model.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:6e7d881191a6708f1597c5f554fe87f5126032168f39c01527f56dc41ff21976
+size 7879632

models/feature_columns.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:1de2e899bd0973534279487769ebeff0422283eed3ede266db7c6ad4d50e4dfa
+size 1406

models/tfidf_vectorizer.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:c902fd5973c55d702cf0b9390d674236f8c86872d0d1c441f419235c403117fc
+size 1941

requirements.txt

@@ -2,4 +2,7 @@ fastapi
 uvicorn
 pandas
 numpy
-python-multipart
+scikit-learn
+lightgbm
+joblib
+shap