Tools: Simulation¶

def build_simulation_report() -> SimulationReportResult:
    """Build the full tabular report from the simulation SQL output."""
    state = get_state()
    result = state.require_simulation_result()

    # Single SQL connection for both tabular data and metadata extraction.
    energyplus_version = ""
    environment = ""
    timestamp = ""
    with _open_sql_result(result) as sql:
        report_sections, table_count = _collect_tabular_sections(sql)
        try:
            sim_rows = sql.query("SELECT EnergyPlusVersion, TimeStamp FROM Simulations LIMIT 1")
            if sim_rows:
                energyplus_version = str(sim_rows[0][0] or "")
                timestamp = str(sim_rows[0][1] or "")
            envs = sql.list_environments()
            if envs:
                environment = ", ".join(e.name for e in envs)
        except Exception:
            logger.debug("Could not extract simulation metadata from SQL", exc_info=True)

    building = "Unknown"
    doc = state.document
    if doc is not None and "Building" in doc:
        bldg = doc["Building"].first()
        if bldg is not None:
            building = bldg.name or "Unknown"

    return SimulationReportResult(
        building_name=building,
        environment=environment,
        energyplus_version=energyplus_version,
        timestamp=timestamp,
        report_count=len(report_sections),
        table_count=table_count,
        reports=report_sections,
    )

@tool(annotations=_EXPORT)
def export_timeseries(
    variable_name: Annotated[str, Field(description="Variable name.")],
    key_value: Annotated[str, Field(description='Zone/surface or "*".')] = "*",
    frequency: Annotated[ReportingFrequency | None, Field(description="Reporting frequency.")] = None,
    environment: Annotated[Literal["sizing", "annual"] | None, Field(description="Environment filter.")] = None,
    output_path: Annotated[str | None, Field(description="CSV path (default: output dir).")] = None,
) -> ExportTimeseriesResult:
    """Export time series to CSV."""
    import csv
    from pathlib import Path

    # Validate output path early, before the (potentially slow) SQL query.
    validated_output_path: Path | None = None
    if output_path is not None:
        from idfkit_mcp.tools._path_validation import validate_output_path

        validated_output_path = validate_output_path(Path(output_path), label="Export path")

    state = get_state()
    result = state.require_simulation_result()

    try:
        with _open_sql_result(result) as sql:
            ts = sql.get_timeseries(
                variable_name=variable_name,
                key_value=key_value,
                frequency=frequency,
                environment=environment,
            )
    except OperationalError as e:
        raise ToolError(
            f"SQL query failed: {e}. "
            "The simulation may not have completed successfully, or Output:SQLite was not configured in the model. "
            "Check run_simulation results for errors."
        ) from e

    if validated_output_path is not None:
        csv_path = validated_output_path
    else:
        safe_name = re.sub(r"[^\w]+", "_", variable_name).strip("_").lower()
        csv_path = result.run_dir / f"timeseries_{safe_name}.csv"

    with csv_path.open("w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["timestamp", ts.variable_name + f" [{ts.units}]"])
        for i in range(len(ts.values)):
            writer.writerow([ts.timestamps[i].isoformat(), ts.values[i]])

    logger.info("Exported timeseries %r to %s (%d rows)", variable_name, csv_path, len(ts.values))
    return ExportTimeseriesResult(
        path=str(csv_path),
        variable_name=ts.variable_name,
        key_value=ts.key_value,
        units=ts.units,
        frequency=ts.frequency,
        rows=len(ts.values),
    )

def get_results_summary() -> GetResultsSummaryResult:
    """Build results summary with QA diagnostics from the last simulation.

    This function powers the ``idfkit://simulation/results`` resource — the primary
    feedback signal for the agent QA loop.
    """
    state = get_state()
    result = state.require_simulation_result()

    errors = result.errors
    fatal_msgs = [{"message": m.message, "details": list(m.details)} for m in errors.fatal]
    severe_msgs = [{"message": m.message, "details": list(m.details)} for m in errors.severe[:10]]

    classified = _classify_warnings(errors)

    # --- SQL-based diagnostics (defensive: degrade gracefully if SQL unavailable) ---
    sql_available = False
    unmet_hours: list[UnmetHoursRow] = []
    total_unmet_heating = 0.0
    total_unmet_cooling = 0.0
    end_uses: list[EndUseRow] = []
    notes: list[str] = []

    if result.sql_path is not None:
        try:
            from idfkit.simulation.parsers.sql import SQLResult

            with SQLResult(result.sql_path) as sql:
                sql_available = True
                try:
                    unmet_hours, total_unmet_heating, total_unmet_cooling = _query_unmet_hours(sql)
                except Exception as e:
                    notes.append(f"Unmet hours unavailable: {e}")
                try:
                    end_uses = _query_end_uses(sql)
                except Exception as e:
                    notes.append(f"End-use data unavailable: {e}")
        except Exception as e:
            notes.append(f"SQL data unavailable: {e}")
    else:
        notes.append("No SQL output file. Add Output:SQLite to the model for energy diagnostics.")

    qa_flags = _build_qa_flags(errors, total_unmet_heating, total_unmet_cooling, classified, sql_available)

    return GetResultsSummaryResult.model_validate({
        "success": result.success,
        "runtime_seconds": round(result.runtime_seconds, 2),
        "output_directory": str(result.run_dir),
        "errors": {
            "fatal": errors.fatal_count,
            "severe": errors.severe_count,
            "warnings": errors.warning_count,
            "summary": errors.summary(),
        },
        "fatal_messages": fatal_msgs if fatal_msgs else None,
        "severe_messages": severe_msgs if severe_msgs else None,
        "sql_available": sql_available,
        "unmet_hours": [u.model_dump() for u in unmet_hours] if sql_available else None,
        "total_unmet_heating_hours": total_unmet_heating if sql_available else None,
        "total_unmet_cooling_hours": total_unmet_cooling if sql_available else None,
        "end_uses": [e.model_dump() for e in end_uses] if sql_available else None,
        "classified_warnings": [w.model_dump() for w in classified] if classified else None,
        "qa_flags": [f.model_dump() for f in qa_flags] if qa_flags else None,
        "notes": notes if notes else None,
    })

@tool(annotations=_READ_ONLY)
def list_output_variables(
    search: Annotated[str | None, Field(description="Regex filter on name (case-insensitive).")] = None,
    limit: Annotated[int, Field(description="Max results.")] = 50,
) -> ListOutputVariablesResult:
    """List output variables and meters from last simulation."""
    state = get_state()
    result = state.require_simulation_result()

    limit = min(limit, 200)

    variables = result.variables
    if variables is not None and (variables.variables or variables.meters):
        from idfkit.simulation.parsers.rdd import OutputVariable

        all_items = variables.search(search) if search else [*variables.variables, *variables.meters]
        serialized: list[dict[str, str | None]] = []
        for item in all_items:
            entry: dict[str, str | None] = {"name": item.name, "units": item.units}
            if isinstance(item, OutputVariable):
                entry["key"] = item.key
                entry["type"] = "variable"
            else:
                entry["type"] = "meter"
            serialized.append(entry)

        total = len(variables.variables) + len(variables.meters)
        return _build_output_variable_result(serialized, total_available=total, limit=limit)

    with _open_sql_result(result) as sql:
        if search:
            try:
                regex = re.compile(search, re.IGNORECASE)
            except re.error as exc:
                raise ToolError(f"Invalid regex pattern: {exc}") from None
        else:
            regex = None
        all_items = sql.list_variables()
        serialized = [
            {
                "name": item.name,
                "units": item.units,
                "key": item.key_value or None,
                "type": "meter" if item.is_meter else "variable",
            }
            for item in all_items
            if regex is None or regex.search(item.name)
        ]
    return _build_output_variable_result(serialized, total_available=len(all_items), limit=limit)

@tool(annotations=_READ_ONLY)
def list_simulation_reports() -> list[str]:
    """List all tabular report names available in the last simulation's SQL output.

    Use the returned names with ``query_simulation_table`` to retrieve specific tables.

    Preconditions: simulation completed with SQL output available.
    Side effects: none — read-only.
    """
    state = get_state()
    result = state.require_simulation_result()

    with _open_sql_result(result) as sql:
        return sql.list_reports()

@tool(annotations=_READ_ONLY)
def query_simulation_table(
    report_name: Annotated[
        str,
        Field(
            description="Report name (e.g. 'AnnualBuildingUtilityPerformanceSummary', 'SystemSummary'). Use list_simulation_reports to discover available names."
        ),
    ],
    table_name: Annotated[
        str | None,
        Field(
            description="Table name within the report (e.g. 'End Uses', 'Time Setpoint Not Met'). Omit to return all tables in the report."
        ),
    ] = None,
    row_name: Annotated[str | None, Field(description="Filter to a specific row label.")] = None,
    column_name: Annotated[str | None, Field(description="Filter to a specific column label.")] = None,
) -> QuerySimulationTableResult:
    """Query tabular report data from the last simulation's SQL output.

    Use this for deeper analysis beyond the structured diagnostics in
    ``idfkit://simulation/results``. Tabular data covers every EnergyPlus
    summary report: energy use, envelope, HVAC sizing, comfort, and more.

    Omit ``table_name`` to retrieve all tables within a report at once.
    To discover available report names call ``list_simulation_reports`` first.
    Common report names:
      - ``AnnualBuildingUtilityPerformanceSummary`` — site/source energy, end uses, EUI
      - ``SystemSummary`` — unmet hours, HVAC sizing
      - ``EnvelopeSummary`` — U-values, areas, orientations
      - ``EquipmentSummary`` — HVAC component sizing
      - ``ZoneComponentLoadSummary`` — peak heating/cooling loads by zone
      - ``LightingSummary`` — lighting power density

    Preconditions: simulation completed with SQL output available (``sql_available: true``
    in ``idfkit://simulation/results``).
    Side effects: none — read-only.
    """
    state = get_state()
    result = state.require_simulation_result()

    with _open_sql_result(result) as sql:
        raw_rows = sql.get_tabular_data(
            report_name=report_name,
            table_name=table_name,
            row_name=row_name,
            column_name=column_name,
        )

    if not raw_rows:
        msg = f"No data found for report '{report_name}'"
        if table_name is not None:
            msg += f", table '{table_name}'"
        msg += ". Use list_simulation_reports to see available reports."
        raise ToolError(msg)

    rows = [
        TabularRow(
            report_name=r.report_name,
            report_for=r.report_for,
            table_name=r.table_name,
            row_name=r.row_name,
            column_name=r.column_name,
            units=r.units or "",
            value=r.value.strip(),
        )
        for r in raw_rows
    ]
    return QuerySimulationTableResult(
        report_name=report_name,
        table_name=table_name,
        row_count=len(rows),
        rows=rows,
    )

@tool(annotations=_READ_ONLY)
def query_timeseries(
    variable_name: Annotated[str, Field(description="Variable name.")],
    key_value: Annotated[str, Field(description='Zone/surface or "*".')] = "*",
    frequency: Annotated[ReportingFrequency | None, Field(description="Reporting frequency.")] = None,
    environment: Annotated[Literal["sizing", "annual"] | None, Field(description="Environment filter.")] = None,
    limit: Annotated[int, Field(description="Max data points.")] = 24,
) -> QueryTimeseriesResult:
    """Query time series data from simulation SQL output."""
    limit = min(limit, 500)

    state = get_state()
    result = state.require_simulation_result()

    try:
        with _open_sql_result(result) as sql:
            ts = sql.get_timeseries(
                variable_name=variable_name,
                key_value=key_value,
                frequency=frequency,
                environment=environment,
            )
    except OperationalError as e:
        raise ToolError(
            f"SQL query failed: {e}. "
            "The simulation may not have completed successfully, or Output:SQLite was not configured in the model. "
            "Check run_simulation results for errors."
        ) from e

    rows = [
        {"timestamp": ts.timestamps[i].isoformat(), "value": ts.values[i]} for i in range(min(limit, len(ts.values)))
    ]

    logger.debug(
        "query_timeseries: %s key=%s freq=%s total=%d returned=%d",
        variable_name,
        key_value,
        frequency,
        len(ts.values),
        len(rows),
    )
    return QueryTimeseriesResult.model_validate({
        "variable_name": ts.variable_name,
        "key_value": ts.key_value,
        "units": ts.units,
        "frequency": ts.frequency,
        "total_points": len(ts.values),
        "returned": len(rows),
        "data": rows,
    })

@resource(
    "ui://idfkit/report-viewer.html",
    name="report_viewer",
    title="Simulation Report Viewer",
    description="Interactive browser for EnergyPlus tabular simulation reports.",
    meta={
        "ui": app_config_to_meta_dict(
            AppConfig(
                csp=ResourceCSP(resourceDomains=["https://unpkg.com"]),
                prefersBorder=False,
            )
        )
    },
)
def report_viewer_html() -> str:
    """Return the self-contained report viewer HTML."""
    from idfkit_mcp.report_viewer import REPORT_VIEWER_HTML

    return REPORT_VIEWER_HTML

@tool(annotations=_RUN)
async def run_simulation(
    weather_file: Annotated[str | None, Field(description="EPW path (default: last downloaded).")] = None,
    design_day: Annotated[bool, Field(description="Design-day only.")] = False,
    annual: Annotated[bool, Field(description="Annual simulation.")] = False,
    energyplus_dir: Annotated[str | None, Field(description="EnergyPlus install dir.")] = None,
    energyplus_version: Annotated[str | None, Field(description='Version filter "X.Y.Z".')] = None,
    output_directory: Annotated[str | None, Field(description="Output dir.")] = None,
    ctx: Context | None = None,
) -> RunSimulationResult:
    """Execute EnergyPlus on the loaded model — the authoritative runtime validation gate.

    Fatal or severe errors mean the model did not simulate correctly. A clean exit does
    not guarantee physically reasonable results. After this call, read the resource
    ``idfkit://simulation/results`` for full QA diagnostics: unmet hours by zone,
    end-use energy breakdown, classified warnings, and QA flags that drive the fix loop.

    Preconditions: model loaded; weather file set via download_weather_file, or design_day=True.
    Side effects: writes outputs to output_directory; updates session simulation result.
    Next step: read idfkit://simulation/results to assess result quality.
    """
    from idfkit.simulation import async_simulate
    from idfkit.simulation.config import find_energyplus

    state = get_state()

    if state.simulation_lock.locked():
        raise ToolError("A simulation is already in progress for this session. Wait for it to finish.")

    async with state.simulation_lock:
        doc = state.require_model()
        weather = _resolve_weather_path(weather_file, design_day)

        # Simulate on a copy so pre-flight injections (Output:SQLite,
        # Output:Table:SummaryReports) do not mutate the user's loaded model.
        sim_doc = doc.copy()
        _ensure_sqlite_output(sim_doc)
        _ensure_summary_reports(sim_doc)

        config = find_energyplus(path=energyplus_dir, version=energyplus_version)
        resolved_output_dir = _resolve_simulation_output_dir(output_directory, state.session_id)
        logger.info(
            "Starting simulation (EnergyPlus %s, weather=%s, design_day=%s, annual=%s)",
            ".".join(str(p) for p in config.version),
            weather,
            design_day,
            annual,
        )

        result = await async_simulate(
            sim_doc,
            weather="" if weather is None else weather,
            design_day=design_day,
            annual=annual,
            energyplus=config,
            output_dir=resolved_output_dir,
            on_progress=_build_progress_handler(ctx),
        )

        state.simulation_result = result
        state.save_session()

        if result.success:
            logger.info("Simulation completed in %.1fs", result.runtime_seconds)
        else:
            logger.warning("Simulation failed after %.1fs", result.runtime_seconds)

        errors = result.errors

        return RunSimulationResult.model_validate({
            "success": result.success,
            "runtime_seconds": round(result.runtime_seconds, 2),
            "output_directory": str(result.run_dir),
            "energyplus": {
                "version": ".".join(str(part) for part in config.version),
                "install_dir": str(config.install_dir),
                "executable": str(config.executable),
            },
            "errors": _serialize_simulation_errors(errors),
            "simulation_complete": errors.simulation_complete,
        })

@tool(
    annotations=_READ_ONLY,
    meta={
        "ui": app_config_to_meta_dict(
            AppConfig(
                resourceUri="ui://idfkit/report-viewer.html",
                prefersBorder=False,
            )
        )
    },
)
def view_simulation_report() -> SimulationReportResult:
    """Browse the full EnergyPlus tabular report in an interactive viewer.

    Returns all tabular data from the simulation SQL output organized by
    report, section, and table. The companion viewer provides a searchable,
    browsable interface with a table-of-contents sidebar.

    Requires a completed simulation with SQL output.
    """
    return build_simulation_report()