Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 67 additions & 1 deletion src/supy/data_model/core/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -2192,6 +2192,67 @@ def validate_hourly_profile_hours(self) -> "SUEWSConfig":

return self

@classmethod
def _transform_validation_error(
cls, error: ValidationError, config_data: dict
) -> ValidationError:
"""Transform Pydantic validation errors to use GRIDID instead of array indices.

Uses structured error data to avoid string replacement collisions when
GRIDID values overlap with array indices (e.g., site 0 has GRIDID=1).
"""

# Extract GRIDID mapping from sites
sites = config_data.get("sites", [])
site_gridid_map = {}
for idx, site in enumerate(sites):
if isinstance(site, dict):
gridiv = site.get("gridiv")
if isinstance(gridiv, dict) and "value" in gridiv:
site_gridid_map[idx] = gridiv["value"]
elif gridiv is not None:
site_gridid_map[idx] = gridiv
else:
site_gridid_map[idx] = idx # Fallback to index
else:
site_gridid_map[idx] = idx # Fallback to index

# Process structured errors (not string manipulation!)
modified_errors = []
for err in error.errors():
err_copy = err.copy()
loc_list = list(err_copy["loc"])

# Replace numeric site index with GRIDID in location tuple
if (
len(loc_list) >= 2
and loc_list[0] == "sites"
and isinstance(loc_list[1], int)
):
site_idx = loc_list[1]
if site_idx in site_gridid_map:
loc_list[1] = site_gridid_map[site_idx]

err_copy["loc"] = tuple(loc_list)
modified_errors.append(err_copy)

# Format into readable message
error_lines = [
f"{error.error_count()} validation error{'s' if error.error_count() > 1 else ''} for SUEWSConfig"
]

for err in modified_errors:
loc_str = ".".join(str(x) for x in err["loc"])
error_lines.append(loc_str)
error_lines.append(
f" {err['msg']} [type={err['type']}, input_value={err['input']}, input_type={type(err['input']).__name__}]"
)
if "url" in err:
error_lines.append(f" For further information visit {err['url']}")

error_msg = "\n".join(error_lines)
raise ValueError(f"SUEWS Configuration Validation Error:\n{error_msg}")

@classmethod
def from_yaml(
cls,
Expand Down Expand Up @@ -2240,7 +2301,12 @@ def from_yaml(
logger_supy.info(
"Running comprehensive Pydantic validation with conditional checks."
)
return cls(**config_data)
try:
return cls(**config_data)
except ValidationError as e:
# Transform Pydantic validation error messages to use GRIDID instead of array indices
transformed_error = cls._transform_validation_error(e, config_data)
raise transformed_error
else:
logger_supy.info("Validation disabled by user. Loading without checks.")
return cls.model_construct(**config_data)
Expand Down
16 changes: 15 additions & 1 deletion src/supy/data_model/validation/pipeline/orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -167,7 +167,21 @@ def parameter_exists_in_standard(param_path: str, standard_data: dict) -> bool:
for i, (orig_item, proc_item) in enumerate(
zip(original_value, processed_value)
):
list_path = f"{current_path}[{i}]"
# Use GRIDID for sites array instead of numeric index
if (
current_path == "sites"
and isinstance(orig_item, dict)
and "gridiv" in orig_item
):
gridiv = orig_item["gridiv"]
# Handle RefValue objects
if isinstance(gridiv, dict) and "value" in gridiv:
gridid = gridiv["value"]
else:
gridid = gridiv
list_path = f"{current_path}.{gridid}"
else:
list_path = f"{current_path}[{i}]"
nested_critical, nested_defaults = detect_pydantic_defaults(
orig_item, proc_item, list_path, standard_data
)
Expand Down
Loading
Loading