Add artifact.scatter_plot/src/artifact_scatter_plot/cli.py
This commit is contained in:
parent
30e75a240d
commit
976929ba27
1 changed files with 60 additions and 0 deletions
60
artifact.scatter_plot/src/artifact_scatter_plot/cli.py
Normal file
60
artifact.scatter_plot/src/artifact_scatter_plot/cli.py
Normal file
|
|
@ -0,0 +1,60 @@
|
||||||
|
import argparse
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import List
|
||||||
|
|
||||||
|
import pandas as pd
|
||||||
|
|
||||||
|
from artifact_scatter_plot.core import ScatterData, create_scatter_plot
|
||||||
|
|
||||||
|
|
||||||
|
def _load_data(json_path: Path) -> List[ScatterData]:
|
||||||
|
"""Lädt und validiert Eingabedaten aus einer JSON-Datei."""
|
||||||
|
if not json_path.exists():
|
||||||
|
raise FileNotFoundError(f"Eingabedatei nicht gefunden: {json_path}")
|
||||||
|
|
||||||
|
with json_path.open('r', encoding='utf-8') as f:
|
||||||
|
data_list = json.load(f)
|
||||||
|
|
||||||
|
if not isinstance(data_list, list):
|
||||||
|
raise ValueError("Eingabe-JSON muss eine Liste von Objekten enthalten.")
|
||||||
|
|
||||||
|
validated_data: List[ScatterData] = []
|
||||||
|
for i, item in enumerate(data_list):
|
||||||
|
if not isinstance(item, dict):
|
||||||
|
raise ValueError(f"Eintrag {i} ist kein JSON-Objekt.")
|
||||||
|
try:
|
||||||
|
band_width = float(item['band_width'])
|
||||||
|
near_expiry_unpinned = float(item['near_expiry_unpinned'])
|
||||||
|
except (KeyError, TypeError, ValueError) as e:
|
||||||
|
raise ValueError(f"Ungültiger Dateneintrag an Index {i}: {e}") from e
|
||||||
|
validated_data.append(ScatterData(band_width=band_width, near_expiry_unpinned=near_expiry_unpinned))
|
||||||
|
return validated_data
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
"""CLI-Einstiegspunkt für das Erstellen eines Scatter-Plots aus JSON-Daten."""
|
||||||
|
parser = argparse.ArgumentParser(description="Erzeugt einen Scatter-Plot aus Mix-Freeze-Messdaten.")
|
||||||
|
parser.add_argument("--input", required=True, help="Pfad zur Eingabe-JSON-Datei mit den Messdaten.")
|
||||||
|
parser.add_argument("--output", required=True, help="Pfad zur Ausgabe-PNG-Datei für den erzeugten Scatter-Plot.")
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
input_path = Path(args.input)
|
||||||
|
output_path = Path(args.output)
|
||||||
|
|
||||||
|
data = _load_data(input_path)
|
||||||
|
|
||||||
|
# Übergibt Daten an die Core-Funktion zur Plot-Erstellung.
|
||||||
|
create_scatter_plot(data=data)
|
||||||
|
|
||||||
|
# Speichere optional den Plot, falls er eine Datei erzeugt.
|
||||||
|
# Dies hängt von der Implementierung in core ab.
|
||||||
|
# In einem CI-Setup prüfen wir, ob der Output-Pfad geschrieben werden soll.
|
||||||
|
import matplotlib.pyplot as plt
|
||||||
|
|
||||||
|
plt.savefig(output_path)
|
||||||
|
print(f"Scatter-Plot erfolgreich gespeichert unter: {output_path}")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
Loading…
Reference in a new issue