Best Python code snippet using locust
couchpotato.py
Source:couchpotato.py  
...48        print data49    if use_json:50        data = json.JSONDecoder().decode(data)51    return data52def log_exception(e):53    logger.log('CouchPotato :: EXCEPTION -- %s' % e, 'DEBUG')54def couchpotato_image(path):55    if path.startswith('/'):56        path = path[1:]57    return '%s/xhr/couchpotato/image/%s' % (WEBROOT, path)58FILTERS['cp_img'] = couchpotato_image59@app.route('/xhr/couchpotato/image/<path:url>')60def couchpotato_proxy(url):61    username = get_setting_value('couchpotato_user')62    password = get_setting_value('couchpotato_password')63    url = '%s/file.cache/%s' % (couchpotato_url(), url)64    req = urllib2.Request(url)65    if username and password:66        base64string = base64.encodestring('%s:%s' % (username, password)).replace('\n', '')67        req.add_header("Authorization", "Basic %s" % base64string)68    img = StringIO.StringIO(urllib2.urlopen(req).read())69    logger.log('CouchPotato :: Fetching image from %s' % (url), 'DEBUG')70    return send_file(img, mimetype='image/jpeg')71@app.route('/xhr/couchpotato/')72@app.route('/xhr/couchpotato/<status>/')73def xhr_couchpotato(status=False):74    status_string = 'status=active'75    template = 'couchpotato.html'76    if status:77        status_string = 'status=%s' % status78        template = 'couchpotato/all.html'79    try:80        logger.log('CouchPotato :: Fetching "%s movies" list' % status, 'INFO')81        couchpotato = couchpotato_api('movie.list', params=status_string)82        if couchpotato['success'] and not couchpotato['empty']:83            couchpotato = couchpotato['movies']84    except Exception as e:85        log_exception(e)86        couchpotato = None87    logger.log('CouchPotato :: Fetching "%s movies" list (DONE)' % status, 'INFO')88    if status == 'wanted' and not type(couchpotato) is list:89        logger.log('CouchPotato :: Wanted movies list is empty', 'INFO')90        return cp_search('There are no movies in your wanted list.')91    return render_template(template,92        url=couchpotato_url(),93        couchpotato=couchpotato,94        compact_view=get_setting_value('couchpotato_compact') == '1',95    )96@app.route('/xhr/couchpotato/history/')97def xhr_couchpotato_history():98    unread = 099    try:100        couchpotato = couchpotato_api('notification.list')101        couchpotato = couchpotato['notifications']102        for notification in couchpotato:103            if not notification['read']:104                unread = unread + 1105    except Exception as e:106        logger.log('CouchPotato :: Could not retrieve Couchpotato - %s' % (e), 'WARNING')107        couchpotato = None108    return render_template('couchpotato/history.html',109        couchpotato=couchpotato,110        unread=unread,111    )112@app.route('/xhr/couchpotato/search/')113def cp_search(message=None):114    couchpotato = {}115    params = False116    profiles = {}117    try:118        query = request.args['name']119        params = 'q=' + query120    except:121        pass122    if params:123        try:124            logger.log('CouchPotato :: Searching for movie: %s' % (query), 'INFO')125            couchpotato = couchpotato_api('movie.search', params=params)126            amount = len(couchpotato['movies'])127            logger.log('CouchPotato :: found %i movies for %s' % (amount, query), 'INFO')128            if couchpotato['success'] and amount != 0:129                couchpotato = couchpotato['movies']130                try:131                    # logger.log('CouchPotato :: Getting quality profiles', 'INFO')132                    profiles = couchpotato_api('profile.list')133                except Exception as e:134                    log_exception(e)135            else:136                return render_template('couchpotato/search.html', error='No movies with "%s" were found' % (query), couchpotato='results')137        except Exception as e:138            log_exception(e)139            couchpotato = None140    else:141        logger.log('CouchPotato :: Loading search template', 'DEBUG')142        couchpotato = None143    return render_template('couchpotato/search.html',144        data=couchpotato,145        couchpotato='results',146        profiles=profiles,147        error=message148    )149@app.route('/xhr/couchpotato/add_movie/<imdbid>/<title>/')150@app.route('/xhr/couchpotato/add_movie/<imdbid>/<title>/<profile>/')151def add_movie(imdbid, title, profile=False):152    if profile:153        params = 'identifier=%s&title=%s&profile_id=%s' % (imdbid, title, profile)154    else:155        params = 'identifier=%s&title=%s' % (imdbid, title)156    try:157        logger.log('CouchPotato :: Adding %s (%s) to wanted list' % (title, imdbid), 'INFO')158        result = couchpotato_api('movie.add', params)159        return jsonify(result)160    except Exception as e:161        log_exception(e)162    return jsonify({'success': False})163@app.route('/xhr/couchpotato/restart/')164@requires_auth165def cp_restart():166    try:167        logger.log('CouchPotato :: Restarting', 'INFO')168        result = couchpotato_api('app.restart', use_json=False)169        if 'restarting' in result:170            return jsonify({'success': True})171    except Exception as e:172        log_exception(e)173    return jsonify({'success': False})174@app.route('/xhr/couchpotato/available/')175@requires_auth176def cp_available():177    try:178        logger.log('CouchPotato :: Checking if CouchPotato is available', 'INFO')179        result = couchpotato_api('app.available')180        return jsonify(result)181    except Exception as e:182        log_exception(e)183    return jsonify({'success': False})184@app.route('/xhr/couchpotato/shutdown/')185@requires_auth186def cp_shutdown():187    try:188        logger.log('CouchPotato :: Shutting down', 'INFO')189        result = couchpotato_api('app.shutdown', use_json=False)190        if 'shutdown' in result:191            return jsonify({'success': True})192    except Exception as e:193        log_exception(e)194    return jsonify({'success': False})195@app.route('/xhr/couchpotato/version/')196@requires_auth197def cp_version():198    try:199        result = couchpotato_api('app.version')200        return jsonify(result)201    except Exception as e:202        log_exception(e)203    return jsonify({'success': False})204@app.route('/xhr/couchpotato/profiles/')205@requires_auth206def cp_profiles():207    try:208        logger.log('CouchPotato :: Getting profiles', 'INFO')209        result = couchpotato_api('profile.list')210        return jsonify(result)211    except Exception as e:212        log_exception(e)213    return jsonify({'success': False})214@app.route('/xhr/couchpotato/quality/')215@requires_auth216def cp_quality():217    try:218        logger.log('CouchPotato :: Getting quality', 'INFO')219        result = couchpotato_api('quality.list')220        return jsonify(result)221    except Exception as e:222        log_exception(e)223    return jsonify({'success': False})224@app.route('/xhr/couchpotato/update/check/')225@requires_auth226def cp_update_check():227    try:228        logger.log('CouchPotato :: Getting update', 'INFO')229        result = couchpotato_api('updater.check')230        return jsonify(result)231    except Exception as e:232        log_exception(e)233    return jsonify({'success': False})234@app.route('/xhr/couchpotato/delete_movie/<id>/')235@requires_auth236def movie_delete(id):237    """238    Delete a movie from list239    ----- Params -----240    id                  int (comma separated)                       Movie ID(s) you want to delete.241    delete_from         string: all (default), wanted, manage       Delete movie from this page242    """243    try:244        logger.log('CouchPotato :: Deleting movie %s' % id, 'INFO')245        result = couchpotato_api('movie.delete', 'id=%s' % id)246        return jsonify(result)247    except Exception as e:248        log_exception(e)249    return jsonify({'success': False})250@app.route('/xhr/couchpotato/refresh_movie/<id>/')251def movie_refresh(id):252    """253    Refresh a movie from list254    ----- Params -----255    id                  int (comma separated)                       Movie ID(s) you want to refresh.256    """257    try:258        logger.log('CouchPotato :: Refreshing movie %s' % id, 'INFO')259        result = couchpotato_api('movie.refresh', 'id=%s' % id)260        return jsonify(result)261    except Exception as e:262        log_exception(e)263    return jsonify({'success': False})264@app.route('/xhr/couchpotato/settings/')265def cp_settings():266    """267    Retrieve settings from CP268    """269    try:270        logger.log('CouchPotato :: Retrieving settings', 'INFO')271        result = couchpotato_api('settings')272        logger.log('CouchPotato :: Retrieving settings (DONE)', 'INFO')273        return render_template('couchpotato/settings.html',274            couchpotato=result,275        )276    except Exception as e:277        log_exception(e)278    return jsonify({'success': False})279@app.route('/xhr/couchpotato/get_movie/<id>/')280def cp_get_movie(id):281    """282    Retrieve movie from CP283    ---- Params -----284    id                  int (comma separated)                       The id of the movie285    """286    try:287        logger.log('CouchPotato :: Retrieving movie info', 'INFO')288        result = couchpotato_api('movie.get', 'id=%s' % id)289        try:290            logger.log('CouchPotato :: Getting quality profiles', 'INFO')291            profiles = couchpotato_api('profile.list')292        except Exception as e:293            log_exception(e)294        logger.log('CouchPotato :: Retrieving movie info (DONE)', 'INFO')295        return render_template('couchpotato/info.html',296            couchpotato=result,297            profiles=profiles,298        )299    except Exception as e:300        log_exception(e)301    return jsonify({'success': False})302@app.route('/xhr/couchpotato/edit_movie/<movieid>/<profileid>/')303def cp_edit_movie(movieid, profileid):304    """305    Edit movie in CP306    ---- Params -----307    movieid                  int (comma separated)                       The id of the movie308    profileid                int                                         Id of the profile to go to309    """310    try:311        logger.log('CouchPotato :: Retrieving movie info', 'INFO')312        result = couchpotato_api('movie.edit', 'id=%s&profile_id=%s' % (movieid, profileid))313        if result['success']:314            logger.log('CouchPotato :: Retrieving movie info (DONE)', 'INFO')315            return jsonify({'success': True})316    except Exception as e:317        log_exception(e)318    return jsonify({'success': False})319@app.route('/xhr/couchpotato/log/')320@app.route('/xhr/couchpotato/log/<type>/<lines>/')321def cp_log(type='all', lines=30):322    """323    Edit movie in CP324    ---- Params -----325    type <optional>          all, error, info, debug                     Type of log326    lines <optional>         int                                         Number of lines - last to first327    """328    try:329        logger.log('CouchPotato :: Retrieving "%s" log' % type, 'INFO')330        result = couchpotato_api('logging.partial', 'type=%s&lines=%s' % (type, lines))331        if result['success']:332            logger.log('CouchPotato :: Retrieving "%s" log (DONE)' % type, 'INFO')333            return render_template('couchpotato/log.html',334                couchpotato=result,335                level=type,336            )337    except Exception as e:338        log_exception(e)339    return jsonify({'success': False})340@app.route('/xhr/couchpotato/notification/read/')341@app.route('/xhr/couchpotato/notification/read/<int:id>/')342def cp_notification_read(id=False):343    """344    Mark notification as read in CP345    ---- Params -----346    ids <optional>      int         Notification id - if empty will mark all notifications347    """348    try:349        logger.log('CouchPotato :: Marking notification "%i" as read' % id, 'INFO')350        if id:351            couchpotato_api('notification.markread', 'ids=%i' % id)352        else:353            couchpotato_api('notification.markread')354        return jsonify({'success': True})355    except Exception as e:356        log_exception(e)357    return jsonify({'success': False})358@app.route('/xhr/couchpotato/release/<action>/<id>/')359@requires_auth360def release_action(action, id):361    if id.isdigit():362        id = int(id)363    try:364        logger.log('CouchPotato :: %sing release %s' % (action.title()[:-1], id), 'INFO')365        result = couchpotato_api('release.%s' % action, 'id=%s' % id)366        return jsonify(result)367    except Exception as e:368        log_exception(e)...robot.py
Source:robot.py  
...19            str(sys.exc_info()[0]),20            str(sys.exc_info()[1])21        )22        print(full_msg, file=sys.stderr)23def log_exception(src, locstr):24    # i.e. caught {ValueError} {in my_method}: {could not cast X to Y}25    log(src, "Caught {} {}: {}".format(26        str(sys.exc_info()[0]), locstr, str(sys.exc_info()[1])27    ))28class Robot(wpilib.IterativeRobot):29    def robotInit(self):30        constants.load_control_config()31        wpilib.CameraServer.launch('driver_vision.py:main')32        self.autoPositionSelect = wpilib.SendableChooser()33        self.autoPositionSelect.addDefault('Middle-Baseline', 'Middle-Baseline')34        self.autoPositionSelect.addObject('Middle-Placement', 'Middle-Placement')  # noqa: E50135        self.autoPositionSelect.addObject('Left', 'Left')36        self.autoPositionSelect.addObject('Right', 'Right')37        wpilib.SmartDashboard.putData(38            'Robot Starting Position',39            self.autoPositionSelect)40        self.drivetrain = swerve.SwerveDrive(41            constants.chassis_length,42            constants.chassis_width,43            constants.swerve_config44        )45        self.drivetrain.load_config_values()46        self.lift = lift.ManualControlLift(47            constants.lift_ids['left'],48            constants.lift_ids['right'],49            constants.lift_limit_channel,50            constants.start_limit_channel51        )52        self.winch = winch.Winch(53            constants.winch_id54        )55        self.throttle = wpilib.Joystick(1)56        self.claw = lift.Claw(57            constants.claw_id,58            constants.claw_follower_id59        )60        self.imu = IMU(wpilib.SPI.Port.kMXP)61        self.sd_update_timer = wpilib.Timer()62        self.sd_update_timer.reset()63        self.sd_update_timer.start()64    def disabledInit(self):65        pass66    def disabledPeriodic(self):67        try:68            self.lift.load_config_values()69            self.drivetrain.load_config_values()70        except:  # noqa: E77271            log_exception('disabled', 'when loading config')72        try:73            self.drivetrain.update_smart_dashboard()74            self.imu.update_smart_dashboard()75            self.lift.update_smart_dashboard()76            self.winch.update_smart_dashboard()77            wpilib.SmartDashboard.putNumber(78                "Throttle Pos", self.throttle.getRawAxis(constants.liftAxis)79            )80        except:  # noqa: E77281            log_exception('disabled', 'when updating SmartDashboard')82        try:83            self.lift.checkLimitSwitch()84            pass85        except:  # noqa: E77286            log_exception('disabled', 'when checking lift limit switch')87        self.drivetrain.update_smart_dashboard()88    def autonomousInit(self):89        try:90            self.drivetrain.load_config_values()91            self.lift.load_config_values()92        except:  # noqa: E77293            log_exception('auto-init', 'when loading config')94        self.autoPos = None95        try:96            self.autoPos = self.autoPositionSelect.getSelected()97        except:  # noqa: E77298            self.autoPos = None99            log_exception('auto-init', 'when getting robot start position')100        try:101            if self.autoPos is not None and self.autoPos != 'None':102                self.auto = Autonomous(self, self.autoPos)103            else:104                log('auto-init', 'Disabling autonomous...')105        except:  # noqa: E772106            log_exception('auto-init', 'in Autonomous constructor')107        try:108            self.lift.checkLimitSwitch()109            pass110        except:  # noqa: E772111            log_exception('auto-init', 'when checking lift limit switch')112    def autonomousPeriodic(self):113        try:114            if self.sd_update_timer.hasPeriodPassed(0.5):115                self.auto.update_smart_dashboard()116                self.imu.update_smart_dashboard()117                self.drivetrain.update_smart_dashboard()118                self.lift.update_smart_dashboard()119                self.winch.update_smart_dashboard()120        except:  # noqa: E772121            log_exception('auto', 'when updating SmartDashboard')122        try:123            if self.autoPos is not None and self.autoPos != 'None':124                self.auto.periodic()125        except:  # noqa: E772126            # Stop everything.127            self.drivetrain.immediate_stop()128            self.lift.setLiftPower(0)129            self.claw.set_power(0)130            self.winch.stop()131            log_exception('auto', 'in auto :periodic()')132        try:133            self.lift.checkLimitSwitch()134            pass135        except:  # noqa: E772136            log_exception('auto', 'when checking lift limit switch')137    def teleopInit(self):138        try:139            self.teleop = Teleop(self)140        except:  # noqa: E772141            log_exception('teleop-init', 'in Teleop constructor')142        try:143            self.drivetrain.load_config_values()144            self.lift.load_config_values()145            constants.load_control_config()146        except:  # noqa: E772147            log_exception('teleop-init', 'when loading config')148        try:149            self.lift.checkLimitSwitch()150            pass151        except:  # noqa: E772152            log_exception('teleop-init', 'when checking lift limit switch')153    def teleopPeriodic(self):154        try:155            self.teleop.drive()156        except:  # noqa: E772157            log_exception('teleop', 'in drive control')158            self.drivetrain.immediate_stop()159        try:160            self.teleop.buttons()161        except:  # noqa: E772162            log_exception('teleop', 'in button handler')163        try:164            self.teleop.lift_control()165        except:  # noqa: E772166            log_exception('teleop', 'in lift_control')167            self.lift.setLiftPower(0)168        try:169            self.teleop.claw_control()170        except:  # noqa: E772171            log_exception('teleop', 'in claw_control')172            self.claw.set_power(0)173        try:174            self.teleop.winch_control()175        except:  # noqa: E772176            log_exception('teleop', 'in winch_control')177            self.winch.stop()178        try:179            self.lift.checkLimitSwitch()180            pass181        except:  # noqa: E772182            log_exception('teleop', 'in lift.checkLimitSwitch')183        if self.sd_update_timer.hasPeriodPassed(0.5):184            try:185                constants.load_control_config()186                self.drivetrain.load_config_values()187                self.lift.load_config_values()188            except:  # noqa: E772189                log_exception('teleop', 'when loading config')190            try:191                self.drivetrain.update_smart_dashboard()192                self.teleop.update_smart_dashboard()193                self.imu.update_smart_dashboard()194                self.lift.update_smart_dashboard()195                self.winch.update_smart_dashboard()196            except:  # noqa: E772197                log_exception('teleop', 'when updating SmartDashboard')198        # for module in self.drivetrain.modules:199        #     module.set_steer_angle(0)200if __name__ == "__main__":...HSAPI.py
Source:HSAPI.py  
1from Commons import *2from logger import Logger3from HSAController import *4import flask_restful5from flask_cors import CORS, cross_origin6from flask import Flask, request, jsonify7from flask_restful import Api8from werkzeug.utils import secure_filename9logger = Logger()10app = Flask(__name__)11api = Api(app)12app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER13app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 102414CORS(app)15class CustomApi(flask_restful.Api):16    def handle_error(self, e):17        user_exception, log_exception = logger.get_exception()18        logger.log(str(log_exception) + "" + str(user_exception), logger.ERROR)19        flask_restful.abort(str(e))20@app.errorhandler(500)21def internal_server_error(error):22    user_exception, log_exception = logger.get_exception()23    logger.log(str(log_exception) + "" + str(user_exception), logger.ERROR)24    print(user_exception)25    print(log_exception)26    return jsonify({'error': user_exception})27@app.errorhandler(Exception)28def unhandled_exception(e):29    user_exception, log_exception = logger.get_exception()30    logger.log(str(log_exception) + "" + str(user_exception), logger.ERROR)31    print(user_exception)32    print(log_exception)33    return jsonify({'error': user_exception})34@app.route('/uploader', methods=['GET', 'POST'])35def upload_file():36    logger.log('upload_file()')37    logger.log('reset data')38    if request.method == 'POST':39        try:40            user_file = request.files['file']41            print(user_file)42        except:43            user_exception, log_exception = logger.get_exception()44            logger.log(log_exception, logger.ERROR)45            return 'File not found'46        data_source_type = get_extension(user_file.filename)47        logger.log('upload file name: ' + str(user_file.filename))48        logger.log('data_source_type: ' + str(data_source_type))49        if user_file.filename == '':50            user_exception, log_exception = logger.get_exception()51            logger.log(log_exception, logger.ERROR)52            return 'Error: File not found'53        logger.log('Check if allowed file')54        if user_file and allowed_file_HSA(user_file.filename):55            filename = secure_filename(user_file.filename)56            file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)57            logger.log('saving user file in upload folder')58            user_file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))59            logger.log('user file path: ' + str(file_path))60            try:61                logger.log('Analyzing file')62                result = ExtractAnalyzeImageFromFile(file_path)63                print(result)64                logger.log('Analyzation successful')65                logger.log(result)66            except Exception as e:67                user_exception, log_exception = logger.get_exception()68                logger.log(str(log_exception) + "" + str(user_exception), logger.ERROR)69                return str(e)70            logger.log('file uploaded successfully')71            return result72        else:73            user_exception, log_exception = logger.get_exception()74            logger.log(str(log_exception) + "" + str(user_exception), logger.ERROR)75            logger.log('ERROR: Exception raised in upload_file()')76            return 'Unsupported File Format'77if __name__ == '__main__':...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
