"""
Data export functionality for FUSION simulator.
This module provides unified data export capabilities supporting multiple formats
and export destinations.
"""
import csv
import json
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Any
class BaseExporter(ABC):
"""Abstract base class for data exporters."""
def __str__(self) -> str:
"""String representation of exporter."""
return self.__class__.__name__
@abstractmethod
def export(self, data: Any, output_path: Path) -> None:
"""Export data to the specified path."""
class JSONExporter(BaseExporter):
"""Export data to JSON format."""
def __str__(self) -> str:
"""String representation of JSONExporter."""
return "JSONExporter"
def export(self, data: Any, output_path: Path) -> None:
"""Export data as JSON."""
output_path.parent.mkdir(parents=True, exist_ok=True)
with output_path.open("w", encoding="utf-8") as f:
json.dump(data, f, indent=2, default=str)
class CSVExporter(BaseExporter):
"""Export data to CSV format."""
def __str__(self) -> str:
"""String representation of CSVExporter."""
return "CSVExporter"
def export(self, data: list[dict], output_path: Path) -> None:
"""Export list of dictionaries as CSV."""
if not data:
return
output_path.parent.mkdir(parents=True, exist_ok=True)
# Get all unique keys from all dictionaries
fieldnames_set: set[str] = set()
for row in data:
fieldnames_set.update(row.keys())
fieldnames = sorted(fieldnames_set)
with output_path.open("w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(data)
[docs]
class ExporterRegistry:
"""Registry for data exporters."""
[docs]
def __init__(self) -> None:
self._exporters = {
"json": JSONExporter(),
"csv": CSVExporter(),
}
[docs]
def register_exporter(self, format_name: str, exporter: BaseExporter) -> None:
"""Register a new exporter."""
self._exporters[format_name] = exporter
[docs]
def get_exporter(self, format_name: str) -> BaseExporter:
"""Get an exporter by format name."""
if format_name not in self._exporters:
raise ValueError(f"Unknown export format: {format_name}")
exporter: BaseExporter = self._exporters[format_name]
return exporter
@property
def supported_formats(self) -> list[str]:
"""Get list of supported export formats."""
return list(self._exporters.keys())
[docs]
class SimulationDataExporter:
"""Main data export interface for simulation data."""
[docs]
def __init__(self) -> None:
self.registry = ExporterRegistry()
[docs]
def export_topology(self, topology_data: dict, output_path: str | Path, output_format: str = "json") -> None:
"""Export network topology data."""
output_path = Path(output_path)
exporter = self.registry.get_exporter(output_format)
exporter.export(topology_data, output_path)
[docs]
def export_results(self, results_data: dict, output_path: str | Path, output_format: str = "json") -> None:
"""Export simulation results data."""
output_path = Path(output_path)
exporter = self.registry.get_exporter(output_format)
exporter.export(results_data, output_path)
[docs]
def export_metrics(
self,
metrics_data: dict | list[dict],
output_path: str | Path,
output_format: str = "csv",
) -> None:
"""Export simulation metrics data."""
output_path = Path(output_path)
exporter = self.registry.get_exporter(output_format)
# Convert dict to list of dicts for CSV export
if output_format == "csv" and isinstance(metrics_data, dict):
metrics_data = [{"metric": k, "value": v} for k, v in metrics_data.items()]
exporter.export(metrics_data, output_path)