KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use eval_thread method in fMBT

Best Python code snippet using fMBT_python

evaluation.py

Source:evaluation.py

1import re2import time3from typing import List4from threading import Thread5from lxml import etree6import PySimpleGUI as sg7from . import s8def get_eval_token() -> str:9    index_resp = s.get('/student/teachingEvaluation/evaluation/index')10    if (token_regex := re.search(r'name="tokenValue" value="(.*?)"', index_resp.text)) is None:11        raise Exception('è¯·åç»å½')12    token = token_regex.group(1)13    return token14def get_eval_list() -> List[dict]:15    resp = s.post(16        '/student/teachingEvaluation/teachingEvaluation/search',17        data={18            'optType': 1,19            'pageSize': 5020        })21    return resp.json()['data']22def get_eval_form(23    evaluatedPeople: str,24    evaluatedPeopleNumber: str,25    questionnaireCode: str,26    questionnaireName: str,27    coureSequenceNumber: str,28    evaluationContentNumber: str,29    tokenValue: str30) -> dict:31    evaluation_page = s.post(32        '/student/teachingEvaluation/teachingEvaluation/evaluationPage',33        data={34            'count': '',35            'evaluatedPeople': evaluatedPeople,36            'evaluatedPeopleNumber': evaluatedPeopleNumber,37            'questionnaireCode': questionnaireCode,38            'questionnaireName': questionnaireName,39            'coureSequenceNumber': coureSequenceNumber,40            'evaluationContentNumber': evaluationContentNumber,41            'evaluationContentContent': '',42            'tokenValue': tokenValue43        }44    )45    base_data = {46        'optType': 'submit',47        'tokenValue': tokenValue,48        'questionnaireCode': questionnaireCode,49        'evaluationContent': evaluationContentNumber,50        'evaluatedPeopleNumber': evaluatedPeopleNumber,51        'count': '',52        'zgpj': 'èå¸ä¸çè¯¾å¾å¥½'53    }54    html = etree.HTML(evaluation_page.text)55    options = html.xpath('//tr/td/div[1]/label/input')56    if (len(options) == 0):57        raise Exception('è¯¥é®å·æ²¡æéé¡¹')58    for option in options:59        base_data[option.attrib['name']] = option.attrib['value']60    61    return base_data62def submit_eval(form: dict) -> bool:63    assessment = s.post(64        '/student/teachingEvaluation/teachingEvaluation/assessment',65        data=form66    )67    return assessment.json()['result'] == 'success'68data = []69layout = [70    [sg.Table(71        data,72        headings=[73            'è¯¾ç¨å·', # id.evaluationContentNumber74            'è¯¾åºå·', # id.coureSequenceNumber75            'è¯¾ç¨å', # evaluationContent76            'æå¸', # evaluatedPeople77            'æ¯å¦å·²è¯æ', # isEvaluated78        ],79        auto_size_columns=False,80        col_widths=[81            8,82            5,83            30,84            8,85            1086        ],87        justification='center',88        num_rows=10,89        key="table"90    )],91    [92        sg.Button('å¼å§', size=(5, 1)), 93        sg.ProgressBar(0, key='progress_bar', size=(30, 30)),94        sg.Text('æªå¼å§', key='evaluating_name', size=(30, 1))95    ]96]97window = sg.Window('è¯æåè¡¨', layout, font=("å¾®è½¯éé»", 12))98def show_evaluation_window():99    window.finalize()100    101    try:102        elist = get_eval_list()103        data = [104            [105                e['id']['evaluationContentNumber'],106                e['id']['coureSequenceNumber'],107                e['evaluationContent'],108                e['evaluatedPeople'],109                e['isEvaluated'],110            ]111            for e in elist112        ]113        window['table'].update(data)114    except Exception as e:115        sg.popup_error_with_traceback('è·åè¯æåè¡¨å¤±è´¥', e)116        return117    eval_thread = None118    thread_stop = False119    120    def _start_evaluation(elist: List[dict], window: sg.Window):121        global eval_thread122        error_msg = ''123        try:124            token = get_eval_token()125        except Exception as e:126            error_msg = 'è·åtokenå¤±è´¥'127            window['å¼å§'].update(disabled=False)128            sg.popup_error_with_traceback(error_msg, e)129            window['evaluating_name'].update(error_msg)130            eval_thread = None131            return132        for i, e in enumerate(elist):133            window['evaluating_name'].update(e['evaluationContent'])134            window['progress_bar'].update(i + 1)135            if e['isEvaluated'] == 'æ¯':136                continue137            try:138                form = get_eval_form(139                    evaluatedPeople=e['evaluatedPeople'],140                    evaluatedPeopleNumber=e['id']['evaluatedPeople'],141                    questionnaireCode=e['questionnaire']['questionnaireNumber'],142                    questionnaireName=e['questionnaire']['questionnaireName'],143                    coureSequenceNumber=e['id']['coureSequenceNumber'],144                    evaluationContentNumber=e['id']['evaluationContentNumber'],145                    tokenValue=token146                )147            except Exception as e:148                error_msg = 'è·åè¯æè¡¨åå¤±è´¥'149                window['å¼å§'].update(disabled=False)150                sg.popup_error_with_traceback(error_msg, e)151                window['evaluating_name'].update(error_msg)152                eval_thread = None153                return154            sec = 121155            while sec > 0 and not thread_stop:156                sec -= 1157                window['evaluating_name'].update(e['evaluationContent'] + f'(çå¾: {sec}s)')158                time.sleep(1)159            if thread_stop:160                return 161            try:162                result = submit_eval(form)163            except Exception as e:164                error_msg = 'æäº¤è¯æå¤±è´¥'165                window['å¼å§'].update(disabled=False)166                sg.popup_error_with_traceback(error_msg, e)167                window['evaluating_name'].update(error_msg)168                eval_thread = None169                return170            if result:171                data[i][4] = 'æ¯'172                window['table'].update(data)173            else:174                error_msg = f'{e["evaluationContent"]} è¯æå¤±è´¥, è¯æå·²åæ¢'175                sg.popup_error(error_msg)176                window['evaluating_name'].update(error_msg)177                window['å¼å§'].update(disabled=False)178                eval_thread = None179                return180        181        window['evaluating_name'].update('å·²å®æ')182        window['å¼å§'].update(disabled=False)183        eval_thread = None184    while True:185        event, values = window.read()186        if event == sg.WIN_CLOSED:187            break188        if event == 'å¼å§':189            if eval_thread is not None:190                sg.popup('è¯æå·²å¨è¿è¡')191                continue192            eval_thread = Thread(target=_start_evaluation, args=(elist, window))193            eval_thread.start()194            window['å¼å§'].update(disabled=True)195    if eval_thread is not None:196        thread_stop = True197        eval_thread.join()...

evaluator_ros.py

Source:evaluator_ros.py

1#!/usr/bin/env python2from logging import INFO3import sys4import rospy5import rosnode6import rosgraph7import re8import time9import datetime10import glog11import os12import evaluator as evaluator13import eval_plotting as plotting14ID = '/rosnode'15class Evaluator:16    def __init__(self):17        self.node_eval_threads = {}18        self.topic_eval_threads = {}19        self.eval_rate_s = rospy.get_param('~eval_rate_s', default=0.5)20        self.node_names = rospy.get_param('~node_names')21        for i in range(0, len(self.node_names)):22            self.node_names[i] = rosgraph.names.script_resolve_name(23                '/', self.node_names[i])24        node_eval_mode = rospy.get_param('~node_eval_mode')25        if self.node_names is not None or node_eval_mode is not None:26            glog.check_eq(len(self.node_names), len(node_eval_mode))27            self.eval_mode = {}28            for name, mode in zip(self.node_names, node_eval_mode):29                self.eval_mode[name] = mode30            for node_name in self.node_names:31                self.node_eval_threads[node_name] = {}32        self.plot_dir = os.path.join(rospy.get_param('~plot_dir', '.'), datetime.datetime.now().strftime(33            '%x').replace('/', '-')+'-'+datetime.datetime.now().strftime('%X').replace(':', '-'))34        if not os.path.exists(self.plot_dir):35            os.mkdir(self.plot_dir)36        print("Saving results to "+self.plot_dir)37        self.topic_names = rospy.get_param('~topic_names', default=None)38        topic_eval_mode = rospy.get_param('~topic_eval_mode', default=None)39        if self.topic_names is not None or self.topic_names is not None:40            glog.check_eq(len(self.topic_names), len(topic_eval_mode))41            self.topic_eval_mode = {}42            for name, mode in zip(self.topic_names, topic_eval_mode):43                self.topic_eval_mode[name] = mode44            for topic in self.topic_names:45                self.topic_eval_threads[topic] = {}46        self.sys_eval_mode = rospy.get_param('~sys_eval_mode', default=None)47        self.sys_eval_threads = {}48        self.master = rosgraph.Master(ID)49        self.node_pid = {}50        self.plot_threads = {}51        self.start_eval()52    def start_eval(self):53        rate = rospy.Rate(2)54        while self.node_names is not None and not rospy.is_shutdown():55            rate.sleep()56            all_node_names = rosnode.get_node_names()57            for node_name in self.node_names:58                # check if node is running59                if node_name not in all_node_names:60                    rospy.logwarn('Node %s is not running' % node_name)61                    continue62                if node_name in self.node_pid:63                    continue64                rospy.loginfo('Looking for pid of node %s' % node_name)65                node_api = rosnode.get_api_uri(self.master, node_name)66                while True:67                    try:68                        node_con_info = rosnode.get_node_connection_info_description(69                            node_api, self.master)70                    except rosnode.ROSNodeIOException as e:71                        time.sleep(0.1)72                        rospy.loginfo_throttle(1, e)73                        continue74                    else:75                        break76                pid_match = re.search('Pid: (\d+)', node_con_info)77                if pid_match is None:78                    rospy.logwarn('Not found pid in description of node %s' %79                                  node_name)80                    continue81                self.node_pid[node_name] = int(pid_match.group(1))82                rospy.loginfo('Pid: %d' % self.node_pid[node_name])83            if len(self.node_pid) == len(self.node_names):84                break85        rospy.loginfo('Catched pid of every node, start evaluating')86        self._start_eval_threads()87        rospy.on_shutdown(self.stop_threads)88        self._plot_loop()89    def _start_eval_threads(self):90        if self.node_names is not None:91            for node_name in self.node_names:92                for eval_mode in self.eval_mode[node_name]:93                    eval_thread = evaluator.EvaluatorFactory.create_evaluator(94                        eval_mode,95                        node_name=node_name,96                        node_pid=self.node_pid[node_name],97                        eval_rate_s=self.eval_rate_s)98                    eval_thread.start()99                    self.node_eval_threads[node_name][eval_mode] = eval_thread100                    self._add_to_plotting(eval_mode, eval_thread)101        if self.topic_names is not None:102            for topic in self.topic_names:103                for eval_mode in self.topic_eval_mode[topic]:104                    eval_thread = evaluator.EvaluatorFactory.create_evaluator(105                        eval_mode,106                        topic=topic,107                        eval_rate_s=self.eval_rate_s)108                    eval_thread.start()109                    self.topic_eval_threads[topic][eval_mode] = eval_thread110                    self._add_to_plotting(eval_mode, eval_thread)111        if self.sys_eval_mode is not None:112            for eval_mode in self.sys_eval_mode:113                eval_thread = evaluator.EvaluatorFactory.create_evaluator(114                    eval_mode,115                    eval_rate_s=self.eval_rate_s)116                eval_thread.start()117                self.sys_eval_threads[eval_mode] = eval_thread118                self._add_to_plotting(eval_mode, eval_thread)119        for plot_mode in self.plot_threads:120            self.plot_threads[plot_mode].start()121    def _add_to_plotting(self, eval_mode, eval_thread):122        if eval_mode == 'sys_bw' or eval_mode == 'bw_from_msg':123            eval_mode = 'topic_bw'124        if eval_mode not in self.plot_threads:125            self.plot_threads[eval_mode] = plotting.PlottingFactory.create_plotting(126                eval_mode, plot_dir=self.plot_dir, plot_rate_s=1.0)127        self.plot_threads[eval_mode].add_stat_update_callback(128            eval_thread.get_eval_stat)129    def _plot_loop(self):130        start_time = time.time()131        while not rospy.is_shutdown():132            for plot_mode in self.plot_threads:133                self.plot_threads[plot_mode].plot()134                time_to_sleep = start_time + self.eval_rate_s - time.time()135                if time_to_sleep > 0:136                    time.sleep(time_to_sleep)137                start_time = time.time()138    def stop_threads(self):139        for node_name in self.node_names:140            for eval_mode in self.eval_mode[node_name]:141                self.node_eval_threads[node_name][eval_mode].stop()142        for topic in self.topic_eval_threads:143            for eval_mode in self.topic_eval_threads[topic]:144                self.topic_eval_threads[topic][eval_mode].stop()145        for key in self.plot_threads:146            self.plot_threads[key].stop()147if __name__ == "__main__":148    rospy.init_node('evaluator', anonymous=True)149    evaluator = Evaluator()...

main.py

Source:main.py

1# coding=utf-82import base643import os4import sys5import cv26from flask import Flask, render_template, request, session, redirect, jsonify7from werkzeug.utils import secure_filename8from web.db.mysqlpool import Mysql9from web.eval_thread import EvalThread10from web.login import login_api, getLastDate11reload(sys)12sys.setdefaultencoding('utf-8')13app = Flask(__name__, static_url_path='')14app.register_blueprint(login_api)15app.secret_key = 'vac_key'16class ThreadPool:17    def __init__(self):18        self.index = -119        self.threads = []20    def add(self, thread):21        self.threads.append(thread)22        self.index = self.index + 123    def get_thread(self):24        return self.threads[self.index]25thread_pool = ThreadPool()26mysql = Mysql()27@app.route('/')28def index():29    if len(session) == 0:30        name = "ç»å½"31    else:32        name = "Hello! " + session['username']33    return redirect('/index.html')34@app.route('/upload')35def upload():36    if 'username' in session:37        return redirect('/upload/upload.html')38    else:39        return redirect('/login')40@app.route('/uploader', methods=['POST'])41def upload_file():42    if 'username' in session:43        username = session['username']44        f = request.files['file']45        filename = secure_filename(f.filename)46        f.save(filename)47        abs_path = os.path.abspath('.') + "/"48        eval_thread = EvalThread()49        eval_thread.set_path(abs_path, filename)50        eval_thread.set_username(username)51        eval_thread.start()52        thread_pool.add(eval_thread)53        return redirect('/widget/analysis.html')54    else:55        return redirect('/login')56@app.route('/analysis')57def info():58    rst = thread_pool.get_thread().get_info()59    alive = thread_pool.get_thread().is_alive()60    detail = thread_pool.get_thread().get_info_detail()61    cv = thread_pool.get_thread().get_cv_output()62    code = False63    if len(cv) == 2:64        code = [cv2_base64(cv[0]), cv2_base64(cv[1])]65    t = {66        'alive': alive,67        'result': rst,68        'filename': thread_pool.get_thread().get_video_name(),69        'detail': detail,70        'summary': thread_pool.get_thread().analysis_result(),71        'code': code72    }73    return jsonify(t)74@app.route('/stop', methods=['post'])75def stop():76    thread_pool.get_thread().stop()77    return 'stop'78@app.route('/user')79def user():80    if len(session) == 0:81        return redirect('/login')82    return redirect('/widget/account.html')83def cv2_base64(image):84    base64_str = cv2.imencode('.jpg',image)[1].tostring()85    base64_str = base64.b64encode(base64_str)86    return base64_str87if __name__ == '__main__':...

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.