v0.15.2

version 0.15.2

Author: thusser

.pre-commit-config.yaml

@@ -0,0 +1,6 @@
+repos:
+  - repo: https://github.com/psf/black
+    rev: 21.12b0
+    hooks:
+      - id: black
+        language_version: python3.9

pyobs_aravis/__init__.py

@@ -1,4 +1,4 @@
 from .araviscamera import AravisCamera
 
 
-__all__ = ['AravisCamera']
+__all__ = ["AravisCamera"]

pyobs_aravis/aravis.py

@@ -9,27 +9,31 @@
 import numpy as np
 import ctypes
 import gi
-gi.require_version('Aravis', '0.8')
+
+gi.require_version("Aravis", "0.8")
 from gi.repository import Aravis
 
 __author__ = "Olivier Roulet-Dubonnet, Morten Lind"
 __copyright__ = "Copyright 2011-2013, Sintef Raufoss Manufacturing"
 __license__ = "GPLv3"
 __version__ = "0.5"
 
+
 class AravisException(Exception):
     pass
 
+
 class Camera(object):
     """
-    Create a Camera object. 
+    Create a Camera object.
     name is the camera ID in aravis.
     If name is None, the first found camera is used.
     If no camera is found an AravisException is raised.
     """
+
     def __init__(self, name=None, loglevel=logging.WARNING):
         self.logger = logging.getLogger(self.__class__.__name__)
-        if len(logging.root.handlers) == 0: #dirty hack
+        if len(logging.root.handlers) == 0:  # dirty hack
             logging.basicConfig()
         self.logger.setLevel(loglevel)
         self.name = name
@@ -50,15 +54,17 @@ def __init__(self, name=None, loglevel=logging.WARNING):
         self._last_payload = 0
 
     def __getattr__(self, name):
-        if hasattr(self.cam, name): # expose methods from the aravis camera object which is also relatively high level
+        if hasattr(
+            self.cam, name
+        ):  # expose methods from the aravis camera object which is also relatively high level
             return getattr(self.cam, name)
-        #elif hasattr(self.dev, name): #epose methods from the aravis device object, this might be confusing
+        # elif hasattr(self.dev, name): #epose methods from the aravis device object, this might be confusing
         #    return getattr(self.dev, name)
         else:
             raise AttributeError(name)
 
     def __dir__(self):
-        tmp = list(self.__dict__.keys()) + self.cam.__dir__()# + self.dev.__dir__()
+        tmp = list(self.__dict__.keys()) + self.cam.__dir__()  # + self.dev.__dir__()
         return tmp
 
     def load_config(self, path):
@@ -85,7 +91,9 @@ def get_feature_type(self, name):
         genicam = self.dev.get_genicam()
         node = genicam.get_node(name)
         if not node:
-            raise AravisException("Feature {} does not seem to exist in camera".format(name))
+            raise AravisException(
+                "Feature {} does not seem to exist in camera".format(name)
+            )
         return node.get_node_name()
 
     def get_feature(self, name):
@@ -134,7 +142,9 @@ def get_feature_vals(self, name):
         if ntype == "Enumeration":
             return self.dev.get_available_enumeration_feature_values_as_strings(name)
         else:
-            raise AravisException("{} is not an enumeration but a {}".format(name, ntype))
+            raise AravisException(
+                "{} is not an enumeration but a {}".format(name, ntype)
+            )
 
     def read_register(self, address):
         return self.dev.read_register(address)
@@ -150,7 +160,9 @@ def create_buffers(self, nb=10, payload=None):
             self.stream.push_buffer(Aravis.Buffer.new_allocate(payload))
 
     def pop_frame(self, timestamp=False):
-        while True: #loop in python in order to allow interrupt, have the loop in C might hang
+        while (
+            True
+        ):  # loop in python in order to allow interrupt, have the loop in C might hang
             if timestamp:
                 ts, frame = self.try_pop_frame(timestamp)
             else:
@@ -186,7 +198,7 @@ def _array_from_buffer_address(self, buf):
         if not buf:
             return None
         pixel_format = buf.get_image_pixel_format()
-        bits_per_pixel = pixel_format >> 16 & 0xff
+        bits_per_pixel = pixel_format >> 16 & 0xFF
         if bits_per_pixel == 8:
             INTP = ctypes.POINTER(ctypes.c_uint8)
         else:
@@ -208,42 +220,41 @@ def __str__(self):
 
     def __repr__(self):
         return self.__str__()
-    
+
     def start_acquisition(self, nb_buffers=10):
         self.logger.info("starting acquisition")
         payload = self.cam.get_payload()
         if payload != self._last_payload:
-            #FIXME should clear buffers
-            self.create_buffers(nb_buffers, payload) 
+            # FIXME should clear buffers
+            self.create_buffers(nb_buffers, payload)
             self._last_payload = payload
         self.cam.start_acquisition()
 
     def start_acquisition_trigger(self, nb_buffers=1):
-        self.set_feature("AcquisitionMode", "Continuous") #no acquisition limits
-        self.set_feature("TriggerSource", "Software") #wait for trigger t acquire image
-        self.set_feature("TriggerMode", "On") #Not documented but necesary
+        self.set_feature("AcquisitionMode", "Continuous")  # no acquisition limits
+        self.set_feature(
+            "TriggerSource", "Software"
+        )  # wait for trigger t acquire image
+        self.set_feature("TriggerMode", "On")  # Not documented but necesary
         self.start_acquisition(nb_buffers)
 
     def start_acquisition_continuous(self, nb_buffers=20):
-        self.set_feature("AcquisitionMode", "Continuous") #no acquisition limits
-        #self.set_feature("TriggerSource", "Freerun") #as fast as possible
-        #self.set_string_feature("TriggerSource", "FixedRate") 
-        #self.set_feature("TriggerMode", "On") #Not documented but necesary
+        self.set_feature("AcquisitionMode", "Continuous")  # no acquisition limits
+        # self.set_feature("TriggerSource", "Freerun") #as fast as possible
+        # self.set_string_feature("TriggerSource", "FixedRate")
+        # self.set_feature("TriggerMode", "On") #Not documented but necesary
         self.start_acquisition(nb_buffers)
 
     def stop_acquisition(self):
         self.cam.stop_acquisition()
 
     def shutdown(self):
-	#Delete the objects on shutdown: socket will be closed!
+        # Delete the objects on shutdown: socket will be closed!
         del self.stream
         del self.dev
         del self.cam
 
 
-
-
-
 def get_device_ids():
     Aravis.update_device_list()
     n = Aravis.get_n_devices()
@@ -252,60 +263,59 @@ def get_device_ids():
 
 def show_frame(frame):
     import cv2
+
     cv2.imshow("capture", frame)
     cv2.waitKey(0)
 
+
 def save_frame(frame, path="frame.png"):
     print("Saving frame to ", path)
     np.save(path, frame)
 
+
 def sfn(cam, path="frame.png"):
     from PIL import Image
+
     cam.start_acquisition()
     frame = cam.pop_frame()
     cam.stop_acquisition()
     im = Image.fromarray(frame)
     print("Saving image to ", path)
     im.save(path)
 
+
 def get_frame(cam):
     cam.start_acquisition()
     frame = cam.pop_frame()
     cam.stop_acquisition()
     return frame
 
 
-
-
-
-
-
 if __name__ == "__main__":
-    #cam = Camera("Prosilica-02-2110A-06145")
-    #cam = Camera("AT-Automation Technology GmbH-20805103")
+    # cam = Camera("Prosilica-02-2110A-06145")
+    # cam = Camera("AT-Automation Technology GmbH-20805103")
     cam = Camera(None)
     try:
-        #Aravis.enable_interface ("Fake")
-        #x, y, width, height = cam.get_region()
+        # Aravis.enable_interface ("Fake")
+        # x, y, width, height = cam.get_region()
         print("Camera model: ", cam.get_model_name())
         print("Vendor Name: ", cam.get_vendor_name())
         print("Device id: ", cam.get_device_id())
-        #print("Image size: ", width, ",", height)
-        print("Sensor size: ", cam.get_sensor_size()) 
+        # print("Image size: ", width, ",", height)
+        print("Sensor size: ", cam.get_sensor_size())
         print("Exposure: ", cam.get_exposure_time())
         print("Frame rate: ", cam.get_frame_rate())
         print("Payload: ", cam.get_payload())
         print("AcquisitionMode: ", cam.get_feature("AcquisitionMode"))
         print("Acquisition vals: ", cam.get_feature_vals("AcquisitionMode"))
-        #print("TriggerMode: ", cam.get_feature("TriggerMode"))
-        #print("Bandwidth: ", cam.get_feature("StreamBytesPerSecond"))
+        # print("TriggerMode: ", cam.get_feature("TriggerMode"))
+        # print("Bandwidth: ", cam.get_feature("StreamBytesPerSecond"))
         print("PixelFormat: ", cam.get_feature("PixelFormat"))
-        #print("ExposureAuto: ", cam.get_feature("ExposureAuto"))
+        # print("ExposureAuto: ", cam.get_feature("ExposureAuto"))
         print("PacketSize: ", cam.get_feature("GevSCPSPacketSize"))
 
-
         from IPython import embed
+
         embed()
     finally:
         cam.shutdown()
-

pyobs_aravis/araviscamera.py

@@ -1,7 +1,8 @@
 import asyncio
 import logging
 import time
-from typing import Any, Dict, Optional
+import numpy.typing as npt
+from typing import Any, Dict, Optional, List
 
 from . import aravis
 
@@ -14,9 +15,16 @@
 
 class AravisCamera(BaseVideo, IExposureTime):
     """A pyobs module for Aravis cameras."""
-    __module__ = 'pyobs_aravis'
 
-    def __init__(self, device: str, settings: Optional[Dict[str, Any]] = None, **kwargs: Any):
+    __module__ = "pyobs_aravis"
+
+    def __init__(
+        self,
+        device: str,
+        settings: Optional[Dict[str, Any]] = None,
+        buffers: int = 5,
+        **kwargs: Any,
+    ):
         """Initializes a new AravisCamera.
 
         Args:
@@ -29,21 +37,24 @@ def __init__(self, device: str, settings: Optional[Dict[str, Any]] = None, **kwa
         self._camera: Optional[aravis.Camera] = None
         self._settings: Dict[str, Any] = {} if settings is None else settings
         self._camera_lock = asyncio.Lock()
+        self._buffers = buffers
 
         # thread
         if device is not None:
             self.add_background_task(self._capture)
         else:
-            log.error('No device name given, not connecting to any camera.')
+            log.error("No device name given, not connecting to any camera.")
 
     async def open(self) -> None:
         """Open module."""
         await BaseVideo.open(self)
 
         # list devices
-        ids = aravis.get_device_ids()
+        ids: List[str] = aravis.get_device_ids()  # type: ignore
         if self._device_name not in ids:
-            raise ValueError('Could not find given device name in list of available cameras.')
+            raise ValueError(
+                "Could not find given device name in list of available cameras."
+            )
 
         # open camera
         await self.activate_camera()
@@ -58,25 +69,25 @@ def _open_camera(self) -> None:
         """Open camera."""
 
         # open camera
-        log.info('Connecting to camera %s...', self._device_name)
-        self._camera = aravis.Camera(self._device_name)
-        log.info('Connected.')
+        log.info("Connecting to camera %s...", self._device_name)
+        self._camera = aravis.Camera(self._device_name)  # type: ignore
+        log.info("Connected.")
 
         # settings
         for key, value in self._settings.items():
-            log.info(f'Setting value {key}={value}...')
-            self._camera.set_feature(key, value)
+            log.info(f"Setting value {key}={value}...")
+            self._camera.set_feature(key, value)  # type: ignore
 
         # start acquisition
-        self._camera.start_acquisition_continuous(nb_buffers=5)
+        self._camera.start_acquisition_continuous(nb_buffers=self._buffers)  # type: ignore
 
     def _close_camera(self) -> None:
         """Close camera."""
         # stop camera
         if self._camera is not None:
-            log.info('Closing camera...')
-            self._camera.stop_acquisition()
-            self._camera.shutdown()
+            log.info("Closing camera...")
+            self._camera.stop_acquisition()  # type: ignore
+            self._camera.shutdown()  # type: ignore
         self._camera = None
 
     async def _activate_camera(self) -> None:
@@ -107,7 +118,7 @@ async def _capture(self) -> None:
                 continue
 
             # read frame
-            frame = self._camera.pop_frame()
+            frame: npt.NDArray[float] = self._camera.pop_frame()  # type: ignore
             last = time.time()
 
             # process it
@@ -123,7 +134,7 @@ async def set_exposure_time(self, exposure_time: float, **kwargs: Any) -> None:
             ValueError: If exposure time could not be set.
         """
         await self.activate_camera()
-        self._camera.set_exposure_time(exposure_time * 1e6)
+        self._camera.set_exposure_time(exposure_time * 1e6)  # type: ignore
 
     async def get_exposure_time(self, **kwargs: Any) -> float:
         """Returns the exposure time in seconds.
@@ -132,7 +143,10 @@ async def get_exposure_time(self, **kwargs: Any) -> float:
             Exposure time in seconds.
         """
         await self.activate_camera()
-        return self._camera.get_exposure_time() / 1e6
+        return self._camera.get_exposure_time() / 1e6  # type: ignore
+
+    async def get_exposure_time_left(self, **kwargs: Any) -> float:
+        return 0.0
 
 
-__all__ = ['AravisCamera']
+__all__ = ["AravisCamera"]

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pyobs-aravis"
-version = "0.15.1"
+version = "0.15.2"
 description = "pyobs module for Aravis cameras"
 authors = ["Tim-Oliver Husser <[email protected]>"]
 license = "MIT"
@@ -11,6 +11,8 @@ numpy = "^1.21"
 PyGObject = "^3.42"
 
 [tool.poetry.dev-dependencies]
+black = "^21.12b0"
+pre-commit = "^2.16.0"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]