Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| from typing import Dict, Any, List, Literal, Tuple | |
| from . import tools | |
| StepRole = Literal["user", "intent_agent", "planning_agent", "narrator_agent", "tool"] | |
| class Step: | |
| role: StepRole | |
| content: str | |
| meta: Dict[str, Any] | |
| def to_dict(self) -> Dict[str, Any]: | |
| return { | |
| "role": self.role, | |
| "content": self.content, | |
| "meta": self.meta, | |
| } | |
| class IntentAgent: | |
| def classify(self, query: str) -> Dict[str, Any]: | |
| text = query.lower() | |
| if any(k in text for k in ["what if", "simulate", "scenario", "contingency", "impact of"]): | |
| intent = "simulation" | |
| elif any(k in text for k in ["host", "hosting capacity", "add pv", "add ev", "der"]): | |
| intent = "hosting_capacity" | |
| elif any(k in text for k in ["explain", "why", "how does"]): | |
| intent = "explanation" | |
| else: | |
| intent = "simulation" | |
| feeder = "F1" | |
| if "f2" in text or "feeder 2" in text: | |
| feeder = "F2" | |
| elif "f3" in text or "feeder 3" in text: | |
| feeder = "F3" | |
| added_pv = 0.0 | |
| added_load = 0.0 | |
| import re | |
| mw_matches = re.findall(r"(\d+(?:\.\d+)?)\s*mw", text) | |
| if mw_matches: | |
| value = float(mw_matches[0]) | |
| if "pv" in text or "solar" in text: | |
| added_pv = value | |
| else: | |
| added_load = value | |
| return { | |
| "intent": intent, | |
| "feeder": feeder, | |
| "added_pv_mw": added_pv, | |
| "added_load_mw": added_load, | |
| } | |
| class PlanningAgent: | |
| def plan_and_analyze(self, query: str, intent_info: Dict[str, Any]) -> Tuple[str, Dict[str, Any], List[Step]]: | |
| feeder = intent_info["feeder"] | |
| added_pv = float(intent_info.get("added_pv_mw", 0.0)) | |
| added_load = float(intent_info.get("added_load_mw", 0.0)) | |
| steps: List[Step] = [] | |
| summary = ( | |
| f"Analyzing feeder {feeder} with added PV={added_pv:.1f} MW, " | |
| f"added load={added_load:.1f} MW using a simplified power-flow stub." | |
| ) | |
| steps.append(Step(role="planning_agent", content=summary, meta={"feeder": feeder})) | |
| pf_result = tools.run_power_flow_scenario(feeder, added_pv_mw=added_pv, added_load_mw=added_load) | |
| steps.append( | |
| Step( | |
| role="tool", | |
| content="Ran simplified power-flow scenario.", | |
| meta=pf_result.to_dict(), | |
| ) | |
| ) | |
| feeder_meta = tools.get_feeder_summary(feeder) | |
| steps.append( | |
| Step( | |
| role="tool", | |
| content="Retrieved static feeder metadata.", | |
| meta=feeder_meta, | |
| ) | |
| ) | |
| technical_summary = { | |
| "intent": intent_info["intent"], | |
| "feeder": feeder, | |
| "added_pv_mw": added_pv, | |
| "added_load_mw": added_load, | |
| "power_flow": pf_result.to_dict(), | |
| "feeder_meta": feeder_meta, | |
| } | |
| return "ok", technical_summary, steps | |
| class NarratorAgent: | |
| def narrate(self, query: str, technical: Dict[str, Any]) -> str: | |
| pf = technical["power_flow"] | |
| meta = technical["feeder_meta"] | |
| lines: List[str] = [] | |
| lines.append(f"You asked: {query.strip()}") | |
| lines.append("") | |
| lines.append(f"Here's what Grid-Gent found for {meta['name']}:") | |
| lines.append( | |
| f"- Base peak demand (demo data): {meta['peak_mw']} MW with about {meta['num_customers']} customers." | |
| ) | |
| lines.append( | |
| f"- In this scenario, we assumed +{technical['added_load_mw']:.1f} MW of extra load and " | |
| f"+{technical['added_pv_mw']:.1f} MW of additional PV." | |
| ) | |
| lines.append("") | |
| lines.append("Simplified power-flow-style results (demo model):") | |
| lines.append(f"- Peak loading: {pf['peak_loading_pct']:.1f}% of an approximate rating.") | |
| lines.append(f"- Minimum voltage: {pf['min_voltage_pu']:.3f} pu") | |
| lines.append(f"- Maximum voltage: {pf['max_voltage_pu']:.3f} pu") | |
| if pf["overload_elements"]: | |
| lines.append("") | |
| lines.append("Potential issues flagged:") | |
| for item in pf["overload_elements"]: | |
| lines.append(f" • {item}") | |
| else: | |
| lines.append("") | |
| lines.append("No major issues were flagged in this simplified view.") | |
| lines.append("") | |
| lines.append(pf["notes"]) | |
| lines.append("") | |
| lines.append( | |
| "Note: this is a deliberately simplified demo model. A real deployment would " | |
| "use your actual network model, load/DER data, and planning criteria, and would " | |
| "treat these results as advisory, subject to engineer review." | |
| ) | |
| return "\n".join(lines) | |