Available devices count (#107)

* Add ability to querying devices count and read serial numbers

* Fix test

* Rename installed_count => connected_device_count

example/devices.py

@@ -0,0 +1,13 @@
+from pyk4a import PyK4A, connected_device_count
+
+
+cnt = connected_device_count()
+if not cnt:
+    print("No devices available")
+    exit()
+print(f"Available devices: {cnt}")
+for device_id in range(cnt):
+    device = PyK4A(device_id=device_id)
+    device.open()
+    print(f"{device_id}: {device.serial}")
+    device.close()

pyk4a/__init__.py

@@ -1,3 +1,5 @@
+import k4a_module
+
 from .calibration import Calibration, CalibrationType
 from .capture import PyK4ACapture
 from .config import (
@@ -21,6 +23,10 @@
 )
 
 
+def connected_device_count() -> int:
+    return k4a_module.device_get_installed_count()
+
+
 __all__ = (
     "Calibration",
     "CalibrationType",
@@ -43,4 +49,5 @@
     "depth_image_to_point_cloud",
     "depth_image_to_color_camera",
     "depth_image_to_color_camera_custom",
+    "connected_device_count",
 )

pyk4a/pyk4a.cpp

@@ -100,6 +100,48 @@ static PyObject *device_open(PyObject *self, PyObject *args) {
   return Py_BuildValue("IN", result, capsule);
 }
 
+static PyObject *device_get_installed_count(PyObject *self, PyObject *args) {
+  uint32_t count;
+  count = k4a_device_get_installed_count();
+  return Py_BuildValue("I", count);
+}
+
+static PyObject *device_get_serialnum(PyObject *self, PyObject *args) {
+  k4a_device_t *device_handle;
+  PyObject *capsule;
+  int thread_safe;
+  PyThreadState *thread_state;
+  k4a_buffer_result_t result;
+  size_t data_size;
+
+  PyArg_ParseTuple(args, "Op", &capsule, &thread_safe);
+  device_handle = (k4a_device_t *)PyCapsule_GetPointer(capsule, CAPSULE_DEVICE_NAME);
+
+  thread_state = _gil_release(thread_safe);
+  result = k4a_device_get_serialnum(*device_handle, NULL, &data_size);
+  if (result == K4A_BUFFER_RESULT_FAILED) {
+    _gil_restore(thread_state);
+    return Py_BuildValue("s", "");
+  }
+  char *data = (char *)malloc(data_size);
+  if (data == NULL) {
+    _gil_restore(thread_state);
+    fprintf(stderr, "Cannot allocate memory");
+    return Py_BuildValue("s", "");
+  }
+  result = k4a_device_get_serialnum(*device_handle, data, &data_size);
+  if (result != K4A_BUFFER_RESULT_SUCCEEDED) {
+    free(data);
+    return Py_BuildValue("s", "");
+  }
+  _gil_restore(thread_state);
+
+  PyObject *res = Py_BuildValue("s", data);
+  free(data);
+
+  return res;
+}
+
 static PyObject *device_close(PyObject *self, PyObject *args) {
   k4a_device_t *device_handle;
   PyObject *capsule;
@@ -1197,6 +1239,8 @@ static PyMethodDef Pyk4aMethods[] = {
     {"playback_get_next_capture", playback_get_next_capture, METH_VARARGS, "Get next capture from playback"},
     {"playback_get_previous_capture", playback_get_previous_capture, METH_VARARGS,
      "Get previous capture from playback"},
+    {"device_get_installed_count", device_get_installed_count, METH_VARARGS, "Gets the number of connected devices"},
+    {"device_get_serialnum", device_get_serialnum, METH_VARARGS, "Get the Azure Kinect device serial number."},
 
     {NULL, NULL, 0, NULL}};
 

pyk4a/pyk4a.py

@@ -140,6 +140,14 @@ def get_imu_sample(self, timeout: int = TIMEOUT_WAIT_INFINITE) -> Optional["ImuS
         _verify_error(res)
         return imu_sample
 
+    @property
+    def serial(self) -> str:
+        self._validate_is_opened()
+        ret = k4a_module.device_get_serialnum(self._device_handle, self.thread_safe)
+        if ret == "":
+            raise K4AException("Cannot read serial")
+        return ret
+
     @property
     def calibration_raw(self) -> str:
         self._validate_is_opened()

tests/functional/test_device.py

@@ -1,6 +1,6 @@
 import pytest
 
-from pyk4a import K4AException, PyK4A
+from pyk4a import K4AException, PyK4A, connected_device_count
 
 
 class TestOpenClose:
@@ -56,6 +56,13 @@ def test_get_calibration(device: PyK4A):
         calibration = device.calibration
         assert calibration._calibration_handle
 
+    @staticmethod
+    @pytest.mark.device
+    def test_serial(device: PyK4A):
+        device.open()
+        serial = device.serial
+        assert len(serial) > 5
+
 
 class TestCameras:
     @staticmethod
@@ -108,3 +115,10 @@ def test_calibration_raw(device: PyK4A):
 
         print(raw, file=sys.stderr)
         assert raw
+
+
+class TestInstalledCount:
+    @staticmethod
+    def test_count():
+        count = connected_device_count()
+        assert count >= 0

tests/plugins/device.py

@@ -20,6 +20,7 @@ class DeviceMeta:
     id: int
     jack_in: bool = False
     jack_out: bool = False
+    serial: str = "123456789"
     color_controls: Tuple[ColorControlCapabilities, ...] = (
         ColorControlCapabilities(
             color_control_command=ColorControlCommand.EXPOSURE_TIME_ABSOLUTE,
@@ -246,6 +247,10 @@ def device_get_raw_calibration(self) -> Optional[str]:
             assert self._opened is True
             return "{}"
 
+        def device_get_serialnum(self) -> str:
+            assert self._opened is True
+            return self._meta.serial
+
     def _device_open(device_id: int, thread_safe: bool) -> Tuple[int, object]:
         if device_id not in DEVICE_METAS:
             return Result.Failed.value, None
@@ -310,6 +315,12 @@ def _device_get_calibration(
     def _device_get_raw_calibration(capsule: DeviceHandle, thread_safe) -> Optional[str]:
         return capsule.device_get_raw_calibration()
 
+    def _device_get_installed_count() -> int:
+        return 1
+
+    def _device_get_serialnum(capsule: DeviceHandle, thread_safe) -> Optional[str]:
+        return capsule.device_get_serialnum()
+
     monkeypatch.setattr("k4a_module.device_open", _device_open)
     monkeypatch.setattr("k4a_module.device_close", _device_close)
     monkeypatch.setattr("k4a_module.device_get_sync_jack", _device_get_sync_jack)
@@ -324,6 +335,8 @@ def _device_get_raw_calibration(capsule: DeviceHandle, thread_safe) -> Optional[
     monkeypatch.setattr("k4a_module.device_get_imu_sample", _device_get_imu_sample)
     monkeypatch.setattr("k4a_module.device_get_calibration", _device_get_calibration)
     monkeypatch.setattr("k4a_module.device_get_raw_calibration", _device_get_raw_calibration)
+    monkeypatch.setattr("k4a_module.device_get_installed_count", _device_get_installed_count)
+    monkeypatch.setattr("k4a_module.device_get_serialnum", _device_get_serialnum)
 
 
 @pytest.fixture()

tests/unit/test_device.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from pyk4a import K4AException, PyK4A
+from pyk4a import K4AException, PyK4A, connected_device_count
 
 
 DEVICE_ID = 0
@@ -139,6 +139,12 @@ def test_calibration(device: PyK4A):
         calibration = device.calibration
         assert calibration
 
+    @staticmethod
+    def test_serial(device: PyK4A):
+        device.open()
+        serial = device.serial
+        assert serial == "123456789"
+
 
 class TestCameras:
     @staticmethod
@@ -179,3 +185,10 @@ def test_calibration_raw_on_closed_device(device: PyK4A):
     def test_calibration_raw(device: PyK4A):
         device.open()
         assert device.calibration_raw
+
+
+class TestInstalledCount:
+    @staticmethod
+    def test_count(patch_module_device):
+        count = connected_device_count()
+        assert count == 1