feat!: migrate to pypeline steps

.vscode/settings.json

@@ -27,5 +27,15 @@
     "editor.defaultFormatter": "charliermarsh.ruff"
   },
   "python.analysis.typeCheckingMode": "basic",
-  "cSpell.words": ["yanga", "typer", "cookiecutter", "hammocking"]
+  "cSpell.words": [
+    "yanga",
+    "typer",
+    "cookiecutter",
+    "hammocking",
+    "mashumaro",
+    "pypeline",
+    "slurper",
+    "subcomponent",
+    "subcomponents"
+  ]
 }

pyproject.toml

@@ -28,15 +28,16 @@ yanga = "yanga.ymain:main"
 [tool.poetry.dependencies]
 python = "^3.10,<3.13"
 mashumaro = "^3.5"
-loguru = "^0.7.0"
-cookiecutter = "^2.1.1"
-py-app-dev = "^2.0.0"
-customtkinter = "^5.2.0"
-pillow = "^10.1.0"
-kspl = "^1.0.0"
-pick = "^2.2.0"
-typer = "^0.11.1"
-hammocking = "^0.2.5"
+loguru = "^0.7"
+cookiecutter = "^2.1"
+py-app-dev = "^2.0"
+customtkinter = "^5.2"
+pillow = "^10.1"
+kspl = "^1.0"
+pick = "^2.2"
+typer = "^0.12"
+hammocking = "^0.2"
+pypeline-runner = "^1.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.0"

src/yanga/commands/project_templates/mini/yanga.yaml

@@ -28,13 +28,13 @@ variants:
 pipeline:
   install:
     - step: CreateVEnv
-      module: yanga.steps.create_venv
+      module: pypeline.steps.create_venv
     - step: ScoopInstall
-      module: yanga.steps.scoop_install
+      module: pypeline.steps.scoop_install
       description: Install scoop dependencies
       timeout_sec: 120
     - step: WestInstall
-      module: yanga.steps.west_install
+      module: pypeline.steps.west_install
       description: Install west dependencies
   gen:
     - step: KConfigGen

src/yanga/commands/run.py

@@ -1,15 +1,16 @@
 from argparse import ArgumentParser, Namespace
 from dataclasses import dataclass, field
+from pathlib import Path
 from typing import List, Optional
 
 from py_app_dev.core.cmd_line import Command, register_arguments_for_config_dataclass
 from py_app_dev.core.exceptions import UserNotificationException
 from py_app_dev.core.logging import logger, time_it
+from pypeline.pypeline import PipelineScheduler, PipelineStepsExecutor
 
 from yanga.domain.config import PlatformConfig, VariantConfig
-from yanga.domain.execution_context import UserRequest, UserRequestScope
+from yanga.domain.execution_context import ExecutionContext, UserRequest, UserRequestScope
 from yanga.domain.project_slurper import YangaProjectSlurper
-from yanga.yrun import PipelineScheduler, PipelineStepsExecutor
 
 from .base import CommandConfigBase, CommandConfigFactory, prompt_user_to_select_option
 
@@ -24,9 +25,7 @@ class RunCommandConfig(CommandConfigBase):
         default=None,
         metadata={"help": "SPL variant name. If none is provided, it will prompt to select one."},
     )
-    component_name: Optional[str] = field(
-        default=None, metadata={"help": "Restrict the scope to one specific component."}
-    )
+    component_name: Optional[str] = field(default=None, metadata={"help": "Restrict the scope to one specific component."})
     target: Optional[str] = field(default=None, metadata={"help": "Define a specific target to execute."})
     step: Optional[str] = field(
         default=None,
@@ -35,8 +34,7 @@ class RunCommandConfig(CommandConfigBase):
     single: bool = field(
         default=False,
         metadata={
-            "help": "If provided, only the provided step will run,"
-            " without running all previous steps in the pipeline.",
+            "help": "If provided, only the provided step will run," " without running all previous steps in the pipeline.",
             "action": "store_true",
         },
     )
@@ -74,54 +72,74 @@ def do_run(self, config: RunCommandConfig) -> int:
             return 0
         variant_name = self.determine_variant_name(config.variant_name, project_slurper.variants)
         platform_name = self.determine_platform_name(config.platform, project_slurper.platforms)
+        user_request = UserRequest(
+            scope=(UserRequestScope.COMPONENT if config.component_name else UserRequestScope.VARIANT),
+            variant_name=variant_name,
+            component_name=config.component_name,
+            target=config.target,
+        )
+        self.execute_pipeline_steps(
+            project_dir=config.project_dir,
+            project_slurper=project_slurper,
+            user_request=user_request,
+            variant_name=variant_name,
+            platform_name=platform_name,
+            step=config.step,
+            single=config.single,
+            force_run=config.force_run,
+        )
+        return 0
+
+    @staticmethod
+    def execute_pipeline_steps(
+        project_dir: Path,
+        project_slurper: YangaProjectSlurper,
+        user_request: UserRequest,
+        variant_name: Optional[str] = None,
+        platform_name: Optional[str] = None,
+        step: Optional[str] = None,
+        force_run: bool = False,
+        single: bool = False,
+    ) -> None:
         if not project_slurper.pipeline:
             raise UserNotificationException("No pipeline found in the configuration.")
         # Schedule the steps to run
-        steps_references = PipelineScheduler(project_slurper.pipeline, config.project_dir).get_steps_to_run(
-            config.step, config.single
-        )
+        steps_references = PipelineScheduler[ExecutionContext](project_slurper.pipeline, project_dir).get_steps_to_run(step, single)
         if not steps_references:
-            if config.step:
-                raise UserNotificationException(f"Step '{config.step}' not found in the pipeline.")
-            self.logger.info("No steps to run.")
-            return 0
-        user_request = UserRequest(
-            (UserRequestScope.COMPONENT if config.component_name else UserRequestScope.VARIANT),
-            variant_name,
-            config.component_name,
-            config.target,
+            if step:
+                raise UserNotificationException(f"Step '{step}' not found in the pipeline.")
+            logger.info("No steps to run.")
+            return
+        execution_context = ExecutionContext(
+            project_root_dir=project_dir,
+            variant_name=variant_name,
+            user_request=user_request,
+            components=(project_slurper.get_variant_components(variant_name) if variant_name else []),
+            user_config_files=project_slurper.user_config_files,
+            config_file=(project_slurper.get_variant_config_file(variant_name) if variant_name else None),
+            platform=project_slurper.get_platform(platform_name),
         )
-        PipelineStepsExecutor(
-            project_slurper,
-            variant_name,
-            platform_name,
-            user_request,
+        PipelineStepsExecutor[ExecutionContext](
+            execution_context,
             steps_references,
-            config.force_run,
+            force_run,
         ).run()
-        return 0
 
-    def determine_variant_name(
-        self, variant_name: Optional[str], variant_configs: List[VariantConfig]
-    ) -> Optional[str]:
+    def determine_variant_name(self, variant_name: Optional[str], variant_configs: List[VariantConfig]) -> Optional[str]:
         selected_variant_name: Optional[str]
         if not variant_name:
             if len(variant_configs) == 1:
                 selected_variant_name = variant_configs[0].name
                 self.logger.info(f"Only one variant found. Using '{selected_variant_name}'.")
             else:
-                selected_variant_name = prompt_user_to_select_option(
-                    [variant.name for variant in variant_configs], "Select variant: "
-                )
+                selected_variant_name = prompt_user_to_select_option([variant.name for variant in variant_configs], "Select variant: ")
         else:
             selected_variant_name = variant_name
         if not selected_variant_name:
             self.logger.warning("No variant selected. This might cause some steps to fail.")
         return selected_variant_name
 
-    def determine_platform_name(
-        self, platform_name: Optional[str], platform_configs: List[PlatformConfig]
-    ) -> Optional[str]:
+    def determine_platform_name(self, platform_name: Optional[str], platform_configs: List[PlatformConfig]) -> Optional[str]:
         selected_platform_name: Optional[str]
         if not platform_name:
             if len(platform_configs) == 1:

src/yanga/domain/config.py

@@ -10,7 +10,7 @@
 from mashumaro.config import TO_DICT_ADD_OMIT_NONE_FLAG, BaseConfig
 from mashumaro.mixins.json import DataClassJSONMixin
 from py_app_dev.core.exceptions import UserNotificationException
-from py_app_dev.core.pipeline import PipelineConfig, PipelineStepConfig
+from pypeline.domain.pipeline import PipelineConfig, PipelineStepConfig
 from yaml.parser import ParserError
 from yaml.scanner import ScannerError
 
@@ -84,8 +84,8 @@ class YangaUserConfig(DataClassDictMixin):
     platforms: List[PlatformConfig] = field(default_factory=list)
     variants: List[VariantConfig] = field(default_factory=list)
     components: List[ComponentConfig] = field(default_factory=list)
-    # This field is intended to keep track of where configuration was loaded from and
-    # it is automatically added when configuration is loaded from file
+    # This field is intended to keep track of where the configuration was loaded from and
+    # it is automatically added when the configuration is loaded from the file
     file: Optional[Path] = None
 
     @classmethod

src/yanga/domain/execution_context.py

@@ -1,11 +1,12 @@
 import os
 from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from enum import Enum, auto
 from pathlib import Path
 from typing import List, Optional, Union
 
 from py_app_dev.core.subprocess import SubprocessExecutor
+from pypeline.domain.execution_context import ExecutionContext as _ExecutionContext
 
 from .artifacts import ProjectArtifactsLocator
 from .components import Component
@@ -60,18 +61,25 @@ def get_include_directories(self) -> List[Path]: ...
 
 
 @dataclass
-class ExecutionContext:
-    project_root_dir: Path
-    variant_name: Optional[str]
-    user_request: UserRequest
-    components: List[Component] = field(default_factory=list)
-    user_config_files: List[Path] = field(default_factory=list)
-    config_file: Optional[Path] = None
-    # Keep track of all install directories, updated by any stage for the subsequent stages
-    install_dirs: List[Path] = field(default_factory=list)
-    # Keep track of all include directory providers
-    include_dirs_providers: List[IncludeDirectoriesProvider] = field(default_factory=list)
-    platform: Optional[PlatformConfig] = None
+class ExecutionContext(_ExecutionContext):
+    def __init__(
+        self,
+        project_root_dir: Path,
+        user_request: UserRequest,
+        variant_name: Optional[str] = None,
+        components: Optional[List[Component]] = None,
+        user_config_files: Optional[List[Path]] = None,
+        config_file: Optional[Path] = None,
+        platform: Optional[PlatformConfig] = None,
+    ) -> None:
+        super().__init__(project_root_dir)
+        self.user_request = user_request
+        self.variant_name = variant_name
+        self.components = components if components else []
+        self.user_config_files = user_config_files if user_config_files else []
+        self.config_file = config_file
+        self.platform = platform
+        self.include_dirs_providers: List[IncludeDirectoriesProvider] = []
 
     @property
     def include_directories(self) -> List[Path]:

src/yanga/domain/project_slurper.py

@@ -3,7 +3,7 @@
 
 from py_app_dev.core.exceptions import UserNotificationException
 from py_app_dev.core.logging import logger
-from py_app_dev.core.pipeline import PipelineConfig
+from pypeline.domain.pipeline import PipelineConfig
 
 from yanga.domain.artifacts import ProjectArtifactsLocator
 
@@ -18,9 +18,8 @@ class YangaProjectSlurper:
     def __init__(self, project_dir: Path) -> None:
         self.logger = logger.bind()
         self.project_dir = project_dir
-        self.user_configs: List[YangaUserConfig] = YangaConfigSlurper(
-            self.project_dir, [".git", ".github", ".vscode", "build", ".venv"]
-        ).slurp()
+        # TODO: Get rid of the exclude directories hardcoded list. Maybe use an ini file?
+        self.user_configs: List[YangaUserConfig] = YangaConfigSlurper(project_dir=self.project_dir, exclude_dirs=[".git", ".github", ".vscode", "build", ".venv"]).slurp()
         self.components_configs_pool: ComponentsConfigsPool = self._collect_components_configs(self.user_configs)
         self.pipeline: Optional[PipelineConfig] = self._find_pipeline_config(self.user_configs)
         self.variants: List[VariantConfig] = self._collect_variants(self.user_configs)
@@ -89,8 +88,6 @@ def _collect_components_configs(self, user_configs: List[YangaUserConfig]) -> Co
         for user_config in user_configs:
             for component_config in user_config.components:
                 if components_config.get(component_config.name, None):
-                    # TODO: throw the UserNotificationException and mention the two files
-                    #  where the components are defined
                     raise UserNotificationException(
                         f"Component '{component_config.name}' is defined in multiple configuration files."
                         f"See {components_config[component_config.name].file} and {user_config.file}"
@@ -117,9 +114,7 @@ def _resolve_subcomponents(
                     if not subcomponent:
                         # TODO: throw the UserNotificationException and mention the file
                         # where the subcomponent was defined
-                        raise UserNotificationException(
-                            f"Component '{subcomponent_name}' not found in the configuration."
-                        )
+                        raise UserNotificationException(f"Component '{subcomponent_name}' not found in the configuration.")
                     component.components.append(subcomponent)
                     subcomponent.is_subcomponent = True