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

How to use typestring method in avocado

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):...

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.