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

How to use write_to_console method in tempest

Best Python code snippet using tempest_python

auto_uploader.py

Source:auto_uploader.py

...385        func(*args, **kwargs)386        console_lock.release()387    return wrapper388@require_console_lock389def write_to_console(fg, bg, message, end='\n'):390    app_view.console.SetDefaultStyle(wx.TextAttr(fg, bg))391    app_view.console.AppendText(f"{message}{end}")392    app_view.console.ShowPosition(app_view.console.GetLastPosition())393def append_if_not_exist(fg, bg, text, append_text=None):394    if append_text is None:395        append_text = text396    if not app_view.console.GetValue().endswith(text):397        write_to_console(fg, bg, append_text, end='')398def pretty_block(func):399    @functools.wraps(func)400    def wrapper(*args, **kwargs):401        append_if_not_exist(*COLOR_GRAY, '\n\n', '\n')402        write_to_console(*COLOR_GRAY, f"[{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] =====>")403        func(*args, **kwargs)404        write_to_console(*COLOR_GRAY, f'==============================', end='\n\n')405    return wrapper406def bind_events():407    app_view.btn_dev_ip_addr.Bind(wx.EVT_BUTTON, lambda evt: on_btn_dev_ip_addr())408    app_view.btn_watch_path.Bind(wx.EVT_BUTTON, lambda evt: on_btn_mon_path())409    app_view.btn_show_ver.Bind(wx.EVT_BUTTON, lambda evt: pool.submit(on_btn_show_ver))410    app_view.btn_list_video.Bind(wx.EVT_BUTTON, lambda evt: pool.submit(on_btn_list_video))411    app_view.btn_clear_console.Bind(wx.EVT_BUTTON, lambda evt: on_btn_clear_console())412    app_view.btn_upload_video.Bind(wx.EVT_BUTTON, lambda evt: pool.submit(on_btn_upload_video))413    app_view.btn_clear_video.Bind(wx.EVT_BUTTON, lambda evt: pool.submit(on_btn_clear_video))414    app_view.btn_pop_video.Bind(wx.EVT_BUTTON, lambda evt: pool.submit(on_btn_pop_video))415    app_view.btn_start_breath.Bind(wx.EVT_BUTTON, lambda evt: pool.submit(on_btn_start_breath))416    app_view.btn_stop_breath.Bind(wx.EVT_BUTTON, lambda evt: pool.submit(on_btn_stop_breath))417    app_view.btn_start_fan.Bind(wx.EVT_BUTTON, lambda evt: pool.submit(on_btn_start_fan))418    app_view.btn_stop_fan.Bind(wx.EVT_BUTTON, lambda evt: pool.submit(on_btn_stop_fan))419    app_view.btn_reset_device.Bind(wx.EVT_BUTTON, lambda evt: pool.submit(on_btn_reset_device))420    app_view.btn_start_service.Bind(wx.EVT_BUTTON, lambda evt: pool.submit(on_btn_start_service))421    app_view.btn_stop_service.Bind(wx.EVT_BUTTON, lambda evt: pool.submit(on_btn_stop_service))422def on_btn_dev_ip_addr():423    global dev_ip_addr424    dev_ip_addr = get_dev_ip_addr()425    write_to_console(*COLOR_WHITE, f'Set device IP to {dev_ip_addr}')426def on_btn_mon_path():427    global watch_path428    watch_path = get_watch_path()429    write_to_console(*COLOR_WHITE, f'Set monitor path to {watch_path}')430@require_fan_lock431@pretty_block432def on_btn_show_ver():433    try:434        write_to_console(*COLOR_BLUE, 'Started to fetch version info...')435        resp = FanControl.show_version(dev_ip_addr)436        write_to_console(*COLOR_WHITE, 'Version:')437        write_to_console(*COLOR_WHITE, f" - FPGA: {resp['FPGA']}")438        write_to_console(*COLOR_WHITE, f" - ARM: {resp['ARM']}")439        write_to_console(*COLOR_WHITE, f" - MCU: {resp['MCU']}")440        write_to_console(*COLOR_GREEN, 'Successfully fetched version info')441    except Exception as e:442        write_to_console(*COLOR_RED, f'Failed to fetch version info:')443        write_to_console(*COLOR_RED, f' - {str(e)}')444@require_fan_lock445@pretty_block446def on_btn_list_video():447    try:448        write_to_console(*COLOR_BLUE, 'Started to list videos...')449        names = FanControl.list_video(dev_ip_addr)450        if len(names) == 0:451            write_to_console(*COLOR_WHITE, 'No video to list')452        else:453            write_to_console(*COLOR_WHITE, f'Found {len(names)} videos:')454            for name in names:455                write_to_console(*COLOR_WHITE, f' - {name}')456        write_to_console(*COLOR_GREEN, 'Successfully listed videos')457    except Exception as e:458        write_to_console(*COLOR_RED, f'Failed to list videos:')459        write_to_console(*COLOR_RED, f' - {str(e)}')460def on_btn_clear_console():461    app_view.console.SetValue('')462@pretty_block463def on_btn_upload_video():464    try:465        write_to_console(*COLOR_BLUE, 'Started to upload videos...')466        # Disable all buttons467        enable_non_service_button(False)468        enable_service_button(False)469        if not os.path.exists(MEDIA_DIR):470            os.mkdir(MEDIA_DIR)471            write_to_console(*COLOR_WHITE, f'Created media folder "{MEDIA_DIR}"')472        elif not os.path.isdir(MEDIA_DIR):473            raise Exception(f'"{MEDIA_DIR}" is not a folder')474        upload_count = 0475        for filename in os.listdir(MEDIA_DIR):476            full_path = os.path.join(MEDIA_DIR, filename)477            if not os.path.isfile(full_path):478                continue479            write_to_console(*COLOR_WHITE, f'[{upload_count + 1}] Uploading "{filename}"...', end='')480            perc_count = 0481            for perc in FanControl.upload_video(server_ipaddr='192.168.4.89',482                                                server_port=SERVER_PORT,483                                                web_path_to_file=f'{MEDIA_DIR}/{filename}',484                                                local_path_to_file=full_path,485                                                ip_addr=dev_ip_addr):486                if perc >= 100:487                    # Progress is completed488                    if perc_count > 0:489                        # Non-first progress is completed490                        write_to_console(*COLOR_WHITE, f'{perc}%')491                else:492                    # Progress is uncompleted493                    write_to_console(*COLOR_WHITE, f'{perc}%...', end='')494                perc_count += 1495            append_if_not_exist(*COLOR_WHITE, '\n')496            if perc_count != 1:497                write_to_console(*COLOR_WHITE, f'[{upload_count + 1}] Uploaded "{filename}"')498            else:499                write_to_console(*COLOR_WHITE, f'[{upload_count + 1}] "{filename}" already exists')500            upload_count += 1501        write_to_console(*COLOR_GREEN, f'Successfully uploaded {upload_count} videos')502        if upload_count == 0:503            write_to_console(*COLOR_YELLOW, f'Please put videos in "{MEDIA_DIR}" folder')504    except Exception as e:505        append_if_not_exist(*COLOR_WHITE, '\n')506        write_to_console(*COLOR_RED, f'Failed to upload videos:')507        write_to_console(*COLOR_RED, f' - {str(e)}')508    finally:509        # Enable all buttons510        enable_non_service_button(True)511        enable_service_button(True)512@pretty_block513def on_btn_clear_video():514    try:515        write_to_console(*COLOR_BLUE, 'Started to clear videos...')516        num_video_cleared = FanControl.clear_video(dev_ip_addr)517        if num_video_cleared == -1:518            write_to_console(*COLOR_WHITE, 'No video to clear')519        else:520            write_to_console(*COLOR_WHITE, f'Cleared {num_video_cleared} videos')521        write_to_console(*COLOR_GREEN, 'Successfully cleared videos')522    except Exception as e:523        write_to_console(*COLOR_RED, f'Failed to clear all videos:')524        write_to_console(*COLOR_RED, f' - {str(e)}')525@pretty_block526def on_btn_pop_video():527    try:528        write_to_console(*COLOR_BLUE, 'Started to pop videos...')529        num_video_remained = FanControl.pop_video(dev_ip_addr)530        if num_video_remained == -1:531            write_to_console(*COLOR_WHITE, 'No video to pop')532        else:533            write_to_console(*COLOR_WHITE,534                             f'Popped the first video and {num_video_remained} videos remain')535        write_to_console(*COLOR_GREEN, 'Successfully popped videos')536    except Exception as e:537        write_to_console(*COLOR_RED, f'Failed to pop videos:')538        write_to_console(*COLOR_RED, f' - {str(e)}')539@pretty_block540def on_btn_start_breath():541    try:542        write_to_console(*COLOR_BLUE, 'Started to enable breath...')543        FanControl.start_breath(dev_ip_addr)544        write_to_console(*COLOR_GREEN, 'Successfully enabled breath')545    except Exception as e:546        write_to_console(*COLOR_RED, f'Failed to enable breath:')547        write_to_console(*COLOR_RED, f' - {str(e)}')548@pretty_block549def on_btn_stop_breath():550    try:551        write_to_console(*COLOR_BLUE, 'Started to disable breath...')552        FanControl.stop_breath(dev_ip_addr)553        write_to_console(*COLOR_GREEN, 'Successfully disabled breath')554    except Exception as e:555        write_to_console(*COLOR_RED, f'Failed to disable breath:')556        write_to_console(*COLOR_RED, f' - {str(e)}')557@require_fan_lock558@pretty_block559def on_btn_start_fan():560    try:561        write_to_console(*COLOR_BLUE, 'Started to enable fan...')562        FanControl.start_fan(dev_ip_addr)563        write_to_console(*COLOR_GREEN, 'Successfully enabled fan')564    except Exception as e:565        write_to_console(*COLOR_RED, f'Failed to enable fan:')566        write_to_console(*COLOR_RED, f' - {str(e)}')567@require_fan_lock568@pretty_block569def on_btn_stop_fan():570    try:571        write_to_console(*COLOR_BLUE, 'Started to disable fan...')572        FanControl.stop_fan(dev_ip_addr)573        write_to_console(*COLOR_GREEN, 'Successfully disabled fan')574    except Exception as e:575        write_to_console(*COLOR_RED, f'Failed to disable fan:')576        write_to_console(*COLOR_RED, f' - {str(e)}')577@pretty_block578def on_btn_reset_device():579    try:580        write_to_console(*COLOR_BLUE, 'Started to reset device...')581        FanControl.restore_to_factory(dev_ip_addr)582        write_to_console(*COLOR_GREEN, 'Successfully reset device')583    except Exception as e:584        write_to_console(*COLOR_RED, f'Failed to reset device:')585        write_to_console(*COLOR_RED, f' - {str(e)}')586def on_btn_start_service():587    # Start service async588    global service_thread589    service_thread = threading.Thread(target=service_worker, args=(), daemon=True)590    service_thread.start()591def on_btn_stop_service():592    global service_status593    if not service_status:594        return595    service_status = False596    write_to_console(*COLOR_BLUE, 'Requested to stop automation service...')597def web_server_worker():598    try:599        handler = http.server.SimpleHTTPRequestHandler600        with socketserver.TCPServer(('0.0.0.0', SERVER_PORT), handler) as httpd:601            write_to_console(*COLOR_GREEN, f'Web server started at http://127.0.0.1:{SERVER_PORT}/')602            httpd.serve_forever()603    except Exception as e:604        write_to_console(*COLOR_RED, 'General failure in web server:')605        write_to_console(*COLOR_RED, f' - {str(e)}')606def service_worker():607    global service_status608    try:609        service_status = True610        detected_file_names = []611        detected_file_durations = []612        on_device_file_durations = []613        next_pop_time = 0614        write_to_console(*COLOR_GREEN, 'Automation service started')615        if not os.path.exists(watch_path):616            os.mkdir(watch_path)617            write_to_console(*COLOR_GREEN, f'Created watch folder "{watch_path}"')618        elif not os.path.isdir(watch_path):619            raise Exception(f'"{watch_path}" is not a folder')620        archive_path = os.path.join(watch_path, ARCHIVE_DIR)621        if not os.path.exists(archive_path):622            os.mkdir(archive_path)623            write_to_console(*COLOR_GREEN, f'Created archive folder "{archive_path}"')624        elif not os.path.isdir(archive_path):625            raise Exception(f'"{archive_path}" is not a folder')626        # Disable buttons627        enable_non_service_button(False)628        # Scan watch folder629        file_count = 0630        for filename in os.listdir(watch_path):631            full_path = os.path.join(watch_path, filename)632            if not os.path.isfile(full_path):633                continue634            file_count += 1635        if file_count == 0:636            # Clear all videos637            write_to_console(*COLOR_BLUE, 'Found 0 video in watch folder, started to clear all videos on device...')638            num_video_cleared = FanControl.clear_video(dev_ip_addr)639            write_to_console(*COLOR_GREEN, f'Successfully cleared {max(0, num_video_cleared)} videos on device')640        else:641            # Mark videos on device642            write_to_console(*COLOR_BLUE,643                             f'Found {file_count} videos in watch folder, started to reduce the number of videos on '644                             f'device to one...')645            num_videos_on_device = len(FanControl.list_video(dev_ip_addr))646            while num_videos_on_device > 1:647                num_videos_on_device = FanControl.pop_video(dev_ip_addr)648            if num_videos_on_device == 1:649                on_device_file_durations.append(-1)650            write_to_console(*COLOR_GREEN,651                             f'Reduced the number of videos on device to one')652        # Service loop653        write_to_console(*COLOR_BLUE, f'Started to watch "{watch_path}" folder...')654        while service_status:655            fan_lock.acquire()656            # Enqueue all new files657            for filename in os.listdir(watch_path):658                full_path = os.path.join(watch_path, filename)659                if not os.path.isfile(full_path) or filename in detected_file_names:660                    continue661                detected_file_names.append(filename)662                duration = int(pymediainfo.MediaInfo.parse(full_path).tracks[0].duration / 1000)663                detected_file_durations.append(duration)664                write_to_console(*COLOR_GREEN, f'Found new file "{filename}", duration {duration} seconds')665            # Dequeue and upload until no available video or device gets full666            while len(detected_file_names) > 0 and (len(on_device_file_durations) < 2):667                filename = detected_file_names.pop(0)668                duration = detected_file_durations.pop(0)669                full_path = os.path.join(watch_path, filename)670                if not os.path.exists(full_path):671                    continue672                write_to_console(*COLOR_BLUE, f'Uploading "{filename}...', end='')673                perc_count = 0674                for perc in FanControl.upload_video(server_ipaddr=SERVER_IP,675                                                    server_port=SERVER_PORT,676                                                    web_path_to_file=f'{watch_path}/{filename}',677                                                    local_path_to_file=full_path,678                                                    ip_addr=dev_ip_addr):679                    if perc >= 100:680                        # Progress is completed681                        if perc_count > 0:682                            # Non-first progress is completed683                            write_to_console(*COLOR_BLUE, f'{perc}%', end='')684                    else:685                        # Progress is uncompleted686                        write_to_console(*COLOR_BLUE, f'{perc}%...', end='')687                    perc_count += 1688                append_if_not_exist(*COLOR_BLUE, '\n')689                if perc_count == 1:690                    raise Exception(f'"{filename}" already exists on device')691                on_device_file_durations.append(duration)692                if len(on_device_file_durations) == 1:693                    next_pop_time = time.time() + duration + 10694                    write_to_console(*COLOR_GREEN, f'Uploaded "{filename}" as the first video, '695                                                   f'will pop it after {duration + 10} seconds')696                else:697                    write_to_console(*COLOR_GREEN, f'Uploaded "{filename}" as the second video, will pend to play')698                shutil.move(full_path, os.path.join(archive_path, filename))699                write_to_console(*COLOR_GREEN, f'Archived "{filename}" to "{archive_path}" folder')700            # Pop video and play the next video701            while len(on_device_file_durations) >= 2 and time.time() >= next_pop_time:702                write_to_console(*COLOR_BLUE, 'Started to pop the first video...')703                # Pop the first video704                FanControl.pop_video(dev_ip_addr)705                on_device_file_durations.pop(0)706                # Calculate next pop time707                next_pop_time = time.time() + on_device_file_durations[0]708                write_to_console(*COLOR_GREEN, f'Popped the first video and proceeded to the second video, '709                                               f'will pop it after {on_device_file_durations[0]} seconds')710            fan_lock.release()711            time.sleep(1)712    except Exception as e:713        write_to_console(*COLOR_RED, 'General failure in auto upload service:')714        write_to_console(*COLOR_RED, f' - {str(e)}')715    finally:716        if fan_lock.locked():717            fan_lock.release()718        # Enable buttons719        enable_non_service_button(True)720        write_to_console(*COLOR_GREEN, 'Stopped automation service')721if __name__ == '__main__':722    # Start GUI723    app = wx.App()724    app_view = AppView()725    # Locks726    console_lock = threading.Lock()727    fan_lock = threading.Lock()728    # Configure GUI729    dev_ip_addr = '192.168.4.1'730    watch_path = 'auto_upload'731    set_dev_ip_addr(dev_ip_addr)732    set_watch_path(watch_path)733    bind_events()734    app_view.Show()...

function_generator_ui.py

Source:function_generator_ui.py

...88        com_layout = QHBoxLayout()89        com_label = QLabel('Serial Port')9091        if self.dummy_mode == True:92            self.write_to_console('Test Mode Activated!')93            com_ports = ['COM1', 'COM2'] #for testing!94        else:95            com_ports = get_serial_ports()9697        com_box = QComboBox()98        com_box.addItems(com_ports)99        self.com_box = com_box100101        com_layout.addWidget(com_label)102        com_layout.addWidget(com_box)103104        # self.main_layout.addLayout(com_layout)105        io_layout.addLayout(com_layout)106107        connect_button = QPushButton('Connect to Device')108        reset_button = QPushButton('Reset')109        disconnect_button = QPushButton('Disconnect')110        # disconnect_button.setIcon(QtGui.QIcon('disconnect.png'))111        disconnect_button.setToolTip('Disconnect')112113        button_layout = QHBoxLayout()114        button_layout.addWidget(connect_button)115        button_layout.addWidget(reset_button)116        button_layout.addWidget(disconnect_button)117118        # self.main_layout.addLayout(button_layout)119        io_layout.addLayout(button_layout)120        io_box.setLayout(io_layout)121122        self.main_layout.addWidget(io_box)123124125        function_box = QGroupBox('Waveform')126        func_layout = QHBoxLayout()127        sine_wave_button = QPushButton(QtGui.QIcon('ico/sine.png'),'')128        sine_wave_button.setToolTip('Sine Wave')129        square_wave_button = QPushButton(QtGui.QIcon('ico/square.png'),'')130        square_wave_button.setToolTip('Square Wave')131        triangle_button = QPushButton(QtGui.QIcon('ico/triangle.png'),'')132        triangle_button.setToolTip('Triangle Wave')133134        func_layout.addWidget(sine_wave_button)135        func_layout.addWidget(square_wave_button)136        func_layout.addWidget(triangle_button)137        function_box.setLayout(func_layout)138        self.main_layout.addWidget(function_box)139140141        voltage_box = QGroupBox('Voltage (Peak to Peak)')142        voltage_layout = QHBoxLayout()143        voltage_label = QLabel('V')144        voltage_input_box = QDoubleSpinBox()145        voltage_input_box.setRange(0,40)146        voltage_input_box.setValue(10)147        voltage_input_box.setDecimals(2)148        self.voltage_input_box = voltage_input_box149150        voltage_button = QPushButton('Set Voltage')151        high_voltage_button = QPushButton('Toggle HV')152        high_voltage_button.setToolTip('Enable High Voltage Output, only for HP33325B')153154        voltage_layout.addWidget(voltage_input_box)155        voltage_layout.addWidget(voltage_label)156        voltage_layout.addWidget(voltage_button)157        voltage_layout.addWidget(high_voltage_button)158        voltage_box.setLayout(voltage_layout)159160161        freq_box = QGroupBox('Frequency')162        freq_layout = QHBoxLayout()163        freq_label = QLabel('Hz')164165        freq_input_box = QDoubleSpinBox()166        freq_input_box.setRange(0,10)167        freq_input_box.setValue(0.1)168        freq_input_box.setSingleStep(0.001)169        freq_input_box.setDecimals(5)170        self.freq_input_box = freq_input_box171172        freq_button = QPushButton('Set Frequency')173        zero_freq_button = QPushButton('Zero Frequency')174175        freq_layout.addWidget(freq_input_box)176        freq_layout.addWidget(freq_label)177        freq_layout.addWidget(freq_button)178        freq_layout.addWidget(zero_freq_button)179180        freq_box.setLayout(freq_layout)181182        offset_box = QGroupBox('Voltage Offset')183        offset_layout = QHBoxLayout()184        offset_label = QLabel('V')185        offset_input_box = QDoubleSpinBox()186        offset_input_box.setRange(-40,40)187        offset_input_box.setValue(0)188        self.offset_input_box = offset_input_box189190        offset_button = QPushButton('Set Offset Voltage')191192        offset_layout.addWidget(offset_input_box)193        offset_layout.addWidget(offset_label)194        offset_layout.addWidget(offset_button)195        offset_box.setLayout(offset_layout)196197198199        self.main_layout.addWidget(voltage_box)200        self.main_layout.addWidget(freq_box)201        self.main_layout.addWidget(offset_box)202        # self.(self.main_layout)203        # self.setCentralWidget(self.main_layout)204        widget = QWidget()205206        self.super_layout.addLayout(self.main_layout)207208209210211        self.super_layout.addWidget(self.console)212        widget.setLayout(self.super_layout)213214215        self.setCentralWidget(widget)216217218219        """220        Connections221        """222223        # these may change224        # self.connected_to_33120A = False225        # self.connected_to_3325B = False226        self.selected_FG = None227        self.selected_serial_port = ''228        self.inst = None229        self.hv_state = 0 # high voltage on 3325b is off by default230231        #menu connections232        dump_console_action.triggered.connect(self.dump_console_to_disk)233        clear_console_action.triggered.connect(self.clear_console)234        self.about.triggered.connect(self.show_about)235236        #instrument connections237        # self.inst = instrument.dummy_instrument('/dev/ttyUSB0',9600,)238        # action_HP_33120A.triggered.connect(self.connect_to_33120A)239        # action_HP_3325B.triggered.connect(self.connect_to_3325B)240241        self.action_HP_33120A.triggered.connect(lambda: self.set_FG('HP33120A'))242        self.action_HP_3325B.triggered.connect(lambda: self.set_FG('HP3325B'))243244245246        #initialize com_box with something247        self.set_serial(com_ports[0])248        #for if the user changes something249        self.com_box.currentTextChanged.connect(lambda: self.set_serial(self.com_box.currentText()))250251252        connect_button.clicked.connect(self.connect_to_inst)253        disconnect_button.clicked.connect(self.disconnect_inst)254        reset_button.clicked.connect(self.reset_inst)255256        #waveform connections257        sine_wave_button.clicked.connect(self.set_sine)258        triangle_button.clicked.connect(self.set_triangle)259        square_wave_button.clicked.connect(self.set_square)260261        #A-F-O connectioned262        voltage_button.clicked.connect(self.set_voltage)263        high_voltage_button.clicked.connect(self.toggle_high_voltage)264        freq_button.clicked.connect(self.set_frequency)265        offset_button.clicked.connect(self.set_offset)266        zero_freq_button.clicked.connect(self.zero_frequency)267268269270    def write_to_console(self, content):271        """272        Write whatever to the console and terminal for debugging273        and clarity.274        """275        console_str = '[{}] '.format(self.console_idx) + str(content) #+ '\n'276        # self.console.append(console_str)277        print(console_str)278        self.console.appendHtml(console_str)279        self.console_idx += 1280281    def dump_console_to_disk(self):282        """283        Dump the console to a text file with a random number as its name284        """285        console_dump_name = 'fg_console_dump_{}.txt'.format(str(random.randint(100,1000)))286        self.write_to_console('Dumping Console to Disk at: {}'.format(console_dump_name))287        with open(console_dump_name, 'a') as f:288            f.write(self.console.toPlainText())289290    def clear_console(self):291        """292        delete contents of the console so it looks nice again293        """294        self.console_idx = 0295        self.console.clear()296        self.write_to_console('Console Cleared!')297298299    def set_FG(self, FG):300        """301        select the function generator from the UI302        """303        self.selected_FG = FG304        self.write_to_console('Selected {} Function Generator'.format(FG))305306    def get_FG(self):307        """308        not used309        """310        return self.selected_FG311312313    def set_serial(self, serial_port):314        """315        store the user seleced serial port for316        later work317        """318        self.serial_port = serial_port319        # self.serial_port = self.com_box.currentText()320        self.write_to_console('Serial Port Set to: {}'.format(self.serial_port))321322    def get_serial(self):323        return self.serial_port324325326327328    def connect_to_inst(self):329        """330        Initialize a connection to the function generator331        and return the serial connection object as332        self.inst333        """334        if self.selected_FG is None:335            self.write_to_console('Select Function Generator Model First!')336            return337338        # print(self.com_box.currentText())339        if str(self.com_box.currentText()) == '':340            self.write_to_console('Please select a Serial Port to Start!')341            return342343        if self.selected_FG == 'HP33120A':344            byte_size = HP33120A.BYTE_SIZE345            baud_rate = HP33120A.BAUD_RATE346            parity = HP33120A.PARITY347348            self.inst = HP33120A.HP33120A(self.serial_port,349                                          baud_rate,350                                          byte_size,351                                          parity,352                                          dummy_mode = self.dummy_mode)353354        elif self.selected_FG == 'HP3325B':355            byte_size = HP3325B.BYTE_SIZE356            baud_rate = HP3325B.BAUD_RATE357            parity = HP3325B.PARITY358            self.inst = HP3325B.HP3325B(self.serial_port,359                                          baud_rate,360                                          byte_size,361                                          parity,362                                          dummy_mode = self.dummy_mode)363364365366367        setup_out = self.inst.setup_connection()368        self.write_to_console(setup_out)369370        connect_out = self.inst.connect()371        self.write_to_console(connect_out)372373    def disconnect_inst(self):374        """375        close out the connection376        """377        disconnect_result = self.inst.disconnect()378        self.write_to_console(disconnect_result)379380    def reset_inst(self):381        """382        send the reset command383        """384        self.check_connection()385        rst_rslt = self.inst.reset()386        self.write_to_console(rst_rslt)387388    def get_error_msg(self):389        """390        handle error messages by printing391        to console392        """393        if self.inst is None:394            return395396        if self.selected_FG is None:397            return398399        err_out = self.inst.get_error()400        self.write_to_console(err_out)401402    def check_connection(self):403        """404        quick check before doing anything to see if405        there is someone to talk to.406        """407        # check if instrument and fg are set properly408409        if self.inst is None:410            self.write_to_console('No connection Found')411            raise ValueError('No Connection Found!')412        if self.selected_FG is None:413            self.write_to_console('Function Generator Method Not Set!')414            raise ValueError('Function Generator Method Not Set!')415416        else:417            return 0418419    def set_sine(self):420        self.check_connection()421        x = self.inst.set_sine_shape()422        self.write_to_console(x)423424    def set_square(self):425        self.check_connection()426        x = self.inst.set_square_shape()427        self.write_to_console(x)428429430    def set_triangle(self):431        self.check_connection()432        x = self.inst.set_triangle_shape()433        self.write_to_console(x)434435    def set_voltage(self):436        voltage = self.voltage_input_box.value()437        self.check_connection()438        x = self.inst.set_amplitude(voltage)439        self.write_to_console(x)440441442    def set_frequency(self):443        freq = self.freq_input_box.value()444        self.check_connection()445        x = self.inst.set_frequency(freq)446        self.write_to_console(x)447448    def zero_frequency(self):449        self.check_connection()450        x = self.inst.set_frequency(0)451        self.write_to_console(x)452        self.write_to_console('Frequency set to 0 Hz!')453454    def set_offset(self):455        offset = self.offset_input_box.value()456        x = self.inst.set_offset(offset)457        self.write_to_console(x)458459    def toggle_high_voltage(self):460        """461        The 3325B has a useful high voltage function462        which enables voltages up to 40 V p2p (+/- 20 V)463        this button allows us to use that.464        """465        self.check_connection()466        if self.selected_FG == 'HP3325B':467            if self.hv_state == 0:468                self.hv_state = 1469                self.inst.write('HV {}'.format(self.hv_state))470                self.write_to_console('Turning HV on!')471            elif self.hv_state == 1:472                self.hv_state = 0473                self.inst.write('HV {}'.format(self.hv_state))474                self.write_to_console('Turning HV Off!')475476        else:477            self.write_to_console('HV not available for selected FG')478479    def show_about(self):480        """481        An about window that needs some work.482        """483        dlg = QDialog()484        dlg.setWindowTitle('About Function Generator')485        # dlg_btn = QPushButton('Close')486487        dlg_label = QLabel('A Simple Function Generator Control Panel!\nFor use with HP33120A and HP3325B')488        dlg_layout = QHBoxLayout()489        dlg_layout.addWidget(dlg_label)490        dlg.setLayout(dlg_layout)491        dlg.setWindowModality(Qt.ApplicationModal)
...

gettxbyeth.py

Source:gettxbyeth.py

...10f_html = None11configfile = "config.ini"12p = SimpleNamespace() # parameters to work with13                      # populate from config file then overwrite from command line14def write_to_console(line, end="\n"):15    """Write line to console and the same to file if necessary"""16    17    print(line,end=end)18    19    if f_console is not None:20        print(line, file=f_console, end=end, flush=True)21csv_columns = ("#","#b","Value (ETH)","Value (wei)","Tx id","Tx link","Block number","Block link")22csv_delimiter = ";"23def getCSVLine(data):24    """Data is a tuple of (n,nb,value_wei,tx_id,block_number)."""25    """26        n = data[0]  # number in all27        nb = data[1]  # number in block28        eth = w3.fromWei(wei,'ether')29        wei = data[2]30        tx = data[1]31        txlink = getTxLink(tx)32        block = data[2]33        blockLink = getBlockLink(block)34    """35   36    # csv_columns = ("#","#b","Value (ETH)","Value (wei)","Tx id","Tx link","Block number","Block link")37    #  0 1    2         3       438    # (n,nb,value_wei,tx_id,block_number)39    return F"{data[0]}{csv_delimiter}\40{data[1]}{csv_delimiter}\41{w3.fromWei(data[2],'ether')}{csv_delimiter}\42{data[2]}{csv_delimiter}\43{data[3]}{csv_delimiter}\44{getTxLink(data[3])}{csv_delimiter}\45{data[4]}{csv_delimiter}\46{getBlockLink(data[4])}"  # end of getCSVLine47# html start48#                    0               1           2       3        4         549# addHtmlStart((latestblock.number,blockmin,blockmax,ethmin_wei,ethmax_wei,start))50def addHtmlStart(settings):51    latestblock = settings[0]52    blockmin = settings[1]53    blockmax = settings[2]54    ethmin_wei = settings[3] 55    ethmax_wei = settings[4] if settings[4] is not None else "infinite"56    start = settings[5]57    ethmin = w3.fromWei(ethmin_wei,'ether')58    ethmax = w3.fromWei(ethmax_wei,'ether') if settings[4] is not None else "infinite"59    60    return F"""61<html><title>ETH range: {ethmin} - {ethmax} | Blockrange: {blockmin}-{blockmax}</title><body>62<h1>Settings</h1>63<table>64<tr>65    <td>Latest block:</td>66    <td>{latestblock}</td>67</tr>68<tr>69    <td>Block range:</td>70    <td>{blockmin} - {blockmax} ({blockmax-blockmin+1} {'blocks' if blockmax-blockmin>0 else 'block'})</td>71</tr>72<tr>73    <td>ETH range:</td>74    <td>{ethmin}-{ethmax} ETH ({ethmin_wei}-{ethmax_wei} wei) </td>75</tr>76</table>77<hr/>78Started @{start} <br/><br/>79<table>80<tr>81<td>#</td>82<td># in block</td>83<td>Value (eth)</td>84<td>Value (wei)</td>85<td>tx hash</td>86<td>block number</td>87<tr>88"""89#            0       1               2     3      490# data = (numoftx,numoftxinblock,t.value,txstr,block.number)   # (n,nb,value_wei,tx_id,block_number)91def addHtmlLine(data):92    return F"""93<tr>94    <td>{data[0]}</td>95    <td>{data[1]}</td>96    <td>{w3.fromWei(data[2],'ether')}</td>97    <td>{data[2]}</td>98    <td><a href={getTxLink(data[3])}>{data[3]}</a></td>99    <td><a href={getBlockLink(data[4])}>{data[4]}</a></td>100<tr>101"""102def addHtmlBlockSummary(data):103    return F"""104    105    106    107    """108#               0       1           2       3109# results = (numoftx, sumofeth, str(end),str(end-start))110def addHtmlEnd(data):111    return F"""</table>112<br/>113End: {data[2]}<br/>114Elapsed time: {data[3]}115<hr/>116<table>117<tr>118    <td> Total number of txs:</td>119    <td> {data[0]}</td>120</tr>121<tr>122    <td> Total ETHs transfered:</td>123    <td>{w3.fromWei(sumofeth,'ether')} ETH ({sumofeth} wei)</td>124</tr>125</table>126</body></html>127"""128# html end129eio_block_base = "https://etherscan.io/block/" 130eio_tx_base = "https://etherscan.io/tx/"131def getTxLink(tx):132    """Returns the etherscan.io link of the tx"""133    return F"{eio_tx_base}{tx}"134def getBlockLink(b):135    """Returns the etherscan.io link of the block number"""136    return F"{eio_block_base}{b}"137# big strings138description = F"""139Get ethereum transactions by sent ether value. 140"""141epilog = F"""142Have fun!143"""144if __name__ == "__main__":145    146    print(F"For the list of options, run ./{os.path.basename(__file__)} -h")147    148    # parsing command line arguments149    parser = argparse.ArgumentParser(description=description, epilog=epilog) 150    configOptions = parser.add_argument_group("Config options")151    blockOptions = parser.add_argument_group("Block options")152    etherOptions = parser.add_argument_group("Ether options")153    outputOptions = parser.add_argument_group("Output options")154        155    #configOptions.add_argument("-c","--config-file",help="Use this config file (default: "+configfile+")")156    configOptions.add_argument("-p","--provider",help="Provider to use to connect to the chain")157    158    blockOptions.add_argument("-b","--block",help="Filter in this exact block (all other block options are ignored)", type=int)159    blockOptions.add_argument("-l","--latest", help="Filter in the last N blocks or only in the latest block if no value specified.", nargs='?', default=-1, metavar="N", type=int)160    blockOptions.add_argument("-v","--block-min", help="Minimum block number to filter in", type=int)161    blockOptions.add_argument("-n","--block-max", help="Maximum block number to filter in", type=int)162    # in the last x minute/hour/day163    164    etherOptions.add_argument("-s","--skip-zero", help="Don't list transactions with 0 ETH (other ether options still apply)", action='store_true')165    etherOptions.add_argument("-0","--zero-only", help="List transactions with 0 ETH only (other ether options are ignored)", action='store_true')166    etherOptions.add_argument("-E","--exact-eth", help="List txs only with this exact ETH value.")167    # etherOptions.add_argument("-W","--exact-wei", help="List txs only with this exact wei value")168    etherOptions.add_argument("-w","--eth-min", help="Minimum value in ETH to filter for", type=float)169    etherOptions.add_argument("-r","--eth-max", help="Maximum value in ETH to filter for", type=float)170    #etherOptions.add_argument("-q","--wei-min", help="Minimum ether value in wei to filter for", type=float)171    #etherOptions.add_argument("-e","--wei-max", help="Maximum ether value in wei to filter for", type=float)172    173    outputOptions.add_argument("-oT","--out-txlink", help="Generate etherscan.io link for the tx", action="store_true")174    outputOptions.add_argument("-oB","--out-blocklink", help="Generate etherscan.io link for the block", action="store_true")175    outputOptions.add_argument("-oH","--out-html", help="Generate html output", metavar="<filename>")176    outputOptions.add_argument("-oF","--out-console", help="Save console output", metavar="<filename>")177    outputOptions.add_argument("-oC","--out-csv", help="Save csv output", metavar="<filename>")178    outputOptions.add_argument("-oA","--out-files", help="Save to every available file format", metavar="<filename>")179    180    args = parser.parse_args()181    182    p = args183    # print(p)184    # exit(-1)185    186    # output files187    if p.out_files is not None:188        p.out_console = p.out_files + ".txt"189        p.out_csv = p.out_files + ".csv"190        p.out_html = p.out_files + ".html"191    192    if p.out_console is not None:193        f_console = open(p.out_console,"w")194    195    '''196    # parsing config file197    config = configparser.ConfigParser()    198    config.optionxform = str  # to preserve case of parameters in the config file199    200    # set defaults201    p.provider = ""202    p.block_min = None203    p.block_max = None204    p.latest = None205    p.eth_min = 0206    p.eth_max = None207    p.skip_zero = False208    p.zero_only = False209    210    211    # read from config file212    if args.config_file:213        configfile = args.config_file214        config.read(configfile)215        216        p.provider = config["infura"]["url"]  217        p.eth_min = config["ethrange"]["min"]218        p.eth_max = config["ethrange"]["max"]219        p.block_min = config["blockrange"]["min"]220        p.block_max = config["blockrange"]["max"] 221    222    # overwrite settings from command line223    if args.provider:224        p.provider = args.provider225    if args.eth_min:226        p.eth_min = args.eth_min227    if args.eth_max:228        p.eth_max = args.eth_max229    if args.block_min:230        p.block_min = args.block_min231    if args.block_max:232        p.block_max = args.block_max233    '''234   235    ###### connecting to the chain    236    write_to_console(F"[*] Provider to use: ", end="")237    providerselect = [False,False,False,False] # http, ws, ipc, auto238    if len(p.provider)>0: 239        if p.provider[:4].lower() == "http":240            write_to_console(F"{p.provider} (treated as 'http' provider)")241            providerselect[0] = True242        elif p.provider[:2].lower() == "ws":243            write_to_console(F"{p.provider} (treated as 'ws' provider)")244            providerselect[1] = True245        else:246            write_to_console(F"{p.provider} (treated as 'ipc' provider)")247            providerselect[2] = True248    else:249        write_to_console("not specified, trying web3's auto connect")   250        providerselect[3] = True251        252    253    write_to_console("[*] Connecting...", end="")    254    if providerselect[0]: # http255        w3 = Web3(Web3.HTTPProvider(p.provider))        256    if providerselect[1]: # websocket257        w3 = Web3(Web3.WebsocketProvider(p.provider))   258    if providerselect[2]: # ipc259        w3 = Web3(Web3.IPCProvider(p.provider))         260    if providerselect[3]: # auto261        w3 = Web3()                                     262    263    264    if w3.isConnected():265        write_to_console(" success, we are now connected!",end="\n\n")266    else:267        write_to_console(" could not connect. Exiting...")268        if f_console is not None:269            f_console.close()    270        exit(-1)271    272    # let every parameter be fine, set defaults if necessary273    latestblock = w3.eth.getBlock("latest")274    275    # block settings276    if p.block is not None:  # -b/--block277        blockmin = p.block278        blockmax = p.block279    elif p.latest != -1:    280        if p.latest is None:281            blockmin = latestblock.number282            blockmax = latestblock.number283        else:284            blockmin = latestblock.number - p.latest + 1285            blockmax = latestblock.number286    else:287        blockmin = int(p.block_min if p.block_min is not None else latestblock.number) 288        blockmax = int(p.block_max if p.block_max is not None else latestblock.number)289    290    # ether settings291    if p.zero_only:292        ethmin_wei = 0293        ethmax_wei = 0294    elif p.exact_eth is not None:295        ethmin_wei = w3.toWei(p.exact_eth, "ether")296        ethmax_wei = w3.toWei(p.exact_eth, "ether")297    else:298        ethmin_wei = w3.toWei(p.eth_min, "ether") if p.eth_min is not None else 0299        ethmax_wei = w3.toWei(p.eth_max, "ether") if p.eth_max is not None else None300    301    write_to_console(F"Latest block: {latestblock.number}")302    write_to_console(F"Block range: {blockmin} - {blockmax} ({blockmax-blockmin+1} {'blocks' if blockmax-blockmin>0 else 'block'})")303    write_to_console(F"Min ether: {w3.fromWei(ethmin_wei,'ether') } ({ethmin_wei} wei)")304    write_to_console(F"Max ether: {w3.fromWei(ethmax_wei,'ether') if ethmax_wei is not None else 'no upper limit'} ({ethmax_wei if ethmax_wei is not None else 'infinite'} wei)")305    if p.skip_zero:306        write_to_console("Not showing transactions with 0 ETH value") 307    308    # open output files309    if p.out_csv is not None:310        f_csv = open(p.out_csv,"w")311        print(csv_delimiter.join(csv_columns),file=f_csv,flush=True)312    313    if p.out_html is not None:314        f_html = open(p.out_html,"w")315        316    317    start = datetime.now()318    319    if f_html is not None:320        print(addHtmlStart((latestblock.number,blockmin,blockmax,ethmin_wei,ethmax_wei,start)),file=f_html,flush=True)321    322    write_to_console("")323    write_to_console("[*] Start @" + str(start),end="\n\n")324    325    326    sumofeth = 0327    numoftx = 0328    for blocknum in range(blockmin,blockmax+1):329        block = w3.eth.getBlock(blocknum)330        write_to_console(F"### Block no. {block.number} ###")331        write_to_console(F"[*] No. of txs in block: {len(block.transactions)}")332        if p.out_blocklink:333            write_to_console(F"[*] {getBlockLink(block.number)}", end="\n\n")334        sumofethinblock = 0335        numoftxinblock = 0336        for tx in block.transactions:337            t = w3.eth.getTransaction(tx)338            if (t.value >= ethmin_wei and ethmax_wei is None) or \339               (t.value >= ethmin_wei and t.value <= ethmax_wei):340                   341                   if t.value==0 and p.skip_zero and p.zero_only is False:342                       continue343                   344                   txstr = str(t.hash.hex()) # tx hash as string345                   346                   write_to_console(F"{txstr}: {w3.fromWei(t.value,'ether')} ETH ({t.value} wei)")347                   348                   if p.out_txlink:349                       write_to_console(F"{getTxLink(txstr)}", end="\n\n")  350                   351                   sumofeth += t.value352                   numoftx += 1353                   sumofethinblock += t.value354                   numoftxinblock += 1355                   356                   data = (numoftx,numoftxinblock,t.value,txstr,block.number)   # (n,nb,value_wei,tx_id,block_number)357                                        358                   if f_csv is not None:359                       print(getCSVLine(data),file=f_csv)360                       361                   if f_html is not None:362                       print(addHtmlLine(data),file=f_html)363                   364        if f_csv is not None:365            f_csv.flush()366        if f_html is not None:367            f_html.flush()368            369        write_to_console("")370        write_to_console(F"Number of filtered txs in this block: {numoftxinblock}")371        write_to_console(F"Sum transfered in these txs in this block: {w3.fromWei(sumofethinblock,'ether')} ETH ({sumofethinblock} wei)")372        write_to_console("")373        374    write_to_console("")375    write_to_console("------------------------------------------")376    write_to_console(F"Total number of filtered txs: {numoftx}")377    write_to_console(F"Total sum transfered in these txs: {w3.fromWei(sumofeth,'ether')} ETH ({sumofeth} wei)")378            379    end = datetime.now()380    write_to_console("")381    write_to_console("[*] End @"+str(end))382    write_to_console("[*] Elapsed time: " + str(end-start))383    384    if f_console is not None:385        f_console.close()    386    387    if f_csv is not None:388        f_csv.close()389    390    if f_html is not None:391        results = (numoftx, sumofeth, str(end),str(end-start))392        print(addHtmlEnd(results),file=f_html)393        f_html.close()...

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.