Add more tensor-native blocks

Author: PawelPeczek-Roboflow

inference/core/workflows/core_steps/classical_cv/convert_grayscale/v1_tensor.py

@@ -0,0 +1,89 @@
+"""Tensor-native sibling of ``convert_grayscale/v1``.
+
+OpenCV's uint8 BGR2GRAY is a per-pixel fixed-point map. The u8 path of the
+installed OpenCV 4.x uses the 15-bit coefficients (``RY15``/``GY15``/``BY15``
+in ``color_rgb2gray``):
+
+    gray = (9798*R + 19235*G + 3735*B + (1 << 14)) >> 15
+
+Verified exhaustively against cv2 4.10: zero mismatches over ALL 2^24 RGB
+triples (a total proof over the input space, not a sample), including
+odd-width/SIMD-tail layouts and non-contiguous inputs, so the map is
+position-independent. Note the widely quoted legacy constants
+(``4899/9617/1868 >> 14``) are NOT what modern cv2 computes - they disagree on
+~0.26% of triples.
+
+Being a pure function of (R, G, B), the conversion runs fully device-resident
+on the CHW RGB tensor: cast to int32 (max accumulator ``255*32768 + 2^14 =
+8_372_224``, far inside int32), weighted channel sum plus the rounding
+constant, arithmetic shift, cast back to uint8, emitted as the ``(1, H, W)``
+grayscale contract shape - zero host syncs, BIT-EXACT versus the numpy block.
+
+Delegation paths:
+- numpy/base64-born images (no materialised tensor) keep v1's numpy math
+  instead of forcing an eager host->device conversion - the standard
+  materialization-aware rule (this also covers 4-channel BGRA input, which can
+  only arrive numpy-born);
+- tensor-born single-channel images delegate too: materialising ``numpy_image``
+  yields the 2-D view that v1 would feed to ``cv2.cvtColor``, which rejects it
+  with ``cv2.error`` - so the error behaviour matches v1 exactly.
+"""
+
+from typing import Type
+
+import cv2
+import torch
+
+from inference.core.workflows.core_steps.classical_cv.convert_grayscale.v1 import (
+    ConvertGrayscaleManifest,
+)
+from inference.core.workflows.core_steps.visualizations.common.base import (
+    OUTPUT_IMAGE_KEY,
+)
+from inference.core.workflows.execution_engine.entities.base import WorkflowImageData
+from inference.core.workflows.prototypes.block import BlockResult, WorkflowBlock
+
+# OpenCV 4.x u8 BGR2GRAY fixed-point constants (color_rgb2gray RY15/GY15/BY15),
+# exhaustively verified against the installed cv2 - see the module docstring.
+_RY15 = 9798
+_GY15 = 19235
+_BY15 = 3735
+_GRAY_SHIFT = 15
+_GRAY_ROUND = 1 << (_GRAY_SHIFT - 1)
+
+
+class ConvertGrayscaleBlockV1(WorkflowBlock):
+    @classmethod
+    def get_manifest(cls) -> Type[ConvertGrayscaleManifest]:
+        return ConvertGrayscaleManifest
+
+    def run(
+        self,
+        image: WorkflowImageData,
+        *args,
+        **kwargs,
+    ) -> BlockResult:
+        if not image.is_tensor_materialised() or image.tensor_image.shape[0] != 3:
+            # Numpy/base64-born image (or a tensor-born single-channel one,
+            # whose 2-D numpy view cv2 rejects exactly as in v1): keep v1's
+            # numpy one-liner instead of forcing a host->device conversion.
+            gray = cv2.cvtColor(image.numpy_image, cv2.COLOR_BGR2GRAY)
+            output = WorkflowImageData.copy_and_replace(
+                origin_image_data=image, numpy_image=gray
+            )
+            return {OUTPUT_IMAGE_KEY: output}
+        return {OUTPUT_IMAGE_KEY: _convert_grayscale_tensor(image=image)}
+
+
+def _convert_grayscale_tensor(image: WorkflowImageData) -> WorkflowImageData:
+    """Device-resident mirror of ``cv2.cvtColor(..., cv2.COLOR_BGR2GRAY)`` for
+    ``(3, H, W)`` RGB uint8 tensors, emitting the ``(1, H, W)`` contract shape."""
+    chw = image.tensor_image.detach().to(torch.int32)
+    weighted = (
+        chw[0] * _RY15 + chw[1] * _GY15 + chw[2] * _BY15 + _GRAY_ROUND
+    ) >> _GRAY_SHIFT
+    gray_chw = weighted.to(torch.uint8).unsqueeze(0)
+    return WorkflowImageData.copy_and_replace(
+        origin_image_data=image,
+        tensor_image=gray_chw,
+    )

inference/core/workflows/core_steps/classical_cv/dominant_color/v1.py

@@ -1,4 +1,4 @@
-from typing import List, Literal, Optional, Type, Union
+from typing import List, Literal, Optional, Tuple, Type, Union
 
 import numpy as np
 from pydantic import AliasChoices, ConfigDict, Field
@@ -155,41 +155,53 @@ def run(
         scale_factor = max(1, min(width, height) // target_size)
         np_image = np_image[::scale_factor, ::scale_factor]
 
-        pixels = np_image.reshape(-1, 3).astype(np.float32)
-
-        centroids = pixels[
-            np.random.choice(pixels.shape[0], color_clusters, replace=False)
-        ]
-
-        for _ in range(max_iterations):
-            # Assign pixels to nearest centroid
-            distances = np.sqrt(((pixels[:, np.newaxis] - centroids) ** 2).sum(axis=2))
-            labels = np.argmin(distances, axis=1)
-
-            # Update centroids
-            new_centroids = np.zeros_like(centroids)
-            for i in range(color_clusters):
-                cluster_points = pixels[labels == i]
-                if len(cluster_points) > 0:
-                    new_centroids[i] = cluster_points.mean(axis=0)
-                else:
-                    # If cluster is empty, reinitialize to a random point
-                    new_centroids[i] = pixels[np.random.choice(pixels.shape[0])]
-
-            # Check for convergence
-            if np.allclose(centroids, new_centroids):
-                break
-
-            centroids = new_centroids
-
-        # Get the colors and their counts
-        colors = centroids
-        uniq, counts = np.unique(labels, return_counts=True)
-
-        # Find the most dominant color
-        dominant_color = colors[uniq[np.argmax(counts)]]
-        rgb_color = tuple(
-            int(np.clip(round(x), 0, 255)) for x in reversed(dominant_color)
+        rgb_color = find_dominant_color(
+            pixels_image=np_image,
+            color_clusters=color_clusters,
+            max_iterations=max_iterations,
         )
 
         return {"rgb_color": rgb_color}
+
+
+def find_dominant_color(
+    pixels_image: np.ndarray, color_clusters: int, max_iterations: int
+) -> Tuple[int, int, int]:
+    """K-means dominant color of an ALREADY-DOWNSAMPLED HWC BGR uint8 image,
+    returned as an ``(r, g, b)`` tuple.
+
+    Extracted verbatim from ``DominantColorBlockV1.run()`` (pure code motion:
+    the same unseeded global ``np.random`` draws in the same order) so the
+    tensor-native sibling can share the exact clustering trajectory."""
+    pixels = pixels_image.reshape(-1, 3).astype(np.float32)
+
+    centroids = pixels[np.random.choice(pixels.shape[0], color_clusters, replace=False)]
+
+    for _ in range(max_iterations):
+        # Assign pixels to nearest centroid
+        distances = np.sqrt(((pixels[:, np.newaxis] - centroids) ** 2).sum(axis=2))
+        labels = np.argmin(distances, axis=1)
+
+        # Update centroids
+        new_centroids = np.zeros_like(centroids)
+        for i in range(color_clusters):
+            cluster_points = pixels[labels == i]
+            if len(cluster_points) > 0:
+                new_centroids[i] = cluster_points.mean(axis=0)
+            else:
+                # If cluster is empty, reinitialize to a random point
+                new_centroids[i] = pixels[np.random.choice(pixels.shape[0])]
+
+        # Check for convergence
+        if np.allclose(centroids, new_centroids):
+            break
+
+        centroids = new_centroids
+
+    # Get the colors and their counts
+    colors = centroids
+    uniq, counts = np.unique(labels, return_counts=True)
+
+    # Find the most dominant color
+    dominant_color = colors[uniq[np.argmax(counts)]]
+    return tuple(int(np.clip(round(x), 0, 255)) for x in reversed(dominant_color))

inference/core/workflows/core_steps/classical_cv/dominant_color/v1_tensor.py

@@ -0,0 +1,90 @@
+"""Tensor-native sibling of ``dominant_color/v1``.
+
+The block's output is a single ``(r, g, b)`` tuple, not an image, and the
+algorithm is an iterative k-means over a downsampled frame: unseeded global
+``np.random`` initialisation, an empty-cluster reinit that also draws from the
+global RNG, and a host-side convergence check every iteration. Porting the
+loop to the device buys nothing - the clustered array is tiny (~100px min-dim
+after downsampling, k <= 10), every iteration would sync for the convergence
+check, and torch has no drop-in replica of the exact numpy arithmetic (and no
+replica of the global-numpy-RNG draws the trajectory is coupled to). So the
+clustering stays on the CPU in the SAME numpy code both paths share:
+``find_dominant_color``, imported from v1.
+
+What the tensor path DOES win is transfer volume: v1 materialises the full
+frame on the host (megabytes for HD frames) only to immediately discard all
+but every ``scale_factor``-th pixel. For a tensor-materialised image the
+strided downsample runs on the device (a zero-copy strided view) and only the
+small downsampled block (tens of KB) crosses to the host.
+
+Trajectory-preservation subtlety: v1's k-means consumes pixels in HWC
+row-major order with BGR channel order. Feeding RGB-ordered pixel vectors
+instead would permute each coordinate triple - mathematically
+distance-preserving, but the floating-point summation order in the distance
+computation changes, which can flip argmin ties and diverge the trajectory.
+The tensor path therefore flips the channel axis back to BGR and permutes to
+HWC before the single D2H copy, making the host-side array BYTE-IDENTICAL to
+v1's ``numpy_image[::scale_factor, ::scale_factor]``: identical bytes +
+identical RNG state => identical trajectory => identical output.
+
+Numpy/base64-born images delegate to the exact v1 numpy path instead of
+forcing an eager host->device conversion - the same materialization-aware rule
+the other tensor siblings follow. NOTE: v1 seeds nothing, so the block is
+nondeterministic run-to-run on BOTH paths; parity is exact only under a pinned
+global RNG (see the tests).
+"""
+
+from typing import Optional, Type
+
+import numpy as np
+import torch
+
+from inference.core.workflows.core_steps.classical_cv.dominant_color.v1 import (
+    DominantColorManifest,
+    find_dominant_color,
+)
+from inference.core.workflows.execution_engine.entities.base import WorkflowImageData
+from inference.core.workflows.prototypes.block import BlockResult, WorkflowBlock
+
+
+class DominantColorBlockV1(WorkflowBlock):
+    @classmethod
+    def get_manifest(cls) -> Type[DominantColorManifest]:
+        return DominantColorManifest
+
+    def run(
+        self,
+        image: WorkflowImageData,
+        color_clusters: Optional[int],
+        max_iterations: Optional[int],
+        target_size: Optional[int],
+        *args,
+        **kwargs
+    ) -> BlockResult:
+        if not image.is_tensor_materialised():
+            np_image = image.numpy_image
+            height, width = np_image.shape[:2]
+            scale_factor = max(1, min(width, height) // target_size)
+            downsampled = np_image[::scale_factor, ::scale_factor]
+        else:
+            chw = image.tensor_image
+            height, width = int(chw.shape[-2]), int(chw.shape[-1])
+            scale_factor = max(1, min(width, height) // target_size)
+            downsampled = _downsample_to_bgr_numpy(chw=chw, scale_factor=scale_factor)
+        rgb_color = find_dominant_color(
+            pixels_image=downsampled,
+            color_clusters=color_clusters,
+            max_iterations=max_iterations,
+        )
+        return {"rgb_color": rgb_color}
+
+
+def _downsample_to_bgr_numpy(chw: torch.Tensor, scale_factor: int) -> np.ndarray:
+    """Strided downsample on the device, then one small D2H copy of an HWC BGR
+    uint8 array byte-identical to v1's
+    ``numpy_image[::scale_factor, ::scale_factor]`` (see the module docstring
+    for why byte-identity - not mere colour equivalence - is required)."""
+    downsampled = chw.detach()[:, ::scale_factor, ::scale_factor]
+    # CHW RGB -> CHW BGR -> HWC BGR; contiguous on-device so the transfer
+    # copies exactly the downsampled block, nothing more.
+    return downsampled.flip(0).permute(1, 2, 0).contiguous().cpu().numpy()