Générateur de rapports externe

#!/usr/bin/env python3
"""
SolarViz - Mirox External Report Generator

Fetches the data of one plant from the Mirox API and builds an energy
production report with a chart as a PDF file:

  * plant information          GET /api/v1/export/report/{plant}/info
  * events                     GET /api/v1/export/report/{plant}/events.csv
  * monthly production figures GET /api/v1/export/series/template/{template}

Requires: pip install requests pandas matplotlib reportlab
The API token is read from the environment variable MIROX_TOKEN.
"""

import io
import os
import sys
import time
from datetime import datetime, timezone

import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt
import pandas as pd
import requests
from reportlab.lib import colors
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet
from reportlab.lib.units import cm
from reportlab.platypus import Image, Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle

# Configuration - replace with your values
API_URL = "https://service.mirox.io"
API_TOKEN = os.environ.get("MIROX_TOKEN", "YOUR_API_TOKEN")  # group "Metric Export"
PARK_UID = "ABC123DEF456"  # the UID of the plant
TEMPLATE_UID = "ABCD12340002"  # system template "Report Technical"

# The report covers this year so far; the events are those of the current quarter
YEAR = datetime.now(timezone.utc).year
QUARTER = (datetime.now(timezone.utc).month - 1) // 3 + 1

HEADERS = {"Authorization": f"Bearer {API_TOKEN}", "Accept-Encoding": "gzip"}

# One CSV dialect for every request, independent of regional settings
CSV_FORMAT = {"separator_csv": ",", "separator_decimal": "."}

# The columns the report reads, by metric id (header=id)
PRODUCTION = "plant.grid_feedin"
EXPECTED = "plant.expected_weather"
LOSS_GRID = "plant.shutdown_grid"
LOSS_MARKETER = "plant.shutdown_external"
PERFORMANCE_RATIO = "plant.pr_weather"


class ExportRefused(Exception):
    """A coded refusal of the Metric Export: what was wrong, and the way out."""

    def __init__(self, status, detail):
        self.status = status
        self.code = detail.get("code", "")
        self.params = detail.get("params", {})
        self.hint = detail.get("hint", {})
        super().__init__(f"{status} {self.code}: {detail.get('message', '')} "
                         f"params={self.params} hint={self.hint}")


def get(path, params=None, retries=3):
    """GET with the token. Waits and repeats when the export is busy (429)."""
    for _attempt in range(retries):
        response = requests.get(f"{API_URL}{path}", headers=HEADERS, params=params, timeout=90)
        if response.status_code == 429:
            time.sleep(int(response.headers.get("Retry-After", "5")))
            continue
        if response.status_code >= 400:
            try:
                detail = response.json().get("detail")
            except ValueError:
                detail = None
            if isinstance(detail, dict):
                raise ExportRefused(response.status_code, detail)
            response.raise_for_status()
        return response
    raise RuntimeError(f"{path}: still busy after {retries} attempts")


def fetch_park_info():
    """Plant information as a dictionary."""
    return get(f"/api/v1/export/report/{PARK_UID}/info").json()


def fetch_events(year, quarter):
    """The important events of a quarter as a DataFrame."""
    response = get(f"/api/v1/export/report/{PARK_UID}/events.csv",
                   {"year": year, "quarter": quarter, **CSV_FORMAT})
    events = pd.read_csv(io.StringIO(response.content.decode("utf-8")))
    if events.empty:
        return events
    # A priority of 1000 or more marks an important event
    events = events[events["Priority"] >= 1000]
    return events.drop(columns=["Event ID", "Priority"], errors="ignore")


def fetch_metrics(year):
    """Monthly figures of the template for the year so far.

    header=id gives column names that never change. A year in progress ends
    now; the notes of the answer say so.
    """
    response = get(f"/api/v1/export/series/template/{TEMPLATE_UID}", {
        "park": PARK_UID, "year": year, "step": "month", "header": "id", **CSV_FORMAT})
    for note in filter(None, response.headers.get("X-Mirox-Notes", "").split(";")):
        print(f"  note: {note}")
    metrics = pd.read_csv(io.StringIO(response.content.decode("utf-8")))
    # An empty cell means "no data"; it is read as NaN and left out of the sums
    return metrics.rename(columns={"timestamp": "month"})


def create_production_chart(metrics):
    """Monthly production with the shutdown losses stacked, against the expectation."""
    plt.figure(figsize=(8, 4))
    positions = range(len(metrics))
    production = metrics[PRODUCTION].fillna(0)
    loss_grid = metrics[LOSS_GRID].fillna(0)
    loss_marketer = metrics[LOSS_MARKETER].fillna(0)

    plt.bar(positions, production, color="#2986cc", label="Grid feed-in")
    plt.bar(positions, loss_grid, bottom=production, color="#e06666",
            label="Shutdown loss, grid operator")
    plt.bar(positions, loss_marketer, bottom=production + loss_grid, color="#f1c232",
            label="Shutdown loss, direct marketer")
    plt.plot(positions, metrics[EXPECTED], marker="o", color="#6aa84f", linewidth=2,
             label="Expected production")

    plt.title("Monthly production and shutdown losses")
    plt.xlabel("Month")
    plt.ylabel("Energy (kWh)")
    plt.xticks(positions, metrics["month"], rotation=45)
    plt.grid(axis="y", alpha=0.3)
    plt.legend(loc="upper left")
    plt.tight_layout()

    buffer = io.BytesIO()
    plt.savefig(buffer, format="png", dpi=300)
    buffer.seek(0)
    plt.close()
    return buffer


def calculate_performance(metrics):
    """The figures of the summary table."""
    production = metrics[PRODUCTION].sum()
    expected = metrics[EXPECTED].sum()
    loss_grid = metrics[LOSS_GRID].sum()
    loss_marketer = metrics[LOSS_MARKETER].sum()
    losses = loss_grid + loss_marketer
    return {
        "production": production,
        "expected": expected,
        "production_vs_expected": production / expected * 100 if expected else None,
        "loss_grid": loss_grid,
        "loss_marketer": loss_marketer,
        "losses": losses,
        "loss_share": losses / (production + losses) * 100 if production + losses else None,
    }


def number(value, unit="", digits=2):
    if value is None or pd.isna(value):
        return "-"
    return f"{value:,.{digits}f}{unit}"


def plain_table(rows, widths):
    table = Table(rows, colWidths=widths)
    table.setStyle(TableStyle([
        ("FONTNAME", (0, 0), (0, -1), "Helvetica-Bold"),
        ("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
        ("TOPPADDING", (0, 0), (-1, -1), 3),
        ("BOTTOMPADDING", (0, 0), (-1, -1), 3),
    ]))
    return table


def grid_table(rows, font_size=9):
    table = Table(rows, repeatRows=1)
    table.setStyle(TableStyle([
        ("BACKGROUND", (0, 0), (-1, 0), colors.lightgrey),
        ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
        ("ALIGN", (0, 0), (-1, 0), "CENTER"),
        ("ALIGN", (1, 1), (-1, -1), "RIGHT"),
        ("GRID", (0, 0), (-1, -1), 0.5, colors.grey),
        ("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
        ("FONTSIZE", (0, 0), (-1, -1), font_size),
    ]))
    return table


def generate_pdf_report(park_info, events, metrics, performance, chart):
    """Build the PDF and return its file name."""
    output_file = f"SolarViz_Report_{park_info['name'].replace(' ', '_')}_{YEAR}_Q{QUARTER}.pdf"
    doc = SimpleDocTemplate(output_file, pagesize=A4)
    styles = getSampleStyleSheet()
    styles["Title"].fontSize = 18
    address = park_info.get("address") or {}
    portfolio = park_info.get("portfolio") or {}
    organization = park_info.get("organization") or {}

    elements = [
        Paragraph(f"Energy Production Report: {park_info['name']}", styles["Title"]),
        Spacer(1, 0.5 * cm),
        Paragraph("Plant Information", styles["Heading2"]),
        plain_table([
            ["Name:", park_info["name"]],
            ["Type:", park_info.get("type", "")],
            ["Location:", f"{address.get('city', '')}, {address.get('country', '')}"],
            ["Peak power:", number((park_info.get("peak_production_w") or 0) / 1e6, " MW")],
            ["Portfolio:", portfolio.get("name", "")],
            ["Organization:", organization.get("name", "")],
        ], [3 * cm, 10 * cm]),
        Spacer(1, 0.5 * cm),
        Paragraph(f"Summary ({YEAR} to date)", styles["Heading2"]),
        plain_table([
            ["Grid feed-in:", number(performance["production"], " kWh")],
            ["Expected production:", number(performance["expected"], " kWh")],
            ["Feed-in against expectation:", number(performance["production_vs_expected"], " %")],
            ["Shutdown losses:", number(performance["losses"], " kWh")],
            ["  by the grid operator:", number(performance["loss_grid"], " kWh")],
            ["  by the direct marketer:", number(performance["loss_marketer"], " kWh")],
            ["Share of the losses:", number(performance["loss_share"], " %")],
        ], [6 * cm, 7 * cm]),
        Spacer(1, 0.5 * cm),
        Paragraph("Monthly Production", styles["Heading2"]),
        Image(chart, width=16 * cm, height=8 * cm),
        Spacer(1, 0.5 * cm),
    ]

    rows = [["Month", "Feed-in (kWh)", "Expected (kWh)", "Loss grid (kWh)",
             "Loss marketer (kWh)", "PR (%)"]]
    for _, row in metrics.iterrows():
        rows.append([row["month"], number(row[PRODUCTION]), number(row[EXPECTED]),
                     number(row[LOSS_GRID]), number(row[LOSS_MARKETER]),
                     number(row[PERFORMANCE_RATIO], digits=1)])
    elements += [grid_table(rows), Spacer(1, 0.5 * cm),
                 Paragraph("Important Events", styles["Heading2"])]

    if events.empty:
        elements.append(Paragraph("No important events in this quarter.", styles["Normal"]))
    else:
        elements.append(grid_table([events.columns.tolist()] + events.values.tolist(), 8))

    elements += [
        Spacer(1, 1 * cm),
        Paragraph("Data Source", styles["Heading2"]),
        Paragraph(
            "Data exported from the Mirox platform through the Metric Export API.<br/>"
            f"Created: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M')} UTC<br/><br/>"
            "<i>Generated by SolarViz - External Reporting Tool</i>", styles["Normal"]),
    ]
    doc.build(elements)
    return output_file


def main():
    try:
        print("Fetching plant information...")
        park_info = fetch_park_info()

        print("Fetching events...")
        events = fetch_events(YEAR, QUARTER)

        print("Fetching monthly figures...")
        metrics = fetch_metrics(YEAR)

        print("Building the report...")
        performance = calculate_performance(metrics)
        chart = create_production_chart(metrics)
        output_file = generate_pdf_report(park_info, events, metrics, performance, chart)
        print(f"Report created: {output_file}")
    except ExportRefused as refused:
        print(f"The export was refused: {refused}")
        if refused.code == "over_budget":
            print("Use a coarser step or a shorter range, see hint.step and hint.max_days.")
        sys.exit(1)
    except requests.RequestException as error:
        print(f"The request failed: {error}")
        sys.exit(1)


if __name__ == "__main__":
    main()