Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _ensure_valid method in autotest

Best Python code snippet using autotest_python

zwo_camera.py

Source:zwo_camera.py

1from python_zwoasi import zwoasi2from PyQt5.QtCore import QThread, QMutex, QMutexLocker, pyqtSignal, pyqtSlot, qDebug, QObject, QTimer, QSettings3import numpy as np4from camera import Camera5from camera_settings_widget import CameraSettingsWidget6def clamp(x, minn, maxx):7    return min(max(x, minn), maxx)8class ZwoCamera(Camera):9    class CaptureThread(QThread):10        ndarray_available = pyqtSignal(np.ndarray)11        def __init__(self, camera: zwoasi.Camera):12            super().__init__()13            self._camera = camera14            self._mutex = QMutex()15            self._abort = False16        def stop(self):17            with QMutexLocker(self._mutex):18                self._abort = True19        def run(self):20            self._abort = False21            self._camera.start_video_capture()22            while True:23                with QMutexLocker(self._mutex):24                    if self._abort:25                        self._camera.stop_video_capture()26                        break27                data = self._camera.get_video_data()28                whbi = self._camera.get_roi_format()29                shape = [whbi[1], whbi[0]]30                if whbi[3] == zwoasi.ASI_IMG_RAW8 or whbi[3] == zwoasi.ASI_IMG_Y8:31                    img = np.frombuffer(data, dtype=np.uint8)32                elif whbi[3] == zwoasi.ASI_IMG_RAW16:33                    img = np.frombuffer(data, dtype=np.uint16)34                elif whbi[3] == zwoasi.ASI_IMG_RGB24:35                    img = np.frombuffer(data, dtype=np.uint8)36                    shape.append(3)37                else:38                    raise ValueError('Unsupported image type')39                img = img.reshape(shape)40                img = np.rot90(img, 2)41                self.ndarray_available.emit(img)42    class AutoSettingsObject(QObject):43        gain_changed = pyqtSignal(float)44        exposure_time_changed = pyqtSignal(float)45        def __init__(self, **kwargs):46            super().__init__(**kwargs)47            self._saturation = 048            self._enabled = False49            self._piUi = 050            self._last_error = 051            self._gain = None52            self._exposure_time = None53            self._min_gain = None54            self._max_gain = None55            self._min_exposure_time = None56            self._max_exposure_time = None57            self._setpoint = 85.58            self.Kp = 0.559            self.Kd = 1060            self.Ki = 0.361            self._interval = 20062            self._timer = QTimer()63            self._timer.timeout.connect(self._run)64        @pyqtSlot(int)65        def set_saturation(self, sat):66            self._saturation = sat67        @pyqtSlot(float)68        def set_exposure_time(self, exposure):69            self._exposure_time = exposure70        @pyqtSlot(float)71        def set_gain(self, gain):72            self._gain = gain73        def set_exposure_range(self, rng):74            self._min_exposure_time, self._max_exposure_time = rng75        def set_gain_range(self, rng):76            self._min_gain, self._max_gain = rng77        def start(self):78            self._timer.start(self._interval)79        def stop(self):80            self._timer.stop()81        def _run(self):82            error = self._saturation - self._setpoint83            ui = self._piUi + error * self._interval / 1000 * self.Ki84            self._piUi = ui85            ud = self._last_error / self._interval * self.Kd86            output = - self.Kp * (error + ui + ud)87            self._last_error = error88            previous_gain = self._gain89            previous_exposure_time = self._exposure_time90            if (error > 0 and self._min_gain < self._gain) or (error < 0 and self._gain < self._max_gain):91                # adjust gain92                db_increase = output / 593                self._gain = clamp(self._gain + db_increase, self._min_gain, self._max_gain)94            elif (error > 0 and self._min_exposure_time < self._exposure_time) or \95                    (error < 0 and self._exposure_time < self._max_exposure_time):96                self._exposure_time = clamp(self._exposure_time + output, self._min_exposure_time,97                                            self._max_exposure_time)98            else:99                # stuck at the edge...100                self._piUi = 0101            if self._exposure_time != previous_exposure_time:102                self.exposure_time_changed.emit(self._exposure_time)103            if self._gain != previous_gain:104                self.gain_changed.emit(self._gain)105    # to convert the values from ZWO into nice units:106    _gain_factor = 10107    _exposure_factor = 1000108    exposure_time_changed = pyqtSignal(float)109    gain_changed = pyqtSignal(float)110    saturation_changed = pyqtSignal(int)111    def __init__(self, parent=None, cam_number=0):112        super().__init__()113        self._camera = None114        self.maxval = 2**16115        self._last_saturations = []116        self._saturation = 0117        self._camera = zwoasi.Camera(cam_number)118        self._active = True119        self._manualMode = True120        self._controls = self._camera.get_controls()121        self._info = self._camera.get_camera_property()122        self._camera.set_image_type(zwoasi.ASI_IMG_RAW16)123        self._camera.set_control_value(zwoasi.ASI_GAIN, 20)124        self._camera.set_control_value(zwoasi.ASI_EXPOSURE, 3000)125        self._camera.set_control_value(zwoasi.ASI_FLIP, 0)126        self._camera.disable_dark_subtract()127        self._camera.set_control_value(zwoasi.ASI_COOLER_ON, 1)128        self._camera.set_control_value(zwoasi.ASI_TARGET_TEMP, self._controls["TargetTemp"]["MinValue"])129        self.capture_thread = self.CaptureThread(self._camera)130        self.capture_thread.ndarray_available.connect(self.ndarray_available)131        self.ndarray_available.connect(self.calculate_saturation)132        self.auto_settings_thread = self.AutoSettingsObject()133        self.auto_settings_thread.set_gain(self.get_gain())134        self.auto_settings_thread.set_exposure_time(self.get_exposure())135        self.auto_settings_thread.set_gain_range(self.get_gain_range())136        self.auto_settings_thread.set_exposure_range(self.get_exposure_range())137        self.gain_changed.connect(self.auto_settings_thread.set_gain)138        self.exposure_time_changed.connect(self.auto_settings_thread.set_exposure_time)139        self.saturation_changed.connect(self.auto_settings_thread.set_saturation)140        self.auto_settings_thread.gain_changed.connect(self.set_gain)141        self.auto_settings_thread.exposure_time_changed.connect(self.set_exposure)142    def __del__(self):143        self.capture_thread.stop()144        self.capture_thread.wait()145    @pyqtSlot()146    def start(self):147        self.capture_thread.start()148    @pyqtSlot()149    def stop(self):150        self.capture_thread.stop()151    @pyqtSlot(np.ndarray)152    def calculate_saturation(self, array):153        sat = array.max() / self.maxval * 100154        self._last_saturations.append(sat)155        if len(self._last_saturations) > 1:156            self._last_saturations.pop(0)157        weight = np.arange(1, len(self._last_saturations) + 1)158        self._saturation = np.sum(weight * np.array(self._last_saturations)) / np.sum(weight)159        self.saturation_changed.emit(self._saturation)160    @pyqtSlot()161    def get_controls(self):162        controls = CameraSettingsWidget()163        min_exp, max_exp = self.get_exposure_range()164        max_exp = min(max_exp, 2000)165        controls.set_exposure_range(min_exp, max_exp)166        controls.set_gain_range(*self.get_gain_range())167        controls.set_gain(self.get_gain())168        controls.set_exposure(self.get_exposure())169        controls.exposure_changed.connect(self.set_exposure)170        controls.gain_changed.connect(self.set_gain)171        controls.auto_changed.connect(self.enable_auto)172        self.exposure_time_changed.connect(controls.set_exposure)173        self.gain_changed.connect(controls.set_gain)174        return controls175    @pyqtSlot()176    def has_controls(self):177        return True178    @pyqtSlot(bool)179    def enable_auto(self, auto):180        if auto:181            self.auto_settings_thread.set_gain = self.get_gain()182            self.auto_settings_thread.exposure_time = self.get_exposure()183            self.auto_settings_thread.start()184        else:185            self.auto_settings_thread.stop()186    @staticmethod187    def get_number_cameras():188        return zwoasi.get_num_cameras()189    @staticmethod190    def get_available_cameras():191        return list(range(ZwoCamera.get_number_cameras()))192    @staticmethod193    def init_library(path):194        zwoasi.init(path)195    def _clear_interface(self):196        pass197        # if self._control.DeviceValid:198        #     self._control.LiveStop()199        #     self._control.DeviceFrameFilters.Clear()200        #     self._control.Device = ""201    def _valid(self):202        return self._active203    @Camera._ensure_valid204    def get_exposure(self):205        return self._camera.get_control_value(zwoasi.ASI_EXPOSURE)[0] / self._exposure_factor206    @Camera._ensure_valid207    def get_exposure_range(self):208        rng = (self._controls["Exposure"]["MinValue"], self._controls["Exposure"]["MaxValue"])209        scaled_rng = [bound / self._exposure_factor for bound in rng]210        return scaled_rng211    @Camera._ensure_valid212    def is_auto_exposure(self):213        return self._camera.get_control_value(zwoasi.ASI_EXPOSURE)[1]214    @Camera._ensure_valid215    def set_auto_exposure(self, auto):216        current_exposure = self._camera.get_control_value(zwoasi.ASI_EXPOSURE)[0]217        self._camera.set_control_value(zwoasi.ASI_EXPOSURE, current_exposure, auto=auto)218    @Camera._ensure_valid219    def set_exposure(self, exposure):220        true_exposure = int(exposure * self._exposure_factor)221        rng = (self._controls["Exposure"]["MinValue"], self._controls["Exposure"]["MaxValue"])222        if not rng[0] <= true_exposure <= rng[1]:223            raise ValueError("Exposure parameter {} is outside of allowed range {}".format(true_exposure, rng))224        self._camera.set_control_value(zwoasi.ASI_EXPOSURE, true_exposure)225        self.exposure_time_changed.emit(exposure)226    @Camera._ensure_valid227    def get_gain(self):228        return self._camera.get_control_value(zwoasi.ASI_GAIN)[0] / self._gain_factor229    @Camera._ensure_valid230    def get_gain_range(self):231        rng = (self._controls["Gain"]["MinValue"], self._controls["Gain"]["MaxValue"])232        scaled_rng = [bound / self._gain_factor for bound in rng]233        return scaled_rng234    @Camera._ensure_valid235    def is_auto_gain(self):236        return self._camera.get_control_value(zwoasi.ASI_GAIN)[1]237    @Camera._ensure_valid238    def set_auto_gain(self, auto):239        current_gain = self._camera.get_control_value(zwoasi.ASI_GAIN)[0]240        self._camera.set_control_value(zwoasi.ASI_GAIN, current_gain, auto=auto)241    @Camera._ensure_valid242    def set_gain(self, gain):243        true_gain = int(gain * self._gain_factor)244        rng = (self._controls["Gain"]["MinValue"], self._controls["Gain"]["MaxValue"])245        if not rng[0] <= true_gain <= rng[1]:246            raise ValueError("Gain parameter {} is outside of allowed range {}".format(true_gain, rng))247        self._camera.set_control_value(zwoasi.ASI_GAIN, true_gain)248        self.gain_changed.emit(gain)249    def get_pid(self):250        p = self.auto_settings_thread.Kp251        i = self.auto_settings_thread.Ki252        d = self.auto_settings_thread.Kd253        return p, i, d254    def set_pid(self, p, i, d):255        self.auto_settings_thread.Kp = p256        self.auto_settings_thread.Ki = i257        self.auto_settings_thread.Kd = d258        self.save_pid_values(p, i, d)259        print("pid:", p, i, i)260    def load_pid_values(self):261        settings = QSettings("Fringes", "Fringes")262        settings.beginGroup("ZwoCamera")263        if settings.value("P") is not None:264            p = settings.value("P")265            i = settings.value("I")266            d = settings.value("D")267        else:268            p, i, d = self.read_pid_settings("ZwoCamera")269        settings.endGroup("ZwoCamera")270        return p, i, d271    def save_pid_values(self, p, i, d):272        settings = QSettings("Fringes", "Fringes")273        settings.beginGroup("ZwoCamera")274        settings.setValue("P", p)275        settings.setValue("I", i)276        settings.setValue("D", d)...

precise_ephemeris.py

Source:precise_ephemeris.py

...24        self._validity_interval = _EmptyInterval()25        self._loaded_interval = _EmptyInterval()26        self._week = None27        self._positions = {}28    def _ensure_valid(self, week, time):29        """30        If the time isn't in validity interval,31        then download and parse a new sp3 file.32        """33        if time in self._validity_interval and self._week == week:34            return35        rounded_time_2 = (time // (2 * 3600)) * (2 * 3600)36        rounded_time_6 = (time // (6 * 3600)) * (6 * 3600)37        38        file_name = self._get_file_name(week, time)39        remote_file = urlparse.urljoin(self._server, file_name)40        if time not in self._loaded_interval:41            logging.info("Downloading file %s", remote_file)42            f = urllib2.urlopen(remote_file)43            with _decompress_Z_file(f) as f2:44                self._sp3 = Sp3(f2)45            self._loaded_interval = _Interval(rounded_time_6, rounded_time_6 + 6 * 3600)46        logging.info("Recalculating polynomials")47        # find start index for the new validity interval in sp3 file48        start = 049        for i, epoch in enumerate(self._sp3.values):50            if epoch.time >= rounded_time_2:51                start = i52                break53        epochs = self._sp3.values[start - 8: start + 17]54        for sv in self._sp3.svs:55            if sv[0] != 'G':56                continue # only gps57            prn = int(sv[1:])58            times = numpy.fromiter(59                (x.time for x in epochs),60                dtype=numpy.float)61            positions = []62            for i in range(4):63                values = numpy.fromiter(64                    (x[sv][i] for x in epochs if sv in x),65                    dtype=numpy.float)66                assert len(times) == len(values) == 2567                if i == 4:68                    values = values * 1e-669                else:70                    values = values * 1e371                poly = numpy.polynomial.polynomial.Polynomial.fit(times, values, 12)72                #poly = numpy.polyfit(times, values, 12) * multiplier73                positions.append(poly)74            self._positions[prn] = positions75        self._validity_interval = _Interval(rounded_time_2, rounded_time_2 + 2 * 3600)76        self._week = week77    def _get_file_name(self, week, time):78        d, rest = divmod(time, 24 * 3600)79        h = rest // 360080        h = (h // 6) * 681        return "{w:04}/igu{w:04}{d:01}_{h:02}.sp3.Z".format(82            w=int(week), d=int(d), h=int(h))83    def sv_pos(self, prn, week, time):84        self._ensure_valid(week, time)85        return numpy.matrix([[86            self._positions[prn][i](time) for i in range(3)]])87    def sv_clock_offset(self, prn, week, time):88        self._ensure_valid(week, time)89        return self._positions[prn][3](time)90    def sv_velocity(prn, week, time):91        self._ensure_valid(week, time)92        raise NotImplementedError()93        #return numpy.matrix([[94        #    numpy.polyval(self._positions[prn][i], time) for i in range(3)]])95    def sv_clock_drift(self, prn, week, time):96        self._ensure_valid(week, time)97        raise NotImplementedError()98    def observed_message_types(self):99        """Return a set of sirf message classes that this observer is interested in."""100        return [sirf_messages.GeodeticNavigationData]101    def __call__(self, message):102        """Notification that a message was received.103        Only the messages specified in observed_message_types() are received."""104        self._current_week = message.extended_gps_week105class _EmptyInterval:106    def __contains__(self, x):107        return False108    def __str__(self):109        return "Epty interval"110class _Interval:...

verify_credentials.py

Source:verify_credentials.py

...15        self._verification_model = verification_model16        self._user_repository = user_repository17        self._ensure_valid = validator18    def execute(self, request):19        self._ensure_valid(request)20        user = self._user_repository.find_one(request.username)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.