Best Python code snippet using avocado_python
ros_loader.py
Source:ros_loader.py
1import time2#!/usr/bin/env python3# Software License Agreement (BSD License)4#5# Copyright (c) 2012, Willow Garage, Inc.6# All rights reserved.7#8# Redistribution and use in source and binary forms, with or without9# modification, are permitted provided that the following conditions10# are met:11#12# * Redistributions of source code must retain the above copyright13# notice, this list of conditions and the following disclaimer.14# * Redistributions in binary form must reproduce the above15# copyright notice, this list of conditions and the following16# disclaimer in the documentation and/or other materials provided17# with the distribution.18# * Neither the name of Willow Garage, Inc. nor the names of its19# contributors may be used to endorse or promote products derived20# from this software without specific prior written permission.21#22# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS23# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT24# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS25# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE26# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,27# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,28# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;29# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER30# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT31# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN32# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE33# POSSIBILITY OF SUCH DAMAGE.34import roslib35import rospy36from threading import Lock37""" ros_loader contains methods for dynamically loading ROS message classes at38runtime. It's achieved by using roslib to load the manifest files for the39package that the respective class is contained in.40Methods typically return the requested class or instance, or None if not found41"""42# Variable containing the loaded classes43_loaded_msgs = {}44_loaded_srvs = {}45_msgs_lock = Lock()46_srvs_lock = Lock()47_manifest_lock = Lock()48class InvalidTypeStringException(Exception):49 def __init__(self, typestring):50 Exception.__init__(self, "%s is not a valid type string" % typestring)51class InvalidPackageException(Exception):52 def __init__(self, package, original_exception):53 Exception.__init__(self,54 "Unable to load the manifest for package %s. Caused by: %s"55 % (package, str(original_exception))56 )57class InvalidModuleException(Exception):58 def __init__(self, modname, subname, original_exception):59 Exception.__init__(self,60 "Unable to import %s.%s from package %s. Caused by: %s"61 % (modname, subname, modname, str(original_exception))62 )63class InvalidClassException(Exception):64 def __init__(self, modname, subname, classname, original_exception):65 Exception.__init__(self,66 "Unable to import %s class %s from package %s. Caused by %s"67 % (subname, classname, modname, str(original_exception))68 )69def get_message_class(typestring):70 """ Loads the message type specified.71 Returns the loaded class, or throws exceptions on failure """72 return _get_msg_class(typestring)73def get_service_class(typestring):74 """ Loads the service type specified.75 Returns the loaded class, or None on failure """76 return _get_srv_class(typestring)77def get_message_instance(typestring):78 """ If not loaded, loads the specified type.79 Then returns an instance of it, or None. """80 cls = get_message_class(typestring)81 return cls()82def get_service_instance(typestring):83 """ If not loaded, loads the specified type.84 Then returns an instance of it, or None. """85 cls = get_service_class(typestring)86 return cls()87def get_service_request_instance(typestring):88 cls = get_service_class(typestring)89 return cls._request_class()90def get_service_response_instance(typestring):91 cls = get_service_class(typestring)92 return cls._response_class()93def _get_msg_class(typestring):94 """ If not loaded, loads the specified msg class then returns an instance95 of it96 Throws various exceptions if loading the msg class fails """97 global _loaded_msgs, _msgs_lock98 return _get_class(typestring, "msg", _loaded_msgs, _msgs_lock)99def _get_srv_class(typestring):100 """ If not loaded, loads the specified srv class then returns an instance101 of it102 Throws various exceptions if loading the srv class fails """103 global _loaded_srvs, _srvs_lock104 return _get_class(typestring, "srv", _loaded_srvs, _srvs_lock)105def _get_class(typestring, subname, cache, lock):106 """ If not loaded, loads the specified class then returns an instance107 of it.108 Loaded classes are cached in the provided cache dict109 Throws various exceptions if loading the msg class fails """110 # First, see if we have this type string cached111 cls = _get_from_cache(cache, lock, typestring)112 if cls is not None:113 return cls114 # Now normalise the typestring115 modname, classname = _splittype(typestring)116 norm_typestring = modname + "/" + classname117 # Check to see if the normalised type string is cached118 cls = _get_from_cache(cache, lock, norm_typestring)119 if cls is not None:120 return cls121 # Load the class122 cls = _load_class(modname, subname, classname)123 # Cache the class for both the regular and normalised typestring124 _add_to_cache(cache, lock, typestring, cls)125 _add_to_cache(cache, lock, norm_typestring, cls)126 return cls127def _load_class(modname, subname, classname):128 """ Loads the manifest and imports the module that contains the specified129 type.130 Logic is similar to that of roslib.message.get_message_class, but we want131 more expressive exceptions.132 Returns the loaded module, or None on failure """133 global loaded_modules134 try:135 with _manifest_lock:136 # roslib maintains a cache of loaded manifests, so no need to duplicate137 roslib.launcher.load_manifest(modname)138 except Exception as exc:139 raise InvalidPackageException(modname, exc)140 try:141 pypkg = __import__('%s.%s' % (modname, subname))142 except Exception as exc:143 raise InvalidModuleException(modname, subname, exc)144 try:145 return getattr(getattr(pypkg, subname), classname)146 except Exception as exc:147 raise InvalidClassException(modname, subname, classname, exc)148def _splittype(typestring):149 """ Split the string the / delimiter and strip out empty strings150 Performs similar logic to roslib.names.package_resource_name but is a bit151 more forgiving about excess slashes152 """153 splits = [x for x in typestring.split("/") if x]154 if len(splits) == 2:155 return splits156 raise InvalidTypeStringException(typestring)157def _add_to_cache(cache, lock, key, value):158 lock.acquire()159 cache[key] = value160 lock.release()161def _get_from_cache(cache, lock, key):162 """ Returns the value for the specified key from the cache.163 Locks the lock before doing anything. Returns None if key not in cache """164 lock.acquire()165 ret = None166 if key in cache:167 ret = cache[key]168 lock.release()...
updater.py
Source:updater.py
1#this is a ghetto method to update a patient by submitting an xform directly via.2#<Phone1>857-334-1329</Phone1>3#<Phone1Type>Terae (daughter)</Phone1Type>4#5#<Phone2/>6#7#<Phone3>857-249-3684</Phone3>8#<Phone3Type>client's cell (CURRENTLY OUT OF SERVICE)</Phone3Type>9#10#<address1>11 East Main St. Mattapan, MA 02126</address1>11#<address1type>Mother's address</address1type>12#13#<address2>50 Blue Hill Ave. 3rd fl. Roxbury, 0</address2>14#<address2type>CURRENT ADDRESS</address2type>15from datetime import datetime16import re17import uuid18from casexml.apps.case.models import CommCareCase19xml_template = """20<data xmlns:jrm="http://dev.commcarehq.org/jr/xforms" xmlns="http://dev.commcarehq.org/pact/patientupdate" version="3" uiVersion="3">21 <Meta>22 <DeviceID>PatientUpdateForm</DeviceID>23 <TimeStart>%(time_start)s</TimeStart>24 <TimeEnd>%(time_end)s</TimeEnd>25 <username>%(username)s</username>26 <chw_id>%(chw_id)s</chw_id>27 <uid>%(uid)s</uid>28 </Meta>29 <pact_id>%(pact_id)s</pact_id>30 <case>31 <case_id>%(case_id)s</case_id>32 <date_modified>%(date_modified)s</date_modified>33 <update>34 %(update_block)s35 </update>36 </case>37</data>38"""39phone_template = """40<Phone%(num)s>%(number)s</Phone%(num)s>41<Phone%(num)sType>%(typestring)s</Phone%(num)sType>42"""43address_template_full = """44<address%(num)s>%(street)s %(city)s, %(state)s %(postal_code)s</address%(num)s>45<address%(num)stype>%(typestring)s</address%(num)stype>46"""47address_template = """48<address%(num)s>%(address)s</address%(num)s>49<address%(num)stype>%(typestring)s</address%(num)stype>50"""51def generate_submission_from_cpatient(couchdoc):52 """53 This is the method to first push an unconverted patient54 """55 pass56def update_patient_casexml(user, case_id, pact_id, active_phones, active_addresses):57 """58 Update casexml59 """60 data_dict = {}61 data_dict['time_start'] = datetime.utcnow()#.isoformat()62 filtered_phones = filter(lambda x: len(x['number']) > 0 and len(x['description']) >= 0, active_phones)63 phone_xml = []64 #for i, p in enumerate(active_phones, start=1):65 for i in range(0,5):66 if i < len(filtered_phones):67 p = filtered_phones[i]68 else:69 p = {'number':'', 'description':''}70 phone_xml.append(get_phone_xml(i+1, p['number'], typestring=p['description']))71 address_xml = []72 filtered_addresses = filter(lambda x: len(x['address']) > 0 and len(x['description']) >= 0, active_addresses)73 #for i, a in enumerate(active_addresses, start=1):74 for i in range(0,5):75 if i < len(filtered_addresses):76 a = filtered_addresses[i]77 else:78 a = {'address': '', 'description': ''}79 address_xml.append(get_address_xml(i+1, a['address'], typestring=a['description']))80 #print phone_xml81 #print address_xml82 #case = CommCareCase.get(patient_doc.case_id)83 data_dict['update_block'] = ''.join(phone_xml + address_xml)84 data_dict['uid'] = uuid.uuid4().hex85 data_dict['case_id'] = case_id86 data_dict['username'] = user.username87 data_dict['chw_id'] = user.id88 data_dict['date_modified'] = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.000Z") # this is != to isoformat()!!!89 data_dict['pact_id'] = pact_id90 data_dict['time_end'] = datetime.utcnow() #datetime.utcnow().isoformat() + "Z"91 return xml_template % data_dict92def generate_update_xml_old(user, patient_doc, cphones, caddresses):93 """94 For a given patient and in memory phone and address objects, generate the xml.95 This is using the old cphone and cpatient models. This is a migration method.96 """97 data_dict = {}98 data_dict['time_start'] = datetime.utcnow()#.isoformat()99 phone_xml = []100 for i, p in enumerate(cphones, start=1):101 if p == None:102 continue103 phone_xml.append(get_phone_xml(i, p.number, typestring=p.description))104 address_xml = []105 for i, a in enumerate(caddresses, start=1):106 if a == None:107 continue108 address_xml.append(get_address_xml_full(i, a.street, a.city, a.state, a.postal_code, typestring=a.description))109 case = CommCareCase.get(patient_doc.case_id)110 data_dict['update_block'] = ''.join(phone_xml + address_xml)111 data_dict['uid'] = uuid.uuid4().hex112 data_dict['case_id'] = case._id113 data_dict['username'] = user.username114 data_dict['chw_id'] = user.id115 data_dict['date_modified'] = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.000Z") # this is != to isoformat()!!!116 data_dict['pact_id'] = patient_doc.pact_id117 data_dict['time_end'] = datetime.utcnow() #datetime.utcnow().isoformat() + "Z"118 return xml_template % data_dict119def get_phone_xml(n, phone, typestring=None):120 """121 Generage a singular casexml phone block.122 123 """124 dict = {'num': n, 'number': phone,125 'typestring': typestring if typestring is not None or len(typestring) > 0 else ''}126 return phone_template % dict127def get_address_xml(n, address, typestring=None):128 if typestring == None:129 typestring = ''130 dict = { 'num': n, 'address': address,131 'typestring': typestring if typestring is not None or len(typestring) > 0 else ''}132 return address_template % dict133def get_address_xml_full(n, street, city, state, postal_code, typestring=None):134 dict = {'num': n, 'street': street, 'city': city, 'state': state, 'postal_code': postal_code,135 'typestring': typestring if typestring is not None or len(typestring) > 0 else ''}136 return address_template_full % dict137def get_blank_phone_xml(n):138 return "<Phone%(num)s/><Phone%(num)sType/>" % {'num': n}139def get_blank_address_xml(n):...
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!!