Adapting integration time to measured intensity

[jtmbeta]

Ocean Optics recommend setting the spectrometer integration time so that the max reported intensity values are within 80-90% of the intensity range. I've got a simple algorithm for this, but its not very efficient and sometimes gets stuck. Is an adaptive measurement algorithm a feature that might be considered for seabreeze? Does anyone have a better algorithm for adapting integration time to the measured intensity values?

from time import sleep

from seabreeze.spectrometers import Spectrometer

class OceanOptics(Spectrometer):

    def __init__(self, device):
        super().__init__(device)
        
    def get_temperatures(self):
        """Get temperatures of the printed circuit board and microcontroller.

        """
        
        if not self.features['temperature'] == []:
            temps = self.f.temperature.temperature_get_all()
            sleep(.01)
            board_temp = temps[0]
            micro_temp = temps[2]
        else:
            board_temp = 'NA'
            micro_temp = 'NA'
            
        return (board_temp, micro_temp)
            
    # User defined methods
    def measurement(self,
                    integration_time: int = None,
                    correct_dark_counts: bool = True
                    setting: dict = {}) -> Tuple[np.array, dict]:
        """Obtain a measurement with an Ocean Optics spectrometer.

        If `integration_time` is not specified, will use an adaptive procedure
        that avoids saturation by aiming for a maximum reported value of
        80-90% of the maximum intensity value for the device. Can take up to a
        maximum of ~3.5 mins for lower light levels, though this could be
        reduced somewhat by optimising the algorithm.

        Parameters
        ----------
        integration_time : int
            The integration time to use for the measurement. Leave as None to
            adaptively set the integration time based on spectral measurements.
        correct_dark_counts : bool, optional
            Pass True to remove the noise floor in measurements
        setting : dict, optional
             Current setting of the light source (if known), to be included in
             the `info`. For example ``{'led' : 5, 'intensity' : 3000}``, or
             ``{'intensities' : [0, 0, 0, 300, 4000, 200, 0, 0, 0, 0]}``.
             The default is ``{}``.

        Returns
        -------
        counts : np.array
            Raw intensity counts from the Ocean Optics spectrometer.
        info : dict
            Companion info for measurement.

        """
        if integration_time:
            # Set the spectrometer integration time
            self.integration_time_micros(int(integration_time))
            sleep(.01)

            # Obtain temperature measurements
            board_temp, micro_temp = self.get_temperatures()

            # Obtain intensity measurements
            counts = self.intensities(correct_dark_counts=True)

            # Get the maximum reported value
            max_reported = max(counts)
            print(f'\tIntegration time: {int(integration_time)} micros')
            print(f'\tMaximum value: {max_reported}')
            intgt = integration_time

        else:
            # Initial parameters
            maximum_intensity = self.max_intensity
            lower_intgt = None
            upper_intgt = None
            lower_bound = maximum_intensity * .7
            upper_bound = maximum_intensity * .9

            # Start with the minimum integration time available
            intgt = min(self.integration_time_micros_limits)
            intgt_hist = []
            max_reported = 0

            # Keep sampling with different integration times until the maximum
            # reported value is within 80-90% of the maximum intensity value
            # for the device
            while max_reported < lower_bound or max_reported > upper_bound:
                
                intgt_hist.append(intgt)

                # If current integration time is greater than the upper limit,
                # set it too the upper limit
                if intgt >= self.integration_time_micros_limits[1]:
                    intgt = self.integration_time_micros_limits[1]

                # Set the spectrometer integration time
                self.integration_time_micros(int(intgt))
                sleep(.01)

                # Obtain temperature measurements if available
                board_temp, micro_temp = self.get_temperatures()

                # Obtain intensity measurements
                counts = self.intensities(correct_dark_counts=True)
                sleep(.01)

                # Get the maximum reported value
                max_reported = max(counts)
                print(f'\tIntegration time: {int(intgt)} micros')
                print(f'\tMaximum value: {max_reported}')

                # If the integration time has reached the upper limit for the
                # spectrometer, exit the while loop, having obtained the final
                # measurement
                if intgt == self.integration_time_micros_limits[1]:
                    break

                # If the max_reported value is less than the lower_bound and
                # the upper_ingt is not yet known, update the lower_intgt and
                # double intgt ready for the next iteration
                elif max_reported < lower_bound and upper_intgt is None:
                    lower_intgt = intgt
                    intgt *= 2.0

                # If the max_reported value is greater than the upper_bound,
                # update the upper_intgt and subtract half of the difference
                # between upper_intgt and lower_intgt from intgt ready for the
                # next iteration
                elif max_reported > upper_bound:
                    upper_intgt = intgt
                    intgt -= (upper_intgt - lower_intgt) / 2

                # If the max_reported value is less than the lower_bound and
                # the value of upper_intgt is known, update the lower_intgt and
                # add half of the difference between upper_intgt and
                # lower_intgt to intgt ready for the next iteration
                elif max_reported < lower_bound and upper_intgt is not None:
                    lower_intgt = intgt
                    intgt += (upper_intgt - lower_intgt) / 2
                
                # TODO: fix
                # Sometimes the algo gets stuck so we need to do something else
                if len(intgt_hist) > 10:
                    if all(val == intgt_hist[-1] for val in intgt_hist[-10:]):
                        intgt = min(self.integration_time_micros_limits)
                        intgt_hist = []
                        max_reported = 0
                        
        info = {
            'board_temp': board_temp,
            'micro_temp': micro_temp,
            'integration_time': int(intgt),
            'model': self.model
        }
        info = {**info, **setting}

        return (counts, info)

[IvanPrzOl]

Hi,
I'm also trying to implement a function for setting automatically the integration time (or optimization as is known in other software), at least for my application, user need to do it from time to time during data collection,

My first approach is to get the maximum count value (after non-linear correction) using a default time (let's say 1 or 10 ms), if detector got saturated, integration time should be reduced to half to get a new spectra,
Then, as max value could be over or under the 50 000 counts (but not saturated) and assuming a linear response, the optimal (or proximate optimal time) time could be solved using a simple fractional formula, but regarding on the non-linear behavior I don't know how to deal with it. May be this integration time could be used as a "seed" to converge over the best optimal integration time using some sort of iterative algorithm.

What do you think?