Commit 74cae1ce authored by vertighel's avatar vertighel
Browse files

feat: loop mechanism + docstring/import cleanup across stx, atik, mako 



- Add looping property/setter (starts background thread), _run_loop,
  _start (private), start (with loop guard) to stx.Camera and atik.Camera
- mako.Guider: add looping/start via get_frame(); remove streaming
  callback machinery (_frame_handler, streaming property, stream_opencv,
  _range, FrameStatus); acquisition = 'capture'
- Fix atik.Camera.image: was incorrectly decorated as @property
- Move make_png import to module top in stx.py and atik.py
- Add numpy-style docstrings to all public methods and properties
  in all three files

Co-Authored-By: default avatarClaude Sonnet 4.6 <noreply@anthropic.com>
parent 75abfa2c

noctua/devices/atik.py

+173 −17
Original line number Diff line number Diff line
@@ -18,6 +18,7 @@ sudo cp ../include/Atik* /usr/include/ 


# System modules

import ctypes

import threading

import time

import numpy as np

from datetime import datetime
@@ -27,6 +28,7 @@ from astropy.io import fits 


# Custom modules

from .basedevice import BaseDevice

from ..utils.image import make_png

from ..utils.logger import log
@@ -45,6 +47,13 @@ class ArtemisProperties(ctypes.Structure): 




class Camera(BaseDevice):

    """High-level interface for Atik cameras via the Artemis SDK.



    Trigger-mode camera: call ``start()`` to begin an exposure, poll

    ``ready`` until the image is available, then read ``matrix``.

    Use ``looping`` to run continuous acquisitions in a background thread.

    """



    dtype = np.float32

    acquisition = 'trigger'
@@ -60,6 +69,10 @@ class Camera(BaseDevice): 
        self._last_datetime = None

        self._setpoint = None



        self._looping = False

        self._loop_thread = None

        self.loop_exposure = 1.0



        try:

            self._lib = ctypes.CDLL("/usr/lib/libatikcameras.so")

            self._lib.ArtemisConnect.restype = ctypes.c_void_p
@@ -129,11 +142,43 @@ class Camera(BaseDevice): 
    # --- Exposure Methods ---



    def abort(self):

        """Abort the current exposure."""

        h = self._check_connection()

        if h: self._lib.ArtemisAbortExposure(h)



        

    def start(self, duration, imagetype, datetime=None):

    @property

    def looping(self):

        """bool : True if the camera is continuously acquiring frames in a background loop."""

        return self._looping



    @looping.setter

    def looping(self, value):

        value = bool(value)

        if value == self._looping:

            return

        self._looping = value

        if value:

            self._loop_thread = threading.Thread(target=self._run_loop, daemon=True)

            self._loop_thread.start()



    def _run_loop(self):

        while self._looping:

            self._start(self.loop_exposure, 1)

            while self._looping and self.ready != 1:

                time.sleep(0.3)



    def _start(self, duration, imagetype, datetime=None):

        """Trigger an exposure unconditionally (no loop guard).



        Parameters

        ----------

        duration : float

            Exposure time in seconds.

        imagetype : int or str

            Frame type (0/2/'Dark'/'Bias' → dark mode; else light mode).

        datetime : str, optional

            UTC ISO-8601 string stored for later FITS header use.

        """

        h = self._check_connection()

        if not h: return

        if self.state != 0:
@@ -147,8 +192,40 @@ class Camera(BaseDevice): 
        self._lib.ArtemisSetDarkMode(h, is_dark)

        self._lib.ArtemisStartExposure(h, ctypes.c_float(duration))



    def start(self, duration, imagetype, datetime=None):

        """Start a single exposure.



        Refuses with an error if a loop is active; set ``looping = False`` first.



        Parameters

        ----------

        duration : float

            Exposure time in seconds.

        imagetype : int or str

            Frame type (0/2/'Dark'/'Bias' → dark mode; else light mode).

        datetime : str, optional

            UTC ISO-8601 string stored for later FITS header use.

        """

        if self._looping:

            log.error("Camera is looping. Set looping=False first.")

            self.error.append("Camera is looping")

            return

        self._start(duration, imagetype, datetime)



        

    def download(self, filepath=None):

        """Wait for the current exposure to finish and save it as a FITS file.



        Parameters

        ----------

        filepath : str, optional

            Destination path. Defaults to the path from cameras.ini.



        Returns

        -------

        str or None

            Absolute path to the saved FITS file, or None on failure.

        """

        if filepath is None:

            from ..config.constants import viewer_fits_path

            filepath = viewer_fits_path(getattr(self, '_viewer_key', None))
@@ -218,7 +295,10 @@ class Camera(BaseDevice): 


    @property

    def matrix(self):

        """Last acquired image as a float32 numpy array (reads SDK buffer)."""

        """ndarray or None : Last acquired frame as a 2-D float32 array.



        Returns None if no image is ready in the SDK buffer.

        """

        h = self._check_connection()

        if not h or not self._lib.ArtemisImageReady(h):

            return None
@@ -233,38 +313,87 @@ class Camera(BaseDevice): 
        buffer = (ctypes.c_uint16 * size).from_address(buf_ptr)

        return np.frombuffer(buffer, dtype=np.uint16).reshape(h_img.value, w.value).astype(np.float32)



    @property

    def image(self):

        """Current frame as viridis PNG bytes."""

    def image(self, vmin=None, vmax=None, color=True):

        """Return the current frame as PNG-encoded bytes.



        Parameters

        ----------

        vmin : float, optional

            Lower clip value for contrast scaling. Defaults to data minimum.

        vmax : float, optional

            Upper clip value for contrast scaling. Defaults to data maximum.

        color : bool, optional

            If True apply viridis colormap, if False use grayscale. Default True.



        Returns

        -------

        bytes or None

            PNG image bytes, or None if no frame is available.

        """

        data = self.matrix

        if data is None:

            return None

        from ..utils.image import fits_to_png

        return fits_to_png(data)

        return make_png(data, vmin=vmin, vmax=vmax, color=color)



    def save_png(self, filepath=None, vmin=None, vmax=None, color=True):

        """Save current frame as PNG. Returns filepath."""

        """Save the current frame as a PNG file.



        Parameters

        ----------

        filepath : str, optional

            Destination path. Defaults to the PNG path from cameras.ini.

        vmin : float, optional

            Lower clip value for contrast scaling.

        vmax : float, optional

            Upper clip value for contrast scaling.

        color : bool, optional

            If True apply viridis colormap, if False use grayscale. Default True.



        Returns

        -------

        str

            Absolute path to the saved PNG file.

        """

        if filepath is None:

            from ..config.constants import viewer_fits_path

            filepath = viewer_fits_path(getattr(self, "_viewer_key", None)).replace(".fits", ".png")

        data = self.matrix

        if data is not None:

            from ..utils.image import make_png

        png = self.image(vmin=vmin, vmax=vmax, color=color)

        if png is not None:

            from pathlib import Path

            Path(filepath).parent.mkdir(parents=True, exist_ok=True)

            with open(filepath, "wb") as f:

                f.write(make_png(data, vmin=vmin, vmax=vmax, color=color))

                f.write(png)

        return filepath



    # --- Windowing Methods ---



    def set_window(self, start_x, start_y, width, height):

        """Set the imaging sub-frame in unbinned pixel coordinates.



        Parameters

        ----------

        start_x : int

            X origin in unbinned pixels.

        start_y : int

            Y origin in unbinned pixels.

        width : int

            Sub-frame width in unbinned pixels.

        height : int

            Sub-frame height in unbinned pixels.

        """

        h = self._check_connection()

        if h:

            self._lib.ArtemisSubframe(h, start_x, start_y, width, height)

            self._subframe = [start_x, start_y, width, height]



    def full_frame(self):

        """Set the camera to use the full sensor area.



        Returns

        -------

        list of int

            Full sensor [width, height] in unbinned pixels.

        """

        h = self._check_connection()

        nx, ny = self._props.nPixelsX, self._props.nPixelsY

        if h:
@@ -273,6 +402,13 @@ class Camera(BaseDevice): 
        return [nx, ny]



    def half_frame(self):

        """Set the camera to use a centered 50% sub-frame.



        Returns

        -------

        list of int

            Sub-frame [width, height] in unbinned pixels.

        """

        h = self._check_connection()

        nx, ny = self._props.nPixelsX, self._props.nPixelsY

        if h:
@@ -281,6 +417,13 @@ class Camera(BaseDevice): 
        return [nx, ny // 2]



    def small_frame(self):

        """Set the camera to use a fixed 500-row strip (rows 1500-2000).



        Returns

        -------

        list of int

            Sub-frame [width, height] in unbinned pixels.

        """

        h = self._check_connection()

        nx = self._props.nPixelsX

        if h:
@@ -292,11 +435,13 @@ class Camera(BaseDevice): 


    @property

    def state(self):

        """int : Current camera state (0=Idle, 1=Waiting, 2=Exposing, 3=Reading, 5=Error)."""

        h = self._check_connection()

        return self._lib.ArtemisCameraState(h) if h else -1



    @property

    def temperature(self):

        """float or None : CCD temperature in degrees Celsius."""

        h = self._check_connection()

        if not h: return None

        temp = ctypes.c_int()
@@ -309,6 +454,7 @@ class Camera(BaseDevice): 


    @property

    def cooler(self):

        """bool : Cooler active state (True=On, False=Off)."""

        h = self._check_connection()

        if not h: return False

        flags = ctypes.c_int()
@@ -329,6 +475,7 @@ class Camera(BaseDevice): 


    @property

    def binning(self):

        """list of int : Current [X, Y] binning factors."""

        h = self._check_connection()

        bx, by = ctypes.c_int(), ctypes.c_int()

        self._lib.ArtemisGetBin(h, ctypes.byref(bx), ctypes.byref(by))
@@ -341,6 +488,7 @@ class Camera(BaseDevice): 


    @property

    def filter(self):

        """int : Always 0; Atik cameras have no filter wheel."""

        return 0



    @filter.setter
@@ -349,6 +497,7 @@ class Camera(BaseDevice): 


    @property

    def xystart(self):

        """list of int : Sub-frame [X, Y] origin in binned pixels."""

        x, y, w, h = self._subframe

        bx, by = self.binning

        bx = bx or 1
@@ -357,6 +506,7 @@ class Camera(BaseDevice): 


    @property

    def xyend(self):

        """list of int : Sub-frame [X, Y] end corner in binned pixels."""

        x, y, w, h = self._subframe

        bx, by = self.binning

        bx = bx or 1
@@ -365,16 +515,19 @@ class Camera(BaseDevice): 


    @property

    def ready(self):

        """int : 1 if an image is ready in the SDK buffer, 0 otherwise."""

        h = self._check_connection()

        if not h: return 0

        return 1 if self._lib.ArtemisImageReady(h) else 0



    @property

    def setpoint(self):

        """float or None : CCD temperature setpoint in degrees Celsius."""

        return self._setpoint



    @property

    def fan(self):

        """int : Cooler power level as a percentage (0-100)."""

        h = self._check_connection()

        if not h: return None

        flags, level, minl, maxl, setp = [ctypes.c_int() for _ in range(5)]
@@ -385,10 +538,12 @@ class Camera(BaseDevice): 


    @property

    def ambient(self):

        """None : Not available on Atik cameras."""

        return None



    @property

    def center(self):

        """list of int : Sensor centre [X, Y] in binned pixels."""

        bx, by = self.binning

        bx = bx or 1

        by = by or 1
@@ -396,6 +551,7 @@ class Camera(BaseDevice): 


    @property

    def xrange(self):

        """list of int : Sub-frame X extent [start, end] in binned pixels."""

        x, y, w, h = self._subframe

        bx, by = self.binning

        bx = bx or 1
@@ -403,6 +559,7 @@ class Camera(BaseDevice): 


    @property

    def yrange(self):

        """list of int : Sub-frame Y extent [start, end] in binned pixels."""

        x, y, w, h = self._subframe

        bx, by = self.binning

        by = by or 1
@@ -410,18 +567,17 @@ class Camera(BaseDevice): 


    @property

    def is_moving(self):

        """int : Always 0; Atik cameras have no filter wheel."""

        return 0



    @property

    def max_range(self):

        """list of int : Full sensor size [width, height] in unbinned pixels."""

        return [self._props.nPixelsX, self._props.nPixelsY]



    @property

    def all(self):

        """

        Get a dictionary of all relevant camera settings and status.

        Matches the structure of the STX driver.

        """

        """dict : Comprehensive camera state matching the STX driver structure."""

        h = self._check_connection()

        if not h:

            return {}

noctua/devices/mako.py

+122 −97
Original line number Diff line number Diff line
@@ -8,10 +8,11 @@ Utilizes external image and streaming utilities. 


# System modules

import threading

import time

import numpy as np

# Third-party modules

from astropy.io import fits

from vmbpy import VmbSystem, FrameStatus

from vmbpy import VmbSystem



# Custom modules

from .basedevice import BaseDevice
@@ -34,7 +35,6 @@ class Mako(BaseDevice): 
        self.id = url

        self.vmb = VmbSystem.get_instance()

        self._cam = None

        self._streaming = False



    def _check_connection(self):

        """
@@ -75,21 +75,21 @@ class Mako(BaseDevice): 
        """Clean shutdown of camera and Vimba system."""



        if self._cam:

            if self._streaming:

                self._cam.stop_streaming()

            self._cam.__exit__(None, None, None)

        if self.vmb:

            self.vmb.__exit__(None, None, None)





class Guider(Mako):

    """

    High-level interface for Mako cameras with streaming and

    capture support.

    """High-level interface for Mako cameras with frame capture and loop support.



    Single-capture mode: call ``start()`` to grab one frame via ``get_frame()``.

    Use ``looping`` to run continuous acquisitions in a background thread.

    The latest frame is always available via ``matrix``.

    """



    dtype = np.uint8        # raw sensor data is 8-bit grayscale

    acquisition = 'stream'  # grab-and-sleep, no exposure trigger needed

    acquisition = 'capture'



    def __init__(self, url):

        """
@@ -99,93 +99,131 @@ class Guider(Mako): 
        super().__init__(url)

        self._last_frame = None

        self._lock = threading.Lock()

        self._range = [0, 255]

        self._streamer = None

        self._looping = False

        self._loop_thread = None





    # def _frame_handler(self, cam, stream, frame):

    #     """Callback to handle frames during streaming."""

        

    #     if frame.get_status() == FrameStatus.Complete:

    #         with self._lock:

    #             self._last_frame = frame.as_numpy_ndarray().copy()

    #     cam.queue_frame(frame)

    def _frame_handler(self, cam, stream, frame):

        """Callback to handle frames during streaming."""

        try:

            status = frame.get_status()

        except ValueError:

            # Frame incompleto (-4 VmbErrorIncomplete) — pacchetti persi in rete

            cam.queue_frame(frame)

    @property

    def looping(self):

        """bool : True if the camera is continuously acquiring frames in a background loop."""

        return self._looping



    @looping.setter

    def looping(self, value):

        value = bool(value)

        if value == self._looping:

            return

        self._looping = value

        if value:

            self._loop_thread = threading.Thread(target=self._run_loop, daemon=True)

            self._loop_thread.start()



        if status == FrameStatus.Complete:

    def _run_loop(self):

        cam = self._check_connection()

        while self._looping:

            try:

                raw = cam.get_frame().as_numpy_ndarray()

                if raw is not None:

                    with self._lock:

                self._last_frame = frame.as_numpy_ndarray().copy()

        cam.queue_frame(frame)

            

    @property

    def streaming(self):

        """Get the current streaming status."""

        

        return self._streaming

    

                        self._last_frame = raw.copy()

            except Exception as e:

                log.error(f"Mako loop error: {e}")

                time.sleep(1.0)



    @streaming.setter

    def streaming(self, b):

        """Start or stop camera streaming."""

    def start(self):

        """Capture a single frame via a blocking get_frame() call.



        if b == self._streaming:

        Refuses with an error if a loop is active; set ``looping = False`` first.

        The result is stored in the internal buffer and accessible via ``matrix``.

        """

        if self._looping:

            log.error("Camera is looping. Set looping=False first.")

            return

        cam = self._check_connection()

        if b:

            cam.start_streaming(handler=self._frame_handler, buffer_count=5)

            self._streaming = True

        else:

            cam.stop_streaming()

            self._streaming = False



        raw = cam.get_frame().as_numpy_ndarray()

        if raw is not None:

            with self._lock:

                self._last_frame = raw.copy()



    @property

    def matrix(self):

        """

        Get the latest image as a 2-D uint8 numpy array ``(H, W)``.

        Works during streaming or via single capture.

        Updated by start() or by the loop.

        """

        if self._streaming:

        with self._lock:

            raw = self._last_frame

        else:

            cam = self._check_connection()

            raw = cam.get_frame().as_numpy_ndarray()

        if raw is None:

            return None

        return np.squeeze(raw)



    @property

    def image(self):

        """Current frame as viridis PNG bytes."""

    def image(self, vmin=None, vmax=None, color=True):

        """Return the current frame as PNG-encoded bytes.



        Parameters

        ----------

        vmin : float, optional

            Lower clip value for contrast scaling. Defaults to data minimum.

        vmax : float, optional

            Upper clip value for contrast scaling. Defaults to data maximum.

        color : bool, optional

            If True apply viridis colormap, if False use grayscale. Default True.



        Returns

        -------

        bytes or None

            PNG image bytes, or None if no frame is available.

        """

        data = self.matrix

        if data is None:

            return None

        return make_png(data)

        return make_png(data, vmin=vmin, vmax=vmax, color=color)



    def save_png(self, filepath=None, vmin=None, vmax=None, color=True):

        """Save current frame as PNG. Returns filepath."""

        """Save the current frame as a PNG file.



        Parameters

        ----------

        filepath : str, optional

            Destination path. Defaults to the PNG path from cameras.ini.

        vmin : float, optional

            Lower clip value for contrast scaling.

        vmax : float, optional

            Upper clip value for contrast scaling.

        color : bool, optional

            If True apply viridis colormap, if False use grayscale. Default True.



        Returns

        -------

        str

            Absolute path to the saved PNG file.

        """

        if filepath is None:

            from ..config.constants import viewer_fits_path

            filepath = viewer_fits_path(getattr(self, "_viewer_key", None)).replace(".fits", ".png")

        data = self.matrix

        if data is not None:

        png = self.image(vmin=vmin, vmax=vmax, color=color)

        if png is not None:

            from pathlib import Path

            Path(filepath).parent.mkdir(parents=True, exist_ok=True)

            with open(filepath, "wb") as f:

                f.write(make_png(data, vmin=vmin, vmax=vmax, color=color))

                f.write(png)

        return filepath



    

    def download(self, filename=None):

        """Save the current frame as a FITS file."""

        """Save the current frame as a FITS file.



        Parameters

        ----------

        filename : str, optional

            Destination path. Defaults to the path from cameras.ini.



        Returns

        -------

        str

            Absolute path to the saved FITS file.

        """

        if filename is None:

            from ..config.constants import viewer_fits_path

            filename = viewer_fits_path(getattr(self, '_viewer_key', None))
@@ -198,29 +236,41 @@ class Guider(Mako): 
    

    @property

    def autoexpose(self):

        """Get exposure auto status and options as a tuple."""

        

        # Accessing the feature object directly to use as_tuple()

        """tuple : ExposureAuto feature current value and allowed options."""

        cam = self._check_connection()

        return cam.ExposureAuto.as_tuple()



    

    @autoexpose.setter

    def autoexpose(self, value):

        """Set exposure auto status ('Continuous', 'Off', 'Once')."""

        """Set ExposureAuto ('Continuous', 'Off', or 'Once')."""

        

        self.put("ExposureAuto", value)



        

    def stream(self, active=True, host='0.0.0.0', port=5534, fps=2):

        """Manage the HTTP stream using the Streamer utility."""

        """Start or stop the HTTP MJPEG frame server.



        Serves ``matrix`` frames while ``looping`` is active.



        Parameters

        ----------

        active : bool, optional

            True to start, False to stop. Default True.

        host : str, optional

            Bind address. Default '0.0.0.0'.

        port : int, optional

            TCP port. Default 5534.

        fps : int, optional

            Target frame rate. Default 2.

        """

        if active:

            if not self._streamer:

                self._streamer = Streamer(

                    host=host,

                    port=port,

                    image_provider=lambda: self.matrix,

                    status_provider=lambda: self.streaming,

                    status_provider=lambda: self.looping,

                    fps=fps

                )

                self._streamer.start()
@@ -233,28 +283,3 @@ class Guider(Mako): 
                self._streamer = None



            
    def stream_opencv(self):

        """

        Open an OpenCV window in a separate thread for local monitoring.

        """

    

        import cv2



        if not self._streaming:

            print("Error: Streaming is False. Start it first.")

            return



        def loop():

            win_name = f"Mako Stream: {self.id}"

            while self._streaming:

                if self._last_frame is not None:

                    fmax = self._last_frame.max()

                    disp = (self._last_frame / (fmax if fmax > 0 else 1) * 255).astype(np.uint8)

                    cv2.imshow(win_name, disp)

                

                if cv2.waitKey(1) & 0xFF == ord('q'):

                    break

            cv2.destroyAllWindows()



        threading.Thread(target=loop, daemon=True).start()

        print("OpenCV Window started. Press 'q' in the window to close.")

noctua/devices/stx.py

+102 −21

File changed.

Preview size limit exceeded, changes collapsed.