"""Studio (What-If Analysis) resource for the CorePlexML SDK."""
from __future__ import annotations
from coreplexml._http import HTTPClient
[docs]
class StudioResource:
"""What-If Analysis -- scenario-based model exploration.
Create sessions tied to a deployed model, define alternative
scenarios by changing input features, and compare predictions
side-by-side.
"""
def __init__(self, http: HTTPClient):
self._http = http
@staticmethod
def _extract_prediction(payload: dict) -> object:
if not isinstance(payload, dict):
return None
if "prediction" in payload:
return payload.get("prediction")
output = payload.get("output")
if isinstance(output, dict):
return output.get("prediction")
return None
[docs]
def create_session(
self, project_id: str, deployment_id: str, baseline_input: dict
) -> dict:
"""Create a new What-If Analysis session.
Args:
project_id: UUID of the project.
deployment_id: UUID of the deployment to analyze.
baseline_input: Feature values for the baseline scenario.
Returns:
Created session dictionary.
"""
data = self._http.post(
"/api/studio/sessions",
json={
"project_id": project_id,
"deployment_id": deployment_id,
"baseline_input": baseline_input,
},
)
if isinstance(data, dict) and "id" not in data and data.get("session_id"):
data["id"] = data["session_id"]
return data
[docs]
def get_session(self, session_id: str) -> dict:
"""Get session details.
Args:
session_id: UUID of the studio session.
Returns:
Session dictionary with baseline and scenarios.
"""
return self._http.get(f"/api/studio/sessions/{session_id}")
[docs]
def list_sessions(self, project_id: str | None = None) -> dict:
"""List What-If sessions, optionally filtered by project."""
params = {"project_id": project_id} if project_id else None
return self._http.get("/api/studio/sessions", params=params)
[docs]
def schema(self, deployment_id: str) -> dict:
"""Get the feature schema for a deployment.
Args:
deployment_id: UUID of the deployment to inspect.
Returns:
Schema dictionary with feature names, types, and ranges.
"""
return self._http.get(f"/api/studio/deployments/{deployment_id}/schema")
[docs]
def create_scenario(
self, session_id: str, name: str, changes: dict | None = None
) -> dict:
"""Add a new scenario to a session.
Args:
session_id: UUID of the studio session.
name: Scenario name.
changes: Feature overrides compared to baseline (optional).
Returns:
Created scenario dictionary.
"""
body: dict = {"name": name}
if changes:
body["changes"] = changes
data = self._http.post(
f"/api/studio/sessions/{session_id}/scenarios", json=body
)
if isinstance(data, dict) and "id" not in data and data.get("scenario_id"):
data["id"] = data["scenario_id"]
return data
[docs]
def run_scenario(self, scenario_id: str) -> dict:
"""Execute a scenario to get predictions.
Args:
scenario_id: UUID of the scenario.
Returns:
Scenario results with predictions.
"""
return self._http.post(f"/api/studio/scenarios/{scenario_id}/run")
[docs]
def compare(self, session_id: str) -> dict:
"""Compare all scenarios in a session.
Args:
session_id: UUID of the studio session.
Returns:
Comparison results across all scenarios.
"""
data = self._http.get(f"/api/studio/sessions/{session_id}/compare")
if not isinstance(data, dict):
return {}
baseline = data.get("baseline")
if isinstance(baseline, dict):
baseline["prediction"] = self._extract_prediction(baseline)
scenarios = data.get("scenarios")
if isinstance(scenarios, list):
for scenario in scenarios:
if isinstance(scenario, dict):
scenario["prediction"] = self._extract_prediction(scenario)
return data