Config Loader¶

def load_from_dict(self, config: dict[str, Any]) -> Diagram:
    """
    Load a diagram from a configuration dictionary.

    Expected structure:
    {
        "title": "My Diagram",
        "board": "raspberry_pi_5" or {"name": "...", ...},
        "devices": [
            {"type": "bh1750", "name": "Light Sensor"},
            {"name": "Custom Device", "pins": [...], ...}
        ],
        "connections": [
            {"board_pin": 1, "device": "Light Sensor", "device_pin": "VCC"},
            ...
        ]
    }

    Args:
        config: Configuration dictionary

    Returns:
        Diagram object

    Raises:
        ValueError: If configuration fails schema validation
    """
    # Check for None or non-dict config
    if config is None or not isinstance(config, dict):
        log.error("invalid_config_type", config_type=type(config).__name__)
        raise ValueError(
            "Configuration validation failed:\n"
            "  • Config must be a dictionary with required fields "
            "(title, board, devices, connections)"
        )

    # Validate configuration against schema
    try:
        validated_config = validate_config(config)
        log.debug(
            "config_schema_validated",
            title=validated_config.title,
            device_count=len(validated_config.devices),
            connection_count=len(validated_config.connections),
        )
    except ValidationError as e:
        # Format validation errors for better readability
        error_messages = []
        for error in e.errors():
            field_path = " -> ".join(str(loc) for loc in error["loc"])
            error_messages.append(f"  • {field_path}: {error['msg']}")

        log.error(
            "config_schema_validation_failed",
            error_count=len(error_messages),
            errors=error_messages,
        )
        raise ValueError(
            "Configuration validation failed:\n" + "\n".join(error_messages)
        ) from e

    # Load board
    board_config = validated_config.board
    if isinstance(board_config, str):
        board = self._load_board_by_name(board_config)
        log.debug("board_loaded", board_name=board.name)
    else:
        log.error("custom_board_not_supported")
        raise ValueError("Custom board definitions not yet supported")

    # Load devices
    device_configs = validated_config.devices
    diagram_devices = []

    log.debug("loading_devices", device_count=len(device_configs))
    for dev_config in device_configs:
        device = self._load_device(dev_config)
        diagram_devices.append(device)
        log.debug(
            "device_loaded",
            device_name=device.name,
            pin_count=len(device.pins),
            device_type=dev_config.get("type", "custom"),
        )

    # Load connections with smart pin assignment
    connection_configs = [
        connection.model_dump(by_alias=True, exclude_none=True)
        for connection in validated_config.connections
    ]
    connections = []

    # Create pin assigner for automatic role-based pin distribution
    pin_assigner = PinAssigner(board)

    log.debug("loading_connections", connection_count=len(connection_configs))
    for conn_config in connection_configs:
        connection = self._load_connection(conn_config, pin_assigner)
        connections.append(connection)

    # Validate graph structure
    log.debug("validating_graph_structure")
    graph = ConnectionGraph(diagram_devices, connections)
    validation_issues = self._validate_graph(graph, diagram_devices, connections)

    # Check for critical errors
    errors = [issue for issue in validation_issues if issue.level == ValidationLevel.ERROR]
    warnings = [issue for issue in validation_issues if issue.level == ValidationLevel.WARNING]

    if errors:
        if self.emit_validation_output:
            self._report_validation_errors(errors)
        log.error("config_validation_failed", error_count=len(errors))
        raise ValueError("Configuration has critical errors")

    if warnings and self.emit_validation_output:
        self._report_validation_warnings(warnings)

    # Parse theme
    theme_str = validated_config.theme
    try:
        theme = Theme(theme_str.lower())
    except ValueError:
        log.warning("invalid_theme", theme=theme_str, using_default="light")
        theme = Theme.LIGHT

    # Create diagram via the shared builder so all entry points assemble the
    # same core model and presentation options.
    options = DiagramOptions(
        show_legend=validated_config.show_legend,
        show_gpio_diagram=validated_config.show_gpio_diagram,
        show_title=validated_config.show_title,
        show_board_name=validated_config.show_board_name,
        theme=theme,
    )
    diagram = (
        DiagramBuilder(self._board_selection_strategy)
        .with_title(validated_config.title)
        .with_board(board)
        .with_devices(diagram_devices)
        .with_connections(connections)
        .with_options(options)
        .build()
    )

    log.info(
        "diagram_config_loaded",
        title=diagram.title,
        board=board.name,
        device_count=len(diagram_devices),
        connection_count=len(connections),
    )

    return diagram

def load_from_file(self, config_path: str | Path) -> Diagram:
    """
    Load a diagram from a YAML or JSON configuration file.

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

    Returns:
        Diagram object

    Raises:
        ValueError: If file format is not supported or config is invalid
    """
    path = Path(config_path)

    log.debug("loading_config_file", config_path=str(path), format=path.suffix)

    if not path.exists():
        log.error("config_file_not_found", config_path=str(path))
        raise FileNotFoundError(f"Configuration file not found: {config_path}")

    # Validate file size to prevent memory issues
    file_size = path.stat().st_size
    if file_size > MAX_CONFIG_FILE_SIZE:
        log.error(
            "config_file_too_large",
            config_path=str(path),
            size_bytes=file_size,
            max_bytes=MAX_CONFIG_FILE_SIZE,
        )
        max_mb = MAX_CONFIG_FILE_SIZE // (1024 * 1024)
        size_mb = file_size / (1024 * 1024)
        raise ValueError(
            format_config_error(
                "file_too_large",
                detail=f"{size_mb:.1f}MB (max: {max_mb}MB)",
            )
        )

    # Load file based on extension
    if path.suffix in [".yaml", ".yml"]:
        with open(path) as f:
            config = yaml.safe_load(f)
        log.debug("yaml_config_parsed", config_path=str(path))
    elif path.suffix == ".json":
        with open(path) as f:
            config = json.load(f)
        log.debug("json_config_parsed", config_path=str(path))
    else:
        log.error("unsupported_file_format", format=path.suffix, config_path=str(path))
        raise ValueError(
            format_config_error(
                "invalid_yaml" if path.suffix in [".yaml", ".yml"] else "invalid_json",
                detail=f"Unsupported file extension: {path.suffix}",
            )
        )

    return self.load_from_dict(config)

def assign_pin(self, role: str | PinRole) -> int:
    """
    Assign next available pin of the specified role.

    Uses round-robin distribution to spread connections across multiple
    pins of the same role.

    Args:
        role: Pin role (e.g., "GND", "3V3") as string or PinRole enum

    Returns:
        Physical pin number

    Raises:
        ValueError: If no pins of the specified role are available
    """
    # Convert string to PinRole
    if isinstance(role, str):
        try:
            pin_role = PinRole(role.upper())
        except ValueError:
            # Try without upper() in case it's already correct case
            try:
                pin_role = PinRole(role)
            except ValueError as e:
                raise ValueError(
                    f"Invalid pin role '{role}'. Must be one of: "
                    f"{', '.join(r.value for r in PinRole)}"
                ) from e
    else:
        pin_role = role

    # Check if this role exists on the board
    if pin_role not in self._pins_by_role:
        available_roles = sorted(r.value for r in self._pins_by_role)
        raise ValueError(
            f"Board '{self.board.name}' has no pins with role '{pin_role.value}'. "
            f"Available roles: {', '.join(available_roles)}"
        )

    available_pins = self._pins_by_role[pin_role]

    # Get current index for this role (default to 0)
    current_index = self._role_assignment_index.get(pin_role, 0)

    # Round-robin: cycle through available pins
    assigned_pin = available_pins[current_index % len(available_pins)]

    # Update index for next assignment
    self._role_assignment_index[pin_role] = current_index + 1

    log.debug(
        "pin_assigned",
        role=pin_role.value,
        assigned_pin=assigned_pin,
        available_count=len(available_pins),
        assignment_index=current_index + 1,
    )

    return assigned_pin