Refactor capture and transformation handles

.github/workflows/ci.yml

@@ -65,4 +65,4 @@ jobs:
           pip install -e .
       - name: Run tests
         run: |
-          make test
+          make test-no-hardware

Makefile

@@ -1,5 +1,5 @@
 SOURCES=pyk4a example tests
-
+TESTS=tests
 .PHONY: setup fmt lint test help build
 .SILENT: help
 help:
@@ -9,7 +9,10 @@ help:
 		"- build: Build and install pyk4a package\n" \
 		"- fmt: Format all code\n" \
 		"- lint: Lint code syntax and formatting\n" \
-		"- test: Run tests"
+		"- test: Run tests\n"\
+		"- test-hardware: Run tests related from connected kinect"
+		"- test-no-hardware: Run tests without connected kinect"
+
 
 setup:
 	pip install -r requirements-dev.txt
@@ -27,4 +30,10 @@ lint:
 	mypy $(SOURCES)
 
 test:
-	pytest --cov=pyk4a
+	pytest --cov=pyk4a --verbose $(TESTS)
+
+test-hardware:
+	pytest --cov=pyk4a  -m "device" --verbose $(TESTS)
+
+test-no-hardware:
+	pytest --cov=pyk4a  -m "not device" --verbose $(TESTS)

example/bench.py

@@ -0,0 +1,159 @@
+from argparse import Action, ArgumentParser, Namespace
+from enum import Enum
+from time import monotonic
+
+from pyk4a import FPS, ColorResolution, Config, DepthMode, ImageFormat, PyK4A, WiredSyncMode
+
+
+class EnumAction(Action):
+    """
+    Argparse action for handling Enums
+    """
+
+    def __init__(self, **kwargs):
+        # Pop off the type value
+        enum = kwargs.pop("type", None)
+
+        # Ensure an Enum subclass is provided
+        if enum is None:
+            raise ValueError("type must be assigned an Enum when using EnumAction")
+        if not issubclass(enum, Enum):
+            raise TypeError("type must be an Enum when using EnumAction")
+
+        # Generate choices from the Enum
+        kwargs.setdefault("choices", tuple(e.name for e in enum))
+
+        super(EnumAction, self).__init__(**kwargs)
+
+        self._enum = enum
+
+    def __call__(self, parser, namespace, values, option_string=None):
+        # Convert value back into an Enum
+        setattr(namespace, self.dest, self._enum(values))
+
+
+class EnumActionTuned(Action):
+    """
+    Argparse action for handling Enums
+    """
+
+    def __init__(self, **kwargs):
+        # Pop off the type value
+        enum = kwargs.pop("type", None)
+
+        # Ensure an Enum subclass is provided
+        if enum is None:
+            raise ValueError("type must be assigned an Enum when using EnumAction")
+        if not issubclass(enum, Enum):
+            raise TypeError("type must be an Enum when using EnumAction")
+
+        # Generate choices from the Enum
+        kwargs.setdefault("choices", tuple(e.name.split("_")[-1] for e in enum))
+
+        super(EnumActionTuned, self).__init__(**kwargs)
+
+        self._enum = enum
+
+    def __call__(self, parser, namespace, values, option_string=None):
+        # Convert value back into an Enum
+        items = {item.name.split("_")[-1]: item.value for item in self._enum}
+        setattr(namespace, self.dest, self._enum(items[values]))
+
+
+def parse_args() -> Namespace:
+    parser = ArgumentParser(
+        description="Bench camera captures transfer speed. \n"
+        "You can check if you USB controller/cable has enough performance."
+    )
+    parser.add_argument("--device-id", type=int, default=0, help="Device ID, from zero. Default: 0")
+    parser.add_argument(
+        "--color-resolution",
+        type=ColorResolution,
+        action=EnumActionTuned,
+        default=ColorResolution.RES_720P,
+        help="Color sensor resoultion. Default: 720P",
+    )
+    parser.add_argument(
+        "--color-color_format",
+        type=ImageFormat,
+        action=EnumActionTuned,
+        default=ImageFormat.COLOR_BGRA32,
+        help="Color color_image color_format. Default: BGRA32",
+    )
+    parser.add_argument(
+        "--depth-mode",
+        type=DepthMode,
+        action=EnumAction,
+        default=DepthMode.NFOV_UNBINNED,
+        help="Depth sensor mode. Default: NFOV_UNBINNED",
+    )
+    parser.add_argument(
+        "--camera-fps", type=FPS, action=EnumActionTuned, default=FPS.FPS_30, help="Camera FPS. Default: 30"
+    )
+    parser.add_argument(
+        "--synchronized-images-only",
+        action="store_true",
+        dest="synchronized_images_only",
+        help="Only synchronized color and depth images, default",
+    )
+    parser.add_argument(
+        "--no-synchronized-images",
+        action="store_false",
+        dest="synchronized_images_only",
+        help="Color and Depth images can be non synced.",
+    )
+    parser.set_defaults(synchronized_images_only=True)
+    parser.add_argument(
+        "--wired-sync-mode",
+        type=WiredSyncMode,
+        action=EnumActionTuned,
+        default=WiredSyncMode.STANDALONE,
+        help="Wired sync mode. Default: STANDALONE",
+    )
+    return parser.parse_args()
+
+
+def bench(config: Config, device_id: int):
+    device = PyK4A(config=config, device_id=device_id)
+    device.connect()
+    depth = color = depth_period = color_period = 0
+    print("Press CTRL-C top stop benchmark")
+    started_at = started_at_period = monotonic()
+    while True:
+        try:
+            capture = device.get_capture()
+            if capture.color is not None:
+                color += 1
+                color_period += 1
+            if capture.depth is not None:
+                depth += 1
+                depth_period += 1
+            elapsed_period = monotonic() - started_at_period
+            if elapsed_period >= 2:
+                print(
+                    f"Color: {color_period / elapsed_period:0.2f} FPS, Depth: {depth_period / elapsed_period: 0.2f} FPS"
+                )
+                color_period = depth_period = 0
+                started_at_period = monotonic()
+        except KeyboardInterrupt:
+            break
+    elapsed = monotonic() - started_at
+    device.disconnect()
+    print()
+    print(f"Result: Color: {color / elapsed:0.2f} FPS, Depth: {depth / elapsed: 0.2f} FPS")
+
+
+def main():
+    args = parse_args()
+    config = Config(
+        color_resolution=args.color_resolution,
+        color_format=args.color_format,
+        depth_mode=args.depth_mode,
+        synchronized_images_only=args.synchronized_images_only,
+        wired_sync_mode=args.wired_sync_mode,
+    )
+    bench(config, args.device_id)
+
+
+if __name__ == "__main__":
+    main()

example/color_formats.py

@@ -25,27 +25,27 @@ def get_color_image_size(config, imshow=True):
 
 def convert_to_bgra_if_required(k4a, img_color):
     # examples for all possible pyk4a.ColorFormats
-    if k4a._config.color_format == pyk4a.ColorFormat.MJPG:
+    if k4a._config.color_format == pyk4a.ImageFormat.COLOR_MJPG:
         img_color = cv2.imdecode(img_color, cv2.IMREAD_COLOR)
-    elif k4a._config.color_format == pyk4a.ColorFormat.NV12:
+    elif k4a._config.color_format == pyk4a.ImageFormat.COLOR_NV12:
         img_color = cv2.cvtColor(img_color, cv2.COLOR_YUV2BGRA_NV12)
-        # this also works and it explains how the NV12 color format is stored in memory
-        # h, w = img_color.shape[0:2]
+        # this also works and it explains how the COLOR_NV12 color color_format is stored in memory
+        # h, w = color_image.shape[0:2]
         # h = h // 3 * 2
-        # luminance = img_color[:h]
-        # chroma = img_color[h:, :w//2]
-        # img_color = cv2.cvtColorTwoPlane(luminance, chroma, cv2.COLOR_YUV2BGRA_NV12)
-    elif k4a._config.color_format == pyk4a.ColorFormat.YUY2:
+        # luminance = color_image[:h]
+        # chroma = color_image[h:, :w//2]
+        # color_image = cv2.cvtColorTwoPlane(luminance, chroma, cv2.COLOR_YUV2BGRA_NV12)
+    elif k4a._config.color_format == pyk4a.ImageFormat.COLOR_YUY2:
         img_color = cv2.cvtColor(img_color, cv2.COLOR_YUV2BGRA_YUY2)
     return img_color
 
 
 if __name__ == "__main__":
     imshow = True
-    config_BGRA32 = Config(color_format=pyk4a.ColorFormat.BGRA32)
-    config_MJPG = Config(color_format=pyk4a.ColorFormat.MJPG)
-    config_NV12 = Config(color_format=pyk4a.ColorFormat.NV12)
-    config_YUY2 = Config(color_format=pyk4a.ColorFormat.YUY2)
+    config_BGRA32 = Config(color_format=pyk4a.ImageFormat.COLOR_BGRA32)
+    config_MJPG = Config(color_format=pyk4a.ImageFormat.COLOR_MJPG)
+    config_NV12 = Config(color_format=pyk4a.ImageFormat.COLOR_NV12)
+    config_YUY2 = Config(color_format=pyk4a.ImageFormat.COLOR_YUY2)
 
     nbytes_BGRA32 = get_color_image_size(config_BGRA32, imshow=imshow)
     nbytes_MJPG = get_color_image_size(config_MJPG, imshow=imshow)
@@ -57,4 +57,4 @@ def convert_to_bgra_if_required(k4a, img_color):
 
     # output:
     # nbytes_BGRA32=3686400 nbytes_MJPG=229693
-    # BGRA32 is 16.04924834452944 larger
+    # COLOR_BGRA32 is 16.04924834452944 larger

example/playback.py

@@ -1,16 +1,61 @@
 from argparse import ArgumentParser
-from json import dumps, loads
+from typing import Optional, Tuple
 
-from pyk4a import PyK4APlayback
+import cv2
+import numpy as np
+
+from pyk4a import ImageFormat, PyK4APlayback
+
+
+def colorize(
+    image: np.ndarray,
+    clipping_range: Tuple[Optional[int], Optional[int]] = (None, None),
+    colormap: int = cv2.COLORMAP_HSV,
+) -> np.ndarray:
+    if clipping_range[0] or clipping_range[1]:
+        img = image.clip(clipping_range[0], clipping_range[1])
+    else:
+        img = image.copy()
+    img = cv2.normalize(img, None, 0, 255, cv2.NORM_MINMAX, dtype=cv2.CV_8U)
+    img = cv2.applyColorMap(img, colormap)
+    return img
+
+
+def convert_to_bgra_if_required(color_format, color_image):
+    # examples for all possible pyk4a.ColorFormats
+    if color_format == ImageFormat.COLOR_MJPG:
+        color_image = cv2.imdecode(color_image, cv2.IMREAD_COLOR)
+    elif color_format == ImageFormat.COLOR_NV12:
+        color_image = cv2.cvtColor(color_image, cv2.COLOR_YUV2BGRA_NV12)
+        # this also works and it explains how the COLOR_NV12 color color_format is stored in memory
+        # h, w = color_image.shape[0:2]
+        # h = h // 3 * 2
+        # luminance = color_image[:h]
+        # chroma = color_image[h:, :w//2]
+        # color_image = cv2.cvtColorTwoPlane(luminance, chroma, cv2.COLOR_YUV2BGRA_NV12)
+    elif color_format == ImageFormat.COLOR_YUY2:
+        color_image = cv2.cvtColor(color_image, cv2.COLOR_YUV2BGRA_YUY2)
+    return color_image
 
 
 def info(playback: PyK4APlayback):
     print(f"Record length: {playback.length / 1000000: 0.2f} sec")
 
-    calibration_str = playback.calibration_json
-    calibration_formatted = dumps(loads(calibration_str), indent=2)
-    print("=== Calibration ===")
-    print(calibration_formatted)
+
+def play(playback: PyK4APlayback):
+    while True:
+        try:
+            capture = playback.get_next_capture()
+            if capture.color is not None:
+                cv2.imshow("Color", convert_to_bgra_if_required(playback.configuration["color_format"], capture.color))
+            if capture.depth is not None:
+                cv2.imshow("Depth", colorize(capture.depth, (None, 5000)))
+            key = cv2.waitKey(10)
+            if key != -1:
+                break
+        except EOFError:
+            break
+    cv2.destroyAllWindows()
 
 
 def main() -> None:
@@ -29,6 +74,7 @@ def main() -> None:
 
     if offset != 0.0:
         playback.seek(int(offset * 1000000))
+    play(playback)
 
     playback.close()
 

example/viewer_depth.py

@@ -8,9 +8,9 @@
 def main():
     k4a = PyK4A(
         Config(
-            color_resolution=pyk4a.ColorResolution.RES_720P,
+            color_resolution=pyk4a.ColorResolution.OFF,
             depth_mode=pyk4a.DepthMode.NFOV_UNBINNED,
-            synchronized_images_only=True,
+            synchronized_images_only=False,
         )
     )
     k4a.connect()
@@ -31,8 +31,6 @@ def main():
             # coloring image by choosed color map
             colored_depth = cv2.applyColorMap(normalized_depth, cv2.COLORMAP_HSV)
             cv2.imshow("k4a", colored_depth)
-            # cv2.imshow('k4a', capture.ir)
-            # cv2.imshow('k4a', capture.color)
             key = cv2.waitKey(10)
             if key != -1:
                 cv2.destroyAllWindows()

example/viewer_transformation.py

@@ -0,0 +1,50 @@
+from typing import Optional, Tuple
+
+import cv2
+import numpy as np
+
+import pyk4a
+from pyk4a import Config, PyK4A
+
+
+def colorize(
+    image: np.ndarray,
+    clipping_range: Tuple[Optional[int], Optional[int]] = (None, None),
+    colormap: int = cv2.COLORMAP_HSV,
+) -> np.ndarray:
+    if clipping_range[0] or clipping_range[1]:
+        img = image.clip(clipping_range[0], clipping_range[1])
+    else:
+        img = image.copy()
+    img = cv2.normalize(img, None, 0, 255, cv2.NORM_MINMAX, dtype=cv2.CV_8U)
+    img = cv2.applyColorMap(img, colormap)
+    return img
+
+
+def main():
+    k4a = PyK4A(Config(color_resolution=pyk4a.ColorResolution.RES_720P, depth_mode=pyk4a.DepthMode.NFOV_UNBINNED,))
+    k4a.connect()
+
+    while True:
+        capture = k4a.get_capture()
+        if capture.depth is not None:
+            cv2.imshow("Depth", colorize(capture.depth, (None, 5000)))
+        if capture.ir is not None:
+            cv2.imshow("IR", colorize(capture.ir, (None, 500), colormap=cv2.COLORMAP_JET))
+        if capture.color is not None:
+            cv2.imshow("Color", capture.color)
+        if capture.transformed_depth is not None:
+            cv2.imshow("Transformed Depth", colorize(capture.transformed_depth, (None, 5000)))
+        if capture.transformed_color is not None:
+            cv2.imshow("Transformed Color", capture.transformed_color)
+
+        key = cv2.waitKey(10)
+        if key != -1:
+            cv2.destroyAllWindows()
+            break
+
+    k4a.disconnect()
+
+
+if __name__ == "__main__":
+    main()