Convert chipflow.toml parsing to pydantic

robtaylor · robtaylor · commit 5453d235978d · 2025-03-16T11:44:11.000Z
diff --git a/chipflow_lib/__init__.py b/chipflow_lib/__init__.py
@@ -1,8 +1,9 @@
 import importlib.metadata
-import jsonschema
 import os
 import sys
 import tomli
+from pathlib import Path
+from pydantic import ValidationError
 
 __version__ = importlib.metadata.version("chipflow_lib")
 
@@ -32,117 +33,31 @@ def _ensure_chipflow_root():
     return os.environ["CHIPFLOW_ROOT"]
 
 
-# TODO: convert to pydantic, one truth of source for the schema
-config_schema = {
-    "$schema": "https://json-schema.org/draft/2020-12/schema",
-    "$id": "https://chipflow.io/meta/chipflow.toml.schema.json",
-    "title": "chipflow.toml",
-    "type": "object",
-    "required": [
-        "chipflow"
-    ],
-    "properties": {
-        "chipflow": {
-            "type": "object",
-            "required": [
-                "steps",
-                "silicon"
-            ],
-            "additionalProperties": False,
-            "properties": {
-                "project_name": {
-                    "type": "string",
-                },
-                "top": {
-                    "type": "object",
-                },
-                "steps": {
-                    "type": "object",
-                },
-                "clocks": {
-                    "type": "object",
-                    "patternPropertues": {
-                        ".+": {"type": "string"}
-                    },
-                },
-                "resets": {
-                    "type": "object",
-                    "patternPropertues": {
-                        ".+": {"type": "string"}
-                    },
-                },
-                "silicon": {
-                    "type": "object",
-                    "required": [
-                        "processes",
-                        "package",
-                    ],
-                    "additionalProperties": False,
-                    "properties": {
-                        "processes": {
-                            "type": "array",
-                            "items": {
-                                "type": "string",
-                                "enum": ["sky130", "gf180", "customer1", "gf130bcd", "ihp_sg13g2"]
-                            }
-                        },
-                        "package": {
-                            "enum": ["caravel", "cf20", "pga144"]
-                        },
-                        "pads": {"$ref": "#/$defs/pin"},
-                        "power": {"$ref": "#/$defs/pin"},
-                        "debug": {
-                            "type": "object",
-                            "properties": {
-                                "heartbeat": {"type": "boolean"}
-                            }
-                        }
-                    },
-                },
-            },
-        },
-    },
-    "$defs": {
-        "pin": {
-            "type": "object",
-            "additionalProperties": False,
-            "minProperties": 1,
-            "patternProperties": {
-                ".+": {
-                    "type": "object",
-                    "required": [
-                        "type",
-                        "loc",
-                    ],
-                    "additionalProperties": False,
-                    "properties": {
-                        "type": {
-                            "enum": ["io", "i", "o", "oe", "clock", "reset", "power", "ground"]
-                        },
-                        "loc": {
-                            "type": "string",
-                            "pattern": "^[NSWE]?[0-9]+$"
-                        },
-                    }
-                }
-            }
-        }
-    }
-}
-
-
 def _parse_config():
+    """Parse the chipflow.toml configuration file."""
     chipflow_root = _ensure_chipflow_root()
-    config_file = f"{chipflow_root}/chipflow.toml"
+    config_file = Path(chipflow_root) / "chipflow.toml"
     return _parse_config_file(config_file)
 
 
 def _parse_config_file(config_file):
+    """Parse a specific chipflow.toml configuration file."""
+    from .config_models import Config
+    
     with open(config_file, "rb") as f:
         config_dict = tomli.load(f)
 
     try:
-        jsonschema.validate(config_dict, config_schema)
-        return config_dict
-    except jsonschema.ValidationError as e:
-        raise ChipFlowError(f"Syntax error in `chipflow.toml` at `{'.'.join(e.path)}`: {e.message}")
+        # Validate with Pydantic
+        config = Config.model_validate(config_dict)
+        return config_dict  # Return the original dict for backward compatibility
+    except ValidationError as e:
+        # Format Pydantic validation errors in a user-friendly way
+        error_messages = []
+        for error in e.errors():
+            location = ".".join(str(loc) for loc in error["loc"])
+            message = error["msg"]
+            error_messages.append(f"Error at '{location}': {message}")
+        
+        error_str = "\n".join(error_messages)
+        raise ChipFlowError(f"Validation error in chipflow.toml:\n{error_str}")
diff --git a/chipflow_lib/config_models.py b/chipflow_lib/config_models.py
@@ -0,0 +1,79 @@
+# SPDX-License-Identifier: BSD-2-Clause
+import enum
+import re
+from typing import Dict, List, Optional, Union, Literal, Any
+
+from pydantic import BaseModel, Field, model_validator, ValidationInfo, field_validator
+
+from .platforms.utils import Process
+
+
+class PadConfig(BaseModel):
+    """Configuration for a pad in chipflow.toml."""
+    type: Literal["io", "i", "o", "oe", "clock", "reset", "power", "ground"]
+    loc: str
+
+    @model_validator(mode="after")
+    def validate_loc_format(self):
+        """Validate that the location is in the correct format."""
+        if not re.match(r"^[NSWE]?[0-9]+$", self.loc):
+            raise ValueError(f"Invalid location format: {self.loc}, expected format: [NSWE]?[0-9]+")
+        return self
+    
+    @classmethod
+    def validate_pad_dict(cls, v: dict, info: ValidationInfo):
+        """Custom validation for pad dicts from TOML that may not have all fields."""
+        if isinstance(v, dict):
+            # Handle legacy format - if 'type' is missing but should be inferred from context
+            if 'loc' in v and 'type' not in v:
+                if info.field_name == 'power':
+                    v['type'] = 'power'
+                    
+            # Map legacy 'clk' type to 'clock' to match our enum
+            if 'type' in v and v['type'] == 'clk':
+                v['type'] = 'clock'
+                
+            return v
+        return v
+
+
+class SiliconConfig(BaseModel):
+    """Configuration for silicon in chipflow.toml."""
+    processes: List[Process]
+    package: Literal["caravel", "cf20", "pga144"]
+    pads: Dict[str, PadConfig] = {}
+    power: Dict[str, PadConfig] = {}
+    debug: Optional[Dict[str, bool]] = None
+    
+    @field_validator('pads', 'power', mode='before')
+    @classmethod
+    def validate_pad_dicts(cls, v, info: ValidationInfo):
+        """Pre-process pad dictionaries to handle legacy format."""
+        if isinstance(v, dict):
+            result = {}
+            for key, pad_dict in v.items():
+                # Apply the pad validator with context about which field we're in
+                validated_pad = PadConfig.validate_pad_dict(pad_dict, info)
+                result[key] = validated_pad
+            return result
+        return v
+
+
+class StepsConfig(BaseModel):
+    """Configuration for steps in chipflow.toml."""
+    silicon: str
+
+
+class ChipFlowConfig(BaseModel):
+    """Root configuration for chipflow.toml."""
+    project_name: Optional[str] = None
+    top: Dict[str, Any] = {}
+    steps: StepsConfig
+    silicon: SiliconConfig
+    clocks: Optional[Dict[str, str]] = None
+    resets: Optional[Dict[str, str]] = None
+
+
+class Config(BaseModel):
+    """Root configuration model for chipflow.toml."""
+    chipflow: ChipFlowConfig
diff --git a/chipflow_lib/pin_lock.py b/chipflow_lib/pin_lock.py
@@ -8,7 +8,8 @@
 
 from chipflow_lib import _parse_config, ChipFlowError
 from chipflow_lib.platforms import PACKAGE_DEFINITIONS, PIN_ANNOTATION_SCHEMA, top_interfaces
-from chipflow_lib.platforms.utils import LockFile, Package, PortMap, Port
+from chipflow_lib.platforms.utils import LockFile, Package, PortMap, Port, Process
+from chipflow_lib.config_models import Config
 
 # logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
 logger = logging.getLogger(__name__)
@@ -76,7 +77,13 @@ def allocate_pins(name: str, member: Dict[str, Any], pins: List[str], port_name:
 
 
 def lock_pins() -> None:
-    config = _parse_config()
+    # Get the config as dict for backward compatibility with top_interfaces
+    config_dict = _parse_config()
+    
+    # Parse with Pydantic for type checking and strong typing
+    from chipflow_lib.config_models import Config
+    config_model = Config.model_validate(config_dict)
+    
     used_pins = set()
     oldlock = None
 
@@ -86,35 +93,46 @@ def lock_pins() -> None:
         oldlock = LockFile.model_validate_json(json_string)
 
     print(f"Locking pins: {'using pins.lock' if lockfile.exists() else ''}")
-    processes = config["chipflow"]["silicon"]["process"]
-    package_name = config["chipflow"]["silicon"]["package"]
+    
+    # Use strongly-typed Pydantic model for processes
+    processes = config_model.chipflow.silicon.processes
+    package_name = config_model.chipflow.silicon.package
 
     if package_name not in PACKAGE_DEFINITIONS:
         logger.debug(f"Package '{package_name} is unknown")
     package_type = PACKAGE_DEFINITIONS[package_name]
 
     package = Package(package_type=package_type)
+    
+    # Process pads and power configurations using Pydantic models
     for d in ("pads", "power"):
         logger.debug(f"Checking [chipflow.silicon.{d}]:")
-        _map = {}
-        for k, v in config["chipflow"]["silicon"][d].items():
-            pin = str(v['loc'])
+        silicon_config = getattr(config_model.chipflow.silicon, d, {})
+        for k, v in silicon_config.items():
+            pin = str(v.loc)
             used_pins.add(pin)
-            port = oldlock.package.check_pad(k, v) if oldlock else None
+            
+            # Convert Pydantic model to dict for backward compatibility
+            v_dict = {"type": v.type, "loc": v.loc}
+            port = oldlock.package.check_pad(k, v_dict) if oldlock else None
+            
             if port and port.pins != [pin]:
                 raise ChipFlowError(
                     f"chipflow.toml conflicts with pins.lock: "
                     f"{k} had pin {port.pins}, now {[pin]}."
                 )
-            package.add_pad(k, v)
+                
+            # Add pad to package
+            package.add_pad(k, v_dict)
 
     logger.debug(f'Pins in use: {package_type.sortpins(used_pins)}')
 
     unallocated = package_type.pins - used_pins
 
     logger.debug(f"unallocated pins = {package_type.sortpins(unallocated)}")
 
-    _, interfaces = top_interfaces(config)
+    # Use the raw dict for top_interfaces since it expects the legacy format
+    _, interfaces = top_interfaces(config_dict)
 
     logger.debug(f"All interfaces:\n{pformat(interfaces)}")
 
diff --git a/chipflow_lib/steps/silicon.py b/chipflow_lib/steps/silicon.py
@@ -60,8 +60,12 @@ class SiliconStep:
     """Prepare and submit the design for an ASIC."""
     def __init__(self, config):
         self.config = config
-        self.project_name = config["chipflow"].get("project_name")
-        self.silicon_config = config["chipflow"]["silicon"]
+        
+        # Also parse with Pydantic for type checking and better code structure
+        from chipflow_lib.config_models import Config
+        self.config_model = Config.model_validate(config)
+        self.project_name = self.config_model.chipflow.project_name
+        self.silicon_config = config["chipflow"]["silicon"]  # Keep for backward compatibility
         self.platform = SiliconPlatform(config)
 
     def build_cli_parser(self, parser):
@@ -96,7 +100,7 @@ def prepare(self):
 
         Returns the path to the RTLIL file.
         """
-        return self.platform.build(SiliconTop(self.config), name=self.config["chipflow"]["project_name"])
+        return self.platform.build(SiliconTop(self.config), name=self.config_model.chipflow.project_name)
 
     def submit(self, rtlil_path, *, dry_run=False):
         """Submit the design to the ChipFlow cloud builder.
@@ -149,13 +153,15 @@ def submit(self, rtlil_path, *, dry_run=False):
                              f"dir={port.direction}, width={width}")
                 pads[padname] = {'loc': port.pins[0], 'type': port.direction.value}
  
+        # Use the Pydantic models to access configuration data
+        silicon_model = self.config_model.chipflow.silicon
         config = {
             "dependency_versions": dep_versions,
             "silicon": {
-                "process": self.silicon_config["processes"][0],
-                "pad_ring": self.silicon_config["package"],
+                "process": str(silicon_model.processes[0]),
+                "pad_ring": silicon_model.package,
                 "pads": pads,
-                "power": self.silicon_config.get("power", {})
+                "power": {k: {"type": v.type, "loc": v.loc} for k, v in silicon_model.power.items()}
             }
         }
         if dry_run:
diff --git a/tests/test_config_models.py b/tests/test_config_models.py
diff --git a/tests/test_parse_config.py b/tests/test_parse_config.py
