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,236 @@ 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."""
+        return path.startswith(("oci://", "docker://")) or (":" 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
+            return path, operator if isinstance(operator, dict) else None, oci_url_from_argv
+
+        if self._is_oci_path(path):
+            oci_url = path
+            if target is None:
+                print("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://", "docker://")):
+            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():