ridesx: support OCI flashing

by leveraging FLS, we introduce the option to do `j storage flash oci://...` for ridesx targets.
Currently, this will extract the OCI archive to the exporter host and proceed to flash with fastboot CLI.

This also includes the option to specify which files to use by using -t/--target in combination with oci://
For example:
```
j storage flash -t boot_a:boot_a.simg system_a:system_a.simg oci://quay.io/bzlotnik/ridesx:latest
```

Otherwise it would rely on annotations to figure out the partitions from caib annotations:
"automotive.sdv.cloud.redhat.com/partition"

Signed-off-by: Benny Zlotnik <bzlotnik@redhat.com>

Author: bennyz

python/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/client.py

@@ -77,60 +77,246 @@ def flash_images(self, partitions: Dict[str, str], operators: Optional[Dict[str,
 
         return flash_result
 
-    def flash(
+    def _oci_url_from_argv(self) -> str | None:
+        """Return first OCI-like URL from sys.argv, or None."""
+        import sys
+
+        for arg in sys.argv[1:]:
+            if self._is_oci_path(arg):
+                return arg
+        return None
+
+    def _target_mappings_from_argv(self) -> Dict[str, str] | None:
+        """Return target mappings from sys.argv, if present."""
+        import sys
+
+        mapping: Dict[str, str] = {}
+        args = sys.argv[1:]
+        i = 0
+        while i < len(args):
+            arg = args[i]
+            spec = None
+            if arg in ("-t", "--target"):
+                if i + 1 < len(args):
+                    spec = args[i + 1]
+                    i += 1
+            elif arg.startswith("--target="):
+                spec = arg.split("=", 1)[1]
+
+            if spec and ":" in spec:
+                name, img = spec.split(":", 1)
+                mapping[name] = img
+            i += 1
+
+        return mapping or None
+
+    def _is_oci_path(self, path: str) -> bool:
+        """Return True if path looks like an OCI image reference (oci:// or bare registry URL)."""
+        if path.startswith("oci://"):
+            return True
+        # Reject other schemes (docker://, http://, etc.)
+        if "://" in path:
+            return False
+        # Bare registry URL: has colon (for tag) and slash (for path), not absolute path
+        return ":" in path and "/" in path and not path.startswith("/")
+
+    def _resolve_flash_input(
         self,
         path: str | Dict[str, str],
-        *,
-        target: str | None = None,
-        operator: Operator | Dict[str, Operator] | None = None,
-        compression=None,
-    ):
+        target: str | None,
+        operator: Operator | Dict[str, Operator] | None,
+        oci_url_from_argv: str | None,
+    ) -> tuple[Dict[str, str] | None, Dict[str, Operator] | None, str | None]:
+        """Resolve path/target/operator into (partitions, operators, oci_url)."""
         if isinstance(path, dict):
-            partitions = path
             operators = operator if isinstance(operator, dict) else None
+            oci_url = oci_url_from_argv if oci_url_from_argv is not None else None
+            return path, operators, oci_url
+
+        if self._is_oci_path(path):
+            oci_url = path
+            if target is None:
+                self.logger.info("Auto-detecting partitions from OCI image annotations...")
+                return None, None, oci_url
+            if ":" not in target:
+                raise ValueError(f"Target must be in format 'partition:filename' for OCI flashing, got: {target}")
+            partition_name, filename = target.split(":", 1)
+            partitions = {partition_name: filename}
         else:
             if target is None:
                 raise ValueError(
                     "This driver requires a target partition.\n"
-                    "Usage: j storage flash --target <partition>:<file>\n"
-                    "Example: j storage flash -t boot_a:aboot.img -t system_a:rootfs.simg -t system_b:qm_var.simg"
+                    "Usage: j storage flash --target <partition>:<file> [path]\n"
+                    "For OCI images: j storage flash [oci://image] (auto-detects partitions)\n"
+                    "For OCI with explicit mapping: j storage flash -t boot_a:boot.simg oci://image\n"
+                    "For local files: j storage flash -t boot_a:/path/to/boot.img"
                 )
             partitions = {target: path}
-            operators = {target: operator} if isinstance(operator, Operator) else None
+            oci_url = oci_url_from_argv
+
+        operators = {list(partitions.keys())[0]: operator} if partitions and isinstance(operator, Operator) else None
+        return partitions, operators, oci_url
 
+    def _validate_partition_mappings(self, partitions: Dict[str, str] | None) -> None:
+        """Validate partition mappings; raise ValueError if any path is empty."""
+        if partitions is None:
+            return
         for partition_name, file_path in partitions.items():
             if not file_path or not file_path.strip():
                 raise ValueError(
                     f"Partition '{partition_name}' has an empty file path. "
                     f"Please provide a valid file path (e.g., -t {partition_name}:/path/to/image)"
                 )
 
-        self.logger.info("Starting RideSX flash operation")
+    def _power_off_if_available(self) -> None:
+        """Power off device if power child is present."""
+        if "power" in self.children:
+            self.power.off()
+            self.logger.info("device powered off")
+        else:
+            self.logger.info("device left running")
 
+    def flash(
+        self,
+        path: str | Dict[str, str],
+        *,
+        target: str | None = None,
+        operator: Operator | Dict[str, Operator] | None = None,
+        compression=None,
+        fls_version: str | None = None,
+        fls_binary_url: str | None = None,
+    ):
+        oci_url_from_argv = self._oci_url_from_argv()
+        if oci_url_from_argv:
+            self.logger.info(f"Detected OCI URL from command line: {oci_url_from_argv}")
+
+        partitions, operators, oci_url = self._resolve_flash_input(path, target, operator, oci_url_from_argv)
+        if oci_url and partitions is None:
+            argv_targets = self._target_mappings_from_argv()
+            if argv_targets:
+                self.logger.info("Using target mappings from command line arguments")
+                partitions = argv_targets
+        self._validate_partition_mappings(partitions)
+
+        self.logger.info("Starting RideSX flash operation")
         self.boot_to_fastboot()
 
-        result = self.flash_images(partitions, operators)
+        if oci_url:
+            if partitions is None:
+                self.logger.info(f"Using FLS auto-detection for OCI image: {oci_url}")
+            else:
+                self.logger.info(f"Using FLS OCI flash with explicit mapping for image: {oci_url}")
+            result = self.flash_oci_auto(oci_url, partitions, fls_version=fls_version, fls_binary_url=fls_binary_url)
+        else:
+            self.logger.info("Using traditional file-based flashing")
+            result = self.flash_images(partitions, operators)
 
         self.logger.info("flash operation completed successfully")
+        self._power_off_if_available()
+        return result
 
-        if "power" in self.children:
-            self.power.off()
-            self.logger.info("device powered off")
+    def flash_oci_auto(
+        self,
+        oci_url: str,
+        partitions: Dict[str, str] | None = None,
+        fls_version: str | None = None,
+        fls_binary_url: str | None = None,
+    ):
+        """Flash OCI image using auto-detection or explicit partition mapping
+
+        Args:
+            oci_url: OCI image reference (e.g., "oci://registry.com/image:latest")
+            partitions: Optional mapping of partition -> filename inside OCI image
+            fls_version: Optional FLS version to download from GitHub releases
+            fls_binary_url: Optional custom URL to download FLS binary from
+        """
+        # Normalize OCI URL
+        if not oci_url.startswith("oci://"):
+            if "://" in oci_url:
+                raise ValueError(f"Only oci:// URLs are supported, got: {oci_url}")
+            if ":" in oci_url and "/" in oci_url:
+                oci_url = f"oci://{oci_url}"
+            else:
+                raise ValueError(f"Invalid OCI URL format: {oci_url}")
+
+        if partitions:
+            self.logger.info(f"Flashing OCI image with explicit mapping: {list(partitions.keys())}")
         else:
-            self.logger.info("device left running")
+            self.logger.info(f"Auto-detecting partitions for OCI image: {oci_url}")
 
-        return result
+        self.logger.info("Checking for fastboot devices on Exporter...")
+        detection_result = self.call("detect_fastboot_device", 5, 2.0)
+
+        if detection_result["status"] != "device_found":
+            raise RuntimeError("No fastboot devices found. Make sure device is in fastboot mode.")
+
+        device_id = detection_result["device_id"]
+        self.logger.info(f"Found fastboot device: {device_id}")
+
+        flash_result = self.call("flash_oci_image", oci_url, partitions, fls_version, fls_binary_url)
+
+        return flash_result
 
     def cli(self):
+        import click
+
         generic_cli = FlasherClient.cli(self)
 
         @driver_click_group(self)
         def base():
             """RideSX storage operations"""
             pass
 
+        # Add all generic commands except 'flash' (we override it)
         for name, cmd in generic_cli.commands.items():
-            base.add_command(cmd, name=name)
+            if name != "flash":
+                base.add_command(cmd, name=name)
+
+        @base.command()
+        @click.argument("file")
+        @click.option(
+            "-t",
+            "--target",
+            "target_specs",
+            multiple=True,
+            help="Partition mapping as partition:filename (e.g., boot_a:boot.img)",
+        )
+        @click.option(
+            "--fls-version",
+            type=str,
+            default=None,
+            help="FLS version to download from GitHub releases (e.g., 0.1.9)",
+        )
+        @click.option(
+            "--fls-binary-url",
+            type=str,
+            default=None,
+            help="Custom URL to download FLS binary from (overrides --fls-version)",
+        )
+        def flash(file, target_specs, fls_version, fls_binary_url):
+            """Flash image to device
+
+            For OCI images with auto-detection:
+                j storage flash oci://registry.com/image:tag
+
+            For OCI images with explicit partition mapping:
+                j storage flash -t boot_a:boot.img oci://registry.com/image:tag
+
+            For local files:
+                j storage flash -t boot_a:/path/to/boot.img .
+            """
+            if target_specs:
+                mapping: dict[str, str] = {}
+                for spec in target_specs:
+                    if ":" not in spec:
+                        raise click.ClickException(f"Invalid target spec '{spec}'. Expected format: partition:filename")
+                    name, img = spec.split(":", 1)
+                    mapping[name] = img
+
+                self.flash(mapping, fls_version=fls_version, fls_binary_url=fls_binary_url)
+                return
+
+            self.flash(file, target=None, fls_version=fls_version, fls_binary_url=fls_binary_url)
 
         @base.command()
         def boot_to_fastboot():

python/packages/jumpstarter-driver-ridesx/jumpstarter_driver_ridesx/driver.py

@@ -8,15 +8,17 @@
 from jumpstarter_driver_opendal.driver import Opendal
 
 from jumpstarter.common.exceptions import ConfigurationError
+from jumpstarter.common.fls import get_fls_binary
 from jumpstarter.driver import Driver, export
 
 
 @dataclass(kw_only=True)
 class RideSXDriver(Driver):
     """RideSX Driver"""
-    decompression_timeout: int = field(default=15 * 60) # 15 minutes
-    flash_timeout: int = field(default=30 * 60) # 30 minutes
-    continue_timeout: int = field(default=20 * 60) # 20 minutes
+
+    decompression_timeout: int = field(default=15 * 60)  # 15 minutes
+    flash_timeout: int = field(default=30 * 60)  # 30 minutes
+    continue_timeout: int = field(default=20 * 60)  # 20 minutes
     storage_dir: str = field(default="/var/lib/jumpstarter/ridesx")
 
     def __post_init__(self):
@@ -197,6 +199,61 @@ def flash_with_fastboot(self, device_id: str, partitions: Dict[str, str]):
             self.logger.warning(f"stdout: {e.stdout}")
             self.logger.warning(f"stderr: {e.stderr}")
 
+    @export
+    def flash_oci_image(
+        self,
+        oci_url: str,
+        partitions: Dict[str, str] | None = None,
+        fls_version: str | None = None,
+        fls_binary_url: str | None = None,
+    ):
+        """Flash OCI image using FLS fastboot CLI
+
+        Args:
+            oci_url: OCI image reference (e.g., "quay.io/bzlotnik/ridesx-image:latest")
+            partitions: Optional mapping of partition -> filename inside OCI image
+            fls_version: Optional FLS version to download from GitHub releases
+            fls_binary_url: Optional custom URL to download FLS binary from
+        """
+        if not oci_url.startswith("oci://"):
+            raise ValueError(f"OCI URL must start with oci://, got: {oci_url}")
+
+        fls_binary = get_fls_binary(fls_version, fls_binary_url)
+        fls_cmd = [fls_binary, "fastboot", oci_url]
+
+        if partitions:
+            for partition_name, filename in sorted(partitions.items()):
+                fls_cmd.extend(["-t", f"{partition_name}:{filename}"])
+
+        # Align fastboot timeout with driver timeout
+        fls_cmd.extend(["--timeout", str(self.flash_timeout)])
+
+        self.logger.info(f"Running FLS fastboot: {' '.join(fls_cmd)}")
+
+        try:
+            result = subprocess.run(fls_cmd, capture_output=True, text=True, check=True, timeout=self.flash_timeout)
+
+            self.logger.info("FLS fastboot auto-detection completed successfully")
+            self.logger.debug(f"FLS stdout: {result.stdout}")
+            if result.stderr:
+                self.logger.debug(f"FLS stderr: {result.stderr}")
+
+            return {"status": "success", "output": result.stdout}
+
+        except subprocess.CalledProcessError as e:
+            self.logger.error(f"FLS fastboot auto-detection failed - return code: {e.returncode}")
+            self.logger.error(f"stdout: {e.stdout}")
+            self.logger.error(f"stderr: {e.stderr}")
+            raise RuntimeError(f"FLS fastboot auto-detection failed: {e}") from e
+
+        except subprocess.TimeoutExpired:
+            self.logger.error("FLS fastboot auto-detection timed out")
+            raise RuntimeError("FLS fastboot auto-detection timeout") from None
+
+        except FileNotFoundError:
+            self.logger.error("FLS command not found - ensure FLS is installed and in PATH")
+            raise RuntimeError("FLS command not found") from None
+
     @export
     async def boot_to_fastboot(self):
         """Boot device to fastboot mode"""