Source code for solidipes.validators.curation

import os
from typing import Optional

from ..scanners.scanner import Scanner
from ..scanners.scanner_dtool import ScannerDTool
from ..scanners.scanner_local import ScannerLocal
from ..utils import solidipes_logging as logging
from ..utils.progress import ProgressBar, get_progress_bar
from ..utils.utils import (
    get_path_relative_to_root,
    get_study_root_path,
)
from .validator import Validator

print = logging.invalidPrint
logger = logging.getLogger()




class CurationValidator(Validator):
    """Validator for curation."""

    def __init__(self, description: str = "All files are valid", **kwargs) -> None:
        super().__init__(description=description, **kwargs)
        self._scanner: Optional[Scanner] = None
        self.remote = False

    @property
    def scanner(self) -> Scanner:
        if self._scanner is None:
            type_scanner = ScannerLocal
            if self.remote:
                type_scanner = ScannerDTool
            self._scanner = type_scanner()

        return self._scanner



    def collect_loader_errors(self, progress, loader) -> None:
        path = get_path_relative_to_root(loader.path)
        try:
            if loader.is_cache_invalid():
                logger.info(f"Detected invalid cache:{path}: {loader.is_cache_invalid()}")
            self.is_cache_invalid |= loader.is_cache_invalid()
        except FileNotFoundError:
            msg = "There are missing files: fresh scan needed"
            if msg not in self._errors:
                self.add_validation_error(msg)
            return

        if isinstance(progress, ProgressBar):
            progress.update(advance=1, text=path)

        validation_results = loader.get_validation_results(exclude=["Ontology is matched"])
        if loader.get_is_valid(validation_results=validation_results):
            return

        error = f'"{path}" is not valid:'

        for validation_result in validation_results:
            if validation_result.valid:
                continue

            for validation_error in validation_result.errors:
                error += f"\n  - {validation_error}"

        self.add_validation_error(error)




    def collect_ontology_errors(self, progress, loader) -> None:
        path = get_path_relative_to_root(loader.path)
        if isinstance(progress, ProgressBar):
            progress.update(advance=1, text=path)

        loader.get_is_valid(include=["Ontology is matched"])




    def _validate(self, obj=None) -> bool:
        from ..viewers import backends

        if backends.current_backend == "streamlit":
            import streamlit as st

            with st.spinner("Scanning files"):
                loader_tree = self.scanner.get_loader_tree()
        else:
            loader_tree = self.scanner.get_loader_tree()

        progress = get_progress_bar("Validating files", total=loader_tree.count)
        self.is_cache_invalid = False
        with progress:
            loader_tree.apply(lambda loader: self.collect_loader_errors(progress, loader))
            from ..loaders import rocrate_metadata

            if self.is_cache_invalid:
                rocrate_metadata.rocrate.write_json()

        progress = get_progress_bar("Matching Ontology", total=loader_tree.count)

        with progress:
            loader_tree.apply(lambda loader: self.collect_ontology_errors(progress, loader))

        return len(self._errors) == 0