"""Models resource for the CorePlexML SDK."""
from __future__ import annotations
from coreplexml._http import HTTPClient
[docs]
class ModelsResource:
"""Access trained models and make predictions.
Models are produced by AutoML experiments. Each experiment generates
one or more models ranked by performance.
"""
def __init__(self, http: HTTPClient):
self._http = http
[docs]
def list(
self,
project_id: str | None = None,
experiment_id: str | None = None,
limit: int = 50,
offset: int = 0,
) -> dict:
"""List models, optionally filtered by project or experiment.
Args:
project_id: Filter by project UUID (optional).
experiment_id: Filter by experiment UUID (optional).
limit: Maximum results (default 50).
offset: Pagination offset.
Returns:
Dictionary with ``items`` list and ``total`` count.
"""
params: dict = {"limit": limit, "offset": offset}
if project_id:
params["project_id"] = project_id
if experiment_id:
params["experiment_id"] = experiment_id
data = self._http.get("/api/models", params=params)
items = data.get("items", []) if isinstance(data, dict) else []
for item in items:
if not isinstance(item, dict):
continue
# Normalize common aliases used in SDK docs examples.
if "algorithm" not in item:
item["algorithm"] = item.get("algo") or item.get("model_type")
return data
[docs]
def get(self, model_id: str) -> dict:
"""Get model details.
Args:
model_id: UUID of the model.
Returns:
Model dictionary with metrics, parameters, etc.
"""
return self._http.get(f"/api/models/{model_id}")
[docs]
def predict(
self, model_id: str, inputs: dict | list, options: dict | None = None
) -> dict:
"""Make predictions with a model.
Args:
model_id: UUID of the model.
inputs: Feature values -- a dict for single prediction or list of dicts for batch.
options: Optional prediction options.
Returns:
Prediction results dictionary.
"""
body = {"inputs": inputs, "options": options or {}}
data = self._http.post(f"/api/models/{model_id}/predict", json=body)
if isinstance(inputs, dict) and isinstance(data, dict):
preds = data.get("predictions")
if isinstance(preds, list) and preds:
first = preds[0] if isinstance(preds[0], dict) else {}
if "prediction" in first and "prediction" not in data:
data["prediction"] = first.get("prediction")
return data
[docs]
def explain(self, model_id: str) -> dict:
"""Get model explainability data.
Args:
model_id: UUID of the model.
Returns:
Feature importance and SHAP values.
"""
return self._http.get(f"/api/models/{model_id}/explain")
[docs]
def parameters(self, model_id: str) -> dict:
"""Get model hyperparameters.
Args:
model_id: UUID of the model.
Returns:
Dictionary of model parameters.
"""
return self._http.get(f"/api/models/{model_id}/parameters")
[docs]
def set_threshold(self, model_id: str, threshold: float) -> dict:
"""Persist a binary-classification decision threshold."""
return self._http.post(
f"/api/models/{model_id}/threshold",
json={"threshold": threshold},
)
[docs]
def batch_predict(self, model_id: str, file_path: str) -> dict:
"""Upload a CSV or Parquet file for synchronous batch prediction."""
return self._http.upload(f"/api/models/{model_id}/predict/batch", file_path)
[docs]
def delete(self, model_id: str) -> dict:
"""Delete a model.
Args:
model_id: UUID of the model.
Returns:
Empty dictionary on success.
"""
return self._http.delete(f"/api/models/{model_id}")