cursus.validation.alignment.utils

Utilities and Models Module

This module contains utility functions, data models, and configuration classes for the alignment validation system. It provides common functionality used across all validation components.

Components: - alignment_utils.py: Core utility functions and helper classes - core_models.py: Core data models and type definitions - level3_validation_config.py: Level 3 validation configuration and modes - script_analysis_models.py: Models for script analysis results - utils.py: General utility functions and helpers

Utility Features: - Common data structures and enums - Validation configuration management - Helper functions for alignment operations - Type definitions and model classes - Configuration validation and defaults

class ValidationLevel(*values)

Bases: Enum

The 3 validation BOUNDARIES — what construction can’t self-check (FZ 31e1d3h / Phase D5).

The unified .step.yaml made Contract<->Spec alignment a construction-time Pydantic invariant (StepInterface._sync_and_align), so the former Level-2 (CONTRACT_SPEC=2) validated a tautology and has been removed. The three surviving levels are renamed in SEMANTICS to the 3 boundaries — SCRIPT_CONTRACT is the Script<->Interface fidelity boundary (B1), SPEC_DEPENDENCY is the cross-step DAG-resolvability boundary (B2, which now also owns the SageMaker property-path check folded in from the old L2), and BUILDER_CONFIG is the registry<->handler<->config binding boundary (B3). The member NAMES + integer VALUES are kept (1/3/4, non-contiguous) so ValidationLevel(1/3/4) coercion and every existing reference keep working; only CONTRACT_SPEC (value 2) is gone.

SCRIPT_CONTRACT = 1

SPEC_DEPENDENCY = 3

BUILDER_CONFIG = 4

class ValidationStatus(*values)

Bases: Enum

Status of validation operations.

PASSED = 'PASSED'

FAILED = 'FAILED'

EXCLUDED = 'EXCLUDED'

ERROR = 'ERROR'

ISSUES_FOUND = 'ISSUES_FOUND'

PASSED_WITH_WARNINGS = 'PASSED_WITH_WARNINGS'

NO_VALIDATOR = 'NO_VALIDATOR'

COMPLETED = 'COMPLETED'

class IssueLevel(*values)

Bases: Enum

Severity levels for validation issues.

ERROR = 'ERROR'

WARNING = 'WARNING'

INFO = 'INFO'

class RuleType(*values)

Bases: Enum

Types of validation rules.

UNIVERSAL = 'universal'

STEP_SPECIFIC = 'step_specific'

METHOD_INTERFACE = 'method_interface'

CONFIGURATION = 'configuration'

class ValidationIssue(level, message, method_name=None, rule_type=None, details=None, step_name=None, file_path=None, line_number=None)

Bases: object

Represents a validation issue found during alignment checking.

details: Dict[str, Any] | None = None

file_path: str | None = None

line_number: int | None = None

method_name: str | None = None

rule_type: RuleType | None = None

step_name: str | None = None

to_dict()

Convert issue to dictionary format.

level: IssueLevel

message: str

class ValidationResult(status, step_name, validation_level=None, issues=None, metadata=None, error_message=None)

Bases: object

Represents the result of a validation operation.

add_issue(issue)

Add a validation issue to the result.

property error_count: int

Count of error-level issues.

error_message: str | None = None

property info_count: int

Count of info-level issues.

issues: List[ValidationIssue] = None

metadata: Dict[str, Any] | None = None

to_dict()

Convert result to dictionary format.

property total_issues: int

Total count of all issues.

validation_level: ValidationLevel | None = None

property warning_count: int

Count of warning-level issues.

status: ValidationStatus

step_name: str

class ValidationSummary

Bases: object

Summary of validation results across multiple steps.

add_result(result)

Add a validation result to the summary.

property success_rate: float

Calculate success rate (passed / total non-excluded).

to_dict()

Convert summary to dictionary format.

class MethodValidationInfo(method_name, is_required, expected_signature=None, return_type=None, purpose=None, validation_rules=None)

Bases: object

Information about method validation requirements.

expected_signature: str | None = None

purpose: str | None = None

return_type: str | None = None

validation_rules: List[str] | None = None

method_name: str

is_required: bool

class StepValidationContext(step_name, step_type, builder_class_name=None, workspace_dirs=None, validation_config=None)

Bases: object

Context information for step validation.

builder_class_name: str | None = None

validation_config: Dict[str, Any] | None = None

workspace_dirs: List[str] | None = None

step_name: str

step_type: str

create_validation_issue(level, message, method_name=None, rule_type=None, details=None, **kwargs)

Create a validation issue with proper type conversion.

create_validation_result(status, step_name, validation_level=None, issues=None, **kwargs)

Create a validation result with proper type conversion.

merge_validation_results(results)

Merge multiple validation results into a single result.

filter_issues_by_level(issues, level)

Filter validation issues by severity level.

group_issues_by_method(issues)

Group validation issues by method name.

format_validation_summary(summary)

Format validation summary as a readable string.

FlexibleFileResolver

alias of FlexibleFileResolverAdapter

get_sagemaker_step_type(step_name, workspace_id=None)

Get SageMaker step type with workspace context.

get_canonical_name_from_file_name(file_name, workspace_id=None)

Enhanced file name resolution with workspace context awareness.

Parameters:

• file_name (str) – File-based name (e.g., “model_evaluation_xgb”, “dummy_training”)

• workspace_id (str) – Optional workspace context for resolution

Returns:

Canonical step name (e.g., “XGBoostModelEval”, “DummyTraining”)

Raises:

ValueError – If file name cannot be mapped to a canonical name

Return type:

str

class StepCatalog(workspace_dirs=None)

Bases: object

Unified step catalog addressing all validated user stories (US1-US5).

This single class consolidates the functionality of 16+ discovery systems while maintaining simple, efficient O(1) lookups through dictionary-based indexing.

PHASE 1 ENHANCEMENT: Now includes StepBuilderRegistry functionality: - Config-to-builder resolution - Legacy alias support - Pipeline construction interface - Enhanced registry integration

LEGACY_ALIASES = {'MIMSPackaging': 'Package', 'MIMSPayload': 'Payload', 'ModelRegistration': 'Registration', 'PytorchModel': 'PyTorchModel', 'PytorchTraining': 'PyTorchTraining'}

build_complete_config_classes(project_id=None)

Build complete mapping integrating manual registration with auto-discovery.

This addresses the TODO in the existing build_complete_config_classes() function.

Parameters:

project_id (str | None) – Optional project ID for workspace-specific discovery

Returns:

Complete dictionary of config classes (manual + auto-discovered)

Return type:

Dict[str, Type]

create_unified_specification(contract_name)

Create unified specification from multiple variants using smart selection.

This method delegates to SpecAutoDiscovery for smart specification handling, integrating SmartSpecificationSelector functionality: - Multi-variant specification discovery - Union of dependencies and outputs from all variants - Smart validation logic with detailed feedback - Primary specification selection (training > generic > first available)

Parameters:

contract_name (str) – Name of the contract to find specifications for

Returns:

Unified specification model with metadata

Return type:

Dict[str, Any]

detect_framework(step_name)

DETECTION: Detect ML framework for a step.

Parameters:

step_name (str) – Name of the step to analyze

Returns:

Framework name (e.g., ‘xgboost’, ‘pytorch’) or None if not detected

Return type:

str | None

discover_config_classes(project_id=None)

Auto-discover configuration classes from core and workspace directories.

Parameters:

project_id (str | None) – Optional project ID for workspace-specific discovery

Returns:

Dictionary mapping class names to class types

Return type:

Dict[str, Type]

discover_contracts_with_scripts()

DISCOVERY: Find all steps that have both contract and script components.

Returns:

List of step names that have both contract and script components

Return type:

List[str]

discover_cross_workspace_components(workspace_ids=None)

DISCOVERY: Find components across multiple workspaces.

Parameters:

workspace_ids (List[str] | None) – Optional list of workspace IDs to search (defaults to all)

Returns:

Dictionary mapping workspace IDs to lists of component identifiers

Return type:

Dict[str, List[str]]

discover_script_files(project_id=None)

Discover script files from package and workspaces with prioritization.

This method enables script discovery for interactive runtime testing by finding scripts referenced in config and contract entry points.

Parameters:

project_id (str | None) – Optional project ID for workspace-specific discovery

Returns:

Dictionary mapping script names to ScriptInfo objects

Return type:

Dict[str, Any]

discover_scripts_from_dag(dag)

Discover scripts referenced in a DAG with intelligent node-to-script mapping.

This method enables DAG-guided script discovery for interactive runtime testing by mapping DAG nodes to actual script files using step catalog intelligence.

Parameters:

dag – PipelineDAG object

Returns:

Dictionary mapping script names to ScriptInfo objects

Return type:

Dict[str, Any]

extract_parent_values_for_inheritance(target_config_class_name, completed_configs)

Extract parent values for inheritance from completed configs.

This method enables Smart Default Value Inheritance by extracting field values from the immediate parent config for pre-populating child config forms. It implements cascading inheritance to eliminate redundant user input.

Parameters:

• target_config_class_name (str) – Target config class name

• completed_configs (Dict[str, Any]) – Dictionary of completed config instances by class name

Returns:

Dictionary of field values from immediate parent config

Return type:

Dict[str, Any]

Example

completed_configs = {

“BasePipelineConfig”: base_config_instance, “ProcessingStepConfigBase”: processing_config_instance

}

parent_values = step_catalog.extract_parent_values_for_inheritance(

“TabularPreprocessingConfig”, completed_configs

) # Returns: ALL field values from ProcessingStepConfigBase # (which includes cascaded values from BasePipelineConfig) # Result: {“author”: “lukexie”, “bucket”: “my-bucket”, # “processing_instance_type”: “ml.m5.2xlarge”, …}

find_contracts_by_entry_point(entry_point)

Find contracts that reference a specific script entry point.

Parameters:

entry_point (str) – Script entry point (e.g., “model_evaluation_xgb.py”)

Returns:

Dictionary mapping contract names to contract instances

Return type:

Dict[str, Any]

find_specs_by_contract(contract_name)

Find all specifications that reference a specific contract.

This method enables contract-specification alignment validation by finding specifications that are associated with a given contract name.

Parameters:

contract_name (str) – Name of the contract to find specifications for

Returns:

Dictionary mapping spec names to specification instances

Return type:

Dict[str, Any]

find_step_by_component(component_path)

Find step name from any component file.

Parameters:

component_path (str) – Path to a component file

Returns:

Step name that owns the component, or None if not found

Return type:

str | None

get_all_builders()

Get all available builders with canonical names.

This method provides comprehensive builder discovery for dynamic testing without requiring hard-coded maintenance.

Returns:

Dict mapping canonical names to builder classes

Return type:

Dict[str, Type]

get_builder_class_path(step_name)

Get builder class path for a step using BuilderAutoDiscovery component.

Parameters:

step_name (str) – Name of the step

Returns:

Path to builder class or None if not found

Return type:

str | None

get_builder_for_config(config, node_name=None)

Map a config instance to a builder PROVIDER (the thing the assembler calls with its 5-kwarg signature to get a StepBuilderBase).

Returns a BuilderProvider (FZ 31e1d3g1 Phase 1): TODAY this is the per-step builder CLASS (a class is already such a callable), so behavior is unchanged. The annotation is provider-based so the classless Design-B end-state can return a non-class factory here without breaking the contract — callers must invoke the result, never assume it is a class.

Parameters:

• config – Configuration instance (BasePipelineConfig)

• node_name (str) – Optional DAG node name for context

Returns:

A builder provider (currently a builder class) or None if not found.

Return type:

Callable[[…], Any] | None

get_builder_for_step_type(step_type)

Get the builder PROVIDER for a step type, with legacy alias support.

Returns a BuilderProvider (FZ 31e1d3g1 Phase 1) — currently the builder class; see get_builder_for_config for the dual-mode contract.

Parameters:

step_type (str) – Step type name (may be legacy alias)

Returns:

A builder provider (currently a builder class) or None if not found.

Return type:

Callable[[…], Any] | None

get_builder_map()

Get a complete builder map for pipeline construction.

Returns:

Dictionary mapping step types to builder classes

Return type:

Dict[str, Type]

get_builders_by_step_type(step_type)

Get builders filtered by SageMaker step type.

This method enables step-type-specific testing by filtering builders based on their registered SageMaker step type.

Parameters:

step_type (str) – SageMaker step type (Processing, Training, Transform, CreateModel, etc.)

Returns:

Dict mapping canonical names to builder classes for the step type

Return type:

Dict[str, Type]

get_config_types_for_step_type(step_type)

Get possible config class names for a step type.

Parameters:

step_type (str) – Step type name

Returns:

List of possible configuration class names

Return type:

List[str]

get_contract_entry_points()

Get all contract entry points for validation.

Returns:

Dictionary mapping contract names to their entry points

Return type:

Dict[str, str]

get_immediate_parent_config_class(config_class_name)

Get the immediate parent config class name for inheritance.

This method enables Smart Default Value Inheritance by identifying which parent config class a specific config should inherit from. It implements cascading inheritance where TabularPreprocessingConfig inherits from ProcessingStepConfigBase (not BasePipelineConfig directly) to get all cascaded values.

Parameters:

config_class_name (str) – Target config class name (e.g., “TabularPreprocessingConfig”)

Returns:

Immediate parent class name (e.g., “ProcessingStepConfigBase”) or None if not found

Return type:

str | None

Example

parent = step_catalog.get_immediate_parent_config_class(“TabularPreprocessingConfig”) # Returns: “ProcessingStepConfigBase” (not “BasePipelineConfig”)

parent = step_catalog.get_immediate_parent_config_class(“CradleDataLoadConfig”) # Returns: “BasePipelineConfig” (direct inheritance)

get_job_type_variants(base_step_name)

Get all job_type variants for a base step name.

Parameters:

base_step_name (str) – Base name of the step

Returns:

List of job type variants found

Return type:

List[str]

get_metrics_report()

Get simple metrics report.

get_script_discovery_stats()

Get script discovery statistics.

This method provides discovery statistics for interactive runtime testing to help users understand the script discovery process.

Returns:

Dictionary with script discovery statistics

Return type:

Dict[str, Any]

get_script_info(script_name)

Get information about a script in dictionary format.

This method provides script information for interactive runtime testing in a user-friendly dictionary format.

Parameters:

script_name (str) – Name of the script

Returns:

Dictionary with script information or None if not found

Return type:

Dict[str, Any] | None

get_sdk_output_location(built_step, output_type=None)

Read the output S3 location(s) off a BUILT SDK-delegation step (FZ 31e1d3i).

A generic, DUCK-TYPED accessor for the MODSPredefinedProcessingStep family: any built SDK step that exposes get_output_locations (the SAIS contract) is supported — CradleDataLoadingStep returns a {DATA, SIGNATURE, METADATA} dict (or one entry when output_type is given), DataUploadingStep returns None (a SINK — output goes to Andes/EDX), and a step lacking the method (e.g. RedshiftDataLoadingStep) returns None. This REPLACES the per-step get_output_location / get_step_outputs helpers the cradle builder used to carry (so that builder becomes a pure shell). It reads a method off an ALREADY-CONSTRUCTED step instance, so it introduces NO offline SAIS import — the SAIS class was necessarily importable to build the step.

Parameters:

• built_step (Any) – A step object produced by an SDK-delegation builder’s create_step.

• output_type (str | None) – Optional SAIS output type (e.g. "DATA"); None returns all locations.

Returns:

The step’s output location(s) as the SDK reports them, or None if the step does not expose get_output_locations.

Return type:

Any

get_spec_job_type_variants(base_step_name)

Get all job type variants for a base step name from specifications.

This method discovers different job type variants (training, validation, testing, etc.) for a given base step name by examining specification file naming patterns.

Parameters:

base_step_name (str) – Base name of the step

Returns:

List of job type variants found

Return type:

List[str]

get_step_builder_suggestions(config_class_name)

Get suggestions for step builders based on config class name.

Parameters:

config_class_name (str) – Configuration class name

Returns:

List of suggested step type names

Return type:

List[str]

get_step_info(step_name, job_type=None)

Get complete information about a step, optionally with job_type variant.

Parameters:

• step_name (str) – Name of the step to retrieve

• job_type (str | None) – Optional job type variant (e.g., ‘training’, ‘validation’)

Returns:

StepInfo object with complete step information, or None if not found

Return type:

StepInfo | None

get_step_interface(step_name, job_type=None)

Load a step’s unified .step.yaml interface — the single canonical accessor.

This is THE one load function for interface data (FZ 31e1d3g3 follow-up): both load_contract_class (→ iface.contract) and load_spec_class (→ iface) are thin views over it, so the job_type-suffix fallback below is written once and every consumer inherits it.

On the first (exact) load miss, if job_type was not supplied, we resolve the node’s base step name ROBUSTLY (naming.resolve_base_step_name against the registry — validates the base against actual step names, not a hardcoded suffix list) and retry with the trailing segment as the job_type (ModelCalibration_calibration → ModelCalibration + job_type calibration; CradleDataLoading_munged → CradleDataLoading + munged). Because the base is validated against the registry, ANY suffix resolves (job_type is open now) while a base like XGBoostModel is never mis-stripped (no XGBoost step exists to strip to).

Parameters:

• step_name (str) – PascalCase step name (may be a Base_variant compound).

• job_type (str | None) – Optional job_type variant; when given, no suffix stripping is attempted.

Returns:

The validated StepInterface, or None if no interface resolves (never raises).

Return type:

Any | None

has_builder_provider(step_name)

Whether a step has a buildable builder, WITHOUT importing/instantiating it (FZ 31e1d3g3 D4).

This is a ROUTABILITY check answered from declarative data — the registry’s sagemaker_step_type (+ the .step.yaml patterns.step_assembly) must resolve to a real PatternHandler. It does NOT call load_builder_class (which would import the builder module — and fail for an SDK-bound builder offline even though the step is perfectly routable). So it stays True for a fileless-but-routable step (the factory-shell end-state) and for the SDK steps offline, and False only for genuinely-absent or non-routable rows (Base/Lambda/unknown). Used by the DAG resolver’s component-availability check, which must not encode the deleted-builder_*.py-file assumption.

Parameters:

step_name (str) – Canonical step name (job-type variants strip to the base name).

Returns:

True iff the step’s registry row routes to a construction handler.

Return type:

bool

is_step_type_supported(step_type)

Check if step type is supported (including legacy aliases).

Parameters:

step_type (str) – Step type name

Returns:

True if supported, False otherwise

Return type:

bool

list_available_scripts()

List all available script names.

This method enables script enumeration for interactive runtime testing by listing all discovered script names.

Returns:

List of script names that have been discovered

Return type:

List[str]

list_available_steps(workspace_id=None, job_type=None)

List all available concrete pipeline steps with deduplication.

Excludes base configuration steps (‘Base’, ‘Processing’) and applies canonical name deduplication following standardization rules.

Parameters:

• workspace_id (str | None) – Optional workspace filter

• job_type (str | None) – Optional job type filter

Returns:

List of concrete pipeline step names (PascalCase, canonical)

Return type:

List[str]

list_steps_with_scripts(workspace_id=None, job_type=None)

List all steps that have script components.

This method filters available steps to only return those that have script file components, which is useful for alignment validation and script-based testing frameworks.

Parameters:

• workspace_id (str | None) – Optional workspace filter

• job_type (str | None) – Optional job type filter

Returns:

List of step names that have script components

Return type:

List[str]

list_steps_with_specs(workspace_id=None, job_type=None)

List all steps that have specification components.

This method filters available steps to only return those that have specification file components, which is useful for validation frameworks that need to work specifically with steps that have specifications.

Parameters:

• workspace_id (str | None) – Optional workspace filter

• job_type (str | None) – Optional job type filter

Returns:

List of step names that have specification components

Return type:

List[str]

list_supported_step_types()

List all supported step types including legacy aliases.

Returns:

List of supported step type names

Return type:

List[str]

load_all_specifications()

Load all specification instances from both package and workspace directories.

This method provides comprehensive specification loading for validation frameworks and dependency analysis tools. It discovers and loads all available specifications, serializing them to dictionary format for easy consumption.

Returns:

Dictionary mapping step names to serialized specification dictionaries

Return type:

Dict[str, Dict[str, Any]]

load_builder_class(step_name)

Load builder class for a step with job type variant fallback support.

Parameters:

step_name (str) – Name of the step (may include job type variant)

Returns:

Builder class type or None if not found/loadable

Return type:

Type | None

load_contract_class(step_name)

Load the contract for a step — a VIEW onto its .step.yaml interface.

Parameters:

step_name (str) – Name of the step (PascalCase, e.g., “TabularPreprocessing”)

Returns:

ContractSection (drop-in for the legacy ScriptContract), or None if not found.

Return type:

Any | None

load_script_info(script_name)

Load script information for a specific script with workspace-aware discovery.

This method enables script information retrieval for interactive runtime testing with workspace prioritization support.

Parameters:

script_name (str) – Name of the script to load info for

Returns:

ScriptInfo object or None if not found

Return type:

Any | None

load_spec_class(step_name)

Load the specification for a step — the .step.yaml interface itself.

The StepInterface is a drop-in for the legacy StepSpecification (exposes dependencies/outputs/…), so it IS the spec view.

Parameters:

step_name (str) – Name of the step (PascalCase)

Returns:

StepInterface (drop-in for the legacy StepSpecification), or None if not found.

Return type:

Any | None

resolve_pipeline_node(node_name)

Resolve PipelineDAG node name to StepInfo (handles job_type variants).

Parameters:

node_name (str) – Node name from PipelineDAG

Returns:

StepInfo for the node, or None if not found

Return type:

StepInfo | None

search_steps(query, job_type=None)

Search steps by name with basic fuzzy matching.

Parameters:

• query (str) – Search query string

• job_type (str | None) – Optional job type filter

Returns:

List of search results sorted by relevance

Return type:

List[StepSearchResult]

serialize_contract(contract_instance)

Convert contract instance to dictionary format.

This method provides standardized serialization of ScriptContract objects for use in script-contract alignment validation.

Parameters:

contract_instance (Any) – Contract instance to serialize

Returns:

Dictionary representation of the contract

Return type:

Dict[str, Any]

serialize_spec(spec_instance)

Convert specification instance to dictionary format.

This method provides standardized serialization of StepSpecification objects for use in validation and alignment testing.

Parameters:

spec_instance (Any) – StepSpecification instance to serialize

Returns:

Dictionary representation of the specification

Return type:

Dict[str, Any]

validate_builder_availability(step_types)

Validate that builders are available for step types.

Parameters:

step_types (List[str]) – List of step types to validate

Returns:

Dictionary mapping step types to availability status

Return type:

Dict[str, bool]

validate_contract_script_mapping()

Validate contract-script relationships across the system.

Returns:

Dictionary with validation results and mapping statistics

Return type:

Dict[str, Any]

validate_dag_compatibility(step_types)

Validate DAG compatibility with available builders.

Parameters:

step_types (List[str]) – List of step types in the DAG

Returns:

Dictionary with validation results

Return type:

Dict[str, Any]

validate_logical_names_smart(contract, contract_name)

Smart validation using multi-variant specification logic.

This method delegates to SpecAutoDiscovery for smart validation, implementing the core Smart Specification Selection validation: - Contract input is valid if it exists in ANY variant - Contract must cover intersection of REQUIRED dependencies - Provides detailed feedback about which variants need what

Parameters:

• contract (Dict[str, Any]) – Contract dictionary

• contract_name (str) – Name of the contract

Returns:

List of validation issues

Return type:

List[Dict[str, Any]]

validate_step_name_with_registry(step_name)

Use registry system for step name validation.

Parameters:

step_name (str) – Step name to validate

Returns:

True if valid, False otherwise

Return type:

bool

metrics: Dict[str, Any]

class Level3ValidationConfig(mode=ValidationMode.RELAXED)

Bases: object

Configuration for Level 3 validation thresholds and behavior.

classmethod create_custom_config(pass_threshold, warning_threshold, error_threshold)

Create a custom validation configuration with specific thresholds.

Parameters:

• pass_threshold (float) – Minimum score for passing validation

• warning_threshold (float) – Minimum score for warning level

• error_threshold (float) – Minimum score for error level

Returns:

Custom Level3ValidationConfig instance

Return type:

Level3ValidationConfig

classmethod create_permissive_config()

Create a permissive validation configuration.

classmethod create_relaxed_config()

Create a relaxed validation configuration.

classmethod create_strict_config()

Create a strict validation configuration.

determine_severity_from_score(score, is_required)

Determine issue severity based on compatibility score and requirement.

Parameters:

• score (float) – Compatibility score (0.0 to 1.0)

• is_required (bool) – Whether the dependency is required

Returns:

Severity level string

Return type:

str

get_threshold_description()

Get human-readable description of current thresholds.

Returns:

Dictionary with threshold descriptions

Return type:

Dict[str, Any]

should_pass_validation(score)

Determine if a compatibility score should pass validation.

Parameters:

score (float) – Compatibility score (0.0 to 1.0)

Returns:

True if score meets pass threshold

Return type:

bool

class ValidationMode(*values)

Bases: Enum

Validation modes with different threshold requirements.

STRICT = 'strict'

RELAXED = 'relaxed'

PERMISSIVE = 'permissive'

normalize_path(path)

Normalize a path for comparison purposes.

Parameters:

path (str) – Path to normalize

Returns:

Normalized path string

Return type:

str

extract_logical_name_from_path(path)

Extract logical name from a SageMaker path.

For paths like ‘/opt/ml/processing/input/data’, extracts ‘data’.

Parameters:

path (str) – SageMaker path

Returns:

Logical name or None if not extractable

Return type:

str | None

is_sagemaker_path(path)

Check if a path is a SageMaker container path.

Parameters:

path (str) – Path to check

Returns:

True if this is a SageMaker path

Return type:

bool

format_alignment_issue(issue)

Format a validation issue for display.

Parameters:

issue (ValidationIssue) – The validation issue to format

Returns:

Formatted string representation

Return type:

str

group_issues_by_severity(issues)

Group validation issues by severity level.

Parameters:

issues (List[ValidationIssue]) – List of validation issues

Returns:

Dictionary mapping severity levels to lists of issues

Return type:

Dict[IssueLevel, List[ValidationIssue]]

get_highest_severity(issues)

Get the highest severity level among a list of issues.

Parameters:

issues (List[ValidationIssue]) – List of validation issues

Returns:

Highest severity level or None if no issues

Return type:

IssueLevel | None

validate_environment_setup()

Validate that the environment is properly set up for alignment validation.

Returns:

List of validation issues found

Return type:

List[str]

get_validation_summary_stats(issues)

Get summary statistics for a list of validation issues.

Parameters:

issues (List[ValidationIssue]) – List of validation issues

Returns:

Dictionary with summary statistics

Return type:

Dict[str, Any]

Modules

utils	Utility functions for alignment validation.
validation_models	Validation Models