Skip to content

I/O API

Configuration loading and results export.

Overview

from phased_array_systems.io import (
    load_config,
    export_results,
)

Functions

load_config

load_config(path: str | Path) -> StudyConfig

Load a study configuration from a YAML or JSON file.

PARAMETER DESCRIPTION
path

Path to configuration file (.yaml, .yml, or .json)

TYPE: str | Path

RETURNS DESCRIPTION
StudyConfig

Validated StudyConfig object

RAISES DESCRIPTION
FileNotFoundError

If config file doesn't exist

ValueError

If file format is not supported

ValidationError

If config validation fails

Source code in src/phased_array_systems/io/config_loader.py
def load_config(path: str | Path) -> StudyConfig:
    """Load a study configuration from a YAML or JSON file.

    Args:
        path: Path to configuration file (.yaml, .yml, or .json)

    Returns:
        Validated StudyConfig object

    Raises:
        FileNotFoundError: If config file doesn't exist
        ValueError: If file format is not supported
        ValidationError: If config validation fails
    """
    path = Path(path)

    if not path.exists():
        raise FileNotFoundError(f"Configuration file not found: {path}")

    suffix = path.suffix.lower()

    if suffix in (".yaml", ".yml"):
        with open(path, encoding="utf-8") as f:
            data = yaml.safe_load(f)
    elif suffix == ".json":
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
    else:
        raise ValueError(f"Unsupported config format: {suffix}. Use .yaml, .yml, or .json")

    return StudyConfig.model_validate(data)

export_results

export_results(results: DataFrame, path: str | Path, format: Literal['parquet', 'csv', 'json'] | None = None, include_metadata: bool = True) -> Path

Export evaluation results to file.

PARAMETER DESCRIPTION
results

DataFrame with evaluation results

TYPE: DataFrame

path

Output file path

TYPE: str | Path

format

Output format (auto-detected from extension if None)

TYPE: Literal['parquet', 'csv', 'json'] | None DEFAULT: None

include_metadata

Include export metadata (timestamp, version)

TYPE: bool DEFAULT: True

RETURNS DESCRIPTION
Path

Path to exported file

Source code in src/phased_array_systems/io/exporters.py
def export_results(
    results: pd.DataFrame,
    path: str | Path,
    format: Literal["parquet", "csv", "json"] | None = None,
    include_metadata: bool = True,
) -> Path:
    """Export evaluation results to file.

    Args:
        results: DataFrame with evaluation results
        path: Output file path
        format: Output format (auto-detected from extension if None)
        include_metadata: Include export metadata (timestamp, version)

    Returns:
        Path to exported file
    """
    path = Path(path)

    if format is None:
        suffix = path.suffix.lower()
        if suffix == ".parquet":
            format = "parquet"
        elif suffix == ".csv":
            format = "csv"
        elif suffix == ".json":
            format = "json"
        else:
            format = "parquet"  # Default

    # Ensure parent directory exists
    path.parent.mkdir(parents=True, exist_ok=True)

    if format == "parquet":
        _export_parquet(results, path, include_metadata)
    elif format == "csv":
        _export_csv(results, path)
    elif format == "json":
        _export_json(results, path, include_metadata)
    else:
        raise ValueError(f"Unknown format: {format}")

    return path

Usage Examples

Loading Configuration

from phased_array_systems.io import load_config

# Load from YAML
config = load_config("config.yaml")

# Load from JSON
config = load_config("config.json")

# Access components
arch = config.get_architecture()
scenario = config.get_scenario()
requirements = config.get_requirement_set()

Configuration Schema

# config.yaml
name: "My Trade Study"

architecture:
  array:
    geometry: rectangular
    nx: 8
    ny: 8
    dx_lambda: 0.5
    dy_lambda: 0.5
    scan_limit_deg: 60.0
  rf:
    tx_power_w_per_elem: 1.0
    pa_efficiency: 0.3
    noise_figure_db: 3.0
  cost:
    cost_per_elem_usd: 100.0
    nre_usd: 10000.0

scenario:
  type: comms
  freq_hz: 10.0e9
  bandwidth_hz: 10.0e6
  range_m: 100.0e3
  required_snr_db: 10.0

requirements:
  - id: REQ-001
    name: Minimum EIRP
    metric_key: eirp_dbw
    op: ">="
    value: 40.0
    severity: must

design_space:
  variables:
    - name: array.nx
      type: categorical
      values: [4, 8, 16]
    - name: rf.tx_power_w_per_elem
      type: float
      low: 0.5
      high: 3.0

Exporting Results

from phased_array_systems.io import export_results
import pandas as pd

# Export to Parquet (recommended for large datasets)
export_results(results, "results.parquet")

# Export to CSV
export_results(results, "results.csv", format="csv")

# Loading back
results_loaded = pd.read_parquet("results.parquet")

Format Options

Format Extension Best For
Parquet .parquet Large datasets, fast I/O
CSV .csv Human-readable, Excel

Parquet Benefits

  • Compressed storage
  • Faster read/write
  • Preserves data types
  • Column-oriented (efficient for analysis)
# Parquet is ~5x smaller and ~10x faster for typical results
export_results(results, "results.parquet")  # Recommended

# CSV for compatibility
export_results(results, "results.csv", format="csv")

Config Object

StudyConfig

Bases: BaseModel

Top-level configuration for a study.

Supports both single-case evaluation and DOE trade studies.

Example YAML
name: "Comms Array Study"
architecture:
  array:
    nx: 8
    ny: 8
    dx_lambda: 0.5
  rf:
    tx_power_w_per_elem: 1.0
scenario:
  type: comms
  freq_hz: 10e9
  bandwidth_hz: 10e6
  range_m: 100e3
  required_snr_db: 10.0
requirements:
  - id: REQ-001
    name: Minimum EIRP
    metric_key: eirp_dbw
    op: ">="
    value: 40.0

get_architecture

get_architecture() -> Architecture

Get the Architecture object, building from shorthand if needed.

Source code in src/phased_array_systems/io/schema.py
def get_architecture(self) -> Architecture:
    """Get the Architecture object, building from shorthand if needed."""
    if self.architecture is not None:
        return self.architecture

    # Build from shorthand configs
    array = self.array or ArrayConfig(nx=8, ny=8)
    rf = self.rf or RFChainConfig(tx_power_w_per_elem=1.0)
    cost = self.cost or CostConfig()

    return Architecture(array=array, rf=rf, cost=cost, name=self.name)

get_scenario

get_scenario() -> CommsLinkScenario | RadarDetectionScenario | None

Get the Scenario object from config.

Source code in src/phased_array_systems/io/schema.py
def get_scenario(self) -> CommsLinkScenario | RadarDetectionScenario | None:
    """Get the Scenario object from config."""
    if self.scenario is None:
        return None

    scenario_dict = self.scenario.copy()
    scenario_type = scenario_dict.pop("type", "comms")

    if scenario_type == "comms":
        return CommsLinkScenario(**scenario_dict)
    elif scenario_type == "radar":
        return RadarDetectionScenario(**scenario_dict)
    else:
        raise ValueError(f"Unknown scenario type: {scenario_type}")

get_requirement_set

get_requirement_set() -> RequirementSet

Get RequirementSet from config.

Source code in src/phased_array_systems/io/schema.py
def get_requirement_set(self) -> RequirementSet:
    """Get RequirementSet from config."""
    req_set = RequirementSet(name=f"{self.name} Requirements")
    for req_config in self.requirements:
        req_set.add(req_config.to_requirement())
    return req_set

YAML Schema Reference

Architecture Section

architecture:
  array:
    geometry: rectangular  # rectangular, circular, triangular
    nx: 8                  # Required
    ny: 8                  # Required
    dx_lambda: 0.5         # Default: 0.5
    dy_lambda: 0.5         # Default: 0.5
    scan_limit_deg: 60.0   # Default: 60.0
    max_subarray_nx: 8     # Default: 8
    max_subarray_ny: 8     # Default: 8
    enforce_subarray_constraint: true  # Default: true

  rf:
    tx_power_w_per_elem: 1.0  # Required
    pa_efficiency: 0.3        # Default: 0.3
    noise_figure_db: 3.0      # Default: 3.0
    n_tx_beams: 1             # Default: 1
    feed_loss_db: 1.0         # Default: 1.0
    system_loss_db: 0.0       # Default: 0.0

  cost:
    cost_per_elem_usd: 100.0     # Default: 100.0
    nre_usd: 0.0                 # Default: 0.0
    integration_cost_usd: 0.0    # Default: 0.0

Scenario Section

# Communications scenario
scenario:
  type: comms
  freq_hz: 10.0e9          # Required
  bandwidth_hz: 10.0e6     # Required
  range_m: 100.0e3         # Required
  required_snr_db: 10.0    # Required
  scan_angle_deg: 0.0      # Default: 0.0
  rx_antenna_gain_db: 0.0  # Default: None (isotropic)
  rx_noise_temp_k: 290.0   # Default: 290.0
  atmospheric_loss_db: 0.0 # Default: 0.0
  rain_loss_db: 0.0        # Default: 0.0

# Radar scenario
scenario:
  type: radar
  freq_hz: 10.0e9          # Required
  target_rcs_m2: 1.0       # Required
  range_m: 100.0e3         # Required
  required_pd: 0.9         # Default: 0.9
  pfa: 1.0e-6              # Default: 1e-6
  pulse_width_s: 10.0e-6   # Required
  prf_hz: 1000             # Required
  n_pulses: 1              # Default: 1
  integration_type: coherent  # Default: coherent
  swerling_model: 1        # Default: 1

Requirements Section

requirements:
  - id: REQ-001
    name: Minimum EIRP
    metric_key: eirp_dbw
    op: ">="
    value: 40.0
    units: dBW             # Optional
    severity: must         # must, should, nice

  - id: REQ-002
    name: Maximum Cost
    metric_key: cost_usd
    op: "<="
    value: 50000.0
    severity: must

Design Space Section

design_space:
  variables:
    # Categorical
    - name: array.nx
      type: categorical
      values: [4, 8, 16]

    # Integer range
    - name: array.ny
      type: int
      low: 4
      high: 16

    # Float range
    - name: rf.tx_power_w_per_elem
      type: float
      low: 0.5
      high: 3.0

    # Fixed value (low == high)
    - name: array.dx_lambda
      type: float
      low: 0.5
      high: 0.5

See Also