Merge pull request #4 from nqminds/from-iam3dpo-code-repo

Mockup the `nqm.irimager.IRImager` class with dummy functions

pyproject.toml

@@ -18,6 +18,7 @@ generate-setup-file = true
 [tool.poetry.dependencies]
 python = "^3.8"
 pybind11 = "^2.10.4"
+numpy = "^1.24.3"
 
 
 [tool.poetry.group.dev.dependencies]

src/nqm/irimager/__init__.pyi

@@ -7,10 +7,55 @@ We use the IRImagerDirect SDK
 (see http://documentation.evocortex.com/libirimager2/html/index.html)
 to control these cameras.
 """
+import datetime
+import os
+import types
+import typing
+
+import numpy as np
+import numpy.typing as npt
+
+# TODO: replace with PEP 673 typing.Self once we support Python 3.11
+_SelfIRImager = typing.TypeVar("_SelfIRImager", bound="IRImager")
 
 class IRImager:
     """IRImager object - interfaces with a camera."""
 
-    def __init__(self, *args, **kwargs) -> None: ...
-    def test(self) -> int:
-        """Return the number 42"""
+    @typing.overload
+    def __init__(self) -> None: ...
+    @typing.overload
+    def __init__(self, xml_path: os.PathLike) -> None:
+        """Loads the configuration for an IR Camera from the given XML file"""
+    def start_streaming(self) -> None:
+        """Start video grabbing
+
+        Raises:
+            RuntimeError: If streaming cannot be started, e.g. if the camera is not connected.
+        """
+    def stop_streaming(self) -> None:
+        """Stop video grabbing"""
+    def __enter__(self: _SelfIRImager) -> _SelfIRImager: ...
+    def __exit__(
+        self,
+        exc_type: typing.Optional[typing.Type[BaseException]],
+        exc: typing.Optional[BaseException],
+        traceback: typing.Optional[types.TracebackType],
+    ) -> None: ...
+    def get_frame(self) -> typing.Tuple[npt.NDArray[np.uint16], datetime.datetime]:
+        """Return a frame
+
+        Returns:
+            A tuple containing:
+                - A 2-D numpy array containing the image. This must be adjusted
+                  by :py:meth:`~IRImager.get_temp_range_decimal` to get the
+                  actual temperature in degrees Celcius.
+                - The time the image was taken.
+        """
+    def get_temp_range_decimal(self) -> int:
+        """The number of decimal places in the thermal data
+
+        For example, if :py:meth:`~IRImager.get_frame` returns 18000, you can
+        divide this number by 10 to the power of the result of
+        :py:meth:`~IRImager.get_temp_range_decimal` to get the actual
+        temperature in degrees Celcius.
+        """

src/nqm/irimager/irimager.cpp

@@ -1,10 +1,74 @@
 #include <pybind11/pybind11.h>
+#include <pybind11/numpy.h>
+#include <pybind11/chrono.h>
+#include <pybind11/stl.h>
+#include <pybind11/stl/filesystem.h>
+#include <pybind11/stl_bind.h>
+
+#include <chrono>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <stdexcept>
 
 class IRImager {
     public:
-    int test() {
-        return 42;
+    IRImager() = default;
+    IRImager(const std::filesystem::path &xml_path) {
+        std::ifstream xml_stream(xml_path, std::fstream::in);
+
+        std::string xml_header(5, '\0');
+        xml_stream.read(xml_header.data(), xml_header.size());
+        if (xml_header != std::string("<?xml")) {
+            throw std::runtime_error("Invalid XML file: The given XML file does not start with '<?xml'");
+        }
+    }
+
+    void start_streaming() {
+        streaming = true;
+    }
+
+    void stop_streaming() {
+        streaming = false;
+    }
+
+    IRImager* _enter_() {
+        start_streaming();
+        return this;
+    }
+
+    void _exit_(
+        [[maybe_unused]] const std::optional<pybind11::type> &exc_type,
+        [[maybe_unused]] const std::optional<pybind11::error_already_set> &exc_value,
+        [[maybe_unused]] const pybind11::object &traceback) {
+        stop_streaming();
+    }
+
+    std::tuple<pybind11::array_t<uint16_t>, std::chrono::system_clock::time_point> get_frame() {
+        if (!streaming) {
+            throw std::runtime_error("IRIMAGER_STREAMOFF: Not streaming");
+        }
+
+        auto frame_size = std::array<ssize_t, 2>{128, 128};
+        auto my_array = pybind11::array_t<uint16_t>(frame_size);
+
+        auto r = my_array.mutable_unchecked<frame_size.size()>();
+
+        for (ssize_t i = 0; i < frame_size[0]; i++) {
+            for (ssize_t j = 0; j < frame_size[1]; j++) {
+                r(i, j) = 1800 * std::pow(10, get_temp_range_decimal());
+            }
+        }
+
+        return std::make_tuple(my_array, std::chrono::system_clock::now());
+    }
+
+    short get_temp_range_decimal() {
+        return 1;
     }
+
+private:
+    bool streaming = false;
 };
 
 PYBIND11_MODULE(irimager, m) {
@@ -16,5 +80,34 @@ to control these cameras.)";
 
     pybind11::class_<IRImager>(m, "IRImager", R"(IRImager object - interfaces with a camera.)")
         .def(pybind11::init<>())
-        .def("test", &IRImager::test, "Return the number 42");
+        .def(pybind11::init<const std::filesystem::path &>(), R"(Loads the configuration for an IR Camera from the given XML file)")
+        .def("get_frame", &IRImager::get_frame, R"(Return a frame
+
+Raises:
+    RuntimeError: If a frame cannot be loaded, e.g. if the camera isn't streaming.
+
+Returns:
+    A tuple containing:
+        - A 2-D numpy array containing the image. This must be adjusted
+          by :py:meth:`~IRImager.get_temp_range_decimal` to get the
+          actual temperature in degrees Celcius.
+        - The time the image was taken.
+)")
+        .def("get_temp_range_decimal", &IRImager::get_temp_range_decimal, R"(The number of decimal places in the thermal data
+
+For example, if :py:meth:`~IRImager.get_frame` returns 18000, you can
+divide this number by 10 to the power of the result of
+:py:meth:`~IRImager.get_temp_range_decimal` to get the actual
+temperature in degrees Celcius.
+)")
+        .def("start_streaming", &IRImager::start_streaming, R"(Start video grabbing
+
+Prefer using `with irimager: ...` to automatically start/stop streaming on errors.
+
+Raises:
+    RuntimeError: If streaming cannot be started, e.g. if the camera is not connected.
+)")
+        .def("stop_streaming", &IRImager::stop_streaming, R"(Stop video grabbing)")
+        .def("__enter__", &IRImager::_enter_, pybind11::return_value_policy::reference_internal)
+        .def("__exit__", &IRImager::_exit_);
 }

tests/__fixtures__/382x288@27Hz.xml

@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<imager xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema">
+  <videoformatindex>0</videoformatindex> <!-- index of the used video format (USB endpoint) -->
+  <formatspath>/usr/share/libirimager</formatspath>
+  <calipath>/usr/share/libirimager/cali</calipath>
+  <!-- Uncomment the following lines to specify user-defined parameters for the desired optic
+       and temperature range. Be aware to specify meaningful parameters.
+       See documentation for further information: http://evocortex.com/libirimager2/html/index.html
+       By default, the first available optic and the first meaningful temperature range are selected.
+  -->
+  <fov>25</fov>
+  <temperature>
+    <min>450</min>
+    <max>1800</max>
+  </temperature>
+  <optics_text></optics_text>
+  <framerate>27.0</framerate>             <!-- scaled down frame rate, must be less or equal than camera frame rate -->
+  <bispectral>0</bispectral>              <!-- 0=only thermal sensor, 1=bispectral technology (only PI200/PI230) -->
+  <average>0</average>                    <!-- average callback frames over intermediate frames -->
+  <autoflag>
+    <enable>1</enable>
+    <mininterval>15.0</mininterval>
+    <maxinterval>0.0</maxinterval>
+  </autoflag>
+  <pif_in_mode>0</pif_in_mode>            <!-- 0=Image capture (default), 1=Flag control -->
+  <pif_out_mode>0</pif_out_mode>          <!-- 0=Off (default), 1=Frame sync signal -->
+  <pif_out_voltage>5000</pif_out_voltage> <!-- PIF out voltage in mV -->
+  <tchipmode>0</tchipmode>                <!-- 0=Floating (default), 1=Auto, 2=Fixed value -->
+  <tchipfixedvalue>40.0</tchipfixedvalue> <!-- Fixed value for tchipmode=2 -->
+  <focus>50.0</focus> <!-- position of focus motor in % of range [0; 100] -->
+</imager>

tests/test_irimager.py

@@ -1,9 +1,62 @@
 """Tests for nqm.irimager.IRImager"""
+import datetime
+import pathlib
+
+import numpy as np
+import pytest
+
 from nqm.irimager import IRImager
 
 
-def test_irimager_test():
-    """Tests nqm.irimager.IRImager"""
+def test_irimager_loads_xml():
+    """Tests nqm.irimager.IRImager(xmlPath) can load an XML file"""
+    # should work with a valid XML file
+    IRImager(pathlib.Path("tests/__fixtures__/382x288@27Hz.xml"))
+
+    with pytest.raises(RuntimeError, match="Invalid XML file"):
+        IRImager(pathlib.Path("README.md"))
+
+
+def test_get_frame_fails_when_not_streaming():
+    """Calling `get_frame()` should raise an error when not streaming"""
+    irimager = IRImager()
+
+    with pytest.raises(RuntimeError, match="IRIMAGER_STREAMOFF"):
+        irimager.get_frame()
+
+
+def test_get_frame_in_context_manager():
+    """Calling `get_frame()` should work when starting streaming with `with`"""
+    irimager = IRImager()
+
+    # context manager should auto-start streaming
+    with irimager:
+        irimager.get_frame()
+
+    # context manager should auto-stop streaming
+    with pytest.raises(RuntimeError, match="IRIMAGER_STREAMOFF"):
+        irimager.get_frame()
+
+
+def test_irimager_get_frame():
+    """Tests nqm.irimager.IRImager#get_frame"""
+    irimager = IRImager()
+
+    with irimager:
+        array, timestamp = irimager.get_frame()
+
+        assert array.dtype == np.uint16
+        # should be 2-dimensional
+        assert array.ndim == 2
+        assert array.shape == (128, 128)
+
+        # image should have been taken in the last 30 seconds
+        assert timestamp > datetime.datetime.now() - datetime.timedelta(seconds=30)
+
+
+def test_irimager_get_temp_range_decimal():
+    """Tests that nqm.irimager.IRImager#get_temp_range_decimal returns an int"""
     irimager = IRImager()
 
-    assert irimager.test() == 42
+    assert irimager.get_temp_range_decimal() >= 0
+    assert isinstance(irimager.get_temp_range_decimal(), int)