stl.py: get()/put() al posto di _sbig() diretto, ruota filtri CFW-L

"""

Interface with a SBIG STL-11000M camera via the SBIG Universal Driver

(SBIGUDrv), used directly over ctypes (no filter wheel — the échelle

station's filter selector is decorative, see cameras.ini/api.ini).

(SBIGUDrv), used directly over ctypes. The camera has an internal

5-position CFW-L filter wheel, driven with the CC_CFW command

(cfwModel=CFWSEL_CFWL) — see the ``filter``/``is_moving`` properties.

The driver library ships under different package/file names depending on

the machine — both are the same upstream SBIG driver (identical exported

CC_OPEN_DEVICE                  = 27

CC_CLOSE_DEVICE                  = 28

CC_START_READOUT                 = 35

CC_CFW                            = 43

CC_START_EXPOSURE2                = 50

# SBIG_DEVICE_TYPE

# CAMERA_TYPE

NO_CAMERA = 0xFFFF

# CFW_MODEL_SELECT — the STL-11000M's internal wheel is a CFW-L.

CFWSEL_CFWL = 4

# CFW_COMMAND

CFWC_QUERY = 0

CFWC_GOTO  = 1

# CFW_STATUS

CFWS_IDLE = 1

CFWS_BUSY = 2

# ---------------------------------------------------------------------------

# ctypes structs — field order/types verbatim from sbigudrv.h, all under

    # request = TEMP_STATUS_ADVANCED2 = 2 — temperatures already in °C.

class CFWParams(_Packed):

    _fields_ = [

        ("cfwModel",   ctypes.c_ushort),

        ("cfwCommand", ctypes.c_ushort),

        ("cfwParam1",  ctypes.c_ulong),

        ("cfwParam2",  ctypes.c_ulong),

        ("outLength",  ctypes.c_ushort),

        ("outPtr",     ctypes.c_void_p),

        ("inLength",   ctypes.c_ushort),

        ("inPtr",      ctypes.c_void_p),

    ]

class CFWResults(_Packed):

    _fields_ = [

        ("cfwModel",    ctypes.c_ushort),

        ("cfwPosition", ctypes.c_ushort),

        ("cfwStatus",   ctypes.c_ushort),

        ("cfwError",    ctypes.c_ushort),

        ("cfwResult1",  ctypes.c_ulong),

        ("cfwResult2",  ctypes.c_ulong),

    ]

class MiscellaneousControlParams(_Packed):

    _fields_ = [

        ("fanEnable",      ctypes.c_ushort),

        self._camera_type = NO_CAMERA

    def _sbig(self, command, params=None, results=None):

        """Call SBIGUnivDrvCommand; returns the PAR_ERROR code (0 = OK)."""

        """Raw call to SBIGUnivDrvCommand; returns the PAR_ERROR code (0 = OK)."""

        if self._lib is None:

            return -1

        p = ctypes.byref(params) if params is not None else None

        r = ctypes.byref(results) if results is not None else None

        return self._lib.SBIGUnivDrvCommand(command, p, r)

    # --- Get/Put for compatibility with other devices ---

    #

    # SBIGUnivDrvCommand is a single dispatch function (PAR_COMMAND code +

    # params/results structs), unlike STX's named CGI methods — there's no

    # string key to route on. get()/put() instead split by command

    # semantics: get() for commands that read camera state, put() for

    # commands that change it or trigger an action. Both just forward to

    # _sbig(); the split exists so callers below express intent.

    def get(self, command, params=None, results=None):

        """Issue a PAR_COMMAND that reads state; returns the PAR_ERROR code (0 = OK)."""

        return self._sbig(command, params, results)

    def put(self, command, params=None, results=None):

        """Issue a PAR_COMMAND that changes state/triggers an action; returns the PAR_ERROR code (0 = OK)."""

        return self._sbig(command, params, results)

    def _check_connection(self):

        """Open the driver/device and establish a link, once."""

        if self._opened or self._lib is None:

            return self._opened

        err = self._sbig(CC_OPEN_DRIVER)

        err = self.put(CC_OPEN_DRIVER)

        if err != 0:

            msg = f"STL: Open Driver failed (error {err})"

            if msg not in self.error:

            return False

        dev_params = OpenDeviceParams(deviceType=DEV_USB, lptBaseAddress=0, ipAddress=0)

        err = self._sbig(CC_OPEN_DEVICE, dev_params)

        err = self.put(CC_OPEN_DEVICE, dev_params)

        if err != 0:

            msg = f"STL: Open Device failed (error {err}) — camera not found on USB?"

            if msg not in self.error:

            return False

        link_results = EstablishLinkResults()

        err = self._sbig(CC_ESTABLISH_LINK, EstablishLinkParams(sbigUseOnly=0), link_results)

        err = self.put(CC_ESTABLISH_LINK, EstablishLinkParams(sbigUseOnly=0), link_results)

        if err != 0 or link_results.cameraType == NO_CAMERA:

            msg = f"STL: Establish Link failed (error {err})"

            if msg not in self.error:

        self.error = []

        return True

    # --- Generic Get/Put for compatibility with other devices ---

    def get(self, key):

        return getattr(self, key)

    def put(self, key, value):

        setattr(self, key, value)

class Camera(STL):

    """High-level interface for the SBIG STL-11000M.

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

    ``ready``/``state`` until it's done, then read ``matrix``.

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

    No filter wheel — ``filter``/``is_moving`` are stubs, like Atik.

    Has an internal 5-position CFW-L filter wheel — see ``filter``/``is_moving``.

    """

    dtype = np.float32

        if not self._check_connection():

            return None

        info = GetCCDInfoResults0()

        err = self._sbig(CC_GET_CCD_INFO, GetCCDInfoParams(request=0), info)

        err = self.get(CC_GET_CCD_INFO, GetCCDInfoParams(request=0), info)

        if err != 0:

            msg = f"STL: Get CCD Info failed (error {err})"

            self.error.append(msg)

        if not self._opened:

            return 5

        results = QueryCommandStatusResults()

        err = self._sbig(CC_QUERY_COMMAND_STATUS,

        err = self.get(CC_QUERY_COMMAND_STATUS,

                        QueryCommandStatusParams(command=CC_START_EXPOSURE2), results)

        if err != 0:

            return 5

    def ready(self):

        """int : 1 if the last exposure has completed and is ready to read out."""

        results = QueryCommandStatusResults()

        err = self._sbig(CC_QUERY_COMMAND_STATUS,

        err = self.get(CC_QUERY_COMMAND_STATUS,

                        QueryCommandStatusParams(command=CC_START_EXPOSURE2), results)

        if err != 0:

            return 0

    def abort(self):

        """Abort the current exposure."""

        self._sbig(CC_END_EXPOSURE, EndExposureParams(ccd=CCD_IMAGING))

        self.put(CC_END_EXPOSURE, EndExposureParams(ccd=CCD_IMAGING))

    # --- Looping ---

            readoutMode=_READOUT_MODE.get(b, 0),

            top=0, left=0, height=height, width=width,

        )

        err = self._sbig(CC_START_EXPOSURE2, params)

        err = self.put(CC_START_EXPOSURE2, params)

        if err != 0:

            msg = f"STL: Start Exposure failed (error {err})"

            self.error.append(msg)

        b = self._binning

        mode = _READOUT_MODE.get(b, 0)

        self._sbig(CC_END_EXPOSURE, EndExposureParams(ccd=CCD_IMAGING))

        self._sbig(CC_START_READOUT, StartReadoutParams(

        self.put(CC_END_EXPOSURE, EndExposureParams(ccd=CCD_IMAGING))

        self.put(CC_START_READOUT, StartReadoutParams(

            ccd=CCD_IMAGING, readoutMode=mode, top=0, left=0, height=height, width=width))

        data = np.empty((height, width), dtype=np.uint16)

            ccd=CCD_IMAGING, readoutMode=mode, pixelStart=0, pixelLength=width)

        row_buf = (ctypes.c_ushort * width)()

        for row in range(height):

            err = self._sbig(CC_READOUT_LINE, line_params, row_buf)

            err = self.get(CC_READOUT_LINE, line_params, row_buf)

            if err != 0:

                msg = f"STL: Readout Line failed at row {row} (error {err})"

                self.error.append(msg)

                break

            data[row, :] = np.ctypeslib.as_array(row_buf)

        self._sbig(CC_END_READOUT, EndReadoutParams(ccd=CCD_IMAGING))

        self.put(CC_END_READOUT, EndReadoutParams(ccd=CCD_IMAGING))

        return data.astype(np.float32)

    @property

    @property

    def filter(self):

        """int : Always 0; the STL-11000M has no filter wheel here."""

        return 0

        """int : Current filter wheel position (1-5).

        The STL-11000M has an internal 5-position CFW-L filter wheel,

        driven through CC_CFW (cfwModel=CFWSEL_CFWL).

        """

        if not self._check_connection():

            return None

        results = CFWResults()

        err = self.get(CC_CFW, CFWParams(cfwModel=CFWSEL_CFWL, cfwCommand=CFWC_QUERY), results)

        if err != 0:

            return None

        return results.cfwPosition

    @filter.setter

    def filter(self, n):

        pass

        if self.is_moving != 0:

            log.error(f"STL: cannot change filter, filter wheel is moving. State: {self.is_moving}")

            self.error.append("Filter wheel busy")

            return

        self.put(CC_CFW, CFWParams(cfwModel=CFWSEL_CFWL, cfwCommand=CFWC_GOTO, cfwParam1=int(n)))

    @property

    def is_moving(self):

        """int : Always 0; no filter wheel."""

        """int : Filter wheel state (0=Idle, 1=Moving)."""

        if not self._check_connection():

            return 0

        results = CFWResults()

        err = self.get(CC_CFW, CFWParams(cfwModel=CFWSEL_CFWL, cfwCommand=CFWC_QUERY), results)

        if err != 0:

            return 0

        return 1 if results.cfwStatus == CFWS_BUSY else 0

    @property

    def cooler(self):

        if not self._check_connection():

            return False

        results = QueryTemperatureStatusResults2()

        err = self._sbig(CC_QUERY_TEMPERATURE_STATUS,

        err = self.get(CC_QUERY_TEMPERATURE_STATUS,

                        QueryTemperatureStatusParams(request=2), results)

        if err != 0:

            return False

            regulation=REGULATION_ON if b else REGULATION_OFF,

            ccdSetpoint=setpoint,

        )

        self._sbig(CC_SET_TEMPERATURE_REGULATION2, params)

        self.put(CC_SET_TEMPERATURE_REGULATION2, params)

    @property

    def temperature(self):

        if not self._check_connection():

            return None

        results = QueryTemperatureStatusResults2()

        err = self._sbig(CC_QUERY_TEMPERATURE_STATUS,

        err = self.get(CC_QUERY_TEMPERATURE_STATUS,

                        QueryTemperatureStatusParams(request=2), results)

        if err != 0:

            return None

        if not self._check_connection():

            return None

        results = QueryTemperatureStatusResults2()

        err = self._sbig(CC_QUERY_TEMPERATURE_STATUS,

        err = self.get(CC_QUERY_TEMPERATURE_STATUS,

                        QueryTemperatureStatusParams(request=2), results)

        if err != 0:

            return None

        if not self._check_connection():

            return None

        results = QueryTemperatureStatusResults2()

        err = self._sbig(CC_QUERY_TEMPERATURE_STATUS,

        err = self.get(CC_QUERY_TEMPERATURE_STATUS,

                        QueryTemperatureStatusParams(request=2), results)

        if err != 0:

            return None

[project]

name = "noctua"

version = "0.8.5"

version = "0.8.6"

authors = [

    { name="Davide Ricci", email="davide.ricci@inaf.it" },

]

# sudo apt install libreadline-dev 

# sudo apt install libffi-dev 

# sudo apt install libbz2-dev

# sudo apt install libsbigudrv-dev # or libsbig-dev

# 

readme = "README.md"