|
|
| import numpy as np |
| import os |
| import json |
|
|
| def predict(submodel_outputs: dict, **kwargs) -> dict: |
| ''' |
| Standard Fusion Predict Interface |
| Args: |
| submodel_outputs: dict mapping model_name -> {"prob_fake": float} |
| ''' |
| |
| base_path = os.path.dirname(__file__) |
| with open(os.path.join(base_path, "config.json")) as f: |
| config = json.load(f) |
|
|
| order = config["submodel_order"] |
| probs = [] |
| for name in order: |
| if name not in submodel_outputs: |
| raise ValueError(f"Missing output for {name}") |
| probs.append(submodel_outputs[name]["prob_fake"]) |
|
|
| X = np.array([probs]) |
|
|
| |
| |
| if os.path.exists(os.path.join(base_path, "fusion_logreg.pkl")): |
| import joblib |
| model = joblib.load(os.path.join(base_path, "fusion_logreg.pkl")) |
| prob_fake = model.predict_proba(X)[0, 1] |
| elif os.path.exists(os.path.join(base_path, "fusion_model.pt")): |
| import torch |
| import torch.nn as nn |
| |
| |
| |
| input_dim = len(order) |
| net = nn.Sequential( |
| nn.Linear(input_dim, 32), |
| nn.ReLU(), |
| nn.Dropout(0.3), |
| nn.Linear(32, 1), |
| nn.Sigmoid() |
| ) |
| net.load_state_dict(torch.load(os.path.join(base_path, "fusion_model.pt"))) |
| net.eval() |
| with torch.no_grad(): |
| prob_fake = net(torch.tensor(X, dtype=torch.float32)).item() |
| else: |
| raise FileNotFoundError("No model file found") |
|
|
| return { |
| "pred": "fake" if prob_fake >= config.get("threshold", 0.5) else "real", |
| "pred_int": 1 if prob_fake >= config.get("threshold", 0.5) else 0, |
| "prob_fake": float(prob_fake), |
| "meta": {"min_prob": float(np.min(probs)), "max_prob": float(np.max(probs))} |
| } |
| |
