""" Operation Registry and Dispatcher for Advanced Data Explorer. All post-processing operations the execution engine can dispatch. The LLM QueryPlanner references this registry to build execution plans. Operations are whitelisted — no arbitrary code execution. """ import pandas as pd import numpy as np from typing import Any, Dict, List, Optional from dataclasses import dataclass, field from analysis.advanced_analysis import AdvancedAnalyzer # ── Result Dataclass ────────────────────────────────────────────── @dataclass class OperationOutput: """Result of a single operation step.""" op_name: str label: str data: Any # Primary result (dict, list, DataFrame) metadata: Dict = field(default_factory=dict) # Stats about the operation exportable_df: Optional[pd.DataFrame] = None # If this step produces downloadable data # ── Operation Registry ──────────────────────────────────────────── # Each entry describes one whitelisted operation for the LLM prompt. OPERATION_REGISTRY = { "find_peak_pressure_blocks": { "description": "Find contiguous time blocks where a pressure sensor exceeds a threshold", "params": { "pressure_tag": {"type": "str", "required": True, "description": "Sensor tag (e.g. PT130)"}, "threshold_bar": {"type": "float", "required": True, "description": "Pressure threshold in bar"}, }, "returns": "Dict with blocks list: start_time, end_time, duration, peak, avg for each block", }, "detect_constant_periods": { "description": "Find periods where a single sensor value is constant within tolerance for a minimum duration", "params": { "tag": {"type": "str", "required": True, "description": "Sensor tag name"}, "tolerance": {"type": "float", "required": False, "default": 0.01}, "min_duration_minutes": {"type": "float", "required": False, "default": 2.0}, }, "returns": "List of constant period dicts: start, end, duration_minutes, value", }, "detect_constant_periods_multi": { "description": "Find periods where multiple sensors are simultaneously constant (overlapping constant windows)", "params": { "tags": {"type": "List[str]", "required": True, "description": "Sensor tags that must all be constant"}, "tolerance": {"type": "float", "required": False, "default": 0.01}, "min_duration_minutes": {"type": "float", "required": False, "default": 2.0}, }, "returns": "List of overlapping constant period dicts", }, "downsample_frequency": { "description": "Reduce data frequency (e.g. 10Hz to 2Hz). Requires raw resolution data.", "params": { "target_hz": {"type": "float", "required": True, "description": "Target frequency in Hz"}, }, "returns": "Downsampled DataFrame", }, "detect_pressure_ramps": { "description": "Detect windows where pressure ramps from one level to another", "params": { "pressure_tag": {"type": "str", "required": True, "description": "Pressure sensor tag"}, "start_bar": {"type": "float", "required": True, "description": "Ramp start pressure in bar"}, "end_bar": {"type": "float", "required": True, "description": "Ramp end pressure in bar"}, "max_ramps": {"type": "int", "required": False, "default": None, "description": "Max ramps to return"}, }, "returns": "List of ramp dicts: start_time, end_time, start_pressure, peak_pressure, duration_minutes", }, "extract_windows": { "description": "Extract sub-DataFrames for each time window from a previous operation's result", "params": { "source": {"type": "str", "required": True, "description": "Must be 'previous_result'"}, }, "returns": "List of DataFrames, one per detected window", }, "filter_by_condition": { "description": "Filter data rows where a sensor meets a condition", "params": { "tag": {"type": "str", "required": True, "description": "Sensor tag"}, "operator": {"type": "str", "required": True, "description": "One of: gt, gte, lt, lte, eq"}, "value": {"type": "float", "required": True, "description": "Threshold value"}, }, "returns": "Filtered DataFrame", }, "compute_statistics": { "description": "Compute summary statistics (min, max, mean, std, count) for sensors", "params": { "tags": {"type": "List[str]", "required": False, "default": None, "description": "Tags to summarize (default: all)"}, }, "returns": "Statistics dict per sensor", }, "detect_and_analyze_fills": { "description": "Detect fill/test cycles using motor speed, then compute per-fill metrics: kWh/kg, total mass, pump strokes, peak pressure. Can filter by kWh/kg threshold or specific fill ID.", "params": { "kwh_per_kg_threshold": {"type": "float", "required": False, "default": None, "description": "Only return fills where kWh/kg exceeds this"}, "fill_id": {"type": "int", "required": False, "default": None, "description": "Analyze only this fill number"}, "include_strokes": {"type": "bool", "required": False, "default": True, "description": "Whether to compute stroke count"}, }, "returns": "List of fill dicts: fill_id, start_time, end_time, duration_min, peak_pressure, avg_flow, total_mass_kg, kwh_per_kg, total_strokes", }, "count_pump_strokes": { "description": "Count total pump strokes (motor revolutions) in the time window by integrating RPM over time", "params": { "speed_tag": {"type": "str", "required": False, "default": None, "description": "Motor speed tag (defaults to first available)"}, }, "returns": "Dict: total_revolutions, avg_rpm, active_minutes, peak_rpm, note", }, } def get_registry_for_prompt() -> str: """Serialize the operation registry as text for LLM prompt injection.""" lines = ["AVAILABLE OPERATIONS:"] for op_name, info in OPERATION_REGISTRY.items(): lines.append(f"\n {op_name}: {info['description']}") lines.append(" Parameters:") for pname, pinfo in info["params"].items(): req = "REQUIRED" if pinfo.get("required") else f"optional, default={pinfo.get('default')}" lines.append(f" {pname} ({pinfo['type']}, {req}): {pinfo.get('description', '')}") lines.append(f" Returns: {info['returns']}") return "\n".join(lines) # ── Operation Dispatcher ────────────────────────────────────────── OPERATOR_MAP = {"gt": ">", "gte": ">=", "lt": "<", "lte": "<=", "eq": "=="} MOTOR_SPEED_TAGS = ["M130_Speed", "MC130_VFD_Speed", "M130_RPM"] class OperationDispatcher: """Dispatch whitelisted operations against a DataFrame.""" def __init__(self): self.analyzer = AdvancedAnalyzer() def run( self, op_name: str, df: pd.DataFrame, params: Dict, prev_result: Any = None, db_connector=None, ) -> OperationOutput: """Dispatch a single operation by name.""" handler = getattr(self, f"_run_{op_name}", None) if handler is None: raise ValueError(f"Unknown operation: {op_name}") # Some operations need extra context if op_name in ("detect_and_analyze_fills",): return handler(df, params, db_connector=db_connector) if op_name in ("extract_windows",): return handler(df, params, prev_result=prev_result) return handler(df, params) # ── Handlers ────────────────────────────────────────────────── def _run_find_peak_pressure_blocks(self, df: pd.DataFrame, params: Dict) -> OperationOutput: pressure_tag = params["pressure_tag"] threshold = float(params["threshold_bar"]) result = self.analyzer.find_peak_pressure_blocks(df, pressure_tag, threshold) # Build exportable table from blocks export_df = None blocks = result.get("blocks", []) if blocks: export_df = pd.DataFrame(blocks) return OperationOutput( op_name="find_peak_pressure_blocks", label=f"{pressure_tag} > {threshold} bar", data=result, metadata={"num_blocks": result.get("num_blocks", 0), "total_points": result.get("total_points_above", 0)}, exportable_df=export_df, ) def _run_detect_constant_periods(self, df: pd.DataFrame, params: Dict) -> OperationOutput: tag = params["tag"] tolerance = float(params.get("tolerance", 0.01)) min_dur = float(params.get("min_duration_minutes", 2.0)) periods = self.analyzer.detect_constant_periods(df, tag, tolerance, min_dur) export_df = pd.DataFrame(periods) if periods else None return OperationOutput( op_name="detect_constant_periods", label=f"{tag} constant periods (±{tolerance}, ≥{min_dur} min)", data=periods, metadata={"num_periods": len(periods)}, exportable_df=export_df, ) def _run_detect_constant_periods_multi(self, df: pd.DataFrame, params: Dict) -> OperationOutput: tags = params["tags"] tolerance = float(params.get("tolerance", 0.01)) min_dur = float(params.get("min_duration_minutes", 2.0)) # Detect constant periods per tag per_tag_periods = {} for tag in tags: if tag in df.columns: per_tag_periods[tag] = self.analyzer.detect_constant_periods(df, tag, tolerance, min_dur) else: per_tag_periods[tag] = [] # Find overlapping intervals across all tags overlapping = self._intersect_periods(per_tag_periods, min_dur) export_df = pd.DataFrame(overlapping) if overlapping else None return OperationOutput( op_name="detect_constant_periods_multi", label=f"Simultaneous constant: {', '.join(tags)} (≥{min_dur} min)", data=overlapping, metadata={"num_overlaps": len(overlapping), "per_tag_counts": {t: len(p) for t, p in per_tag_periods.items()}}, exportable_df=export_df, ) def _intersect_periods(self, per_tag: Dict[str, List[Dict]], min_dur: float) -> List[Dict]: """Intersect constant periods across multiple tags — return only overlapping windows.""" tags = list(per_tag.keys()) if not tags or any(len(per_tag[t]) == 0 for t in tags): return [] # Start with first tag's periods current_intervals = [(p["start"], p["end"]) for p in per_tag[tags[0]]] # Intersect with each subsequent tag for tag in tags[1:]: other = [(p["start"], p["end"]) for p in per_tag[tag]] current_intervals = self._pairwise_intersect(current_intervals, other) if not current_intervals: return [] # Filter by minimum duration and build result dicts result = [] for start, end in current_intervals: dur_min = (end - start).total_seconds() / 60 if dur_min >= min_dur: # Gather each tag's value during this window tag_values = {} for tag in tags: for p in per_tag[tag]: if p["start"] <= start and p["end"] >= end: tag_values[tag] = p["value"] break result.append({ "start": start, "end": end, "duration_minutes": dur_min, "tag_values": tag_values, }) return result @staticmethod def _pairwise_intersect(intervals_a, intervals_b): """Compute pairwise intersection of two sorted interval lists.""" result = [] i, j = 0, 0 while i < len(intervals_a) and j < len(intervals_b): a_start, a_end = intervals_a[i] b_start, b_end = intervals_b[j] # Overlap start = max(a_start, b_start) end = min(a_end, b_end) if start < end: result.append((start, end)) # Advance the interval that ends first if a_end <= b_end: i += 1 else: j += 1 return result def _run_downsample_frequency(self, df: pd.DataFrame, params: Dict) -> OperationOutput: target_hz = float(params["target_hz"]) if "timestamp" not in df.columns or len(df) < 2: return OperationOutput("downsample_frequency", f"Downsample to {target_hz} Hz", df, {"error": "Insufficient data"}) # Estimate current frequency sorted_df = df.sort_values("timestamp") dt = sorted_df["timestamp"].diff().dt.total_seconds().dropna() current_hz = 1.0 / dt.median() if dt.median() > 0 else 1.0 if target_hz >= current_hz: # Already at or below target frequency return OperationOutput( "downsample_frequency", f"Data already at {current_hz:.1f} Hz (target: {target_hz} Hz)", sorted_df, {"current_hz": current_hz, "target_hz": target_hz, "rows_before": len(sorted_df), "rows_after": len(sorted_df)}, exportable_df=sorted_df, ) step = max(1, int(round(current_hz / target_hz))) downsampled = sorted_df.iloc[::step].reset_index(drop=True) return OperationOutput( op_name="downsample_frequency", label=f"Downsampled {current_hz:.1f} Hz → {target_hz} Hz", data=downsampled, metadata={"current_hz": current_hz, "target_hz": target_hz, "step": step, "rows_before": len(sorted_df), "rows_after": len(downsampled)}, exportable_df=downsampled, ) def _run_detect_pressure_ramps(self, df: pd.DataFrame, params: Dict) -> OperationOutput: pressure_tag = params["pressure_tag"] start_bar = float(params["start_bar"]) end_bar = float(params["end_bar"]) max_ramps = params.get("max_ramps") if pressure_tag not in df.columns: return OperationOutput("detect_pressure_ramps", f"Ramps in {pressure_tag}", [], {"error": f"Tag {pressure_tag} not found"}) sorted_df = df[["timestamp", pressure_tag]].dropna(subset=[pressure_tag]).sort_values("timestamp") timestamps = sorted_df["timestamp"].values values = sorted_df[pressure_tag].values # State machine: IDLE → RAMPING → COMPLETE ramps = [] state = "IDLE" ramp_start_idx = 0 ramp_max_val = 0.0 for i in range(len(values)): if state == "IDLE": # Look for crossing start_bar from below if values[i] >= start_bar * 0.9 and values[i] <= start_bar * 1.3: state = "RAMPING" ramp_start_idx = i ramp_max_val = values[i] elif state == "RAMPING": ramp_max_val = max(ramp_max_val, values[i]) # Ramp complete: crossed end_bar if values[i] >= end_bar: ramps.append({ "start_time": pd.Timestamp(timestamps[ramp_start_idx]), "end_time": pd.Timestamp(timestamps[i]), "start_pressure": float(values[ramp_start_idx]), "peak_pressure": float(ramp_max_val), "duration_minutes": (pd.Timestamp(timestamps[i]) - pd.Timestamp(timestamps[ramp_start_idx])).total_seconds() / 60, }) state = "IDLE" continue # Ramp aborted: pressure dropped significantly below start if values[i] < start_bar * 0.5: state = "IDLE" if max_ramps and len(ramps) > int(max_ramps): ramps = ramps[:int(max_ramps)] export_df = pd.DataFrame(ramps) if ramps else None return OperationOutput( op_name="detect_pressure_ramps", label=f"{pressure_tag} ramps {start_bar}→{end_bar} bar", data=ramps, metadata={"num_ramps": len(ramps)}, exportable_df=export_df, ) def _run_extract_windows(self, df: pd.DataFrame, params: Dict, prev_result: Any = None) -> OperationOutput: """Slice DataFrame into sub-DataFrames per detected window from previous operation.""" windows = [] if isinstance(prev_result, OperationOutput): data = prev_result.data if isinstance(data, list): windows = data elif isinstance(data, dict): windows = data.get("blocks", data.get("ramps", [])) if not windows: return OperationOutput("extract_windows", "Extract windows", [], {"error": "No windows found in previous result"}) dfs = [] labels = [] for i, w in enumerate(windows): start = w.get("start_time", w.get("start")) end = w.get("end_time", w.get("end")) if start is None or end is None: continue mask = (df["timestamp"] >= pd.Timestamp(start)) & (df["timestamp"] <= pd.Timestamp(end)) window_df = df[mask].copy() if not window_df.empty: dfs.append(window_df) labels.append(f"Window {i + 1}") return OperationOutput( op_name="extract_windows", label=f"Extracted {len(dfs)} windows", data=dfs, metadata={"num_windows": len(dfs), "labels": labels, "rows_per_window": [len(d) for d in dfs]}, exportable_df=None, # Multi-window: handled by export_utils ) def _run_filter_by_condition(self, df: pd.DataFrame, params: Dict) -> OperationOutput: tag = params["tag"] operator = params["operator"] value = float(params["value"]) if tag not in df.columns: return OperationOutput("filter_by_condition", f"Filter {tag}", df, {"error": f"Tag {tag} not found"}) op_str = OPERATOR_MAP.get(operator) if not op_str: return OperationOutput("filter_by_condition", f"Filter {tag}", df, {"error": f"Unknown operator: {operator}. Use: gt, gte, lt, lte, eq"}) if operator == "gt": mask = df[tag] > value elif operator == "gte": mask = df[tag] >= value elif operator == "lt": mask = df[tag] < value elif operator == "lte": mask = df[tag] <= value else: mask = df[tag] == value filtered = df[mask].copy() return OperationOutput( op_name="filter_by_condition", label=f"{tag} {op_str} {value}", data=filtered, metadata={"rows_before": len(df), "rows_after": len(filtered)}, exportable_df=filtered, ) def _run_compute_statistics(self, df: pd.DataFrame, params: Dict) -> OperationOutput: tags = params.get("tags") numeric_cols = [c for c in df.columns if c != "timestamp" and pd.api.types.is_numeric_dtype(df[c])] if tags: numeric_cols = [c for c in numeric_cols if c in tags] stats = {} for col in numeric_cols: data = df[col].dropna() if len(data) > 0: stats[col] = { "count": int(len(data)), "min": float(data.min()), "max": float(data.max()), "mean": float(data.mean()), "std": float(data.std()), } export_df = pd.DataFrame(stats).T.reset_index().rename(columns={"index": "Sensor"}) if stats else None return OperationOutput( op_name="compute_statistics", label=f"Statistics for {len(numeric_cols)} sensors", data=stats, metadata={"num_sensors": len(stats)}, exportable_df=export_df, ) def _run_detect_and_analyze_fills(self, df: pd.DataFrame, params: Dict, db_connector=None) -> OperationOutput: """Detect fill cycles and compute per-fill metrics.""" from analysis.pump_cycle_detector import PumpCycleDetector kwh_threshold = params.get("kwh_per_kg_threshold") target_fill_id = params.get("fill_id") include_strokes = params.get("include_strokes", True) # Find motor speed tag in DataFrame speed_tag = None for tag in MOTOR_SPEED_TAGS: if tag in df.columns: speed_tag = tag break if not speed_tag: return OperationOutput("detect_and_analyze_fills", "Detect fills", [], {"error": "No motor speed tag found in data. Include M130_Speed or MC130_VFD_Speed."}) # Prepare motor speed series for cycle detector motor_df = df[["timestamp", speed_tag]].dropna(subset=[speed_tag]).rename(columns={speed_tag: "value"}) detector = PumpCycleDetector() raw_cycles = detector.detect_cycles(motor_df) if not raw_cycles: return OperationOutput("detect_and_analyze_fills", "Detect fills", [], {"error": "No cycles detected in the time range"}) # Enrich with pressure/flow/temp from the fetched data (no additional DB query needed) enriched = self._enrich_fills_from_df(raw_cycles, df, db_connector) # Compute per-fill advanced metrics fill_results = [] for c in enriched: fill_dict = { "fill_id": c["cycle_id"], "start_time": c["start_time"], "end_time": c["end_time"], "duration_min": c.get("duration_minutes", 0), "peak_pressure_bar": c.get("peak_pressure", None), "avg_flow_kg_min": c.get("avg_flow", None), "total_mass_kg": c.get("total_mass_kg", None), "kwh_per_kg": None, "total_revolutions": None, } # kWh/kg computation if "avg_power_kw" in c and c.get("avg_flow") and c["avg_flow"] > 0: # kWh/kg = AvgPower [kW] * 60 / FT140 [kg/min] kwh_kg = (c["avg_power_kw"] * 60) / c["avg_flow"] if c["avg_flow"] > 0.001 else None fill_dict["kwh_per_kg"] = round(kwh_kg, 4) if kwh_kg is not None else None # Stroke count if include_strokes and "avg_rpm" in c and c.get("duration_minutes"): revolutions = c["avg_rpm"] * c["duration_minutes"] fill_dict["total_revolutions"] = int(round(revolutions)) fill_results.append(fill_dict) # Apply filters if target_fill_id is not None: target_id = int(target_fill_id) matched = [f for f in fill_results if f["fill_id"] == target_id] if not matched: available = [f["fill_id"] for f in fill_results] return OperationOutput("detect_and_analyze_fills", f"Fill {target_id}", [], {"error": f"Fill {target_id} not found. Available fill IDs: {available}"}) fill_results = matched if kwh_threshold is not None: threshold = float(kwh_threshold) fill_results = [f for f in fill_results if f.get("kwh_per_kg") is not None and f["kwh_per_kg"] > threshold] export_df = pd.DataFrame(fill_results) if fill_results else None note = "Revolutions = RPM × minutes. For simplex: strokes = revolutions. For triplex: strokes = 3 × revolutions." return OperationOutput( op_name="detect_and_analyze_fills", label=f"Fill analysis ({len(fill_results)} fills)", data=fill_results, metadata={"num_fills": len(fill_results), "note": note}, exportable_df=export_df, ) def _enrich_fills_from_df(self, cycles: List[Dict], df: pd.DataFrame, db_connector) -> List[Dict]: """Enrich cycle dicts with metrics computed from the already-fetched wide DataFrame.""" enriched = [] for c in cycles: ec = c.copy() mask = (df["timestamp"] >= c["start_time"]) & (df["timestamp"] <= c["end_time"]) cdf = df[mask] if cdf.empty: enriched.append(ec) continue # Peak pressure if "PT130" in cdf.columns: pt130 = cdf["PT130"].dropna() if len(pt130) > 0: ec["peak_pressure"] = float(pt130.max()) ec["initial_pressure"] = float(pt130.iloc[0]) # Average flow if "FT140" in cdf.columns: ft140 = cdf["FT140"].dropna() if len(ft140) > 0: ec["avg_flow"] = float(ft140.mean()) # Total mass via trapezoidal integration flow_sorted = cdf[["timestamp", "FT140"]].dropna().sort_values("timestamp") if len(flow_sorted) > 1: dt_min = flow_sorted["timestamp"].diff().dt.total_seconds().fillna(0) / 60 ec["total_mass_kg"] = float((flow_sorted["FT140"] * dt_min).sum()) # Average power if "AvgPower" in cdf.columns: power = cdf["AvgPower"].dropna() if len(power) > 0: ec["avg_power_kw"] = float(power.mean()) # Motor speed for tag in MOTOR_SPEED_TAGS: if tag in cdf.columns: speed = cdf[tag].dropna() if len(speed) > 0: ec["avg_rpm"] = float(speed.mean()) ec["peak_rpm"] = float(speed.max()) break # Peak temperatures for tag in ["TT110", "TT130"]: if tag in cdf.columns: data = cdf[tag].dropna() if len(data) > 0: ec[f"peak_{tag}"] = float(data.max()) enriched.append(ec) return enriched def _run_count_pump_strokes(self, df: pd.DataFrame, params: Dict) -> OperationOutput: """Count total pump revolutions by integrating RPM over time.""" speed_tag = params.get("speed_tag") if not speed_tag: for tag in MOTOR_SPEED_TAGS: if tag in df.columns: speed_tag = tag break if not speed_tag or speed_tag not in df.columns: return OperationOutput("count_pump_strokes", "Pump strokes", {}, {"error": "No motor speed tag found. Include M130_Speed or MC130_VFD_Speed."}) speed_df = df[["timestamp", speed_tag]].dropna(subset=[speed_tag]).sort_values("timestamp") if len(speed_df) < 2: return OperationOutput("count_pump_strokes", "Pump strokes", {}, {"error": "Insufficient motor speed data"}) rpm_values = speed_df[speed_tag].values timestamps = speed_df["timestamp"].values # Integrate: revolutions = sum(RPM_i × Δt_i) where Δt in minutes dt_minutes = pd.Series(timestamps).diff().dt.total_seconds().fillna(0).values / 60 total_revolutions = float(np.sum(rpm_values * dt_minutes)) active_mask = rpm_values > 10 # RPM > idle threshold active_minutes = float(np.sum(dt_minutes[active_mask])) result = { "total_revolutions": int(round(total_revolutions)), "avg_rpm": float(np.mean(rpm_values[active_mask])) if active_mask.any() else 0, "peak_rpm": float(np.max(rpm_values)) if len(rpm_values) > 0 else 0, "active_minutes": round(active_minutes, 1), "speed_tag_used": speed_tag, "note": "Revolutions = RPM × time. For simplex: strokes = revolutions. For triplex: strokes = 3 × revolutions.", } return OperationOutput( op_name="count_pump_strokes", label=f"Pump strokes ({speed_tag})", data=result, metadata={"total_revolutions": result["total_revolutions"]}, )