diff --git a/DashAI/back/initial_components.py b/DashAI/back/initial_components.py index 97e9efb79..9e19a7bed 100644 --- a/DashAI/back/initial_components.py +++ b/DashAI/back/initial_components.py @@ -350,12 +350,16 @@ from DashAI.back.tasks.translation_task import TranslationTask # Units +from DashAI.back.units.apply_converter_unit import ApplyConverterUnit from DashAI.back.units.build_model_unit import BuildModelUnit from DashAI.back.units.evaluate_model_unit import EvaluateModelUnit +from DashAI.back.units.fit_converter_unit import FitConverterUnit from DashAI.back.units.fit_model_unit import FitModelUnit from DashAI.back.units.load_dataset_unit import LoadDatasetUnit from DashAI.back.units.prepare_and_split_unit import PrepareAndSplitUnit +from DashAI.back.units.save_dataset_unit import SaveDatasetUnit from DashAI.back.units.save_model_unit import SaveModelUnit +from DashAI.back.units.transform_dataset_unit import TransformDatasetUnit logging.basicConfig(level=logging.DEBUG) log = logging.getLogger(__name__) @@ -528,6 +532,10 @@ def get_initial_components(): FitModelUnit, EvaluateModelUnit, SaveModelUnit, + ApplyConverterUnit, + FitConverterUnit, + TransformDatasetUnit, + SaveDatasetUnit, # Explainers ContrastiveShap, DiceCounterfactual, diff --git a/DashAI/back/job/converter_job.py b/DashAI/back/job/converter_job.py index fcbfc4d68..f833dd3b7 100644 --- a/DashAI/back/job/converter_job.py +++ b/DashAI/back/job/converter_job.py @@ -1,123 +1,24 @@ import logging -from typing import TYPE_CHECKING, List +from typing import TYPE_CHECKING from kink import inject from sqlalchemy import exc -from DashAI.back.api.api_v1.schemas.converter_params import ConverterParams from DashAI.back.dependencies.database.models import Converter from DashAI.back.dependencies.database.models import Dataset as DatasetModel from DashAI.back.job.base_job import BaseJob, JobError +from DashAI.back.units.apply_converter_unit import ApplyConverterUnit +from DashAI.back.units.context import ExecutionContext +from DashAI.back.units.load_dataset_unit import LoadDatasetUnit +from DashAI.back.units.save_dataset_unit import SaveDatasetUnit if TYPE_CHECKING: from sqlalchemy.orm import sessionmaker - from DashAI.back.dataloaders.classes.dashai_dataset import DashAIDataset - logging.basicConfig(level=logging.DEBUG) log = logging.getLogger(__name__) -def _rebuild_dataset_with_transformed_columns( - base: "DashAIDataset", - transformed: "DashAIDataset", - scope_column_names: List[str], - scope_column_indexes: List[int], -) -> "DashAIDataset": - """ - Replaces specific columns in the base dataset with columns from the transformed - dataset, preserving their original positions. Also appends any additional columns - that were generated by the transformer at the end. Keeps the features and metadata - consistent. - - Parameters - ---------- - base : DashAIDataset - The original dataset before transformation. - - transformed : DashAIDataset - The dataset resulting from applying a transformer, containing updated and/or - new columns. - - scope_column_names : List[str] - Names of the columns that were originally selected for transformation. - - scope_column_indexes : List[int] - The indices of the columns in the base dataset that were replaced. - Must match the order of scope_column_names. - - Returns - ------- - DashAIDataset - A new dataset with the specified columns replaced in place, new columns - appended, and original metadata and split information preserved. - """ - from DashAI.back.dataloaders.classes.dashai_dataset import modify_table - - original_columns = base.column_names - transformed_cols = transformed.column_names - - transformed_cols_set = set(transformed_cols) - scope_column_names_set = set(scope_column_names) - - removed_cols = [ - col for col in scope_column_names if col not in transformed_cols_set - ] - replacement_cols = [ - col for col in scope_column_names if col in transformed_cols_set - ] - new_cols = [col for col in transformed_cols if col not in scope_column_names_set] - - removed_cols_set = set(removed_cols) - - new_columns_order = [] - seen_cols = set() - for col in original_columns: - if col in removed_cols_set: - continue - if col not in seen_cols: - new_columns_order.append(col) - seen_cols.add(col) - - col_name_mapping = {} - for col in new_cols: - unique_col = col - counter = 1 - while unique_col in seen_cols: - unique_col = f"{col}_{counter}" - counter += 1 - new_columns_order.append(unique_col) - seen_cols.add(unique_col) - col_name_mapping[col] = unique_col - - updated_arrays = {} - for col in replacement_cols: - if col in transformed_cols_set: - updated_arrays[col] = transformed.arrow_table[col] - for col, unique_col in col_name_mapping.items(): - if col in transformed_cols_set: - updated_arrays[unique_col] = transformed.arrow_table[col] - - updated_types = base.types.copy() - - for col in removed_cols: - if col in updated_types: - del updated_types[col] - - for col in replacement_cols: - if col in transformed.types: - updated_types[col] = transformed.types[col] - for col, unique_col in col_name_mapping.items(): - if col in transformed.types: - updated_types[unique_col] = transformed.types[col] - - # Use existing modify_table (imported at module level) - modified_dataset = modify_table(base, updated_arrays, types=updated_types) - modified_dataset = modified_dataset.select_columns(new_columns_order) - - return modified_dataset - - class ConverterJob(BaseJob): """ConverterJob class to modify a dataset by applying a sequence of converters.""" @@ -198,31 +99,9 @@ def run( ) -> None: from kink import di - from DashAI.back.dataloaders.classes.dashai_dataset import ( - load_dataset, - save_dataset, - ) - session_factory = di["session_factory"] - component_registry = di["component_registry"] - - def instantiate_converters( - converter_name: str, - converter_params: ConverterParams, - ) -> object: - # Import the converter - try: - converter_constructor = component_registry[converter_name]["class"] - except KeyError as e: - log.exception(e) - raise JobError( - f"Error importing converter {converter_name}: {e}" - ) from e - - # Get parameters or empty dict if none - converter_parameters = converter_params.get("params", {}) - return converter_constructor(**converter_parameters) + ctx = ExecutionContext() # Extract job parameters converter_id = self.kwargs["converter_id"] @@ -246,38 +125,48 @@ def instantiate_converters( # Get dataset try: + notebook_id = converter.notebook_id dataset_id = converter.notebook.dataset_id dataset = db.get(DatasetModel, dataset_id) - # dataset to edit - dataset_path = f"{converter.notebook.file_path}/dataset" - loaded_dataset = load_dataset(dataset_path) - params = converter.parameters or {} - target_column_index = ( - params["target"].get("idx") - if params.get("target") is not None - else None - ) + # dataset to edit: the notebook's own working copy + LoadDatasetUnit(notebook_id=notebook_id)(ctx) + dataset_path = ctx.require("dataset_path") - if not loaded_dataset: - raise JobError(f"Dataset with path {dataset_path} not found") + # How the converter configuration is stored on the row, not + # part of the transformation itself. + params = converter.parameters or {} except exc.SQLAlchemyError as e: log.exception(e) converter.set_status_as_error() db.commit() raise JobError("Error loading dataset info") from e - - # Load dataset + except Exception: + # Anything the load unit raises (missing notebook, unreadable + # dataset) also has to leave the row in ERROR. Nothing marks it + # otherwise: the Huey error signal only writes to its own + # task_copy table, never to the Converter row, so without this + # the converter would stay STARTED forever. Re-raised as-is so + # the unit's specific message survives. + converter.set_status_as_error() + db.commit() + raise + + apply_converter = ApplyConverterUnit( + converter={ + "component": converter.converter, + "params": params.get("params") or {}, + }, + scope=params.get("scope"), + target=params.get("target"), + ) + + # Validating before the work starts keeps an impossible target + # index reported as a dataset problem, which is where it was + # reported before the job was split into units. try: - # Validate target column index - if target_column_index is not None and ( - int(target_column_index) < 1 - or int(target_column_index) > len(loaded_dataset.features) - ): - raise JobError( - f"Target column index {target_column_index} is out of bounds" - ) + apply_converter.validate(ctx) except Exception as e: log.exception(e) converter.set_status_as_error() @@ -285,142 +174,11 @@ def instantiate_converters( raise JobError(f"Cannot load dataset from {dataset_path}") from e try: - # Get stored converter configurations - converters_stored_info = {converter.converter: converter.parameters} - dataset_original_columns = loaded_dataset.column_names - - # Sort converters by order - converters_sorted_list = sorted( - converters_stored_info.items(), key=lambda x: x[1]["order"] - ) - - i = 0 - converter_instances = [] - - while i < len(converters_sorted_list): - converter_name = converters_sorted_list[i][0] - converter_params = converters_sorted_list[i][1] - # Regular converter - converter_instance = instantiate_converters( - converter_name, - converter_params, - ) - - # Get scope or use default - scope = converter_params.get("scope", {"columns": [], "rows": []}) - - # Add to instances - converter_instances.append( - { - "name": converter_name, - "instance": converter_instance, - "scope": scope, - } - ) - i += 1 - - # Apply each converter in sequence - total_converters = len(converter_instances) - for converter_index, converter_info in enumerate(converter_instances): - converter_instance = converter_info["instance"] - converter_name = converter_info["name"] - converter_scope = converter_info["scope"] - - # Map converter progress onto the 0.2-0.9 band. - self.report_progress( - 0.2 + 0.7 * (converter_index / max(total_converters, 1)), - f"Applying {converter_name}", - ) - log.info(f"Applying converter: {converter_name}") - - columns_scope = [ - column["idx"] - 1 for column in converter_scope["columns"] - ] - scope_column_indexes = sorted(set(columns_scope)) - - if not scope_column_indexes: - scope_column_indexes = list(range(len(loaded_dataset.features))) - - scope_column_names = [ - dataset_original_columns[index] - for index in scope_column_indexes - ] - - rows_scope = [row - 1 for row in converter_scope["rows"]] - scope_rows_indexes = sorted(set(rows_scope)) - - y_dataset_fit = None - target_column_name = None - y_full_transform = None - if target_column_index is not None: - target_column_index_0based = int(target_column_index) - 1 - target_column_name = dataset_original_columns[ - target_column_index_0based - ] - y_dataset_fit = loaded_dataset.select_columns( - [target_column_name] - ) - if scope_rows_indexes: - y_dataset_fit = y_dataset_fit.select(scope_rows_indexes) - y_full_transform = loaded_dataset.select_columns( - [target_column_name] - ) - else: - y_full_transform = y_dataset_fit - - X_dataset_fit = loaded_dataset.select_columns(scope_column_names) - - if scope_rows_indexes: - X_dataset_fit = X_dataset_fit.select(scope_rows_indexes) - - try: - converter_instance = converter_instance.fit( - X_dataset_fit, y_dataset_fit - ) - except ValueError as e: - log.error(f"Validation error in {converter_name}: {e}") - raise JobError( - f"Validation error fitting {converter_name}: {e}" - ) from e - except Exception as e: - log.exception(e) - raise JobError( - f"Error fitting converter {converter_name}: {e}" - ) from e - - if scope_rows_indexes: - X_full_transform = loaded_dataset.select_columns( - scope_column_names - ) - else: - # Same reuse as above: no row-level fit scope means - # X_dataset_fit already covers the full transform scope. - X_full_transform = X_dataset_fit - - try: - transformed_dataset = converter_instance.transform( - X_full_transform, y_full_transform - ) - except Exception as e: - log.exception(e) - raise JobError( - f"Error transforming data with {converter_name}: {e}" - ) from e - - if type(converter_instance).CHANGES_ROW_COUNT: - loaded_dataset = transformed_dataset - else: - loaded_dataset = _rebuild_dataset_with_transformed_columns( - loaded_dataset, - transformed_dataset, - scope_column_names, - scope_column_indexes, - ) - - dataset_original_columns = loaded_dataset.column_names + self.report_progress(0.2, f"Applying {converter.converter}") + apply_converter(ctx) self.report_progress(0.95, "Saving dataset") - save_dataset(loaded_dataset, f"{dataset_path}") + SaveDatasetUnit()(ctx) converter.set_status_as_finished() db.commit() db.refresh(dataset) @@ -432,3 +190,5 @@ def instantiate_converters( raise JobError( f"Error applying converters to dataset {dataset_id}: {e}" ) from e + finally: + ctx.clear_cache() diff --git a/DashAI/back/units/apply_converter_unit.py b/DashAI/back/units/apply_converter_unit.py new file mode 100644 index 000000000..da4bc636d --- /dev/null +++ b/DashAI/back/units/apply_converter_unit.py @@ -0,0 +1,145 @@ +"""Unit that fits a converter and transforms one dataset with it.""" + +import logging +from typing import TYPE_CHECKING + +from DashAI.back.core.schema_fields import BaseSchema +from DashAI.back.job.base_job import JobError +from DashAI.back.units.base_unit import BaseUnit +from DashAI.back.units.context import ExecutionContext +from DashAI.back.units.converter_scope import ( + ConverterScopeMixin, + converter_field, + scope_field, + target_field, +) + +if TYPE_CHECKING: + from DashAI.back.dataloaders.classes.dashai_dataset import DashAIDataset + +log = logging.getLogger(__name__) + + +class ApplyConverterSchema(BaseSchema): + converter: converter_field() # type: ignore + scope: scope_field() # type: ignore + target: target_field() # type: ignore + + +class ApplyConverterUnit(BaseUnit, ConverterScopeMixin): + """Fit one converter on the scoped data and transform the same dataset. + + The single-dataset case, which is what applying a converter to a notebook + means. Equivalent to :class:`FitConverterUnit` followed by + :class:`TransformDatasetUnit` on the same dataset, kept as one unit for two + reasons: it is the common case, and when there is no row scope it hands + ``transform`` the very object ``fit`` saw, which some converters rely on to + skip recomputing (see below). + + Use the split pair instead whenever the converter has to be learned from one + dataset and applied to another — fit on train, transform on test. + + Reads ``dataset`` and writes ``dataset``: the same key on both sides, so any + number of these can be chained in one context, each one seeing what the + previous one produced. + + That is also why nothing about column identity ever crosses the context + boundary. The scope is expressed as 1-based column and row indexes, and + those indexes are resolved against ``dataset.column_names`` read at the top + of ``execute`` — the dataset as it is *right now*. A converter that renames, + drops or adds columns changes what index 3 means, so a resolved column list + published to the context would be stale for the very next converter. + """ + + SCHEMA = ApplyConverterSchema + + REQUIRES = ("dataset",) + PROVIDES = ("dataset", "fitted_converter") + + def __init__(self, **config) -> None: + super().__init__(**config) + # Memoized on the instance, never in the context: two units of this + # class can live in the same context and they are different converters. + self._converter_class = None + + @property + def _converter_name(self) -> str: + return self.config["converter"]["component"] + + def _resolve_converter(self): + """Look the converter class up in the registry, once per instance.""" + if self._converter_class is not None: + return self._converter_class + + from kink import di + + component_registry = di["component_registry"] + converter_name = self._converter_name + + try: + self._converter_class = component_registry[converter_name]["class"] + except KeyError as e: + log.exception(e) + raise JobError(f"Error importing converter {converter_name}: {e}") from e + + return self._converter_class + + def validate(self, ctx: ExecutionContext) -> None: + """Reject an out-of-bounds target before any work is done.""" + self._check_target_bounds(ctx.require("dataset")) + + def execute(self, ctx: ExecutionContext) -> None: + loaded_dataset: "DashAIDataset" = ctx.require("dataset") + converter_name = self._converter_name + + converter_constructor = self._resolve_converter() + converter_instance = converter_constructor( + **(self.config["converter"].get("params") or {}) + ) + + self._check_target_bounds(loaded_dataset) + ( + scope_column_names, + scope_rows_indexes, + target_column_name, + ) = self._resolve_scope(loaded_dataset) + + log.info(f"Applying converter: {converter_name}") + + x_dataset_fit, y_dataset_fit = self._slice_for_fit( + loaded_dataset, + scope_column_names, + scope_rows_indexes, + target_column_name, + ) + + converter_instance = self._fit( + converter_instance, x_dataset_fit, y_dataset_fit, converter_name + ) + + if scope_rows_indexes: + x_full_transform, y_full_transform = self._slice_for_transform( + loaded_dataset, scope_column_names, target_column_name + ) + else: + # Deliberately the *same objects* as the ones passed to fit, not + # equal ones: converters such as TypeCast key a cache off the + # identity of the dataset they were fitted on and skip recomputing + # when transform receives it again. Without a row scope the fit + # slice already covers every row, so reusing it is also correct. + x_full_transform, y_full_transform = x_dataset_fit, y_dataset_fit + + transformed_dataset = self._transform( + converter_instance, x_full_transform, y_full_transform, converter_name + ) + + ctx.put( + "dataset", + self._merge_transformed( + converter_instance, + loaded_dataset, + transformed_dataset, + scope_column_names, + ), + ) + ctx.put("fitted_converter", converter_instance) diff --git a/DashAI/back/units/build_model_unit.py b/DashAI/back/units/build_model_unit.py index 45c14c936..3f39bb1b6 100644 --- a/DashAI/back/units/build_model_unit.py +++ b/DashAI/back/units/build_model_unit.py @@ -120,7 +120,11 @@ class BuildModelUnit(BaseUnit): SCHEMA = BuildModelSchema - REQUIRES = ("x", "y", "n_labels") + # run_id and task_name only appear in the ModelFactory call and in error + # messages, but they are declared all the same: a key read without being + # declared is invisible to any caller — and to any future DAG validator — + # that inspects REQUIRES instead of running the unit. + REQUIRES = ("x", "y", "n_labels", "run_id", "task_name") PROVIDES = ("model", "factory", "optimizable_parameters", "model_parameters") def __init__(self, **config) -> None: @@ -195,6 +199,8 @@ def execute(self, ctx: ExecutionContext) -> None: component_registry = di["component_registry"] parameters = self.model_parameters + run_id = ctx.require("run_id") + task_name = ctx.require("task_name") model_class = self._resolve_model_class() @@ -212,15 +218,14 @@ def execute(self, ctx: ExecutionContext) -> None: except Exception as e: log.exception(e) raise JobError( - "Unable to find metrics associated with" - f"Task {ctx.get('task_name')} in registry", + f"Unable to find metrics associated with Task {task_name} in registry", ) from e try: factory = ModelFactory( model_class, parameters, - ctx.get("run_id"), + run_id, ctx.require("x"), ctx.require("y"), train_metrics, @@ -232,7 +237,7 @@ def execute(self, ctx: ExecutionContext) -> None: except Exception as e: log.exception(e) raise JobError( - f"Unable to instantiate model using run {ctx.get('run_id')}", + f"Unable to instantiate model using run {run_id}", ) from e # The original tree is what the search unit rewrites with the best diff --git a/DashAI/back/units/converter_scope.py b/DashAI/back/units/converter_scope.py new file mode 100644 index 000000000..73a10a712 --- /dev/null +++ b/DashAI/back/units/converter_scope.py @@ -0,0 +1,357 @@ +"""Shared scope resolution and slicing for the converter units. + +The three converter units — fit, transform, and the fused apply — all express +which part of a dataset they touch the same way, and all resolve it the same +way. That logic lives here so the fused path and the split path cannot drift +apart. +""" + +import logging +from typing import TYPE_CHECKING, Dict, List, Optional, Tuple + +from DashAI.back.core.schema_fields import ( + component_field, + none_type, + schema_field, +) +from DashAI.back.core.utils import MultilingualString +from DashAI.back.job.base_job import JobError + +if TYPE_CHECKING: + from DashAI.back.dataloaders.classes.dashai_dataset import DashAIDataset + +log = logging.getLogger(__name__) + +EMPTY_SCOPE: Dict[str, List] = {"columns": [], "rows": []} + + +def converter_field(): + """Schema field for picking a converter and configuring it.""" + return schema_field( + component_field(parent="BaseConverter"), + placeholder={"component": "ColumnRemover", "params": {}}, + description=MultilingualString( + en="Converter to apply, together with its own configuration.", + es="Convertidor a aplicar, junto con su propia configuración.", + pt="Conversor a aplicar, junto com a sua própria configuração.", + de="Anzuwendender Konverter samt seiner eigenen Konfiguration.", + zh="要应用的转换器及其自身的配置。", + ), + alias=MultilingualString( + en="Converter", + es="Convertidor", + pt="Conversor", + de="Konverter", + zh="转换器", + ), + ) + + +def scope_field(): + """Schema field for the part of the dataset a converter touches.""" + return schema_field( + none_type(dict), + placeholder={"columns": [], "rows": []}, + description=MultilingualString( + en="Part of the dataset the converter applies to: 'columns' is a " + "list of {'idx': n} and 'rows' a list of n, both 1-based. An " + "empty column list means every column.", + es="Parte del conjunto de datos a la que se aplica el convertidor: " + "'columns' es una lista de {'idx': n} y 'rows' una lista de n, " + "ambas con base 1. Una lista de columnas vacía significa todas " + "las columnas.", + pt="Parte do conjunto de dados à qual o conversor se aplica: " + "'columns' é uma lista de {'idx': n} e 'rows' uma lista de n, " + "ambas com base 1. Uma lista de colunas vazia significa todas " + "as colunas.", + de="Teil des Datensatzes, auf den der Konverter angewendet wird: " + "'columns' ist eine Liste von {'idx': n} und 'rows' eine Liste " + "von n, beide 1-basiert. Eine leere Spaltenliste bedeutet alle " + "Spalten.", + zh="转换器作用的数据集范围：'columns' 是 {'idx': n} 的列表，" + "'rows' 是 n 的列表，均从 1 开始。空的列列表表示所有列。", + ), + alias=MultilingualString( + en="Scope", + es="Alcance", + pt="Escopo", + de="Geltungsbereich", + zh="范围", + ), + ) + + +def target_field(): + """Schema field for the column handed to the converter as ``y``.""" + return schema_field( + none_type(dict), + placeholder=None, + description=MultilingualString( + en="Target column handed to the converter as y, as {'idx': n} with " + "n 1-based. Leave empty for unsupervised converters.", + es="Columna objetivo entregada al convertidor como y, como " + "{'idx': n} con n en base 1. Dejar vacío para convertidores no " + "supervisados.", + pt="Coluna alvo entregue ao conversor como y, como {'idx': n} com " + "n com base 1. Deixe vazio para conversores não supervisionados.", + de="Zielspalte, die dem Konverter als y übergeben wird, als " + "{'idx': n} mit 1-basiertem n. Für unüberwachte Konverter leer " + "lassen.", + zh="作为 y 传给转换器的目标列，格式为 {'idx': n}，n 从 1 开始。" + "无监督转换器留空。", + ), + alias=MultilingualString( + en="Target column", + es="Columna objetivo", + pt="Coluna alvo", + de="Zielspalte", + zh="目标列", + ), + ) + + +def rebuild_dataset_with_transformed_columns( + base: "DashAIDataset", + transformed: "DashAIDataset", + scope_column_names: List[str], +) -> "DashAIDataset": + """ + Replaces specific columns in the base dataset with columns from the transformed + dataset, preserving their original positions. Also appends any additional columns + that were generated by the transformer at the end. Keeps the features and metadata + consistent. + + Parameters + ---------- + base : DashAIDataset + The original dataset before transformation. + + transformed : DashAIDataset + The dataset resulting from applying a transformer, containing updated and/or + new columns. + + scope_column_names : List[str] + Names of the columns that were originally selected for transformation. + + Returns + ------- + DashAIDataset + A new dataset with the specified columns replaced in place, new columns + appended, and original metadata and split information preserved. + """ + from DashAI.back.dataloaders.classes.dashai_dataset import modify_table + + original_columns = base.column_names + transformed_cols = transformed.column_names + + transformed_cols_set = set(transformed_cols) + scope_column_names_set = set(scope_column_names) + + removed_cols = [ + col for col in scope_column_names if col not in transformed_cols_set + ] + replacement_cols = [ + col for col in scope_column_names if col in transformed_cols_set + ] + new_cols = [col for col in transformed_cols if col not in scope_column_names_set] + + removed_cols_set = set(removed_cols) + + new_columns_order = [] + seen_cols = set() + for col in original_columns: + if col in removed_cols_set: + continue + if col not in seen_cols: + new_columns_order.append(col) + seen_cols.add(col) + + col_name_mapping = {} + for col in new_cols: + unique_col = col + counter = 1 + while unique_col in seen_cols: + unique_col = f"{col}_{counter}" + counter += 1 + new_columns_order.append(unique_col) + seen_cols.add(unique_col) + col_name_mapping[col] = unique_col + + updated_arrays = {} + for col in replacement_cols: + if col in transformed_cols_set: + updated_arrays[col] = transformed.arrow_table[col] + for col, unique_col in col_name_mapping.items(): + if col in transformed_cols_set: + updated_arrays[unique_col] = transformed.arrow_table[col] + + updated_types = base.types.copy() + + for col in removed_cols: + if col in updated_types: + del updated_types[col] + + for col in replacement_cols: + if col in transformed.types: + updated_types[col] = transformed.types[col] + for col, unique_col in col_name_mapping.items(): + if col in transformed.types: + updated_types[unique_col] = transformed.types[col] + + modified_dataset = modify_table(base, updated_arrays, types=updated_types) + modified_dataset = modified_dataset.select_columns(new_columns_order) + + return modified_dataset + + +class ConverterScopeMixin: + """Scope handling shared by the converter units. + + Deliberately **not** named ``Base*`` and deliberately not inheriting from + ``BaseUnit``. ``ComponentRegistry._get_base_type`` walks the MRO for + ancestors whose name contains "Base" and that declare a ``TYPE``, and + rejects a component with more than one candidate — an intermediate class + called ``BaseConverterUnit`` would be a second candidate and break the + registration of every unit inheriting from it. + """ + + def _scope(self) -> Dict[str, List]: + """The configured scope, tolerating an explicit ``None``. + + The API schema always writes the ``scope`` key but allows it to be null, + so a converter saved without a scope arrives here as ``None`` rather + than as a missing key. + """ + return self.config.get("scope") or EMPTY_SCOPE + + def _target_index(self) -> Optional[int]: + """The 1-based target column index, or None when there is no target.""" + target = self.config.get("target") + if target is None: + return None + return target.get("idx") + + def _check_target_bounds(self, dataset: "DashAIDataset") -> None: + """Reject a target index that does not point at a column. + + Checked against the dataset in hand rather than once up front: when + several converters are chained the column count changes underneath, and + an unchecked index would surface as a bare IndexError. + """ + target_column_index = self._target_index() + if target_column_index is None: + return + + if int(target_column_index) < 1 or int(target_column_index) > len( + dataset.features + ): + raise JobError( + f"Target column index {target_column_index} is out of bounds" + ) + + def _resolve_scope( + self, dataset: "DashAIDataset" + ) -> Tuple[List[str], List[int], Optional[str]]: + """Turn the 1-based scope into names and indexes for **this** dataset. + + Always resolved against the dataset in hand, never against a list + carried through the context: a converter that renames, drops or adds + columns changes what index 3 means for whatever runs next. + """ + scope = self._scope() + column_names = dataset.column_names + + columns_scope = [column["idx"] - 1 for column in scope["columns"]] + scope_column_indexes = sorted(set(columns_scope)) + + if not scope_column_indexes: + scope_column_indexes = list(range(len(dataset.features))) + + scope_column_names = [column_names[index] for index in scope_column_indexes] + + rows_scope = [row - 1 for row in scope["rows"]] + scope_rows_indexes = sorted(set(rows_scope)) + + target_column_name = None + target_column_index = self._target_index() + if target_column_index is not None: + target_column_name = column_names[int(target_column_index) - 1] + + return scope_column_names, scope_rows_indexes, target_column_name + + def _slice_for_fit( + self, + dataset: "DashAIDataset", + scope_column_names: List[str], + scope_rows_indexes: List[int], + target_column_name: Optional[str], + ) -> Tuple["DashAIDataset", Optional["DashAIDataset"]]: + """The X and y a converter is fitted on: scoped columns, scoped rows.""" + y_dataset = None + if target_column_name is not None: + y_dataset = dataset.select_columns([target_column_name]) + if scope_rows_indexes: + y_dataset = y_dataset.select(scope_rows_indexes) + + x_dataset = dataset.select_columns(scope_column_names) + if scope_rows_indexes: + x_dataset = x_dataset.select(scope_rows_indexes) + + return x_dataset, y_dataset + + def _slice_for_transform( + self, + dataset: "DashAIDataset", + scope_column_names: List[str], + target_column_name: Optional[str], + ) -> Tuple["DashAIDataset", Optional["DashAIDataset"]]: + """The X and y a converter transforms: scoped columns, **every** row. + + A row scope narrows what the converter learns from, never what it is + applied to. + """ + y_dataset = None + if target_column_name is not None: + y_dataset = dataset.select_columns([target_column_name]) + + return dataset.select_columns(scope_column_names), y_dataset + + def _merge_transformed( + self, + converter_instance, + dataset: "DashAIDataset", + transformed_dataset: "DashAIDataset", + scope_column_names: List[str], + ) -> "DashAIDataset": + """Fold the transform output back into the dataset. + + A converter that changes the row count cannot have its columns merged + back position by position, so its output replaces the dataset outright. + """ + if type(converter_instance).CHANGES_ROW_COUNT: + return transformed_dataset + + return rebuild_dataset_with_transformed_columns( + dataset, + transformed_dataset, + scope_column_names, + ) + + def _fit(self, converter_instance, x_dataset, y_dataset, converter_name): + """Fit a converter, preserving the job's two error messages.""" + try: + return converter_instance.fit(x_dataset, y_dataset) + except ValueError as e: + log.error(f"Validation error in {converter_name}: {e}") + raise JobError(f"Validation error fitting {converter_name}: {e}") from e + except Exception as e: + log.exception(e) + raise JobError(f"Error fitting converter {converter_name}: {e}") from e + + def _transform(self, converter_instance, x_dataset, y_dataset, converter_name): + """Transform with a fitted converter, preserving the job's message.""" + try: + return converter_instance.transform(x_dataset, y_dataset) + except Exception as e: + log.exception(e) + raise JobError(f"Error transforming data with {converter_name}: {e}") from e diff --git a/DashAI/back/units/fit_converter_unit.py b/DashAI/back/units/fit_converter_unit.py new file mode 100644 index 000000000..645f10d52 --- /dev/null +++ b/DashAI/back/units/fit_converter_unit.py @@ -0,0 +1,107 @@ +"""Unit that fits a converter on a dataset without transforming anything.""" + +import logging +from typing import TYPE_CHECKING + +from DashAI.back.core.schema_fields import BaseSchema +from DashAI.back.job.base_job import JobError +from DashAI.back.units.base_unit import BaseUnit +from DashAI.back.units.context import ExecutionContext +from DashAI.back.units.converter_scope import ( + ConverterScopeMixin, + converter_field, + scope_field, + target_field, +) + +if TYPE_CHECKING: + from DashAI.back.dataloaders.classes.dashai_dataset import DashAIDataset + +log = logging.getLogger(__name__) + + +class FitConverterSchema(BaseSchema): + converter: converter_field() # type: ignore + scope: scope_field() # type: ignore + target: target_field() # type: ignore + + +class FitConverterUnit(BaseUnit, ConverterScopeMixin): + """Fit a converter on the dataset in the context and publish it fitted. + + Splits the "fit" half out of :class:`ApplyConverterUnit` so a converter can + be learned from one dataset and applied to another. That is the standard + train/test discipline: a scaler, an encoder or an imputer must learn its + statistics from the training data only, and then be applied unchanged to the + test data — refitting on test would leak information into the evaluation. + + The dataset is left untouched: this unit produces a fitted converter, not + data. Pair it with :class:`TransformDatasetUnit`, once per dataset the + fitted converter has to be applied to. + """ + + SCHEMA = FitConverterSchema + + REQUIRES = ("dataset",) + PROVIDES = ("fitted_converter",) + + def __init__(self, **config) -> None: + super().__init__(**config) + # Memoized on the instance, never in the context: two units of this + # class can live in the same context and they are different converters. + self._converter_class = None + + @property + def _converter_name(self) -> str: + return self.config["converter"]["component"] + + def _resolve_converter(self): + """Look the converter class up in the registry, once per instance.""" + if self._converter_class is not None: + return self._converter_class + + from kink import di + + component_registry = di["component_registry"] + converter_name = self._converter_name + + try: + self._converter_class = component_registry[converter_name]["class"] + except KeyError as e: + log.exception(e) + raise JobError(f"Error importing converter {converter_name}: {e}") from e + + return self._converter_class + + def validate(self, ctx: ExecutionContext) -> None: + """Reject an out-of-bounds target before any work is done.""" + self._check_target_bounds(ctx.require("dataset")) + + def execute(self, ctx: ExecutionContext) -> None: + dataset: "DashAIDataset" = ctx.require("dataset") + converter_name = self._converter_name + + converter_constructor = self._resolve_converter() + converter_instance = converter_constructor( + **(self.config["converter"].get("params") or {}) + ) + + self._check_target_bounds(dataset) + ( + scope_column_names, + scope_rows_indexes, + target_column_name, + ) = self._resolve_scope(dataset) + + log.info(f"Fitting converter: {converter_name}") + + x_dataset, y_dataset = self._slice_for_fit( + dataset, + scope_column_names, + scope_rows_indexes, + target_column_name, + ) + + fitted = self._fit(converter_instance, x_dataset, y_dataset, converter_name) + + ctx.put("fitted_converter", fitted) diff --git a/DashAI/back/units/fit_model_unit.py b/DashAI/back/units/fit_model_unit.py index 8fc16f958..8f9c6bf14 100644 --- a/DashAI/back/units/fit_model_unit.py +++ b/DashAI/back/units/fit_model_unit.py @@ -92,18 +92,26 @@ class FitModelUnit(BaseUnit): "x", "y", "task", + "run_id", ) PROVIDES = ("model", "plot_paths") - def validate(self, ctx: ExecutionContext) -> None: - # ctx.require, not ctx.get: "optimizable_parameters" is one of this - # unit's REQUIRES, so its absence means BuildModelUnit hasn't run yet - # — a call-order mistake, not "there is nothing to optimize". Only an - # empty value (the key present, genuinely no optimizable parameters) - # skips the optimizer/goal-metric checks below, so no registry lookup - # is needed either. - if not ctx.require("optimizable_parameters"): - return + def __init__(self, **config) -> None: + super().__init__(**config) + self._optimizer = None + self._goal_metric = None + + def _resolve_search(self): + """Resolve the optimizer and the goal metric, memoized on this unit. + + Kept on the instance rather than in the context on purpose. These are + this unit's own state, not something it hands to another unit: two + ``FitModelUnit`` instances sharing a context — a DAG with two training + nodes — would otherwise overwrite each other's optimizer, and the + second one would silently run the first one's. + """ + if self._optimizer is not None: + return self._optimizer, self._goal_metric from kink import di @@ -132,8 +140,21 @@ def validate(self, ctx: ExecutionContext) -> None: f"Error instantiating optimizer {optimizer_name}, {e}", ) from e - ctx.put("goal_metric", goal_metric) - ctx.put("optimizer", optimizer) + self._goal_metric = goal_metric + self._optimizer = optimizer + return optimizer, goal_metric + + def validate(self, ctx: ExecutionContext) -> None: + # ctx.require, not ctx.get: "optimizable_parameters" is one of this + # unit's REQUIRES, so its absence means BuildModelUnit hasn't run yet + # — a call-order mistake, not "there is nothing to optimize". Only an + # empty value (the key present, genuinely no optimizable parameters) + # skips the optimizer/goal-metric checks below, so no registry lookup + # is needed either. + if not ctx.require("optimizable_parameters"): + return + + self._resolve_search() def execute(self, ctx: ExecutionContext) -> None: import os @@ -146,6 +167,7 @@ def execute(self, ctx: ExecutionContext) -> None: model = ctx.require("model") x = ctx.require("x") y = ctx.require("y") + run_id = ctx.require("run_id") optimizable_parameters = ctx.require("optimizable_parameters") plot_paths = [] @@ -153,12 +175,10 @@ def execute(self, ctx: ExecutionContext) -> None: if not optimizable_parameters: model.train(x["train"], y["train"], x["validation"], y["validation"]) else: - # __call__ always runs validate() immediately before execute(), - # so "optimizer"/"goal_metric" are already in ctx here. - optimizer = ctx.require("optimizer") - goal_metric = ctx.require("goal_metric") + # Memoized: validate() resolved these already, and resolving + # again here would be the same lookup. + optimizer, goal_metric = self._resolve_search() factory = ctx.require("factory") - run_id = ctx.get("run_id") optimizer.optimize( model, diff --git a/DashAI/back/units/load_dataset_unit.py b/DashAI/back/units/load_dataset_unit.py index dc3244645..cac25e15c 100644 --- a/DashAI/back/units/load_dataset_unit.py +++ b/DashAI/back/units/load_dataset_unit.py @@ -3,9 +3,14 @@ import logging from typing import TYPE_CHECKING -from DashAI.back.core.schema_fields import BaseSchema, int_field, schema_field +from DashAI.back.core.schema_fields import ( + BaseSchema, + int_field, + none_type, + schema_field, +) from DashAI.back.core.utils import MultilingualString -from DashAI.back.dependencies.database.models import Dataset +from DashAI.back.dependencies.database.models import Dataset, Notebook from DashAI.back.job.base_job import JobError from DashAI.back.units.base_unit import BaseUnit from DashAI.back.units.context import ExecutionContext @@ -18,14 +23,18 @@ class LoadDatasetSchema(BaseSchema): dataset_id: schema_field( - int_field(gt=0), - placeholder=1, + none_type(int_field(gt=0)), + placeholder=None, description=MultilingualString( - en="Identifier of the stored dataset to load.", - es="Identificador del conjunto de datos almacenado a cargar.", - pt="Identificador do conjunto de dados armazenado a carregar.", - de="Kennung des zu ladenden gespeicherten Datensatzes.", - zh="要加载的已存储数据集的标识符。", + en="Identifier of the stored dataset to load. Mutually exclusive " + "with the notebook identifier.", + es="Identificador del conjunto de datos almacenado a cargar. " + "Excluyente con el identificador del cuaderno.", + pt="Identificador do conjunto de dados armazenado a carregar. " + "Mutuamente exclusivo com o identificador do caderno.", + de="Kennung des zu ladenden gespeicherten Datensatzes. Schließt " + "die Notebook-Kennung aus.", + zh="要加载的已存储数据集的标识符。与笔记本标识符互斥。", ), alias=MultilingualString( en="Dataset", @@ -35,18 +44,51 @@ class LoadDatasetSchema(BaseSchema): zh="数据集", ), ) # type: ignore + notebook_id: schema_field( + none_type(int_field(gt=0)), + placeholder=None, + description=MultilingualString( + en="Identifier of the notebook whose working copy of the dataset " + "should be loaded. Mutually exclusive with the dataset identifier.", + es="Identificador del cuaderno cuya copia de trabajo del conjunto " + "de datos se debe cargar. Excluyente con el identificador del " + "conjunto de datos.", + pt="Identificador do caderno cuja cópia de trabalho do conjunto de " + "dados deve ser carregada. Mutuamente exclusivo com o " + "identificador do conjunto de dados.", + de="Kennung des Notebooks, dessen Arbeitskopie des Datensatzes " + "geladen werden soll. Schließt die Datensatz-Kennung aus.", + zh="要加载其数据集工作副本的笔记本标识符。与数据集标识符互斥。", + ), + alias=MultilingualString( + en="Notebook", + es="Cuaderno", + pt="Caderno", + de="Notebook", + zh="笔记本", + ), + ) # type: ignore class LoadDatasetUnit(BaseUnit): """Load a dataset from disk into the execution context. - Resolves the dataset row to find where it is stored and materialises it, so - downstream units receive a dataset instead of an identifier. + Takes exactly one of two starting points and materialises the dataset each + one points at, so downstream units receive a dataset instead of an + identifier: + + * ``dataset_id``: the stored dataset itself, read from ``Dataset.file_path``. + * ``notebook_id``: the notebook's own working copy, read from + ``Notebook.file_path``. A notebook holds a private copy precisely so + converters can rewrite it without touching the source dataset. + + Either way the unit publishes ``dataset_path``, which is where a later unit + has to write the dataset back for the change to be visible. """ SCHEMA = LoadDatasetSchema - PROVIDES = ("dataset",) + PROVIDES = ("dataset", "dataset_id", "dataset_path") def execute(self, ctx: ExecutionContext) -> None: from kink import di @@ -54,21 +96,43 @@ def execute(self, ctx: ExecutionContext) -> None: from DashAI.back.dataloaders.classes.dashai_dataset import load_dataset session_factory = di["session_factory"] - dataset_id: int = self.config["dataset_id"] + + dataset_id = self.config.get("dataset_id") + notebook_id = self.config.get("notebook_id") + + if (dataset_id is None) == (notebook_id is None): + raise JobError( + "LoadDatasetUnit needs exactly one of dataset_id or notebook_id." + ) with session_factory() as db: - dataset: Dataset = db.get(Dataset, dataset_id) - if not dataset: - raise JobError(f"Dataset {dataset_id} does not exist in DB.") - file_path = dataset.file_path + if dataset_id is not None: + dataset: Dataset = db.get(Dataset, dataset_id) + if not dataset: + raise JobError(f"Dataset {dataset_id} does not exist in DB.") + file_path = dataset.file_path + else: + notebook: Notebook = db.get(Notebook, notebook_id) + if not notebook: + raise JobError(f"Notebook {notebook_id} does not exist in DB.") + file_path = notebook.file_path + # The notebook's copy still belongs to a source dataset, and + # downstream error messages identify the work by that id. + dataset_id = notebook.dataset_id + + dataset_path = f"{file_path}/dataset" try: - loaded_dataset: "DashAIDataset" = load_dataset(f"{file_path}/dataset") + loaded_dataset: "DashAIDataset" = load_dataset(dataset_path) except Exception as e: log.exception(e) raise JobError( f"Can not load dataset from path {file_path}", ) from e - ctx.put_ref("dataset_id", dataset_id) + if not loaded_dataset: + raise JobError(f"Dataset with path {dataset_path} not found") + ctx.put("dataset", loaded_dataset) + ctx.put_ref("dataset_id", dataset_id) + ctx.put_ref("dataset_path", dataset_path) diff --git a/DashAI/back/units/prepare_and_split_unit.py b/DashAI/back/units/prepare_and_split_unit.py index adfaf5928..17df6c52c 100644 --- a/DashAI/back/units/prepare_and_split_unit.py +++ b/DashAI/back/units/prepare_and_split_unit.py @@ -109,8 +109,11 @@ class PrepareAndSplitUnit(BaseUnit): SCHEMA = PrepareAndSplitSchema - REQUIRES = ("dataset",) - PROVIDES = ("x", "y", "n_labels", "task", "split_indexes") + # dataset_id is declared even though it only decorates an error message: + # an undeclared read is a contract a DAG validator cannot see, and the + # loaders publish it precisely so downstream units can name their input. + REQUIRES = ("dataset", "dataset_id") + PROVIDES = ("x", "y", "n_labels", "task", "split_indexes", "task_name") def execute(self, ctx: ExecutionContext) -> None: from kink import di @@ -129,6 +132,7 @@ def execute(self, ctx: ExecutionContext) -> None: splits = self.config["splits"] loaded_dataset = ctx.require("dataset") + dataset_id = ctx.require("dataset_id") try: task: "BaseTask" = component_registry[task_name]["class"]() @@ -170,7 +174,7 @@ def execute(self, ctx: ExecutionContext) -> None: except Exception as e: log.exception(e) raise JobError( - f"Can not prepare Dataset {ctx.get('dataset_id')} for Task {task_name}", + f"Can not prepare Dataset {dataset_id} for Task {task_name}", ) from e ctx.put_ref("task_name", task_name) diff --git a/DashAI/back/units/save_dataset_unit.py b/DashAI/back/units/save_dataset_unit.py new file mode 100644 index 000000000..50d9c9bc7 --- /dev/null +++ b/DashAI/back/units/save_dataset_unit.py @@ -0,0 +1,34 @@ +"""Unit that persists the dataset in the context back to disk.""" + +import logging + +from DashAI.back.job.base_job import JobError +from DashAI.back.units.base_unit import BaseUnit +from DashAI.back.units.context import ExecutionContext + +log = logging.getLogger(__name__) + + +class SaveDatasetUnit(BaseUnit): + """Write the dataset back to the path it was loaded from. + + Takes no configuration: the destination is ``dataset_path``, published by + whichever unit loaded the dataset, so a save can never land somewhere the + load did not come from. Declares no outputs — its result is on disk, not in + the context. + """ + + REQUIRES = ("dataset", "dataset_path") + PROVIDES = () + + def execute(self, ctx: ExecutionContext) -> None: + from DashAI.back.dataloaders.classes.dashai_dataset import save_dataset + + dataset = ctx.require("dataset") + dataset_path = ctx.require("dataset_path") + + try: + save_dataset(dataset, dataset_path) + except Exception as e: + log.exception(e) + raise JobError(f"Can not save dataset to path {dataset_path}") from e diff --git a/DashAI/back/units/transform_dataset_unit.py b/DashAI/back/units/transform_dataset_unit.py new file mode 100644 index 000000000..cba697af7 --- /dev/null +++ b/DashAI/back/units/transform_dataset_unit.py @@ -0,0 +1,82 @@ +"""Unit that transforms a dataset with an already fitted converter.""" + +import logging +from typing import TYPE_CHECKING + +from DashAI.back.core.schema_fields import BaseSchema +from DashAI.back.units.base_unit import BaseUnit +from DashAI.back.units.context import ExecutionContext +from DashAI.back.units.converter_scope import ( + ConverterScopeMixin, + scope_field, + target_field, +) + +if TYPE_CHECKING: + from DashAI.back.dataloaders.classes.dashai_dataset import DashAIDataset + +log = logging.getLogger(__name__) + + +class TransformDatasetSchema(BaseSchema): + scope: scope_field() # type: ignore + target: target_field() # type: ignore + + +class TransformDatasetUnit(BaseUnit, ConverterScopeMixin): + """Apply an already fitted converter to the dataset in the context. + + Takes no converter configuration: the converter arrives fitted through the + context, from :class:`FitConverterUnit` or :class:`ApplyConverterUnit`. That + is what makes "fit on train, apply to test" possible without refitting — + run this once per dataset, and the converter keeps the state it learned. + + It does carry its own ``scope``, on purpose: the scope is a list of 1-based + indexes, and they are resolved against **this** unit's dataset. Reusing the + column names resolved during fit would break the moment the two datasets + order their columns differently, and would carry stale column identity + across the context boundary. + """ + + SCHEMA = TransformDatasetSchema + + REQUIRES = ("dataset", "fitted_converter") + PROVIDES = ("dataset",) + + def validate(self, ctx: ExecutionContext) -> None: + """Reject an out-of-bounds target before any work is done.""" + self._check_target_bounds(ctx.require("dataset")) + + def execute(self, ctx: ExecutionContext) -> None: + dataset: "DashAIDataset" = ctx.require("dataset") + converter_instance = ctx.require("fitted_converter") + converter_name = type(converter_instance).__name__ + + self._check_target_bounds(dataset) + ( + scope_column_names, + _scope_rows_indexes, + target_column_name, + ) = self._resolve_scope(dataset) + + log.info(f"Transforming with fitted converter: {converter_name}") + + x_dataset, y_dataset = self._slice_for_transform( + dataset, + scope_column_names, + target_column_name, + ) + + transformed_dataset = self._transform( + converter_instance, x_dataset, y_dataset, converter_name + ) + + ctx.put( + "dataset", + self._merge_transformed( + converter_instance, + dataset, + transformed_dataset, + scope_column_names, + ), + ) diff --git a/tests/back/api/test_converter_job.py b/tests/back/api/test_converter_job.py new file mode 100644 index 000000000..925113b65 --- /dev/null +++ b/tests/back/api/test_converter_job.py @@ -0,0 +1,336 @@ +"""End-to-end regression net for ``ConverterJob``. + +Written before the job is decomposed into atomic units, and asserted against the +monolithic implementation, so that the refactor has something to be measured +against. The assertions are deliberately explicit — exact status values, exact +column names on disk, exact error message fragments — instead of the looser +``status in ["finished", "error"]`` style used elsewhere in this suite, which +cannot tell a unit that silently stopped doing part of its work from one that +did it. + +Lives under ``tests/back/api`` to reuse the ``client`` and ``dataset_1`` +fixtures from this package's ``conftest.py``. +""" + +import pytest +from fastapi.testclient import TestClient + +from DashAI.back.core.enums.status import ConverterStatus +from DashAI.back.dataloaders.classes.dashai_dataset import load_dataset +from DashAI.back.dependencies.database.models import Converter +from DashAI.back.job.base_job import JobError +from DashAI.back.job.converter_job import ConverterJob + +IRIS_COLUMNS = [ + "SepalLengthCm", + "SepalWidthCm", + "PetalLengthCm", + "PetalWidthCm", + "Species", +] +IRIS_ROWS = 150 + + +@pytest.fixture(name="notebook") +def create_notebook(client: TestClient, dataset_1): + """A notebook holding its own copy of the iris dataset. + + ``POST /notebook/`` copies the dataset folder, so every converter run + mutates the notebook's copy and never the source dataset. + """ + response = client.post( + "/api/v1/notebook/", + json={"dataset_id": dataset_1.id, "name": "converter job test"}, + ) + assert response.status_code == 201, response.text + return response.json() + + +def _create_converter(client, notebook_id, converter_name, scope=None, target=None): + """Create a Converter row through the API and return its id.""" + response = client.post( + "/api/v1/converter/", + json={ + "notebook_id": notebook_id, + "converter": converter_name, + "parameters": { + "order": 0, + "params": {}, + "scope": scope if scope is not None else {"columns": [], "rows": []}, + "target": target, + }, + }, + ) + assert response.status_code == 201, response.text + return response.json()["id"] + + +def _stored_converter(client, converter_id): + """Read the Converter row straight from the database.""" + session_factory = client.app.container["session_factory"] + with session_factory() as db: + converter = db.get(Converter, converter_id) + return { + "status": converter.status, + "start_time": converter.start_time, + "end_time": converter.end_time, + } + + +def _notebook_dataset(notebook): + return load_dataset(f"{notebook['file_path']}/dataset") + + +def test_the_notebook_starts_as_an_untouched_copy_of_the_dataset(notebook): + """Guards the fixture itself: the assertions below mean nothing if the + notebook copy does not start out as the full iris dataset.""" + dataset = _notebook_dataset(notebook) + + assert dataset.column_names == IRIS_COLUMNS + assert len(dataset) == IRIS_ROWS + + +def test_converter_job_removes_the_scoped_column_and_finishes(client, notebook): + """The happy path, end to end: status transitions and the dataset on disk. + + ``ColumnRemover`` deletes whatever is in scope, so a one-column scope is + directly observable in the saved dataset. + """ + converter_id = _create_converter( + client, + notebook["id"], + "ColumnRemover", + scope={"columns": [{"idx": 2}], "rows": []}, + ) + + ConverterJob(converter_id=converter_id).run() + + stored = _stored_converter(client, converter_id) + assert stored["status"] == ConverterStatus.FINISHED + assert stored["start_time"] is not None + assert stored["end_time"] is not None + + dataset = _notebook_dataset(notebook) + assert dataset.column_names == [ + "SepalLengthCm", + "PetalLengthCm", + "PetalWidthCm", + "Species", + ] + assert len(dataset) == IRIS_ROWS + + +def test_two_converters_chained_see_the_columns_the_previous_one_left(client, notebook): + """Column scope is resolved by index against the *current* dataset. + + Each converter is its own row and its own job invocation, and the second + one's ``idx`` must be read against the four columns the first one left + behind, not against the original five. This is the behaviour any atomic + decomposition has to keep: nothing may cache the original column list. + """ + first = _create_converter( + client, + notebook["id"], + "ColumnRemover", + scope={"columns": [{"idx": 1}], "rows": []}, + ) + ConverterJob(converter_id=first).run() + + assert _notebook_dataset(notebook).column_names == [ + "SepalWidthCm", + "PetalLengthCm", + "PetalWidthCm", + "Species", + ] + + # idx 1 now means SepalWidthCm, not SepalLengthCm. + second = _create_converter( + client, + notebook["id"], + "ColumnRemover", + scope={"columns": [{"idx": 1}], "rows": []}, + ) + ConverterJob(converter_id=second).run() + + assert _stored_converter(client, second)["status"] == ConverterStatus.FINISHED + assert _notebook_dataset(notebook).column_names == [ + "PetalLengthCm", + "PetalWidthCm", + "Species", + ] + + +def test_a_changes_row_count_converter_replaces_the_whole_dataset(client, notebook): + """``CHANGES_ROW_COUNT`` takes the transform output as the new dataset. + + ``NanRemover`` returns only the scoped columns, so the columns outside the + scope are dropped — the documented consequence of replacing the dataset + instead of merging the transformed columns back in. + """ + converter_id = _create_converter( + client, + notebook["id"], + "NanRemover", + scope={"columns": [{"idx": 1}, {"idx": 5}], "rows": []}, + ) + + ConverterJob(converter_id=converter_id).run() + + assert _stored_converter(client, converter_id)["status"] == ( + ConverterStatus.FINISHED + ) + + dataset = _notebook_dataset(notebook) + assert dataset.column_names == ["SepalLengthCm", "Species"] + # iris has no missing values, so no row is dropped. + assert len(dataset) == IRIS_ROWS + + +def test_an_empty_column_scope_means_every_column(client, notebook): + """An empty scope is not "no columns"; it is "all of them".""" + converter_id = _create_converter( + client, + notebook["id"], + "NanRemover", + scope={"columns": [], "rows": []}, + ) + + ConverterJob(converter_id=converter_id).run() + + assert _stored_converter(client, converter_id)["status"] == ( + ConverterStatus.FINISHED + ) + assert _notebook_dataset(notebook).column_names == IRIS_COLUMNS + + +def test_a_row_scope_fits_on_the_subset_and_transforms_the_whole_dataset( + client, notebook +): + """A row scope narrows ``fit`` only; ``transform`` still sees every row.""" + converter_id = _create_converter( + client, + notebook["id"], + "NanRemover", + scope={"columns": [{"idx": 1}], "rows": [1, 2, 3]}, + ) + + ConverterJob(converter_id=converter_id).run() + + assert _stored_converter(client, converter_id)["status"] == ( + ConverterStatus.FINISHED + ) + + dataset = _notebook_dataset(notebook) + assert dataset.column_names == ["SepalLengthCm"] + assert len(dataset) == IRIS_ROWS + + +def test_a_target_column_is_resolved_and_passed_to_the_converter(client, notebook): + """The supervised path builds y and hands it to fit/transform.""" + converter_id = _create_converter( + client, + notebook["id"], + "NanRemover", + scope={"columns": [{"idx": 1}], "rows": []}, + target={"idx": 5}, + ) + + ConverterJob(converter_id=converter_id).run() + + assert _stored_converter(client, converter_id)["status"] == ( + ConverterStatus.FINISHED + ) + assert _notebook_dataset(notebook).column_names == ["SepalLengthCm"] + + +def test_a_missing_converter_row_reports_it_by_id(client, notebook): + with pytest.raises(JobError, match="Converter with id 999999 not found"): + ConverterJob(converter_id=999999).run() + + +def test_an_unknown_converter_name_fails_with_the_wrapped_message(client, notebook): + """The registry lookup error is reported inside the outer wrapper. + + Both halves matter: the import error names the culprit, the wrapper is what + the jobs UI actually shows. + """ + converter_id = _create_converter( + client, notebook["id"], "ThisConverterDoesNotExist" + ) + + with pytest.raises(JobError) as excinfo: + ConverterJob(converter_id=converter_id).run() + + message = str(excinfo.value) + assert "Error applying converters to dataset" in message + assert "Error importing converter ThisConverterDoesNotExist" in message + + assert _stored_converter(client, converter_id)["status"] == ConverterStatus.ERROR + # The dataset must be left untouched when the converter never ran. + assert _notebook_dataset(notebook).column_names == IRIS_COLUMNS + + +def test_an_out_of_bounds_target_index_reports_cannot_load_dataset(client, notebook): + """The "out of bounds" text is swallowed; only the wrapper reaches the user. + + The inner JobError is re-caught by the surrounding ``except Exception`` and + replaced, surviving only as ``__cause__``. Locking this in because it is an + easy detail to "fix" by accident while refactoring. + """ + converter_id = _create_converter( + client, + notebook["id"], + "ColumnRemover", + scope={"columns": [{"idx": 1}], "rows": []}, + target={"idx": 99}, + ) + + with pytest.raises(JobError, match="Cannot load dataset from") as excinfo: + ConverterJob(converter_id=converter_id).run() + + assert "Target column index 99 is out of bounds" in str(excinfo.value.__cause__) + + assert _stored_converter(client, converter_id)["status"] == ConverterStatus.ERROR + assert _notebook_dataset(notebook).column_names == IRIS_COLUMNS + + +def test_a_dataset_that_cannot_be_loaded_still_leaves_the_row_in_error( + client, notebook +): + """A load failure must not leave the converter stuck in STARTED. + + Nothing else would fix it: the Huey error signal writes only to its own + ``task_copy`` table and never touches the ``Converter`` row, and the job + runs with no outer handler. Before this, only ``SQLAlchemyError`` was + caught here, so an unreadable dataset left the row STARTED forever and the + UI showed the converter as still running. + """ + import shutil + + converter_id = _create_converter(client, notebook["id"], "ColumnRemover") + shutil.rmtree(f"{notebook['file_path']}/dataset") + + with pytest.raises(JobError, match="Can not load dataset from path"): + ConverterJob(converter_id=converter_id).run() + + assert _stored_converter(client, converter_id)["status"] == ConverterStatus.ERROR + + +def test_a_failing_converter_leaves_the_dataset_untouched(client, notebook): + """``ColumnRemover`` raises when asked for a column that is not there. + + Reaching that requires a scope index past the end of the dataset, which is + exactly what a stale column list would produce in a chained run. + """ + converter_id = _create_converter( + client, + notebook["id"], + "ColumnRemover", + scope={"columns": [{"idx": 99}], "rows": []}, + ) + + with pytest.raises(JobError, match="Error applying converters to dataset"): + ConverterJob(converter_id=converter_id).run() + + assert _stored_converter(client, converter_id)["status"] == ConverterStatus.ERROR + assert _notebook_dataset(notebook).column_names == IRIS_COLUMNS diff --git a/tests/back/api/test_units_api.py b/tests/back/api/test_units_api.py index 87de3d44a..5286ab22a 100644 --- a/tests/back/api/test_units_api.py +++ b/tests/back/api/test_units_api.py @@ -10,6 +10,10 @@ "FitModelUnit", "EvaluateModelUnit", "SaveModelUnit", + "ApplyConverterUnit", + "FitConverterUnit", + "TransformDatasetUnit", + "SaveDatasetUnit", } @@ -36,7 +40,10 @@ def test_units_expose_a_schema_the_front_can_render(units): def test_unit_schemas_describe_their_configuration(units): - assert "dataset_id" in units["LoadDatasetUnit"]["schema"]["properties"] + assert set(units["LoadDatasetUnit"]["schema"]["properties"]) == { + "dataset_id", + "notebook_id", + } assert set(units["PrepareAndSplitUnit"]["schema"]["properties"]) == { "task_name", "input_columns", @@ -45,6 +52,23 @@ def test_unit_schemas_describe_their_configuration(units): } assert "model" in units["BuildModelUnit"]["schema"]["properties"] assert "optimizer" in units["FitModelUnit"]["schema"]["properties"] + assert set(units["ApplyConverterUnit"]["schema"]["properties"]) == { + "converter", + "scope", + "target", + } + assert set(units["FitConverterUnit"]["schema"]["properties"]) == { + "converter", + "scope", + "target", + } + # No converter to pick: it arrives already fitted through the context. + assert set(units["TransformDatasetUnit"]["schema"]["properties"]) == { + "scope", + "target", + } + # SaveDatasetUnit is configuration-free: it saves where the load said. + assert units["SaveDatasetUnit"]["schema"]["properties"] == {} def test_component_fields_tell_the_front_which_components_to_offer(units): @@ -56,10 +80,13 @@ def test_component_fields_tell_the_front_which_components_to_offer(units): """ model = units["BuildModelUnit"]["schema"]["properties"]["model"] optimizer = units["FitModelUnit"]["schema"]["properties"]["optimizer"] + converter = units["ApplyConverterUnit"]["schema"]["properties"]["converter"] assert model["parent"] == "BaseModel" assert optimizer["parent"] == "BaseOptimizer" + assert converter["parent"] == "BaseConverter" assert set(model["properties"]) == {"component", "params"} + assert set(converter["properties"]) == {"component", "params"} def test_a_component_field_parent_resolves_to_real_components(client: TestClient): diff --git a/tests/back/units/test_apply_converter_unit.py b/tests/back/units/test_apply_converter_unit.py new file mode 100644 index 000000000..cc960504a --- /dev/null +++ b/tests/back/units/test_apply_converter_unit.py @@ -0,0 +1,388 @@ +"""Contract tests for ApplyConverterUnit, isolated from ConverterJob. + +The job only ever runs one of these per invocation, so an end-to-end run cannot +show whether the unit is safe to chain. These tests build the context by hand +and run several units against it, which is what the future DAG will do. +""" + +import pandas as pd +import pyarrow as pa +import pytest +from kink import di + +from DashAI.back.dataloaders.classes.dashai_dataset import to_dashai_dataset +from DashAI.back.job.base_job import JobError +from DashAI.back.types.value_types import Integer +from DashAI.back.units.apply_converter_unit import ApplyConverterUnit +from DashAI.back.units.context import ExecutionContext, UnitContractError + + +def _dataset(**columns): + frame = pd.DataFrame(columns) + types = {name: Integer(arrow_type=pa.int64()) for name in frame.columns} + return to_dashai_dataset(frame, types=types) + + +class _DropScopedColumns: + """Converter that removes whatever is in scope, like ColumnRemover.""" + + CHANGES_ROW_COUNT = False + + def __init__(self, **params): + self.params = params + self.columns = [] + + def fit(self, x, y=None): + self.columns = x.column_names + return self + + def transform(self, x, y=None): + return x.remove_columns(self.columns) + + +class _ReplaceWithScope: + """Converter that replaces the dataset with the scoped columns only.""" + + CHANGES_ROW_COUNT = True + + def __init__(self, **params): + self.params = params + + def fit(self, x, y=None): + return self + + def transform(self, x, y=None): + return x + + +class _RecordingConverter: + """Converter that records what fit and transform were handed.""" + + CHANGES_ROW_COUNT = True + + def __init__(self, **params): + self.params = params + self.fit_x = None + self.fit_y = None + self.transform_x = None + self.transform_y = None + + def fit(self, x, y=None): + self.fit_x = x + self.fit_y = y + return self + + def transform(self, x, y=None): + self.transform_x = x + self.transform_y = y + return x + + +class _FailingFit: + CHANGES_ROW_COUNT = False + + def __init__(self, **params): + pass + + def fit(self, x, y=None): + raise ValueError("bad input") + + def transform(self, x, y=None): # pragma: no cover - never reached + return x + + +class _FailingTransform: + CHANGES_ROW_COUNT = False + + def __init__(self, **params): + pass + + def fit(self, x, y=None): + return self + + def transform(self, x, y=None): + raise RuntimeError("boom") + + +@pytest.fixture(name="registry") +def fixture_registry(): + registry = { + "DropScopedColumns": {"class": _DropScopedColumns}, + "ReplaceWithScope": {"class": _ReplaceWithScope}, + "RecordingConverter": {"class": _RecordingConverter}, + "FailingFit": {"class": _FailingFit}, + "FailingTransform": {"class": _FailingTransform}, + } + di["component_registry"] = registry + yield registry + del di["component_registry"] + + +def _unit(name, scope=None, target=None, params=None): + return ApplyConverterUnit( + converter={"component": name, "params": params or {}}, + scope=scope, + target=target, + ) + + +def _ctx(dataset): + ctx = ExecutionContext() + ctx.put("dataset", dataset) + return ctx + + +def test_the_unit_refuses_to_run_without_a_dataset(registry): + with pytest.raises(UnitContractError, match="'dataset' is not available"): + _unit("DropScopedColumns")(ExecutionContext()) + + +def test_a_scoped_column_is_resolved_by_one_based_index(registry): + ctx = _ctx(_dataset(a=[1, 2], b=[3, 4], c=[5, 6])) + + _unit("DropScopedColumns", scope={"columns": [{"idx": 2}], "rows": []})(ctx) + + assert ctx.require("dataset").column_names == ["a", "c"] + + +def test_an_empty_column_scope_selects_every_column(registry): + ctx = _ctx(_dataset(a=[1, 2], b=[3, 4])) + + _unit("ReplaceWithScope", scope={"columns": [], "rows": []})(ctx) + + assert ctx.require("dataset").column_names == ["a", "b"] + + +def test_a_null_scope_is_treated_as_an_empty_one(registry): + """The API schema writes the ``scope`` key but allows it to be null. + + ``dict.get("scope", default)`` returns ``None`` rather than the default when + the key is present and null, so the unit has to coalesce it explicitly. + """ + ctx = _ctx(_dataset(a=[1, 2], b=[3, 4])) + + _unit("ReplaceWithScope", scope=None)(ctx) + + assert ctx.require("dataset").column_names == ["a", "b"] + + +def test_chained_units_resolve_their_indexes_against_the_current_dataset(registry): + """The reason no column identity may cross the context boundary. + + The first unit drops column ``a``, so index 1 means ``b`` by the time the + second unit runs. A column list resolved once and published to the context + would make the second unit drop the wrong column. + """ + ctx = _ctx(_dataset(a=[1, 2], b=[3, 4], c=[5, 6])) + + _unit("DropScopedColumns", scope={"columns": [{"idx": 1}], "rows": []})(ctx) + assert ctx.require("dataset").column_names == ["b", "c"] + + _unit("DropScopedColumns", scope={"columns": [{"idx": 1}], "rows": []})(ctx) + assert ctx.require("dataset").column_names == ["c"] + + +def test_the_unit_publishes_no_column_state_into_the_context(registry): + """Nothing resolved from the dataset may outlive the unit that resolved it.""" + ctx = _ctx(_dataset(a=[1, 2], b=[3, 4])) + + _unit("DropScopedColumns", scope={"columns": [{"idx": 1}], "rows": []})(ctx) + + for leaked in ( + "column_names", + "scope_column_names", + "scope_column_indexes", + "target_column_name", + "converter", + ): + assert not ctx.has(leaked), leaked + + +def test_two_units_in_one_context_do_not_share_their_resolved_converter(registry): + """Registry lookups are memoized on the instance, never in the context. + + A class cached under a context key would make the second unit silently run + the first one's converter. + """ + ctx = _ctx(_dataset(a=[1, 2], b=[3, 4], c=[5, 6])) + + first = _unit("DropScopedColumns", scope={"columns": [{"idx": 1}], "rows": []}) + second = _unit("ReplaceWithScope", scope={"columns": [{"idx": 1}], "rows": []}) + + first(ctx) + second(ctx) + + assert first._converter_class is _DropScopedColumns + assert second._converter_class is _ReplaceWithScope + # ReplaceWithScope keeps only what is in scope, which is now "b". + assert ctx.require("dataset").column_names == ["b"] + + +def test_a_row_scope_narrows_fit_but_not_transform(registry): + ctx = _ctx(_dataset(a=[1, 2, 3, 4], b=[5, 6, 7, 8])) + recorded = {} + + class _Spy(_RecordingConverter): + def fit(self, x, y=None): + recorded["fit_rows"] = len(x) + return super().fit(x, y) + + def transform(self, x, y=None): + recorded["transform_rows"] = len(x) + recorded["same_object"] = x is self.fit_x + return super().transform(x, y) + + registry["Spy"] = {"class": _Spy} + + _unit("Spy", scope={"columns": [{"idx": 1}], "rows": [1, 2]})(ctx) + + assert recorded["fit_rows"] == 2 + assert recorded["transform_rows"] == 4 + assert recorded["same_object"] is False + + +def test_no_row_scope_hands_transform_the_same_object_as_fit(registry): + """Object identity, not equality, is part of the contract here. + + ``TypeCastConverter`` caches converted columns during ``fit`` and reuses + them in ``transform`` only when handed the same dataset object. Losing that + identity does not fail anything — it just silently recomputes. + """ + ctx = _ctx(_dataset(a=[1, 2, 3, 4], b=[5, 6, 7, 8])) + recorded = {} + + class _Spy(_RecordingConverter): + def transform(self, x, y=None): + recorded["same_object"] = x is self.fit_x + return super().transform(x, y) + + registry["Spy"] = {"class": _Spy} + + _unit("Spy", scope={"columns": [{"idx": 1}], "rows": []})(ctx) + + assert recorded["same_object"] is True + + +def test_a_target_column_is_resolved_and_handed_over_as_y(registry): + ctx = _ctx(_dataset(a=[1, 2], b=[3, 4], label=[5, 6])) + recorded = {} + + class _Spy(_RecordingConverter): + def fit(self, x, y=None): + recorded["fit_y"] = None if y is None else y.column_names + return super().fit(x, y) + + def transform(self, x, y=None): + recorded["transform_y"] = None if y is None else y.column_names + return super().transform(x, y) + + registry["Spy"] = {"class": _Spy} + + _unit( + "Spy", + scope={"columns": [{"idx": 1}], "rows": []}, + target={"idx": 3}, + )(ctx) + + assert recorded["fit_y"] == ["label"] + assert recorded["transform_y"] == ["label"] + + +def test_no_target_means_no_y(registry): + ctx = _ctx(_dataset(a=[1, 2], b=[3, 4])) + recorded = {} + + class _Spy(_RecordingConverter): + def fit(self, x, y=None): + recorded["fit_y"] = y + return super().fit(x, y) + + registry["Spy"] = {"class": _Spy} + + _unit("Spy", scope={"columns": [{"idx": 1}], "rows": []}, target=None)(ctx) + + assert recorded["fit_y"] is None + + +def test_validate_rejects_an_out_of_bounds_target_without_running(registry): + ctx = _ctx(_dataset(a=[1, 2], b=[3, 4])) + unit = _unit("DropScopedColumns", target={"idx": 99}) + + with pytest.raises(JobError, match="Target column index 99 is out of bounds"): + unit.validate(ctx) + + assert ctx.require("dataset").column_names == ["a", "b"] + + +def test_the_target_bound_is_rechecked_against_the_dataset_of_the_moment(registry): + """A target that was valid before the previous converter ran may not be now. + + ``validate`` runs against whatever dataset is in the context when it is + called; chaining means ``execute`` can face a narrower one. Re-checking + turns what would be a bare IndexError into the same JobError. + """ + ctx = _ctx(_dataset(a=[1, 2], b=[3, 4], c=[5, 6])) + unit = _unit( + "DropScopedColumns", + scope={"columns": [{"idx": 1}], "rows": []}, + target={"idx": 3}, + ) + + unit.validate(ctx) # 3 columns: idx 3 is fine right now. + + # Another converter narrows the dataset behind its back. + _unit("ReplaceWithScope", scope={"columns": [{"idx": 1}], "rows": []})(ctx) + + with pytest.raises(JobError, match="Target column index 3 is out of bounds"): + unit(ctx) + + +def test_an_unknown_converter_names_the_culprit(registry): + ctx = _ctx(_dataset(a=[1, 2])) + + with pytest.raises(JobError, match="Error importing converter NotRegistered"): + _unit("NotRegistered")(ctx) + + +def test_a_value_error_during_fit_is_reported_as_a_validation_error(registry): + ctx = _ctx(_dataset(a=[1, 2])) + + with pytest.raises(JobError, match="Validation error fitting FailingFit"): + _unit("FailingFit")(ctx) + + +def test_any_other_error_during_fit_is_reported_as_a_fit_error(registry): + ctx = _ctx(_dataset(a=[1, 2])) + + class _Boom(_FailingFit): + def fit(self, x, y=None): + raise RuntimeError("nope") + + registry["Boom"] = {"class": _Boom} + + with pytest.raises(JobError, match="Error fitting converter Boom"): + _unit("Boom")(ctx) + + +def test_an_error_during_transform_is_reported_as_a_transform_error(registry): + ctx = _ctx(_dataset(a=[1, 2])) + + with pytest.raises(JobError, match="Error transforming data with FailingTransform"): + _unit("FailingTransform")(ctx) + + +def test_converter_params_reach_the_constructor(registry): + ctx = _ctx(_dataset(a=[1, 2])) + unit = _unit("ReplaceWithScope", params={"threshold": 3}) + + unit(ctx) + + assert unit._converter_class is _ReplaceWithScope + + +def test_a_changes_row_count_converter_replaces_the_whole_dataset(registry): + ctx = _ctx(_dataset(a=[1, 2], b=[3, 4], c=[5, 6])) + + _unit("ReplaceWithScope", scope={"columns": [{"idx": 2}], "rows": []})(ctx) + + assert ctx.require("dataset").column_names == ["b"] diff --git a/tests/back/units/test_build_model_unit.py b/tests/back/units/test_build_model_unit.py index 8a4c3f624..a14aee9ab 100644 --- a/tests/back/units/test_build_model_unit.py +++ b/tests/back/units/test_build_model_unit.py @@ -51,6 +51,8 @@ def test_two_build_model_units_in_one_context_resolve_independently(fake_registr ctx.put("x", {"train": None, "validation": None}) ctx.put("y", {"train": None, "validation": None}) ctx.put("n_labels", None) + ctx.put_ref("run_id", 1) + ctx.put_ref("task_name", "ATask") a = _build_unit("ModelA") a(ctx) diff --git a/tests/back/units/test_converter_fit_transform_split.py b/tests/back/units/test_converter_fit_transform_split.py new file mode 100644 index 000000000..00cba74a1 --- /dev/null +++ b/tests/back/units/test_converter_fit_transform_split.py @@ -0,0 +1,278 @@ +"""Fitting a converter on one dataset and applying it to another. + +The point of splitting ``ApplyConverterUnit`` into ``FitConverterUnit`` + +``TransformDatasetUnit``: a scaler, encoder or imputer must learn its statistics +from the training data only and then be applied unchanged to the test data. +Refitting on test would leak the test distribution into the evaluation, which is +exactly what the fused unit forced. +""" + +import pandas as pd +import pyarrow as pa +import pytest +from kink import di + +from DashAI.back.converters.scikit_learn.min_max_scaler import MinMaxScaler +from DashAI.back.dataloaders.classes.dashai_dataset import to_dashai_dataset +from DashAI.back.job.base_job import JobError +from DashAI.back.types.value_types import Float +from DashAI.back.units.apply_converter_unit import ApplyConverterUnit +from DashAI.back.units.context import ExecutionContext, UnitContractError +from DashAI.back.units.fit_converter_unit import FitConverterUnit +from DashAI.back.units.transform_dataset_unit import TransformDatasetUnit + +FULL_SCOPE = {"columns": [], "rows": []} + + +def _dataset(**columns): + frame = pd.DataFrame(columns) + types = {name: Float(arrow_type=pa.float64()) for name in frame.columns} + return to_dashai_dataset(frame, types=types) + + +def _values(dataset, column="a"): + return list(dataset.to_pandas()[column]) + + +@pytest.fixture(name="registry") +def fixture_registry(): + registry = {"MinMaxScaler": {"class": MinMaxScaler}} + di["component_registry"] = registry + yield registry + del di["component_registry"] + + +def test_a_converter_fitted_on_train_is_applied_to_test_without_refitting(registry): + """The headline case. + + MinMaxScaler fitted on [0, 5, 10] learns min=0, max=10. Applied to a test + value of 20 it must yield 2.0 — outside [0, 1] precisely because the range + came from train. A refit on the test data would have produced 0.0 instead, + so the number is what proves the fitted state survived. + """ + ctx = ExecutionContext() + + # Fit on train. + ctx.put("dataset", _dataset(a=[0.0, 5.0, 10.0])) + FitConverterUnit( + converter={"component": "MinMaxScaler", "params": {}}, + scope=FULL_SCOPE, + target=None, + )(ctx) + + # Transform train with it. + TransformDatasetUnit(scope=FULL_SCOPE, target=None)(ctx) + assert _values(ctx.require("dataset")) == [0.0, 0.5, 1.0] + + # Swap in the test dataset and transform with the *same* fitted converter. + ctx.put("dataset", _dataset(a=[20.0])) + TransformDatasetUnit(scope=FULL_SCOPE, target=None)(ctx) + + assert _values(ctx.require("dataset")) == [2.0] + + +def test_fitting_leaves_the_dataset_untouched(registry): + """FitConverterUnit produces a converter, not data.""" + ctx = ExecutionContext() + original = _dataset(a=[0.0, 5.0, 10.0]) + ctx.put("dataset", original) + + FitConverterUnit( + converter={"component": "MinMaxScaler", "params": {}}, + scope=FULL_SCOPE, + target=None, + )(ctx) + + assert ctx.require("dataset") is original + assert _values(ctx.require("dataset")) == [0.0, 5.0, 10.0] + + +def test_the_fitted_converter_is_published_live_not_copied(registry): + """It has to be the same object, or the learned state would be lost.""" + ctx = ExecutionContext() + ctx.put("dataset", _dataset(a=[0.0, 10.0])) + + FitConverterUnit( + converter={"component": "MinMaxScaler", "params": {}}, + scope=FULL_SCOPE, + target=None, + )(ctx) + + fitted = ctx.require("fitted_converter") + assert isinstance(fitted, MinMaxScaler) + assert ctx.require("fitted_converter") is fitted + # The learned statistics are what makes it worth reusing. + assert list(fitted.data_min_) == [0.0] + assert list(fitted.data_max_) == [10.0] + + +def test_transforming_without_a_fitted_converter_is_a_contract_error(): + """A missing converter is a wiring mistake, not "nothing to apply".""" + ctx = ExecutionContext() + ctx.put("dataset", _dataset(a=[1.0])) + + with pytest.raises(UnitContractError, match="'fitted_converter' is not available"): + TransformDatasetUnit(scope=FULL_SCOPE, target=None)(ctx) + + +def test_transforming_without_a_dataset_is_a_contract_error(registry): + ctx = ExecutionContext() + ctx.put("fitted_converter", MinMaxScaler()) + + with pytest.raises(UnitContractError, match="'dataset' is not available"): + TransformDatasetUnit(scope=FULL_SCOPE, target=None)(ctx) + + +def test_the_fused_unit_also_publishes_its_fitted_converter(registry): + """ApplyConverterUnit stays usable as the source of a reusable converter. + + So the single-dataset path and the train/test path are the same mechanism: + whoever fitted the converter publishes it, and any number of + TransformDatasetUnits can then apply it elsewhere. + """ + ctx = ExecutionContext() + ctx.put("dataset", _dataset(a=[0.0, 5.0, 10.0])) + + ApplyConverterUnit( + converter={"component": "MinMaxScaler", "params": {}}, + scope=FULL_SCOPE, + target=None, + )(ctx) + + assert _values(ctx.require("dataset")) == [0.0, 0.5, 1.0] + + ctx.put("dataset", _dataset(a=[20.0])) + TransformDatasetUnit(scope=FULL_SCOPE, target=None)(ctx) + + assert _values(ctx.require("dataset")) == [2.0] + + +def test_the_split_pair_matches_the_fused_unit_on_a_single_dataset(registry): + """The two paths must not drift: same input, same output.""" + fused_ctx = ExecutionContext() + fused_ctx.put("dataset", _dataset(a=[1.0, 2.0, 3.0, 4.0])) + ApplyConverterUnit( + converter={"component": "MinMaxScaler", "params": {}}, + scope=FULL_SCOPE, + target=None, + )(fused_ctx) + + split_ctx = ExecutionContext() + split_ctx.put("dataset", _dataset(a=[1.0, 2.0, 3.0, 4.0])) + FitConverterUnit( + converter={"component": "MinMaxScaler", "params": {}}, + scope=FULL_SCOPE, + target=None, + )(split_ctx) + TransformDatasetUnit(scope=FULL_SCOPE, target=None)(split_ctx) + + assert _values(fused_ctx.require("dataset")) == _values( + split_ctx.require("dataset") + ) + + +def test_a_row_scope_narrows_the_fit_but_the_transform_still_sees_every_row(registry): + """Row scope belongs to fitting; transform always covers the dataset. + + Fitting on rows 1-2 of [0, 10, 100] learns min=0, max=10, so the third row + scales to 10.0 rather than 1.0. + """ + ctx = ExecutionContext() + ctx.put("dataset", _dataset(a=[0.0, 10.0, 100.0])) + + FitConverterUnit( + converter={"component": "MinMaxScaler", "params": {}}, + scope={"columns": [], "rows": [1, 2]}, + target=None, + )(ctx) + TransformDatasetUnit(scope=FULL_SCOPE, target=None)(ctx) + + assert _values(ctx.require("dataset")) == [0.0, 1.0, 10.0] + + +def test_fit_names_the_culprit_when_the_converter_is_not_registered(registry): + ctx = ExecutionContext() + ctx.put("dataset", _dataset(a=[1.0])) + + with pytest.raises(JobError, match="Error importing converter NotRegistered"): + FitConverterUnit( + converter={"component": "NotRegistered", "params": {}}, + scope=FULL_SCOPE, + target=None, + )(ctx) + + +def test_fit_rejects_an_out_of_bounds_target_before_running(registry): + ctx = ExecutionContext() + ctx.put("dataset", _dataset(a=[1.0], b=[2.0])) + unit = FitConverterUnit( + converter={"component": "MinMaxScaler", "params": {}}, + scope=FULL_SCOPE, + target={"idx": 99}, + ) + + with pytest.raises(JobError, match="Target column index 99 is out of bounds"): + unit.validate(ctx) + + assert not ctx.has("fitted_converter") + + +def test_two_fit_units_in_one_context_resolve_their_converters_independently(registry): + """Registry lookups are memoized on the instance, never in the context.""" + + class _Other(MinMaxScaler): + pass + + registry["Other"] = {"class": _Other} + + ctx = ExecutionContext() + ctx.put("dataset", _dataset(a=[0.0, 10.0])) + + first = FitConverterUnit( + converter={"component": "MinMaxScaler", "params": {}}, + scope=FULL_SCOPE, + target=None, + ) + second = FitConverterUnit( + converter={"component": "Other", "params": {}}, + scope=FULL_SCOPE, + target=None, + ) + first(ctx) + second(ctx) + + assert first._converter_class is MinMaxScaler + assert second._converter_class is _Other + + +def test_the_units_register_under_the_unit_type(): + """Guards the mixin's name. + + ``ConverterScopeMixin`` must not be called ``Base*``: the registry rejects a + component whose MRO has more than one "Base" ancestor declaring a TYPE, so + an intermediate ``BaseConverterUnit`` would break registration for all three + converter units at once. + """ + from DashAI.back.dependencies.registry import ComponentRegistry + + units = [ApplyConverterUnit, FitConverterUnit, TransformDatasetUnit] + component_registry = ComponentRegistry(initial_components=units) + + for unit in units: + assert component_registry[unit.__name__]["type"] == "Unit" + + +def test_one_fitted_converter_feeds_several_transforms(registry): + """Nothing about a transform consumes or invalidates the fitted converter.""" + ctx = ExecutionContext() + ctx.put("dataset", _dataset(a=[0.0, 10.0])) + FitConverterUnit( + converter={"component": "MinMaxScaler", "params": {}}, + scope=FULL_SCOPE, + target=None, + )(ctx) + + for value, expected in ((5.0, 0.5), (20.0, 2.0), (-10.0, -1.0)): + ctx.put("dataset", _dataset(a=[value])) + TransformDatasetUnit(scope=FULL_SCOPE, target=None)(ctx) + assert _values(ctx.require("dataset")) == [expected] diff --git a/tests/back/units/test_fit_model_unit.py b/tests/back/units/test_fit_model_unit.py index b4a0bab33..3ce22551c 100644 --- a/tests/back/units/test_fit_model_unit.py +++ b/tests/back/units/test_fit_model_unit.py @@ -1,6 +1,7 @@ """Tests for FitModelUnit's validation, independent of an actual training run.""" import pytest +from kink import di from DashAI.back.units.context import ExecutionContext, UnitContractError from DashAI.back.units.fit_model_unit import FitModelUnit @@ -34,7 +35,40 @@ def test_validate_is_a_noop_when_there_are_genuinely_no_optimizable_parameters() # Should not raise, and should not need the optimizer/goal_metric to # resolve in the registry. - _unit(optimizer_name="DoesNotExist", goal_metric="DoesNotExist").validate(ctx) + unit = _unit(optimizer_name="DoesNotExist", goal_metric="DoesNotExist") + unit.validate(ctx) - assert not ctx.has("optimizer") - assert not ctx.has("goal_metric") + assert unit._optimizer is None + assert unit._goal_metric is None + + +def test_the_optimizer_is_kept_on_the_unit_not_in_the_shared_context(): + """Regression: the optimizer is this unit's own state, not an output. + + It used to be written to the context by ``validate`` and read back by + ``execute``, using the shared context as a scratchpad between one unit's + own two phases. Two FitModelUnits in one context would overwrite each + other's optimizer, and the second would silently run the first one's. + """ + + class _Optimizer: + def __init__(self, **params): + pass + + registry = { + "AnOptimizer": {"class": _Optimizer}, + "Accuracy": {"class": object, "metadata": {"maximize": True}}, + } + di["component_registry"] = registry + try: + ctx = ExecutionContext() + ctx.put("optimizable_parameters", ["lr"]) + + unit = _unit(optimizer_name="AnOptimizer") + unit.validate(ctx) + + assert isinstance(unit._optimizer, _Optimizer) + assert not ctx.has("optimizer") + assert not ctx.has("goal_metric") + finally: + del di["component_registry"] diff --git a/tests/back/units/test_load_dataset_unit.py b/tests/back/units/test_load_dataset_unit.py new file mode 100644 index 000000000..1c5bbc994 --- /dev/null +++ b/tests/back/units/test_load_dataset_unit.py @@ -0,0 +1,151 @@ +"""Contract tests for LoadDatasetUnit, isolated from any orchestrating job. + +The context is built by hand rather than through a job, which is what exposes +composability mistakes: a job always wires the context "correctly", so an +end-to-end run cannot tell a real contract from a lucky one. +""" + +import pytest +from kink import di + +from DashAI.back.dataloaders.classes.dashai_dataset import ( + save_dataset, + to_dashai_dataset, +) +from DashAI.back.job.base_job import JobError +from DashAI.back.units.context import ExecutionContext +from DashAI.back.units.load_dataset_unit import LoadDatasetUnit + + +class _Row: + """Stand-in for a Dataset or Notebook ORM row.""" + + def __init__(self, file_path=None, dataset_id=None): + self.file_path = file_path + self.dataset_id = dataset_id + + +class _FakeSession: + """Session that answers ``get`` from a table -> {id: row} mapping.""" + + def __init__(self, rows): + self._rows = rows + + def get(self, model, row_id): + return self._rows.get(model.__name__, {}).get(row_id) + + def __enter__(self): + return self + + def __exit__(self, *exc_info): + return False + + +class _FakeSessionFactory: + """Stand-in for a ``sessionmaker``. + + A class rather than a lambda on purpose: kink invokes any registered lambda + with the container to resolve it, so a lambda here would be called as a + service factory instead of being handed to the unit as one. + """ + + def __init__(self, rows): + self._rows = rows + + def __call__(self): + return _FakeSession(self._rows) + + +@pytest.fixture(name="stored_dataset") +def fixture_stored_dataset(tmp_path): + """A real two-column dataset written to ``/store/dataset``.""" + import pandas as pd + import pyarrow as pa + + from DashAI.back.types.value_types import Integer + + frame = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}) + types = { + "a": Integer(arrow_type=pa.int64()), + "b": Integer(arrow_type=pa.int64()), + } + root = tmp_path / "store" + save_dataset(to_dashai_dataset(frame, types=types), str(root / "dataset")) + return root + + +@pytest.fixture(name="fake_db") +def fixture_fake_db(stored_dataset): + """Point both a Dataset row and a Notebook row at the stored dataset.""" + rows = { + "Dataset": {7: _Row(file_path=str(stored_dataset))}, + "Notebook": {3: _Row(file_path=str(stored_dataset), dataset_id=7)}, + } + di["session_factory"] = _FakeSessionFactory(rows) + yield rows + del di["session_factory"] + + +def test_loading_by_dataset_id_publishes_the_whole_contract(fake_db): + ctx = ExecutionContext() + + LoadDatasetUnit(dataset_id=7)(ctx) + + assert ctx.require("dataset").column_names == ["a", "b"] + assert ctx.require("dataset_id") == 7 + assert ctx.require("dataset_path").endswith("dataset") + + +def test_loading_by_notebook_id_resolves_the_source_dataset_id(fake_db): + """The notebook branch must still publish a dataset id. + + Downstream error messages identify the work by dataset id, so a notebook + load that left the key unset would report ``None`` instead of the dataset. + """ + ctx = ExecutionContext() + + LoadDatasetUnit(notebook_id=3)(ctx) + + assert ctx.require("dataset").column_names == ["a", "b"] + assert ctx.require("dataset_id") == 7 + + +def test_the_two_starting_points_are_mutually_exclusive(fake_db): + with pytest.raises(JobError, match="exactly one of dataset_id or notebook_id"): + LoadDatasetUnit(dataset_id=7, notebook_id=3)(ExecutionContext()) + + +def test_no_starting_point_at_all_is_rejected(fake_db): + with pytest.raises(JobError, match="exactly one of dataset_id or notebook_id"): + LoadDatasetUnit()(ExecutionContext()) + + +def test_a_missing_dataset_row_is_reported_by_id(fake_db): + with pytest.raises(JobError, match="Dataset 99 does not exist in DB."): + LoadDatasetUnit(dataset_id=99)(ExecutionContext()) + + +def test_a_missing_notebook_row_is_reported_by_id(fake_db): + with pytest.raises(JobError, match="Notebook 99 does not exist in DB."): + LoadDatasetUnit(notebook_id=99)(ExecutionContext()) + + +def test_an_unreadable_path_becomes_a_job_error(fake_db, tmp_path): + fake_db["Notebook"][4] = _Row(file_path=str(tmp_path / "nowhere"), dataset_id=7) + + with pytest.raises(JobError, match="Can not load dataset from path"): + LoadDatasetUnit(notebook_id=4)(ExecutionContext()) + + +def test_the_dataset_is_cached_live_not_copied(fake_db): + """The dataset must come back as the same object, not a copy. + + ``ctx.get`` deep-copies the refs half and returns the cache half by + reference; a dataset that came back copied would mean every unit downstream + transformed a different object than the one that gets saved. + """ + ctx = ExecutionContext() + + LoadDatasetUnit(dataset_id=7)(ctx) + + assert ctx.require("dataset") is ctx.require("dataset") diff --git a/tests/back/units/test_save_dataset_unit.py b/tests/back/units/test_save_dataset_unit.py new file mode 100644 index 000000000..a1e534c08 --- /dev/null +++ b/tests/back/units/test_save_dataset_unit.py @@ -0,0 +1,64 @@ +"""Contract tests for SaveDatasetUnit.""" + +import pandas as pd +import pyarrow as pa +import pytest + +from DashAI.back.dataloaders.classes.dashai_dataset import ( + load_dataset, + to_dashai_dataset, +) +from DashAI.back.job.base_job import JobError +from DashAI.back.types.value_types import Integer +from DashAI.back.units.context import ExecutionContext, UnitContractError +from DashAI.back.units.save_dataset_unit import SaveDatasetUnit + + +def _dataset(**columns): + frame = pd.DataFrame(columns) + types = {name: Integer(arrow_type=pa.int64()) for name in frame.columns} + return to_dashai_dataset(frame, types=types) + + +def test_the_dataset_is_written_where_the_path_says(tmp_path): + destination = str(tmp_path / "notebook" / "dataset") + ctx = ExecutionContext() + ctx.put("dataset", _dataset(a=[1, 2], b=[3, 4])) + ctx.put_ref("dataset_path", destination) + + SaveDatasetUnit()(ctx) + + assert load_dataset(destination).column_names == ["a", "b"] + + +def test_saving_without_a_dataset_is_a_contract_error(tmp_path): + ctx = ExecutionContext() + ctx.put_ref("dataset_path", str(tmp_path / "dataset")) + + with pytest.raises(UnitContractError, match="'dataset' is not available"): + SaveDatasetUnit()(ctx) + + +def test_saving_without_a_path_is_a_contract_error(): + """A missing path is a wiring mistake, not a "nowhere to save" decision. + + The unit has no fallback destination on purpose: silently picking one would + write the dataset somewhere nobody asked for. + """ + ctx = ExecutionContext() + ctx.put("dataset", _dataset(a=[1])) + + with pytest.raises(UnitContractError, match="'dataset_path' is not available"): + SaveDatasetUnit()(ctx) + + +def test_an_unwritable_destination_becomes_a_job_error(tmp_path): + blocker = tmp_path / "blocker" + blocker.write_text("not a directory", encoding="utf-8") + + ctx = ExecutionContext() + ctx.put("dataset", _dataset(a=[1])) + ctx.put_ref("dataset_path", str(blocker / "dataset")) + + with pytest.raises(JobError, match="Can not save dataset to path"): + SaveDatasetUnit()(ctx) diff --git a/tests/back/units/test_unit_contracts.py b/tests/back/units/test_unit_contracts.py new file mode 100644 index 000000000..b472a885a --- /dev/null +++ b/tests/back/units/test_unit_contracts.py @@ -0,0 +1,135 @@ +"""A contract audit over every registered unit, enforced as a test. + +The individual unit tests check behaviour; this one checks that the *declared* +contract matches the code. Undeclared context reads are the recurring mistake in +this design: they never break the job that happens to wire the context by hand, +so they survive every end-to-end test and only surface when something reuses the +unit — which is the whole point of having units. +""" + +import ast +import pathlib + +import pytest + +UNITS_DIR = pathlib.Path(__file__).resolve().parents[3] / "DashAI" / "back" / "units" + +#: Keys a unit reads that its own execution produces, so they need no declaration. +SELF_PRODUCED = {"dataset"} + + +def _unit_modules(): + for path in sorted(UNITS_DIR.glob("*.py")): + if path.name in {"__init__.py", "base_unit.py", "context.py"}: + continue + yield path + + +def _string_literals(node): + return { + element.value + for element in ast.walk(node) + if isinstance(element, ast.Constant) and isinstance(element.value, str) + } + + +def _unit_class(tree): + for node in ast.walk(tree): + if isinstance(node, ast.ClassDef) and any( + isinstance(base, ast.Name) and base.id == "BaseUnit" for base in node.bases + ): + return node + return None + + +def _declared(cls, name): + for node in cls.body: + if isinstance(node, ast.Assign) and any( + isinstance(t, ast.Name) and t.id == name for t in node.targets + ): + return _string_literals(node.value) + return set() + + +def _context_calls(cls, methods): + """Every ``ctx.("key")`` literal inside the class.""" + keys = set() + for node in ast.walk(cls): + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr in methods + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "ctx" + and node.args + and isinstance(node.args[0], ast.Constant) + and isinstance(node.args[0].value, str) + ): + keys.add(node.args[0].value) + return keys + + +def _parsed_units(): + units = [] + for path in _unit_modules(): + tree = ast.parse(path.read_text(encoding="utf-8")) + cls = _unit_class(tree) + if cls is not None: + units.append((path.name, cls)) + return units + + +UNITS = _parsed_units() + + +def test_the_audit_actually_found_the_units(): + """Guards the audit itself: a broken parser would make it vacuously pass.""" + assert {name for name, _ in UNITS} >= { + "load_dataset_unit.py", + "apply_converter_unit.py", + "save_dataset_unit.py", + } + + +@pytest.mark.parametrize(("name", "cls"), UNITS, ids=[name for name, _ in UNITS]) +def test_every_context_key_a_unit_reads_is_declared_in_requires(name, cls): + read = _context_calls(cls, {"require", "get", "has"}) + declared = _declared(cls, "REQUIRES") | _declared(cls, "PROVIDES") | SELF_PRODUCED + + undeclared = read - declared + assert not undeclared, ( + f"{name} reads {sorted(undeclared)} from the context without declaring " + "them in REQUIRES. A caller inspecting the contract cannot know they " + "are needed, and a missing value reads as 'not applicable' instead of " + "'wiring mistake'." + ) + + +@pytest.mark.parametrize(("name", "cls"), UNITS, ids=[name for name, _ in UNITS]) +def test_every_key_a_unit_promises_is_actually_written(name, cls): + written = _context_calls(cls, {"put", "put_ref"}) + promised = _declared(cls, "PROVIDES") + + unwritten = promised - written + assert not unwritten, ( + f"{name} promises {sorted(unwritten)} in PROVIDES but never writes it." + ) + + +@pytest.mark.parametrize(("name", "cls"), UNITS, ids=[name for name, _ in UNITS]) +def test_a_unit_does_not_use_the_context_as_its_own_scratchpad(name, cls): + """A key written and read by the same unit, and promised to nobody. + + That is instance state wearing a context key's clothes: two units of the + same class in one context would overwrite each other. It belongs on + ``self``, memoized, the way the registry lookups are. + """ + written = _context_calls(cls, {"put", "put_ref"}) + read = _context_calls(cls, {"require", "get", "has"}) + promised = _declared(cls, "PROVIDES") + + scratch = (written & read) - promised - SELF_PRODUCED + assert not scratch, ( + f"{name} writes and reads {sorted(scratch)} without promising it. " + "Keep per-instance state on the unit, not in the shared context." + )