API Reference Index

API documentation for the signalJourney libraries.

API Reference

This section provides the API documentation for the signaljourney_validator package.

signaljourney_validator

ValidationErrorDetail dataclass

Represents a detailed validation error.

__post_init__()

Generate suggestions based on error type after initialization.

SignalJourneyValidationError

Bases: Exception

Custom exception for validation errors.

Validator

Validates a signalJourney JSON file or dictionary against the schema. Optionally performs BIDS context validation. Supports version-based schema validation.

__init__(schema=None, schema_version=None, schema_dir=None)

Initializes the Validator.

Parameters:

Name	Type	Description	Default
schema	Optional[Union[Path, str, JsonDict]]	Path to the schema file, the schema dictionary, or None to use version-based schema loading. External file $refs will be automatically inlined during initialization.	None
schema_version	Optional[str]	Specific schema version to use (e.g., "0.1.0"). If None, will auto-detect from data during validation.	None
schema_dir	Optional[Path]	Custom schema directory path. If None, uses default.	None

validate(data, raise_exceptions=True, bids_context=None, auto_detect_version=True)

Validates the given data against the appropriate signalJourney schema.

Parameters:

Name	Type	Description	Default
data	Union[Path, str, JsonDict]	Path to the JSON file, the JSON string, a dictionary representing the JSON data.	required
raise_exceptions	bool	If True (default), raises SignalJourneyValidationError on the first failure. If False, returns a list of all validation errors found.	True
bids_context	Optional[Path]	Optional Path to the BIDS dataset root directory. If provided, enables BIDS context validation checks (e.g., file existence relative to the root).	None
auto_detect_version	bool	If True (default), automatically detects the schema version from the data and uses the appropriate schema. If False, uses the validator's configured schema.	True

Returns:

Type	Description
List[ValidationErrorDetail]	A list of ValidationErrorDetail objects if raise_exceptions is False
List[ValidationErrorDetail]	and validation fails. Returns an empty list if validation succeeds.

Raises:

Type	Description
SignalJourneyValidationError	If validation fails and raise_exceptions is True.
FileNotFoundError	If data file/path does not exist.
TypeError	If data is not a Path, string, or dictionary.
ValueError	If detected schema version is not supported.

get_supported_versions()

Get list of supported schema versions.

get_current_version()

Get the currently configured schema version.

get_latest_version()

Get the latest available schema version.

Validator Module

signaljourney_validator.validator

Validator

Validates a signalJourney JSON file or dictionary against the schema. Optionally performs BIDS context validation. Supports version-based schema validation.

Source code in src/signaljourney_validator/validator.py

class Validator:
    """
    Validates a signalJourney JSON file or dictionary against the schema.
    Optionally performs BIDS context validation.
    Supports version-based schema validation.
    """

    _schema: JsonDict
    _validator: Draft202012Validator
    _registry: SchemaVersionRegistry
    _schema_version: Optional[str]

    def __init__(
        self,
        schema: Optional[Union[Path, str, JsonDict]] = None,
        schema_version: Optional[str] = None,
        schema_dir: Optional[Path] = None,
    ):
        """
        Initializes the Validator.

        Args:
            schema: Path to the schema file, the schema dictionary, or None
                    to use version-based schema loading. External file $refs will be
                    automatically inlined during initialization.
            schema_version: Specific schema version to use (e.g., "0.1.0").
                           If None, will auto-detect from data during validation.
            schema_dir: Custom schema directory path. If None, uses default.
        """
        self._registry = SchemaVersionRegistry(schema_dir)
        self._schema_version = schema_version

        if schema is not None:
            # Use provided schema (legacy behavior)
            schema_path = self._get_schema_path(schema)
            initial_schema = self._load_schema_dict(schema, schema_path)
            base_resolve_path = (
                schema_path.parent if schema_path else DEFAULT_SCHEMA_PATH.parent
            )
        elif schema_version is not None:
            # Use specific version from registry
            if not self._registry.is_version_supported(schema_version):
                raise ValueError(
                    f"Schema version '{schema_version}' is not supported. "
                    f"Available versions: {self._registry.get_supported_versions()}"
                )
            schema_path = self._registry.get_schema_path(schema_version)
            initial_schema = self._registry.load_schema(schema_version)
            base_resolve_path = schema_path.parent
        else:
            # Use latest version as default
            try:
                latest_version = self._registry.get_latest_version()
                self._schema_version = latest_version
                schema_path = self._registry.get_schema_path(latest_version)
                initial_schema = self._registry.load_schema(latest_version)
                base_resolve_path = schema_path.parent
            except Exception:
                # Fallback to legacy schema path if registry fails
                schema_path = DEFAULT_SCHEMA_PATH
                initial_schema = self._load_schema_dict(None, schema_path)
                base_resolve_path = schema_path.parent

        # Inline external $refs
        print("\n--- Inlining schema refs within Validator ---")
        loaded_cache = {}
        self._schema = inline_refs(initial_schema, base_resolve_path, loaded_cache)
        print("--- Inlining complete --- \n")

        # Initialize the validator with the resolved schema.
        try:
            # Check if schema is valid before creating validator
            Draft202012Validator.check_schema(self._schema)
            self._validator = Draft202012Validator(
                schema=self._schema,
            )
        except jsonschema.SchemaError as e:
            raise SignalJourneyValidationError(f"Invalid schema provided: {e}") from e

    def _get_schema_path(
        self, schema_input: Optional[Union[Path, str, JsonDict]]
    ) -> Optional[Path]:
        """Determines the Path object if schema is given as Path or str."""
        if schema_input is None:
            return DEFAULT_SCHEMA_PATH
        if isinstance(schema_input, Path):
            return schema_input
        if isinstance(schema_input, str):
            return Path(schema_input)
        return None

    def _load_schema_dict(
        self,
        schema_input: Optional[Union[Path, str, JsonDict]],
        schema_path: Optional[Path],
    ) -> JsonDict:
        """Loads the schema into a dictionary."""
        if isinstance(schema_input, dict):
            return schema_input.copy()  # Return a copy

        # Determine path to load from
        load_path = schema_path if schema_path else DEFAULT_SCHEMA_PATH

        if not load_path or not load_path.exists():
            raise FileNotFoundError(f"Schema file not found: {load_path}")
        try:
            with open(load_path, "r", encoding="utf-8") as f:
                return json.load(f)
        except Exception as e:
            raise IOError(f"Error reading schema file {load_path}: {e}") from e

    def _create_validator_for_version(self, version: str) -> Draft202012Validator:
        """
        Create a validator for a specific schema version.

        Args:
            version: Schema version string

        Returns:
            Configured validator for the specified version
        """
        if not self._registry.is_version_supported(version):
            raise ValueError(
                f"Schema version '{version}' is not supported. "
                f"Available versions: {self._registry.get_supported_versions()}"
            )

        schema_path = self._registry.get_schema_path(version)
        schema_dict = self._registry.load_schema(version)

        # Inline external $refs
        loaded_cache = {}
        resolved_schema = inline_refs(schema_dict, schema_path.parent, loaded_cache)

        # Create and return validator
        Draft202012Validator.check_schema(resolved_schema)
        return Draft202012Validator(schema=resolved_schema)

    def validate(
        self,
        data: Union[Path, str, JsonDict],
        raise_exceptions: bool = True,
        bids_context: Optional[Path] = None,
        auto_detect_version: bool = True,
    ) -> List[ValidationErrorDetail]:
        """
        Validates the given data against the appropriate signalJourney schema.

        Args:
            data: Path to the JSON file, the JSON string, a dictionary
                  representing the JSON data.
            raise_exceptions: If True (default), raises SignalJourneyValidationError
                              on the first failure. If False, returns a list of
                              all validation errors found.
            bids_context: Optional Path to the BIDS dataset root directory.
                          If provided, enables BIDS context validation checks
                          (e.g., file existence relative to the root).
            auto_detect_version: If True (default), automatically detects the schema
                                version from the data and uses the appropriate schema.
                                If False, uses the validator's configured schema.

        Returns:
            A list of ValidationErrorDetail objects if raise_exceptions is False
            and validation fails. Returns an empty list if validation succeeds.

        Raises:
            SignalJourneyValidationError: If validation fails and
                                      raise_exceptions is True.
            FileNotFoundError: If data file/path does not exist.
            TypeError: If data is not a Path, string, or dictionary.
            ValueError: If detected schema version is not supported.
        """
        instance: JsonDict
        file_path_context: Optional[Path] = None  # For BIDS checks

        # --- Load Instance ---
        if isinstance(data, (Path, str)):
            file_path = Path(data)
            file_path_context = file_path  # Store for BIDS checks
            if not file_path.exists():
                raise FileNotFoundError(f"Data file not found: {file_path}")
            try:
                with open(file_path, "r", encoding="utf-8") as f:
                    instance = json.load(f)
            except json.JSONDecodeError as e:
                raise SignalJourneyValidationError(
                    f"Error decoding data JSON from {file_path}: {e}"
                ) from e
            except Exception as e:
                raise SignalJourneyValidationError(
                    f"Error loading data file from {file_path}: {e}"
                ) from e
        elif isinstance(data, dict):
            instance = data
        else:
            raise TypeError("Data must be a Path, string, or dictionary.")

        # --- Version Detection and Validator Selection ---
        validator_to_use = self._validator
        detected_version = None

        if auto_detect_version:
            try:
                detected_version = self._registry.detect_schema_version(instance)
                if detected_version:
                    if not self._registry.is_version_supported(detected_version):
                        error_msg = (
                            f"Detected schema version '{detected_version}' is not "
                            f"supported. Available versions: "
                            f"{self._registry.get_supported_versions()}"
                        )
                        if raise_exceptions:
                            raise SignalJourneyValidationError(error_msg)
                        else:
                            # Return a validation error
                            return [
                                ValidationErrorDetail(
                                    message=error_msg,
                                    path=["schema_version"],
                                    schema_path=[],
                                    validator="version_support",
                                    validator_value=self._registry.get_supported_versions(),
                                    instance_value=detected_version,
                                )
                            ]

                    # Use version-specific validator if different from current
                    if detected_version != self._schema_version:
                        validator_to_use = self._create_validator_for_version(
                            detected_version
                        )

                elif self._schema_version is None:
                    # No version detected and no default configured
                    error_msg = (
                        "No schema_version field found in data and no default version "
                        "configured. Please specify a schema_version in your "
                        "signalJourney file."
                    )
                    if raise_exceptions:
                        raise SignalJourneyValidationError(error_msg)
                    else:
                        return [
                            ValidationErrorDetail(
                                message=error_msg,
                                path=["schema_version"],
                                schema_path=[],
                                validator="required",
                                validator_value=True,
                                instance_value=None,
                            )
                        ]

            except (FileNotFoundError, IOError) as e:
                # Re-raise file system errors
                raise e
            except Exception as e:
                # Handle version detection errors
                error_msg = f"Error during schema version detection: {e}"
                if raise_exceptions:
                    raise SignalJourneyValidationError(error_msg) from e
                else:
                    return [
                        ValidationErrorDetail(
                            message=error_msg,
                            path=["schema_version"],
                            schema_path=[],
                            validator="version_detection",
                            validator_value=None,
                            instance_value=instance.get("schema_version"),
                        )
                    ]

        schema_errors: List[ValidationErrorDetail] = []
        bids_errors: List[ValidationErrorDetail] = []

        # --- Schema Validation ---
        # Use the selected validator's iter_errors method
        try:
            errors = sorted(
                validator_to_use.iter_errors(instance), key=lambda e: e.path
            )
            if errors:
                for error in errors:
                    # Convert jsonschema error to our custom format
                    detail = ValidationErrorDetail(
                        message=error.message,
                        path=list(error.path),
                        schema_path=list(error.schema_path),
                        validator=error.validator,
                        validator_value=error.validator_value,
                        instance_value=error.instance,
                        # suggestion is added by ValidationErrorDetail constructor
                    )
                    schema_errors.append(detail)
        except jsonschema.RefResolutionError as e:
            # This shouldn't happen if schema is fully resolved, but handle defensively
            print(f"DEBUG: Unexpected RefResolutionError: {e}")
            failed_ref = getattr(e, "ref", "[unknown ref]")
            raise SignalJourneyValidationError(
                f"Schema validation failed: Unexpectedly could not resolve "
                f"reference '{failed_ref}'"
            ) from e
        except jsonschema.SchemaError as e:
            # Catch schema errors separately from resolution errors
            raise SignalJourneyValidationError(f"Invalid schema: {e}") from e
        except Exception as e:
            # Capture the actual exception type for better debugging
            print(
                f"DEBUG: Unexpected validation error type: "
                f"{type(e).__name__}, Error: {e}"
            )
            # Reraise or wrap depending on desired behavior
            raise SignalJourneyValidationError(
                f"An unexpected error occurred during schema validation: {e}"
            ) from e

        # --- BIDS Context Validation (Optional) ---
        if bids_context:
            bids_errors = self._validate_bids_context(
                instance, file_path_context, bids_context
            )

        all_errors = schema_errors + bids_errors

        # --- Result Handling ---
        if all_errors:
            if raise_exceptions:
                raise SignalJourneyValidationError(
                    "Validation failed.", errors=all_errors
                )
            else:
                return all_errors
        else:
            return []  # Success

    def _validate_bids_context(
        self, instance: JsonDict, file_path: Optional[Path], bids_root: Path
    ) -> List[ValidationErrorDetail]:
        """Placeholder for BIDS context validation logic."""
        errors: List[ValidationErrorDetail] = []
        print(
            f"[INFO] BIDS context validation requested for {file_path} "
            f"within {bids_root} (Not implemented)"
        )

        # TODO: Implement BIDS checks using file_path and bids_root
        # Examples:
        # - Check if file_path is correctly placed within bids_root derivatives
        # - Check naming convention against BIDS standards (might need pybids)
        # - Check if files referenced in inputSources/outputTargets exist
        #   relative to bids_root
        # - Differentiate rules for root-level vs derivative-level journey files

        # Example error:
        # if some_bids_check_fails:
        #     errors.append(ValidationErrorDetail(
        #         message="BIDS Check Failed: Reason...",
        #         path=['relevant', 'path']
        #     ))

        return errors

    def get_supported_versions(self) -> List[str]:
        """Get list of supported schema versions."""
        return self._registry.get_supported_versions()

    def get_current_version(self) -> Optional[str]:
        """Get the currently configured schema version."""
        return self._schema_version

    def get_latest_version(self) -> Optional[str]:
        """Get the latest available schema version."""
        return self._registry.get_latest_version()

__init__(schema=None, schema_version=None, schema_dir=None)

Initializes the Validator.

Parameters:

Name	Type	Description	Default
schema	Optional[Union[Path, str, JsonDict]]	Path to the schema file, the schema dictionary, or None to use version-based schema loading. External file $refs will be automatically inlined during initialization.	None
schema_version	Optional[str]	Specific schema version to use (e.g., "0.1.0"). If None, will auto-detect from data during validation.	None
schema_dir	Optional[Path]	Custom schema directory path. If None, uses default.	None

Source code in src/signaljourney_validator/validator.py

def __init__(
    self,
    schema: Optional[Union[Path, str, JsonDict]] = None,
    schema_version: Optional[str] = None,
    schema_dir: Optional[Path] = None,
):
    """
    Initializes the Validator.

    Args:
        schema: Path to the schema file, the schema dictionary, or None
                to use version-based schema loading. External file $refs will be
                automatically inlined during initialization.
        schema_version: Specific schema version to use (e.g., "0.1.0").
                       If None, will auto-detect from data during validation.
        schema_dir: Custom schema directory path. If None, uses default.
    """
    self._registry = SchemaVersionRegistry(schema_dir)
    self._schema_version = schema_version

    if schema is not None:
        # Use provided schema (legacy behavior)
        schema_path = self._get_schema_path(schema)
        initial_schema = self._load_schema_dict(schema, schema_path)
        base_resolve_path = (
            schema_path.parent if schema_path else DEFAULT_SCHEMA_PATH.parent
        )
    elif schema_version is not None:
        # Use specific version from registry
        if not self._registry.is_version_supported(schema_version):
            raise ValueError(
                f"Schema version '{schema_version}' is not supported. "
                f"Available versions: {self._registry.get_supported_versions()}"
            )
        schema_path = self._registry.get_schema_path(schema_version)
        initial_schema = self._registry.load_schema(schema_version)
        base_resolve_path = schema_path.parent
    else:
        # Use latest version as default
        try:
            latest_version = self._registry.get_latest_version()
            self._schema_version = latest_version
            schema_path = self._registry.get_schema_path(latest_version)
            initial_schema = self._registry.load_schema(latest_version)
            base_resolve_path = schema_path.parent
        except Exception:
            # Fallback to legacy schema path if registry fails
            schema_path = DEFAULT_SCHEMA_PATH
            initial_schema = self._load_schema_dict(None, schema_path)
            base_resolve_path = schema_path.parent

    # Inline external $refs
    print("\n--- Inlining schema refs within Validator ---")
    loaded_cache = {}
    self._schema = inline_refs(initial_schema, base_resolve_path, loaded_cache)
    print("--- Inlining complete --- \n")

    # Initialize the validator with the resolved schema.
    try:
        # Check if schema is valid before creating validator
        Draft202012Validator.check_schema(self._schema)
        self._validator = Draft202012Validator(
            schema=self._schema,
        )
    except jsonschema.SchemaError as e:
        raise SignalJourneyValidationError(f"Invalid schema provided: {e}") from e

validate(data, raise_exceptions=True, bids_context=None, auto_detect_version=True)

Validates the given data against the appropriate signalJourney schema.

Parameters:

Name	Type	Description	Default
data	Union[Path, str, JsonDict]	Path to the JSON file, the JSON string, a dictionary representing the JSON data.	required
raise_exceptions	bool	If True (default), raises SignalJourneyValidationError on the first failure. If False, returns a list of all validation errors found.	True
bids_context	Optional[Path]	Optional Path to the BIDS dataset root directory. If provided, enables BIDS context validation checks (e.g., file existence relative to the root).	None
auto_detect_version	bool	If True (default), automatically detects the schema version from the data and uses the appropriate schema. If False, uses the validator's configured schema.	True

Returns:

Type	Description
List[ValidationErrorDetail]	A list of ValidationErrorDetail objects if raise_exceptions is False
List[ValidationErrorDetail]	and validation fails. Returns an empty list if validation succeeds.

Raises:

Type	Description
SignalJourneyValidationError	If validation fails and raise_exceptions is True.
FileNotFoundError	If data file/path does not exist.
TypeError	If data is not a Path, string, or dictionary.
ValueError	If detected schema version is not supported.

Source code in src/signaljourney_validator/validator.py

def validate(
    self,
    data: Union[Path, str, JsonDict],
    raise_exceptions: bool = True,
    bids_context: Optional[Path] = None,
    auto_detect_version: bool = True,
) -> List[ValidationErrorDetail]:
    """
    Validates the given data against the appropriate signalJourney schema.

    Args:
        data: Path to the JSON file, the JSON string, a dictionary
              representing the JSON data.
        raise_exceptions: If True (default), raises SignalJourneyValidationError
                          on the first failure. If False, returns a list of
                          all validation errors found.
        bids_context: Optional Path to the BIDS dataset root directory.
                      If provided, enables BIDS context validation checks
                      (e.g., file existence relative to the root).
        auto_detect_version: If True (default), automatically detects the schema
                            version from the data and uses the appropriate schema.
                            If False, uses the validator's configured schema.

    Returns:
        A list of ValidationErrorDetail objects if raise_exceptions is False
        and validation fails. Returns an empty list if validation succeeds.

    Raises:
        SignalJourneyValidationError: If validation fails and
                                  raise_exceptions is True.
        FileNotFoundError: If data file/path does not exist.
        TypeError: If data is not a Path, string, or dictionary.
        ValueError: If detected schema version is not supported.
    """
    instance: JsonDict
    file_path_context: Optional[Path] = None  # For BIDS checks

    # --- Load Instance ---
    if isinstance(data, (Path, str)):
        file_path = Path(data)
        file_path_context = file_path  # Store for BIDS checks
        if not file_path.exists():
            raise FileNotFoundError(f"Data file not found: {file_path}")
        try:
            with open(file_path, "r", encoding="utf-8") as f:
                instance = json.load(f)
        except json.JSONDecodeError as e:
            raise SignalJourneyValidationError(
                f"Error decoding data JSON from {file_path}: {e}"
            ) from e
        except Exception as e:
            raise SignalJourneyValidationError(
                f"Error loading data file from {file_path}: {e}"
            ) from e
    elif isinstance(data, dict):
        instance = data
    else:
        raise TypeError("Data must be a Path, string, or dictionary.")

    # --- Version Detection and Validator Selection ---
    validator_to_use = self._validator
    detected_version = None

    if auto_detect_version:
        try:
            detected_version = self._registry.detect_schema_version(instance)
            if detected_version:
                if not self._registry.is_version_supported(detected_version):
                    error_msg = (
                        f"Detected schema version '{detected_version}' is not "
                        f"supported. Available versions: "
                        f"{self._registry.get_supported_versions()}"
                    )
                    if raise_exceptions:
                        raise SignalJourneyValidationError(error_msg)
                    else:
                        # Return a validation error
                        return [
                            ValidationErrorDetail(
                                message=error_msg,
                                path=["schema_version"],
                                schema_path=[],
                                validator="version_support",
                                validator_value=self._registry.get_supported_versions(),
                                instance_value=detected_version,
                            )
                        ]

                # Use version-specific validator if different from current
                if detected_version != self._schema_version:
                    validator_to_use = self._create_validator_for_version(
                        detected_version
                    )

            elif self._schema_version is None:
                # No version detected and no default configured
                error_msg = (
                    "No schema_version field found in data and no default version "
                    "configured. Please specify a schema_version in your "
                    "signalJourney file."
                )
                if raise_exceptions:
                    raise SignalJourneyValidationError(error_msg)
                else:
                    return [
                        ValidationErrorDetail(
                            message=error_msg,
                            path=["schema_version"],
                            schema_path=[],
                            validator="required",
                            validator_value=True,
                            instance_value=None,
                        )
                    ]

        except (FileNotFoundError, IOError) as e:
            # Re-raise file system errors
            raise e
        except Exception as e:
            # Handle version detection errors
            error_msg = f"Error during schema version detection: {e}"
            if raise_exceptions:
                raise SignalJourneyValidationError(error_msg) from e
            else:
                return [
                    ValidationErrorDetail(
                        message=error_msg,
                        path=["schema_version"],
                        schema_path=[],
                        validator="version_detection",
                        validator_value=None,
                        instance_value=instance.get("schema_version"),
                    )
                ]

    schema_errors: List[ValidationErrorDetail] = []
    bids_errors: List[ValidationErrorDetail] = []

    # --- Schema Validation ---
    # Use the selected validator's iter_errors method
    try:
        errors = sorted(
            validator_to_use.iter_errors(instance), key=lambda e: e.path
        )
        if errors:
            for error in errors:
                # Convert jsonschema error to our custom format
                detail = ValidationErrorDetail(
                    message=error.message,
                    path=list(error.path),
                    schema_path=list(error.schema_path),
                    validator=error.validator,
                    validator_value=error.validator_value,
                    instance_value=error.instance,
                    # suggestion is added by ValidationErrorDetail constructor
                )
                schema_errors.append(detail)
    except jsonschema.RefResolutionError as e:
        # This shouldn't happen if schema is fully resolved, but handle defensively
        print(f"DEBUG: Unexpected RefResolutionError: {e}")
        failed_ref = getattr(e, "ref", "[unknown ref]")
        raise SignalJourneyValidationError(
            f"Schema validation failed: Unexpectedly could not resolve "
            f"reference '{failed_ref}'"
        ) from e
    except jsonschema.SchemaError as e:
        # Catch schema errors separately from resolution errors
        raise SignalJourneyValidationError(f"Invalid schema: {e}") from e
    except Exception as e:
        # Capture the actual exception type for better debugging
        print(
            f"DEBUG: Unexpected validation error type: "
            f"{type(e).__name__}, Error: {e}"
        )
        # Reraise or wrap depending on desired behavior
        raise SignalJourneyValidationError(
            f"An unexpected error occurred during schema validation: {e}"
        ) from e

    # --- BIDS Context Validation (Optional) ---
    if bids_context:
        bids_errors = self._validate_bids_context(
            instance, file_path_context, bids_context
        )

    all_errors = schema_errors + bids_errors

    # --- Result Handling ---
    if all_errors:
        if raise_exceptions:
            raise SignalJourneyValidationError(
                "Validation failed.", errors=all_errors
            )
        else:
            return all_errors
    else:
        return []  # Success

get_supported_versions()

Get list of supported schema versions.

Source code in src/signaljourney_validator/validator.py

def get_supported_versions(self) -> List[str]:
    """Get list of supported schema versions."""
    return self._registry.get_supported_versions()

get_current_version()

Get the currently configured schema version.

Source code in src/signaljourney_validator/validator.py

def get_current_version(self) -> Optional[str]:
    """Get the currently configured schema version."""
    return self._schema_version

get_latest_version()

Get the latest available schema version.

Source code in src/signaljourney_validator/validator.py

def get_latest_version(self) -> Optional[str]:
    """Get the latest available schema version."""
    return self._registry.get_latest_version()

inline_refs(schema, base_path, loaded_schemas_cache)

Recursively replace $ref keys with the content of the referenced file. Uses a cache (loaded_schemas_cache) to avoid infinite loops with circular refs and redundant file loading. Cache keys should be absolute POSIX paths of the schema files.

Source code in src/signaljourney_validator/validator.py

def inline_refs(
    schema: Union[Dict, list], base_path: Path, loaded_schemas_cache: Dict[str, Dict]
):
    """Recursively replace $ref keys with the content of the referenced file.
    Uses a cache (loaded_schemas_cache) to avoid infinite loops with circular refs
    and redundant file loading.
    Cache keys should be absolute POSIX paths of the schema files.
    """
    if isinstance(schema, dict):
        if (
            "$ref" in schema
            and isinstance(schema["$ref"], str)
            and not schema["$ref"].startswith("#")
        ):
            ref_path_str = schema["$ref"]
            # Resolve relative ref path against the current base path
            ref_path = (base_path / ref_path_str).resolve()

            # Cache key based on resolved absolute path
            cache_key = ref_path.as_posix()

            # Check cache first
            if cache_key in loaded_schemas_cache:
                # Return a copy to prevent modification issues during recursion
                return loaded_schemas_cache[cache_key].copy()

            # If not cached, load the file
            if ref_path.exists() and ref_path.is_file():
                try:
                    # print(
                    #     f"[Inline] Loading $ref: {ref_path_str} "
                    #     f"(from {base_path}) -> {ref_path}"
                    # )
                    with open(ref_path, "r", encoding="utf-8") as f:
                        ref_content = json.load(f)
                    # Store in cache BEFORE recursion to handle circular refs
                    loaded_schemas_cache[cache_key] = ref_content
                    # Recursively resolve refs *within* the loaded content
                    # Use the directory of the *referenced* file as the new base path
                    resolved_content = inline_refs(
                        ref_content, ref_path.parent, loaded_schemas_cache
                    )
                    # Update cache with the fully resolved content
                    loaded_schemas_cache[cache_key] = resolved_content
                    # Return a copy of the resolved content
                    return resolved_content.copy()
                except Exception as e:
                    print(
                        f"Warning: Failed to load or parse $ref: {ref_path_str} "
                        f"from {ref_path}. Error: {e}"
                    )
                    return schema  # Keep original $ref on error
            else:
                print(
                    f"Warning: $ref path does not exist or is not a file: "
                    f"{ref_path_str} -> {ref_path}"
                )
                return schema  # Keep original $ref if file not found
        else:
            # Recursively process other keys in the dictionary
            new_schema = {}
            for key, value in schema.items():
                new_schema[key] = inline_refs(value, base_path, loaded_schemas_cache)
            return new_schema
    elif isinstance(schema, list):
        # Recursively process items in the list
        return [inline_refs(item, base_path, loaded_schemas_cache) for item in schema]
    else:
        # Return non-dict/list items as is
        return schema

CLI Module

signaljourney_validator.cli

cli()

Signal Journey Validator CLI.

Provides tools to validate signalJourney JSON files against the official specification schema. Supports validating single files or entire directories.

Source code in src/signaljourney_validator/cli.py

@click.group(context_settings=dict(help_option_names=["-h", "--help"]))
@click.version_option(
    package_name="signaljourney-validator", prog_name="signaljourney-validate"
)
def cli():
    """
    Signal Journey Validator CLI.

    Provides tools to validate signalJourney JSON files against the official
    specification schema.
    Supports validating single files or entire directories.
    """
    pass

validate(path, schema, schema_version, recursive, output_format, verbose, bids, bids_root)

Validate one or more signalJourney JSON files.

Checks conformance against the official signalJourney schema (or a custom schema if provided via --schema). Supports version-based validation using --schema-version.

Examples:

Validate a single file (auto-detects schema version):

signaljourney-validate path/to/sub-01_task-rest_signalJourney.json

Validate with specific schema version:

signaljourney-validate --schema-version 0.1.0 path/to/file.json

Validate all files in a directory (non-recursively):

signaljourney-validate path/to/derivatives/pipelineX/

Validate all files recursively, outputting JSON:

signaljourney-validate -r -o json path/to/bids_dataset/

Validate with BIDS context checks:

signaljourney-validate --bids --bids-root path/to/bids_dataset \
    path/to/bids_dataset/derivatives/...

Source code in src/signaljourney_validator/cli.py

@cli.command()
@click.argument(
    "path",
    type=click.Path(exists=True, readable=True, resolve_path=True, path_type=Path),
)
@click.option(
    "--schema",
    "-s",
    type=click.Path(
        exists=True, dir_okay=False, readable=True, resolve_path=True, path_type=Path
    ),
    help="Path to a custom JSON schema file to validate against.",
)
@click.option(
    "--schema-version",
    type=str,
    help="Specific schema version to use (e.g., '0.1.0'). If not specified, "
    "auto-detects from each file's schema_version field.",
)
@click.option(
    "--recursive",
    "-r",
    is_flag=True,
    default=False,
    help="Recursively search for *_signalJourney.json files in subdirectories.",
)
@click.option(
    "--output-format",
    "-o",
    type=click.Choice(["text", "json"], case_sensitive=False),
    default="text",
    help='Output format: "text" (human-readable, default) or "json" '
    "(machine-readable).",
)
@click.option(
    "--verbose",
    "-v",
    is_flag=True,
    default=False,
    help='Enable verbose output for the "text" format (shows more error details).',
)
@click.option(
    "--bids",
    is_flag=True,
    default=False,
    help="Enable BIDS context validation checks (experimental).",
)
@click.option(
    "--bids-root",
    type=click.Path(exists=True, file_okay=False, path_type=Path, resolve_path=True),
    help="Path to the BIDS dataset root directory (required if --bids is used).",
)
def validate(
    path: Path,
    schema: Path,
    schema_version: str,
    recursive: bool,
    output_format: str,
    verbose: bool,
    bids: bool,
    bids_root: Path,
):
    """
    Validate one or more signalJourney JSON files.

    Checks conformance against the official signalJourney schema (or a custom schema
    if provided via --schema). Supports version-based validation using --schema-version.

    Examples:

    Validate a single file (auto-detects schema version):

        signaljourney-validate path/to/sub-01_task-rest_signalJourney.json

    Validate with specific schema version:

        signaljourney-validate --schema-version 0.1.0 path/to/file.json

    Validate all files in a directory (non-recursively):

        signaljourney-validate path/to/derivatives/pipelineX/

    Validate all files recursively, outputting JSON:

        signaljourney-validate -r -o json path/to/bids_dataset/

    Validate with BIDS context checks:

        signaljourney-validate --bids --bids-root path/to/bids_dataset \\
            path/to/bids_dataset/derivatives/...
    """
    if bids and not bids_root:
        click.echo(
            "Error: --bids-root is required when using the --bids flag.", err=True
        )
        sys.exit(1)

    if schema and schema_version:
        click.echo(
            "Error: Cannot specify both --schema and --schema-version. "
            "Use --schema for custom schema files or --schema-version for "
            "version-based validation.",
            err=True,
        )
        sys.exit(1)

    files_to_validate: List[Path] = []
    if path.is_file():
        # If a single file is provided, attempt to validate it if it's JSON,
        # regardless of the _signalJourney suffix.
        # Keep the suffix check for directory scanning.
        if path.name.lower().endswith(".json"):
            files_to_validate.append(path)
        elif output_format == "text":  # Only print skip message for non-JSON files
            click.echo(f"Skipping non-JSON file: {path}", err=True)
    elif path.is_dir():
        if output_format == "text":
            scan_mode = " recursively" if recursive else ""
            bids_mode = " (BIDS mode)" if bids else ""
            click.echo(f"Scanning directory: {path}{scan_mode}{bids_mode}")
        if recursive:
            # Keep the suffix check for recursive scanning
            for root, _, filenames in os.walk(path):
                for filename in filenames:
                    if filename.endswith("_signalJourney.json"):
                        files_to_validate.append(Path(root) / filename)
        else:
            # For non-recursive directory scan, find any .json file
            for item in path.iterdir():
                if item.is_file() and item.name.lower().endswith(".json"):
                    files_to_validate.append(item)
    else:
        # This error should occur regardless of format
        click.echo(
            f"Error: Input path is neither a file nor a directory: {path}", err=True
        )
        sys.exit(1)

    # Prepare results structure
    results: Dict[str, Any] = {"overall_success": True, "files": []}
    validator_instance: Optional[Validator] = None

    if not files_to_validate:
        if output_format == "text" and path.is_dir():
            # Only print if input was a directory and no files were found.
            scan_mode = " recursively" if recursive else ""
            click.echo(
                f"No *_signalJourney.json files found to validate in {path}{scan_mode}",
                err=True,
            )
        elif output_format == "json":
            # Output valid JSON even if no files processed
            print(json.dumps({"files": [], "overall_success": True}, indent=2))
        # Exit code 0 if input dir was empty, 1 otherwise (e.g., non-JSON file input)
        sys.exit(0 if path.is_dir() else 1)

    # --- Schema Loading and Resolver Setup (Load ONCE) ---
    # try:
    #     schema_to_use = schema if schema else DEFAULT_SCHEMA_PATH
    #     if not schema_to_use.exists():
    #         raise FileNotFoundError(f"Schema file not found: {schema_to_use}")
    #     with open(schema_to_use, "r", encoding="utf-8") as f:
    #         main_schema_dict = json.load(f)
    #
    #     # Setup resolver similar to conftest.py - REMOVED
    #     ...
    #
    #     resolver = jsonschema.RefResolver(...)
    # except Exception as e:
    #     click.echo(f"Error loading schema or building resolver: {e}", err=True)
    #     sys.exit(1)
    # --- End Schema Loading ---

    overall_success = True
    for filepath in files_to_validate:
        file_result: Dict[str, Any] = {
            "filepath": str(filepath),
            "status": "unknown",
            "errors": [],
        }
        if output_format == "text":
            click.echo(f"Validating: {filepath} ... ", nl=False)

        try:
            if validator_instance is None:
                # Create validator ONCE using the schema path (or None for default)
                # Validator internal __init__ now handles registry setup.
                try:
                    # Create validator with schema version support
                    validator_instance = Validator(
                        schema=schema, schema_version=schema_version
                    )
                except Exception as e:
                    # Handle potential errors during Validator initialization
                    # (e.g., schema loading)
                    click.echo(f"CRITICAL ERROR initializing validator: {e}", err=True)
                    # For JSON output, log the critical error at file level
                    if output_format == "json":
                        file_result["status"] = "critical_error"
                        file_result["errors"] = [
                            {"message": f"Validator init failed: {e}"}
                        ]
                        results["files"].append(file_result)
                        results["overall_success"] = False
                        overall_success = False  # Ensure overall failure
                    # Exit or continue? Maybe continue to report errors for other files?
                    # For now, let's make it a fatal error for the specific file.
                    if output_format == "text":
                        click.echo(" CRITICAL ERROR")
                        click.echo(f"  - Initialization Failed: {e}")
                    # Skip to the next file if validator init fails
                    continue

            # Pass bids_root to validator if bids flag is set
            current_bids_context = bids_root if bids else None
            validation_errors = validator_instance.validate(
                filepath, raise_exceptions=False, bids_context=current_bids_context
            )

            if validation_errors:
                overall_success = False
                file_result["status"] = "failed"
                if output_format == "text":
                    click.secho("FAILED", fg="red")
                for error in validation_errors:
                    # Store structured error
                    error_dict = {
                        "message": error.message,
                        "path": list(error.path) if error.path else [],
                        "schema_path": list(error.schema_path)
                        if error.schema_path
                        else [],
                        "validator": error.validator,
                        "validator_value": repr(error.validator_value),  # Use repr
                        "instance_value": repr(error.instance_value),  # Use repr
                        "suggestion": error.suggestion,
                    }
                    file_result["errors"].append(error_dict)

                    # Print detailed error in text mode
                    if output_format == "text":
                        error_path_list = list(error.path) if error.path else []
                        error_path_str = (
                            "/".join(map(str, error_path_list))
                            if error_path_list
                            else "root"
                        )
                        error_msg = f"  - Error at '{error_path_str}': {error.message}"
                        if verbose:
                            # Add more details in verbose mode
                            error_msg += f" (validator: '{error.validator}')"
                            # Add more details if needed
                        if error.suggestion:
                            error_msg += f" -- Suggestion: {error.suggestion}"
                        click.echo(error_msg)
            else:
                file_result["status"] = "passed"
                if output_format == "text":
                    click.secho("PASSED", fg="green")

        except SignalJourneyValidationError as e:
            overall_success = False
            file_result["status"] = "error"
            file_result["error_message"] = str(e)
            if output_format == "text":
                click.secho("ERROR", fg="yellow")
                click.echo(f"  - Validation Error: {e}", err=True)
            # Include details from SignalJourneyValidationError if available
            detailed_errors = []
            if (
                isinstance(e, SignalJourneyValidationError)
                and hasattr(e, "errors")
                and e.errors
            ):
                detailed_errors = [
                    {
                        "message": detail.get("message", "N/A"),
                        "path": detail.get("path", []),
                        # Add other relevant fields from ValidationErrorDetail if needed
                    }
                    for detail in e.errors
                ]
                file_result["errors"] = detailed_errors  # Overwrite with details
                if output_format == "text":
                    click.echo("    Detailed Errors:", err=True)
                    for detail in detailed_errors:
                        path_str = detail.get("path", "N/A")
                        msg_str = detail.get("message", "N/A")
                        click.echo(f"    - Path: {path_str}, Msg: {msg_str}", err=True)

        except Exception as e:
            overall_success = False
            file_result["status"] = "error"
            file_result["error_message"] = f"An unexpected error occurred: {e}"
            if output_format == "text":
                click.secho("CRITICAL ERROR", fg="red", bold=True)
                click.echo(f"  - Unexpected Error: {e}", err=True)

        results["files"].append(file_result)

    results["overall_success"] = overall_success

    if output_format == "json":
        print(json.dumps(results, indent=2))
    elif output_format == "text" and verbose:
        # Add a summary line in verbose text mode
        status_msg = (
            click.style("PASSED", fg="green")
            if overall_success
            else click.style("FAILED", fg="red")
        )
        click.echo(f"\nOverall validation result: {status_msg}")

    # Exit with appropriate code
    sys.exit(0 if overall_success else 1)

Errors Module

signaljourney_validator.errors

ValidationErrorDetail dataclass

Represents a detailed validation error.

Source code in src/signaljourney_validator/errors.py

@dataclass
class ValidationErrorDetail:
    """Represents a detailed validation error."""

    message: str
    path: Sequence[JsonPath] = field(default_factory=list)
    schema_path: Sequence[JsonPath] = field(default_factory=list)
    validator: str = ""
    validator_value: Any = None
    instance_value: Any = None
    # For nested errors
    context: List["ValidationErrorDetail"] = field(default_factory=list)
    suggestion: Optional[str] = None  # Added field for suggestions

    def __post_init__(self):
        """Generate suggestions based on error type after initialization."""
        self._generate_suggestion()

    def __str__(self) -> str:
        path_str = "/".join(map(str, self.path)) if self.path else "root"
        msg = f"Error at '{path_str}': {self.message}"
        if self.suggestion:
            msg += f" -- Suggestion: {self.suggestion}"
        return msg

    def _generate_suggestion(self):
        """Internal method to populate the suggestion field based on validator type."""
        if self.validator == "required":
            # 'validator_value' usually holds the list of required properties
            missing_props = self.validator_value
            if isinstance(missing_props, list):
                props_str = "', '".join(missing_props)
                self.suggestion = (
                    f"Ensure required property or properties ('{props_str}') "
                    f"are present."
                )
            else:
                self.suggestion = (
                    "Ensure required property is present (check schema for details)."
                )

        elif self.validator == "type":
            expected_types = self.validator_value
            actual_type = type(self.instance_value).__name__
            if isinstance(expected_types, list):
                types_str = "', '".join(expected_types)
                self.suggestion = (
                    f"Change value type from '{actual_type}' to one of: '{types_str}'."
                )
            elif isinstance(expected_types, str):
                self.suggestion = (
                    f"Change value type from '{actual_type}' to '{expected_types}'."
                )
            else:
                self.suggestion = (
                    f"Check schema for expected type(s) instead of '{actual_type}'."
                )

        elif self.validator == "pattern":
            pattern = self.validator_value
            self.suggestion = (
                f"Ensure value matches the required regex pattern: '{pattern}'."
            )

        elif self.validator == "enum":
            allowed_values = self.validator_value
            if isinstance(allowed_values, list):
                suggestion_text = (
                    f"Value must be one of: {', '.join(map(repr, allowed_values))}."
                )
                # Optional: Add fuzzy matching
                if (
                    HAS_FUZZY
                    and isinstance(self.instance_value, str)
                    and self.instance_value
                ):
                    try:
                        # Filter for string choices
                        string_allowed_values = [
                            str(v) for v in allowed_values if isinstance(v, str)
                        ]
                        if string_allowed_values:
                            best_match, score = fuzzy_process.extractOne(
                                self.instance_value, string_allowed_values
                            )
                            if score > 80:  # Threshold
                                suggestion_text += f" Did you mean '{best_match}'?"
                    except Exception:
                        pass  # Ignore fuzzy matching errors
                self.suggestion = suggestion_text
            else:
                self.suggestion = (
                    "Ensure value is one of the allowed options (check schema)."
                )

        # Add suggestions for length/item constraints
        elif self.validator == "minLength":
            min_len = self.validator_value
            actual_len = (
                len(self.instance_value)
                if isinstance(self.instance_value, (str, list))
                else "N/A"
            )
            self.suggestion = (
                f"Ensure value has at least {min_len} characters/items "
                f"(currently {actual_len})."
            )

        elif self.validator == "maxLength":
            max_len = self.validator_value
            actual_len = (
                len(self.instance_value)
                if isinstance(self.instance_value, (str, list))
                else "N/A"
            )
            self.suggestion = (
                f"Ensure value has at most {max_len} characters/items "
                f"(currently {actual_len})."
            )

        elif self.validator == "minItems":
            min_num = self.validator_value
            actual_num = (
                len(self.instance_value)
                if isinstance(self.instance_value, list)
                else "N/A"
            )
            self.suggestion = (
                f"Ensure array has at least {min_num} items (currently {actual_num})."
            )

        elif self.validator == "maxItems":
            max_num = self.validator_value
            actual_num = (
                len(self.instance_value)
                if isinstance(self.instance_value, list)
                else "N/A"
            )
            self.suggestion = (
                f"Ensure array has at most {max_num} items (currently {actual_num})."
            )

        elif self.validator == "minimum":
            min_val = self.validator_value
            self.suggestion = f"Ensure value is at least {min_val}."

        elif self.validator == "maximum":
            max_val = self.validator_value
            self.suggestion = f"Ensure value is at most {max_val}."

        elif self.validator == "exclusiveMinimum":
            ex_min_val = self.validator_value
            self.suggestion = f"Ensure value is strictly greater than {ex_min_val}."

        elif self.validator == "exclusiveMaximum":
            ex_max_val = self.validator_value
            self.suggestion = f"Ensure value is strictly less than {ex_max_val}."

__post_init__()

Generate suggestions based on error type after initialization.

Source code in src/signaljourney_validator/errors.py

def __post_init__(self):
    """Generate suggestions based on error type after initialization."""
    self._generate_suggestion()

SignalJourneyValidationError

Bases: Exception

Custom exception for validation errors.

Source code in src/signaljourney_validator/errors.py

class SignalJourneyValidationError(Exception):
    """Custom exception for validation errors."""

    def __init__(
        self, message: str, errors: Optional[List[ValidationErrorDetail]] = None
    ):
        super().__init__(message)
        self.errors = errors or []