analysis_runtime.py

# analysis_runtime.py
from __future__ import annotations
from typing import Dict, Any, List, Optional
import pandas as pd

class ExecContext:
    def __init__(self, frames: Dict[str,pd.DataFrame]):
        self.frames = frames  # name -> DataFrame

def _find_frame_with_columns(ctx: ExecContext, cols: List[str]) -> Optional[pd.DataFrame]:
    for df in ctx.frames.values():
        if all(c in df.columns for c in cols):
            return df
    return None

def op_summary_table(ctx: ExecContext, metrics: List[str], by: List[str], aggs: List[str]=["median"]) -> Optional[pd.DataFrame]:
    df = _find_frame_with_columns(ctx, metrics + by)
    if df is None: return None
    agg_spec = {m: aggs for m in metrics}
    out = df.groupby(by, dropna=False).agg(agg_spec)
    out.columns = [f"{m}_{a}" for m,a in out.columns]
    return out.reset_index()

def op_rank_top_n(ctx: ExecContext, metric: str, by: str, n: int=5, agg: str="median", ascending=False) -> Optional[pd.DataFrame]:
    df = _find_frame_with_columns(ctx, [metric, by])
    if df is None: return None
    g = df.groupby(by, dropna=False)[metric].agg(agg).sort_values(ascending=ascending).head(n)
    return g.reset_index().rename(columns={metric:f"{metric}_{agg}"})

def op_delta_over_time(ctx: ExecContext, metric: str, by: List[str], time_col: str, t0: str, t1: str, agg: str="median") -> Optional[pd.DataFrame]:
    df = _find_frame_with_columns(ctx, [metric, time_col] + by)
    if df is None: return None
    a = df[df[time_col]==t0].groupby(by, dropna=False)[metric].agg(agg)
    b = df[df[time_col]==t1].groupby(by, dropna=False)[metric].agg(agg)
    out = (b - a).reset_index()
    out = out.rename(columns={0:f"{metric}_delta_{agg}_{t0}_to_{t1}"})
    return out

def op_capacity_calc(clients_per_day: float, teams: int, days: int) -> pd.DataFrame:
    total = float(clients_per_day) * float(teams) * float(days)
    return pd.DataFrame([{"clients_per_day": clients_per_day, "teams": teams, "days": days, "capacity_total": total}])

def op_cost_total(startup_per_client: float, ongoing_per_client: float, volume: int) -> pd.DataFrame:
    total = float(startup_per_client + ongoing_per_client) * float(volume)
    return pd.DataFrame([{
        "startup_per_client": startup_per_client, "ongoing_per_client": ongoing_per_client,
        "volume": volume, "total_cost": total
    }])