[CDF-27102] 📖 Variable parsing

cognite_toolkit/_cdf_tk/commands/build_v2/_module_source_parser.py

@@ -1,4 +1,5 @@
 from collections import defaultdict
+from itertools import groupby
 from pathlib import Path
 from typing import Any
 
@@ -10,7 +11,7 @@
     RelativeDirPath,
 )
 from cognite_toolkit._cdf_tk.commands.build_v2.data_classes._module import BuildVariable
-from cognite_toolkit._cdf_tk.constants import EXCL_FILES
+from cognite_toolkit._cdf_tk.constants import EXCL_FILES, MODULES
 from cognite_toolkit._cdf_tk.cruds import CRUDS_BY_FOLDER_NAME_INCLUDE_ALPHA
 
 
@@ -30,20 +31,30 @@ def parse(self, yaml_files: list[RelativeDirPath], variables: dict[str, Any]) ->
             self.errors.extend(errors)
             return []
         selected_modules = self._select_modules(files_by_module, self.selected_modules)
-        build_variables, errors = self._parse_variables(variables, set(files_by_module.keys()), set(selected_modules))
+        build_variables, errors = self._parse_module_variables(
+            variables, set(files_by_module.keys()), set(selected_modules)
+        )
         if errors:
             self.errors.extend(errors)
             return []
+        return self._create_module_sources(build_variables, files_by_module, selected_modules)
+
+    def _create_module_sources(
+        self,
+        build_variables: dict[Path, dict[int, list[BuildVariable]]],
+        files_by_module: dict[Path, list[Path]],
+        selected_modules: list[Path],
+    ) -> list[ModuleSource]:
         module_sources: list[ModuleSource] = []
         for module in selected_modules:
             source = ModuleSource(
                 path=self.organization_dir / module,
                 id=module,
                 resource_files=[self.organization_dir / resource_file for resource_file in files_by_module[module]],
             )
-            module_build_variables = build_variables.get(module, [])
+            module_build_variables = build_variables.get(module, {})
             if module_build_variables:
-                for iteration, module_variable in enumerate(module_build_variables, start=1):
+                for iteration, module_variable in module_build_variables.items():
                     module_sources.append(
                         source.model_copy(
                             update={
@@ -107,10 +118,86 @@ def _select_modules(
         ]
 
     @classmethod
-    def _parse_variables(
+    def _parse_module_variables(
         cls,
         variables: dict[str, Any],
         available_modules: set[RelativeDirPath],
         selected_modules: set[RelativeDirPath],
-    ) -> tuple[dict[RelativeDirPath, list[list[BuildVariable]]], list[ModelSyntaxError]]:
-        return {}, []
+    ) -> tuple[dict[RelativeDirPath, dict[int, list[BuildVariable]]], list[ModelSyntaxError]]:
+        all_available_paths = (
+            {Path("")} | available_modules | {parent for module in available_modules for parent in module.parents}
+        )
+        selected_paths = (
+            {Path("")} | selected_modules | {parent for module in selected_modules for parent in module.parents}
+        )
+        parsed_variables, errors = cls._parse_variables(variables, all_available_paths, selected_paths)
+        variable_by_module = cls._organize_variables_by_module(parsed_variables, selected_modules)
+        return variable_by_module, errors
+
+    @classmethod
+    def _parse_variables(
+        cls, variables: dict[str, Any], available_paths: set[RelativeDirPath], selected_paths: set[RelativeDirPath]
+    ) -> tuple[dict[RelativeDirPath, list[BuildVariable]], list[ModelSyntaxError]]:
+        variables_by_path: dict[RelativeDirPath, list[BuildVariable]] = defaultdict(list)
+        errors: list[ModelSyntaxError] = []
+        to_check: list[tuple[RelativeDirPath, int | None, dict[str, Any]]] = [(Path(""), None, variables)]
+        while to_check:
+            path, iteration, subdict = to_check.pop()
+            for key, value in subdict.items():
+                subpath = path / key
+                if isinstance(value, str | float | int | bool):
+                    variables_by_path[path].append(
+                        BuildVariable(id=subpath, value=value, is_selected=path in selected_paths, iteration=iteration)
+                    )
+                elif isinstance(value, dict):
+                    if subpath in available_paths:
+                        to_check.append((subpath, iteration, value))
+                    else:
+                        errors.append(
+                            ModelSyntaxError(
+                                code=cls.VARIABLE_ERROR_CODE,
+                                message=f"Invalid variable path: {'.'.join(subpath.parts)}. This does not correspond to the "
+                                f"folder structure inside the {MODULES} directory.",
+                                fix="Ensure that the variable paths correspond to the folder structure inside the modules directory.",
+                            )
+                        )
+                elif isinstance(value, list):
+                    if all(isinstance(item, str | float | int | bool) for item in value):
+                        variables_by_path[path].append(
+                            BuildVariable(
+                                id=subpath, value=value, is_selected=path in selected_paths, iteration=iteration
+                            )
+                        )
+                    elif all(isinstance(item, dict) for item in value):
+                        for idx, item in enumerate(value, start=1):
+                            to_check.append((subpath, idx, item))
+                    else:
+                        errors.append(
+                            ModelSyntaxError(
+                                code=cls.VARIABLE_ERROR_CODE,
+                                message=f"Invalid variable type in list for variable {'.'.join(subpath.parts)}.",
+                                fix="Ensure that all items in the list are of the same supported type either (str, int, float, bool) or dict.",
+                            )
+                        )
+                else:
+                    raise NotImplementedError(f"Unsupported variable type: {type(value)} for variable {subpath}")
+        return variables_by_path, errors
+
+    @classmethod
+    def _organize_variables_by_module(
+        cls, variables_by_path: dict[RelativeDirPath, list[BuildVariable]], selected_modules: set[RelativeDirPath]
+    ) -> dict[RelativeDirPath, dict[int, list[BuildVariable]]]:
+        module_path_by_relative_paths: dict[frozenset[RelativeDirPath], RelativeDirPath] = {
+            frozenset([module, *list(module.parents)]): module for module in selected_modules
+        }
+        variables_by_module: dict[RelativeDirPath, dict[int, list[BuildVariable]]] = defaultdict(
+            lambda: defaultdict(list)
+        )
+        for variable_path, variables in variables_by_path.items():
+            for module_paths, module in module_path_by_relative_paths.items():
+                if variable_path in module_paths:
+                    for iteration, variable in groupby(
+                        sorted(variables, key=lambda v: v.iteration or 0), key=lambda v: v.iteration or 0
+                    ):
+                        variables_by_module[module][iteration or 0].extend(variable)
+        return dict(variables_by_module)

cognite_toolkit/_cdf_tk/commands/build_v2/build_v2.py

@@ -1,7 +1,7 @@
 import os
 import sys
 from collections import defaultdict
-from collections.abc import Iterable
+from collections.abc import Iterable, Sequence
 from pathlib import Path
 
 from pydantic import JsonValue, ValidationError
@@ -13,9 +13,9 @@
 from cognite_toolkit._cdf_tk.commands._base import ToolkitCommand
 from cognite_toolkit._cdf_tk.commands.build_v2._module_source_parser import ModuleSourceParser
 from cognite_toolkit._cdf_tk.commands.build_v2.data_classes import (
-    BuildFiles,
     BuildFolder,
     BuildParameters,
+    BuildSourceFiles,
     BuiltModule,
     ConfigYAML,
     InsightList,
@@ -133,11 +133,8 @@ def _create_suggested_command(cls, display_path: Path, user_args: list[str]) ->
             suggestion.append(f"-o {display_path}")
         return f"'{' '.join(suggestion)}'"
 
-    def _parse_module_sources(self, build: BuildFiles) -> list[ModuleSource]:
-        parser = ModuleSourceParser(
-            build.selected_modules,
-            build.organization_dir,
-        )
+    def _parse_module_sources(self, build: BuildSourceFiles) -> list[ModuleSource]:
+        parser = ModuleSourceParser(build.selected_modules, build.organization_dir)
         module_sources = parser.parse(build.yaml_files, build.variables)
         if parser.errors:
             # Todo: Nicer way of formatting errors.
@@ -147,7 +144,7 @@ def _parse_module_sources(self, build: BuildFiles) -> list[ModuleSource]:
         return module_sources
 
     @classmethod
-    def _read_file_system(cls, parameters: BuildParameters) -> BuildFiles:
+    def _read_file_system(cls, parameters: BuildParameters) -> BuildSourceFiles:
         """Reads the file system to find the YAML files to build along with config.<name>.yaml if it exists."""
         selected: set[RelativeDirPath | str] = {
             parameters.modules_directory.relative_to(parameters.organization_dir)
@@ -183,7 +180,7 @@ def _read_file_system(cls, parameters: BuildParameters) -> BuildFiles:
             yaml_file.relative_to(parameters.organization_dir)
             for yaml_file in parameters.modules_directory.rglob("*.y*ml")
         ]
-        return BuildFiles(
+        return BuildSourceFiles(
             yaml_files=yaml_files,
             selected_modules=selected,
             variables=variables,
@@ -226,7 +223,7 @@ def _parse_user_selection(
         return selected, errors
 
     def _build_modules(
-        self, module_sources: Iterable[ModuleSource], build_dir: Path, max_workers: int = 1
+        self, module_sources: Sequence[ModuleSource], build_dir: Path, max_workers: int = 1
     ) -> BuildFolder:
         folder: BuildFolder = BuildFolder(path=build_dir)
 

cognite_toolkit/_cdf_tk/commands/build_v2/data_classes/__init__.py

@@ -1,14 +1,15 @@
-from ._build import BuildFiles, BuildFolder, BuildParameters, BuiltModule
+from ._build import BuildFolder, BuildParameters, BuildSourceFiles, BuiltModule
 from ._config import ConfigYAML
 from ._insights import ConsistencyError, Insight, InsightList, ModelSyntaxError, Recommendation
-from ._module import Module, ModuleSource, ResourceType
+from ._module import BuildVariable, Module, ModuleSource, ResourceType
 from ._types import AbsoluteDirPath, RelativeDirPath, RelativeFilePath, ValidationType
 
 __all__ = [
     "AbsoluteDirPath",
-    "BuildFiles",
     "BuildFolder",
     "BuildParameters",
+    "BuildSourceFiles",
+    "BuildVariable",
     "BuiltModule",
     "ConfigYAML",
     "ConsistencyError",

cognite_toolkit/_cdf_tk/commands/build_v2/data_classes/_build.py

@@ -29,7 +29,7 @@ def modules_directory(self) -> Path:
         return self.organization_dir / MODULES
 
 
-class BuildFiles(BaseModel):
+class BuildSourceFiles(BaseModel):
     """Intermediate format used when parsing modules"""
 
     yaml_files: list[RelativeFilePath] = Field(

cognite_toolkit/_cdf_tk/commands/build_v2/data_classes/_module.py

@@ -6,7 +6,15 @@
 from ._types import AbsoluteFilePath, RelativeDirPath
 
 
-class BuildVariable(BaseModel): ...
+class BuildVariable(BaseModel):
+    id: RelativeDirPath
+    value: str | bool | int | float | list[str | bool | int | float]
+    is_selected: bool
+    iteration: int | None = None
+
+    @property
+    def name(self) -> str:
+        return self.id.name
 
 
 class ModuleSource(BaseModel):

tests/test_unit/test_cdf_tk/test_commands/test_buildv2/test_command.py

@@ -169,8 +169,8 @@ def test_happy_path(self, tmp_path: Path) -> None:
             config_yaml_name="dev",
             user_selected_modules=["module1", "module2"],
         )
-        parse_input = BuildV2Command._read_file_system(parameters)
-        assert parse_input.model_dump() == {
+        build_files = BuildV2Command._read_file_system(parameters)
+        assert build_files.model_dump() == {
             "yaml_files": [resource_file.relative_to(tmp_path)],
             # Since user_selected_modules are provided, they should be used instead of config selected modules.
             "selected_modules": {"module1", "module2"},