import seabreeze.backends # get the backend and add some functions/classes to this module lib = seabreeze.backends.get_backend() list_devices = lib.device_list_devices SeaBreezeError = lib.SeaBreezeError SeaBreezeDevice = lib.SeaBreezeDevice import numpy class _HelperFeatureAdder(object): def __init__(self, other): self._other = other def add(self, feature): ffunc = getattr(lib, "device_get_%s_feature_id" % feature) fids = ffunc(self._other._dev) if fids: return fids[0] else: return -1 # It seems as if negative numbers are not used for featureIDs class LightSource(object): def __init__(self, device, featureId, index): ident = device, featureId, index self._ident = ident self._has_enable = lib.light_source_has_enable(*ident) self._has_varint = lib.light_source_has_variable_intensity(*ident) def set_enable(self, enable): if self._has_enable: d, f, i = self._ident lib.light_source_set_enable(d, f, i, enable) else: raise SeaBreezeError("Light source #%d can't be enabled or disabled.") def set_intensity(self, intensity): if self._has_varint: d, f, i = self._ident lib.light_source_set_intensity(d, f, i, intensity) else: raise SeaBreezeError("Light source #%d intensity can't be set.") def get_intensity(self): try: return lib.light_source_get_intensity(*self._ident) except SeaBreezeError: raise def __repr__(self): d, f, i = self._ident return "" % (i, d.model, d.serial) class Spectrometer(object): def __init__(self, device): self._open_device(device) @classmethod def from_serial_number(cls, serial=None): if serial is None: # pick first spectrometer for dev in lib.device_list_devices(): if not lib.device_is_open(dev): return cls(dev) else: raise SeaBreezeError("No unopened device found.") else: # pick spectrometer with correct serial for dev in lib.device_list_devices(): if dev.serial == str(serial): if lib.device_is_open(dev): raise SeaBreezeError("Device already opened.") else: return cls(dev) else: raise SeaBreezeError("No device attached with serial number '%s'." % serial) def _open_device(self, device): if not isinstance(device, SeaBreezeDevice): raise SeaBreezeError("Not a SeaBreezeDevice") if lib.device_is_open(device): raise SeaBreezeError("Device already opened.") if hasattr(self, '_dev') and lib.device_is_open(getattr(self, '_dev')): lib.device_close(getattr(self, '_dev')) self._dev = device lib.device_open(self._dev) # get default information self._serial = self._dev.serial self._model = self._dev.model # get features feature = _HelperFeatureAdder(self) self._fidsp = feature.add('spectrometer') self._fidsh = feature.add('shutter') self._fidls = feature.add('light_source') self._fidcs = feature.add('continuous_strobe') self._fidee = feature.add('eeprom') self._fidic = feature.add('irrad_calibration') self._fidla = feature.add('lamp') self._fidte = feature.add('tec') self._fidnc = feature.add('nonlinearity_coeffs') # Added self._fidsl = feature.add('stray_light_coeffs') self._fidspp = feature.add('spectrum_processing') # Not implemented in pyseabreeze # get additional information self._pixels = lib.spectrometer_get_formatted_spectrum_length(self._dev, self._fidsp) self._minimum_integration_time_micros = ( lib.spectrometer_get_minimum_integration_time_micros(self._dev, self._fidsp)) # get wavelengths self._wavelengths = numpy.zeros((self._pixels,), dtype=numpy.double) transfered_N = 0 while True: transfered_N += lib.spectrometer_get_wavelengths(self._dev, self._fidsp, self._wavelengths[transfered_N:]) if transfered_N >= self._pixels: break # get dark pixel indices self._dark = lib.spectrometer_get_electric_dark_pixel_indices(self._dev, self._fidsp) self._has_dark_pixels = True if len(self._dark) > 0 else False # get nonlinearity coefficients try: sbnc = lib.nonlinearity_coeffs_get(self._dev, self._fidnc) self._nc = numpy.poly1d(sbnc[::-1]) self._has_nonlinearity_coeffs = True except SeaBreezeError: self._has_nonlinearity_coeffs = False # if lightsources try: N_light_sources = lib.light_source_get_count(self._dev, self._fidls) self._light_sources = tuple(LightSource(self._dev, self._fidls, i) for i in range(N_light_sources)) except SeaBreezeError: self._light_sources = tuple() def wavelengths(self): return self._wavelengths def intensities(self, correct_dark_counts=False, correct_nonlinearity=False): if correct_dark_counts and not self._has_dark_pixels: raise SeaBreezeError("This device does not support dark count correction.") if correct_nonlinearity and not self._has_nonlinearity_coeffs: raise SeaBreezeError("This device does not support nonlinearity correction.") # Get the intensities out = numpy.empty((self._pixels,), dtype=numpy.double) transfered_N = 0 while True: transfered_N += lib.spectrometer_get_formatted_spectrum(self._dev, self._fidsp, out[transfered_N:]) if transfered_N >= self._pixels: break # Do corrections if requested if (correct_nonlinearity or correct_dark_counts): dark_offset = numpy.mean(out[self._dark]) if self._has_dark_pixels else 0. out -= dark_offset if (correct_nonlinearity): out = out / numpy.polyval(self._nc, out) if correct_nonlinearity and (not correct_dark_counts): out += dark_offset return out def spectrum(self, correct_dark_counts=False, correct_nonlinearity=False): return numpy.vstack((self._wavelengths, self.intensities(correct_dark_counts, correct_nonlinearity))) def integration_time_micros(self, integration_time_micros): # Protect against a bug in libseabreeze: # If integration time is out of bounds, libseabreeze returns Undefined Error # (Probably only for devices with a non micro second time base...) try: lib.spectrometer_set_integration_time_micros(self._dev, self._fidsp, integration_time_micros) except SeaBreezeError as e: if getattr(e, 'error_code', None) == 1: # Only replace if 'Undefined Error' raise SeaBreezeError("FIX: Specified integration time is out of range.") else: raise e def trigger_mode(self, mode): lib.spectrometer_set_trigger_mode(self._dev, self._fidsp, mode) def boxcar_width(self, boxcar_width): lib.spectrum_processing_set_boxcar_width(self._dev, self._fidspp, boxcar_width) # Not implemented in pyseabreeze def scans_to_average(self, scans_to_average): lib.spectrum_processing_set_scans_to_average(self._dev, self._fidspp, scans_to_average) # Not implemented in pyseabreeze def get_boxcar_width(self): lib.spectrum_processing_get_boxcar_width(self._dev, self._fidspp) # Not implemented in pyseabreeze def get_scans_to_average(self): lib.spectrum_processing_get_scans_to_average(self._dev, self._fidspp) # Not implemented in pyseabreeze @property def serial_number(self): return self._serial @property def model(self): return self._model @property def pixels(self): return self._pixels @property def minimum_integration_time_micros(self): return self._minimum_integration_time_micros @property def light_sources(self): return self._light_sources def eeprom_read_slot(self, slot): return lib.eeprom_read_slot(self._dev, self._fidee, slot) def tec_set_enable(self, enable): lib.tec_set_enable(self._dev, self._fidte, enable) def tec_set_temperature_C(self, set_point_C): lib.tec_set_temperature_setpoint_degrees_C(self._dev, self._fidte, set_point_C) def tec_get_temperature_C(self): return lib.tec_read_temperature_degrees_C(self._dev, self._fidte) def lamp_set_enable(self, enable): lib.lamp_set_lamp_enable(self._dev, self._fidla, enable) def shutter_set_open(self, state): lib.shutter_set_shutter_open(self._dev, self._fidsh, state) def stray_light_coeffs(self): return lib.stray_light_coeffs_get(self._dev, self._fidsl) def irrad_calibration(self): out = numpy.empty((self._pixels,), dtype=numpy.float32) lib.irrad_calibration_read(self._dev, self._fidic, out) return out def irrad_calibration_collection_area(self): if lib.irrad_calibration_has_collection_area(self._dev, self._fidic): return lib.irrad_calibration_read_collection_area(self._dev, self._fidic) else: raise SeaBreezeError("Device does not store irrad calibration area.") def continuous_strobe_set_enable(self, enable): lib.continuous_strobe_set_enable(self._dev, self._fidcs, enable) def continuous_strobe_set_period_micros(self, period_micros): lib.continuous_strobe_set_period_micros(self._dev, self._fidcs, period_micros) def close(self): lib.device_close(self._dev) def __enter__(self): return self def __exit__(self, type, value, traceback): self.close() def __repr__(self): return "" % (self.model, self.serial_number)