KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use from_ method in Selene
Best Python code snippet using selene_python
labware_client.py
Source:labware_client.py 
1#!/usr/bin/env python32import asyncio3import time4import json5import uuid6import datetime7import sys8import collections9import copy10from labware_driver import LabwareDriver11from autobahn.asyncio import wamp, websocket12from autobahn.asyncio.wamp import ApplicationSession, ApplicationRunner 13class WampComponent(wamp.ApplicationSession):14 """WAMP application session for OTOne (Overrides protocol.ApplicationSession - WAMP endpoint session)15 """16 def onConnect(self):17 """Callback fired when the transport this session will run over has been established.18 """19 self.join(u"ot_realm")20 @asyncio.coroutine21 def onJoin(self, details):22 """Callback fired when WAMP session has been established.23 May return a Deferred/Future.24 Starts instatiation of robot objects by calling :meth:`otone_client.instantiate_objects`.25 """26 print(datetime.datetime.now(),' - labware_client : WampComponent.onJoin:')27 print('\tdetails: ',str(details))28 if not self.factory._myAppSession:29 self.factory._myAppSession = self30 try:31 self.factory._crossbar_connected = True32 except AttributeError:33 print('ERROR: factory does not have "crossbar_connected" attribute')34 def handshake(client_data):35 """ FACTORY STUB36 """37 print(datetime.datetime.now(),' - labware_client : WampComponent.handshake:')38 print('\n\targs: ',locals(),'\n')39 try:40 self.factory._handshake(client_data)41 except AttributeError:42 print('ERROR: factory does not have "_handshake" attribute')43 def dispatch_message(client_data):44 """ FACTORY STUB45 """46 print(datetime.datetime.now(),' - labware_client : WampComponent.dispatch_message:')47 print('\n\targs: ',locals(),'\n')48 try:49 self.factory._dispatch_message(client_data)50 except AttributeError:51 print('ERROR: factory does not have "_dispatch_message" attribute')52 yield from self.subscribe(handshake, 'com.opentrons.labware_handshake')53 yield from self.subscribe(dispatch_message, 'com.opentrons.labware')54 def onLeave(self, details):55 """Callback fired when WAMP session has been closed.56 :param details: Close information.57 """58 print('driver_client : WampComponent.onLeave:')59 print('\n\targs: ',locals(),'\n')60 if self.factory._myAppSession == self:61 self.factory._myAppSession = None62 try:63 self.disconnect()64 except:65 raise66 67 def onDisconnect(self):68 """Callback fired when underlying transport has been closed.69 """70 print(datetime.datetime.now(),' - labware_client : WampComponent.onDisconnect:')71 asyncio.get_event_loop().stop()72 try:73 self.factory._crossbar_connected = False74 except AttributeError:75 print('ERROR: outer does not have "crossbar_connected" attribute')76class LabwareClient():77 def __init__(self):78 print(datetime.datetime.now(),' - LabwareClient.__init__:')79 print('\n\targs: ',locals(),'\n')80 self.driver_dict = {}81 self.meta_dict = {82 'drivers' : lambda from_,session_id,name,param: self.drivers(from_,session_id,name,param),83 'add_driver' : lambda from_,session_id,name,param: self.add_driver(from_,session_id,name,param),84 'remove_driver' : lambda from_,session_id,name,param: self.remove_driver(from_,session_id,name,param),85 'callbacks' : lambda from_,session_id,name,param: self.callbacks(from_,session_id,name,param),86 'meta_callbacks' : lambda from_,session_id,name,param: self.meta_callbacks(from_,session_id,name,param),87 'set_meta_callback' : lambda from_,session_id,name,param: self.set_meta_callback(from_,session_id,name,param),88 'add_callback' : lambda from_,session_id,name,param: self.add_callback(from_,session_id,name,param),89 'remove_callback' : lambda from_,session_id,name,param: self.remove_callback(from_,session_id,name,param),90 'flow' : lambda from_,session_id,name,param: self.flow(from_,session_id,name,param),91 'clear_queue' : lambda from_,session_id,name,param: self.clear_queue(from_,session_id,name,param),92 'connect' : lambda from_,session_id,name,param: self.driver_connect(from_,session_id,name,param),93 'close' : lambda from_,session_id,name,param: self.driver_close(from_,session_id,name,param),94 'meta_commands' : lambda from_,session_id,name,param: self.meta_commands(from_,session_id,name,param)95 }96 self.in_dispatcher = {97 'command': lambda from_,session_id,data: self.send_command(from_,session_id,data),98 'meta': lambda from_,session_id,data: self.meta_command(from_,session_id,data)99 }100 self.topic = {101 'frontend' : 'com.opentrons.frontend',102 'driver' : 'com.opentrons.driver',103 'labware' : 'com.opentrons.labware',104 'bootstrapper' : 'com.opentrons.bootstrapper'105 }106 self.clients = {107 # uuid : 'com.opentrons.[uuid]'108 }109 self.max_clients = 4110 self.id = str(uuid.uuid4())111 self.session_factory = wamp.ApplicationSessionFactory()112 self.session_factory.session = WampComponent113 self.session_factory._myAppSession = None114 self.session_factory._crossbar_connected = False115 self.transport_factory = None116 self.transport = None117 self.protocol = None118 self.loop = asyncio.get_event_loop()119 # FUNCTIONS FROM SUBSCRIBER120 def dispatch_message(self, message):121 print(datetime.datetime.now(),' - LabwareClient.dispatch_message:')122 print('\n\targs: ',locals(),'\n')123 try:124 dictum = collections.OrderedDict(json.loads(message.strip(), object_pairs_hook=collections.OrderedDict))125 if 'type' in dictum and 'from' in dictum and 'sessionID' in dictum and 'data' in dictum:126 if dictum['type'] in self.in_dispatcher:127 # if self.client_check(dictum['from']):128 # opportunity to filter, not actually used129 self.in_dispatcher[dictum['type']](dictum['from'],dictum['sessionID'],dictum['data'])130 else:131 print(datetime.datetime.now(),' - ERROR:\n\r',sys.exc_info())132 print('type: ',dictum['type'])133 else:134 print(datetime.datetime.now(),' - ERROR:\n\r',sys.exc_info())135 except:136 print(datetime.datetime.now(),' - ERROR:\n\r',sys.exc_info())137 # FUNCTIONS FROM PUBLISHER138 def handshake(self, data):139 print(datetime.datetime.now(),' - LabwareClient.handshake:')140 print('\n\targs: ',locals(),'\n')141 142 data_dict = json.loads(data)143 if isinstance(data_dict, dict):144 if 'from' in data:145 print('* data has "from"')146 client_id = data_dict['from']147 print('client_id: ',client_id)148 if client_id in self.clients:149 print('* from is a client')150 if 'data' in data_dict:151 if 'message' in data_dict['data']:152 if 'extend' in data_dict['data']['message']:153 print('handshake called again on client ',client_id,'. We could have done something here to repopulate data')154 self.publish( client_id , client_id , client_id ,'handshake','labware','result','already_connected')155 if 'shake' in data_dict['data']['message']:156 self.publish_client_ids(client_id,client_id)157 else:158 print('* from is NOT a client')159 if len(self.clients) > self.max_clients:160 self.publish( 'frontend', '' , 'handshake' , '' , 'labware' , 'result' , 'fail' )161 else:162 if client_id != "":163 self.clients[client_id] = 'com.opentrons.'+client_id164 self.publish( 'frontend' , client_id , client_id , 'handshake', 'labware', 'result','success')165 else:166 self.gen_client_id()167 else:168 print('* data does NOT have "from"')169 self.gen_client_id()170 171 if 'get_ids' in data_dict:172 publish_client_ids('','')173 else:174 self.gen_client_id()175 def gen_client_id(self):176 print(datetime.datetime.now(),' - LabwareClient.gen_client_id:')177 print('\n\targs: ',locals(),'\n')178 ret_id = ''179 if len(self.clients) > self.max_clients:180 self.publish( 'frontend', '' , '' , 'handshake' , 'labware' , 'result' , 'fail' )181 else:182 client_id = str(uuid.uuid4())183 self.clients[client_id] = 'com.opentrons.'+client_id184 self.publish( 'frontend' , client_id , client_id , 'handshake' , 'labware' , 'result' , 'success' )185 ret_id = client_id186 return ret_id187 def client_check(self, id_):188 print(datetime.datetime.now(),' - LabwareClient.client_check:')189 print('\n\targs: ',locals(),'\n')190 if id_ in self.clients:191 return True192 else:193 return False194 def publish_client_ids(self, id_, session_id):195 print(datetime.datetime.now(),' - LabwareClient.publish_client_ids:')196 print('\n\targs: ',locals(),'\n')197 if id_ in self.clients:198 self.publish( id_ , id_ , session_id, 'handshake' , 'labware' , 'ids' , list(self.clients) )199 else:200 self.publish( 'frontend' , '' , session_id, 'handshake' , 'labware' , 'ids' , list(self.clients) )201 return list(self.clients)202 def publish(self,topic,to,session_id,type_,name,message,param):203 """204 """205 print(datetime.datetime.now(),' - LabwareClient.publish:')206 print('\n\targs: ',locals(),'\n')207 if self.session_factory is not None and topic is not None and type_ is not None:208 if name is None:209 name = 'None'210 if message is None:211 message = ''212 if param is None:213 param = ''214 if self.session_factory is not None:215 if self.session_factory._myAppSession is not None:216 time_string = str(datetime.datetime.now())217 msg = {'time':time_string,'type':type_,'to':to,'sessionID':session_id,'from':self.id,'data':{'name':name,'message':{message:param}}}218 try:219 if topic in self.topic:220 print('TOPIC: ',self.topic)221 print(datetime.datetime.now(),'url topic: ',self.topic.get(topic))222 self.session_factory._myAppSession.publish(self.topic.get(topic),json.dumps(msg))223 else:224 print('TO:',to)225 url_topic = 'com.opentrons.'+to226 print(datetime.datetime.now(),'url topic: ',url_topic)227 self.session_factory._myAppSession.publish(url_topic,json.dumps(msg))228 except:229 print(datetime.datetime.now(),' - publisher.py - publish - error:\n\r',sys.exc_info())230 else:231 print(datetime.datetime.now(),' - publisher.py - publish - error: caller._myAppSession is None')232 else:233 print(datetime.datetime.now(),' - publisher.py - publish - error: calller, topic, or type_ is None')234 # FUNCTIONS FROM HARNESS235 def drivers(self, from_, session_id, name, param):236 """237 name: n/a238 param: n/a239 """240 print(datetime.datetime.now(),'- LabwareClient.drivers:')241 print('\n\targs: ',locals(),'\n')242 return_list = list(self.driver_dict)243 if name is None:244 name = 'None'245 if from_ == "":246 self.publish('frontend',from_,session_id,'labware',name,'drivers',return_list)247 else:248 self.publish(from_,from_,session_id,'labware',name,'drivers',return_list)249 return return_list250 def add_driver(self, from_, session_id, name, param):251 """252 name: name of driver to add_driver253 param: driver object254 """255 print(datetime.datetime.now(),' - LabwareClient.add_driver:')256 print('\n\targs: ',locals(),'\n')257 self.driver_dict[name] = param258 return_list = list(self.driver_dict)259 if from_ == "":260 self.publish('frontend',from_,session_id,'labware',name,'drivers',return_list)261 else:262 self.publish(from_,from_,session_id,'labware',name,'drivers',return_list)263 return return_list264 def remove_driver(self, from_, session_id, name, param):265 """266 name: name of driver to be driver267 param: n/a268 """269 print(datetime.datetime.now(),' - LabwareClient.remove_driver:')270 print('\n\targs: ',locals(),'\n')271 del self.driver_dict[name]272 return_list = list(self.driver_dict)273 if from_ == "":274 self.publish('frontend',from_,session_id,'labware',name,'drivers',return_list)275 else:276 self.publish(from_,from_,session_id,'labware',name,'drivers',return_list)277 return return_list278 def callbacks(self, from_, session_id, name, param):279 """280 name: name of driver281 param: n/a282 """283 print(datetime.datetime.now(),' - LabwareClient.callbacks:')284 print('\n\targs: ',locals(),'\n')285 return_dict = self.driver_dict[name].callbacks()286 if from_ == "":287 self.publish('frontend',from_,session_id,'labware',name,'callbacks',return_dict)288 else:289 self.publish(from_,from_,session_id,'labware',name,'callbacks',return_dict)290 return return_dict291 def meta_callbacks(self, from_, session_id, name, param):292 """293 name: name of driver294 param: n/a295 """296 print(datetime.datetime.now(),' - labware_harness.meta_callbacks:')297 print('\n\targs: ',locals(),'\n')298 return_dict = self.driver_dict[name].meta_callbacks()299 self.publish(from_,from_,session_id,'labware',name,'meta_callbacks',return_dict)300 return return_dict301 def set_meta_callback(self, from_, session_id, name, param):302 """303 name: name of driver304 param: { meta-callback-name : meta-callback-object }305 """306 print(datetime.datetime.now(),' - LabwareClient.set_meta_callback:')307 print('\n\targs: ',locals(),'\n')308 if isinstance(param,dict):309 return_dict = self.driver_dict.get(name).set_meta_callback(list(param)[0],list(param.values())[0])310 else:311 return_dict = self.driver_dict.get(name).meta_callbacks()312 self.publish(from_,from_,session_id,'labware',name,'meta_callback',return_dict)313 return return_dict314 def add_callback(self, from_, session_id, name, param):315 """316 name: name of driver317 param: { callback obj: [messages list] }318 """319 print(datetime.datetime.now(),' - LabwareClient.add_callback:')320 print('\n\targs: ',locals(),'\n')321 return_dict = self.driver_dict[name].add_callback(list(param)[0],list(param.values())[0])322 if from_ == "":323 self.publish('frontend',from_,session_id,'labware',name,'callbacks',return_dict)324 else:325 self.publish(from_,from_,session_id,'labware',name,'callbacks',return_dict)326 return return_dict327 def remove_callback(self, from_, session_id, name, param):328 """329 name: name of driver330 param: name of callback to remove331 """332 print(datetime.datetime.now(),' - LabwareClient.remove_callback:')333 print('\n\targs: ',locals(),'\n')334 return_dict = self.driver_dict[name].remove_callback(param)335 if from_ == "":336 self.publish('frontend',from_,session_id,'labware',name,'callbacks',return_dict)337 else:338 self.publish(from_,from_,session_id,'labware',name,'callbacks',return_dict)339 return return_dict340 def flow(self, from_, session_id, name, param):341 """342 name: name of driver343 param: n/a344 """345 print(datetime.datetime.now(),' - LabwareClient.flow:')346 print('\n\targs: ',locals(),'\n')347 return_dict = self.driver_dict.get(name).flow()348 if from_ == "":349 self.publish('frontend',from_,session_id,'labware',name,'flow',return_dict)350 else:351 self.publish(from_,from_,session_id,'labware',name,'flow',return_dict)352 return return_dict353 def clear_queue(self, from_, session_id, name, param):354 """355 name: name of driver356 param: n/a357 """358 print(datetime.datetime.now(),' - LabwareClient.clear_queue:')359 print('\n\targs: ',locals(),'\n')360 return_dict = self.driver_dict.get(name).clear_queue()361 if from_ == "":362 self.publish('frontend',from_,session_id,'labware',name,'clear_queue',return_dict)363 else:364 self.publish(from_,from_,session_id,'labware',name,'clear_queue',return_dict)365 return return_dict366 def driver_connect(self, from_, session_id, name, param):367 """368 name: name of driver369 param: n/a370 """371 print(datetime.datetime.now(),' - LabwareClient.driver_connect:')372 print('\n\targs: ',locals(),'\n')373 print('self.driver_dict: ',self.driver_dict)374 print('self.driver_dict[',name,']: ',self.driver_dict[name])375 self.driver_dict[name].connect(from_,session_id)376 def driver_close(self, from_, session_id, name, param):377 """378 name: name of driver379 param: n/a380 """381 print(datetime.datetime.now(),' - LabwareClient.driver_close:')382 print('\n\targs: ',locals(),'\n')383 self.driver_dict.get(name).close(from_,session_id)384 def meta_commands(self, from_, session_id, name, param):385 """386 name: name of driver387 param: n/a388 """389 print(datetime.datetime.now(),' - LabwareClient.meta_commands:')390 print('\n\targs: ',locals(),'\n')391 return_list = list(self.meta_dict)392 if from_ == "":393 self.publish('frontend',from_,session_id,'labware',name,'meta_commands',return_list)394 else:395 self.publish(from_,from_,session_id,'labware',name,'meta_commands',return_list)396 return return_list397 def meta_command(self, from_, session_id, data):398 """399 data should be in the form:400 {401 'name': name,402 'message': value403 }404 where name the name of the driver or None if n/a,405 and value is one of two forms:406 1. string407 2. {command:params}408 params --> {param1:value, ... , paramN:value}409 """410 print(datetime.datetime.now(),' - LabwareClient.meta_command:')411 print('\n\targs: ',locals(),'\n')412 if isinstance(data, dict):413 name = data['name']414 value = data['message']415 if name in self.driver_dict:416 if isinstance(value, dict):417 command = list(value)[0]418 params = value[command]419 try:420 self.meta_dict[command](from_,session_id,name,params)421 except:422 if from_ == "":423 self.publish('frontend',from_,session_id,'labware',name,'error',str(sys.exc_info()))424 else:425 self.publish(from_,from_,session_id,'labware',name,'error',str(sys.exc_info()))426 print(datetime.datetime.now(),' - meta_command error: ',str(sys.exc_info()))427 elif isinstance(value, str):428 command = value429 try:430 self.meta_dict[command](from_,session_id,name,None)431 except:432 if from_ == "":433 self.publish('frontend',from_,session_id,'labware',name,'error',str(sys.exc_info()))434 else:435 self.publish(from_,from_,session_id,'labware',name,'error',str(sys.exc_info()))436 print(datetime.datetime.now(),' - meta_command error: ',sys.exc_info())437 else:438 if isinstance(value, dict):439 command = list(value)[0]440 params = value[command]441 try:442 self.meta_dict[command](from_,session_id,None,params)443 except:444 if from_ == "":445 self.publish('frontend',from_,session_id,'labware',name,'error',sys.exc_info())446 else:447 self.publish(from_,from_,session_id,'labware',name,'error',sys.exc_info())448 print(datetime.datetime.now(),' - meta_command error, name not in drivers: ',sys.exc_info())449 elif isinstance(value, str):450 command = value451 try:452 self.meta_dict[command](from_,session_id,None,None)453 except:454 if from_ == "":455 self.publish('frontend',from_,session_id,'labware','None','error',sys.exc_info())456 else:457 self.publish(from_,from_,session_id,'labware','None','error',sys.exc_info())458 print(datetime.datetime.now(),' - meta_command error, name not in drivers: ',sys.exc_info())459 def send_command(self, from_, session_id, data):460 """461 data:462 {463 'name': <name-of-driver>464 'message': <string> or { message : {param:values} } <--- the part the driver cares about465 }466 """467 print(datetime.datetime.now(),'LabwareClient.send_command:')468 print('\n\targs: ',locals(),'\n')469 if isinstance(data, dict):470 name = data['name']471 value = data['message']472 if name in self.driver_dict:473 try:474 self.driver_dict[name].send_command(session_id,value)475 except:476 if from_ == "":477 self.publish('frontend',from_,session_id,'labware',name,'error',sys.exc_info())478 else:479 self.publish(from_,from_,session_id,'labware',name,'error',sys.exc_info())480 print(datetime.datetime.now(),' - send_command error: '+sys.exc_info())481 else:482 if from_ == "":483 self.publish('frontend',from_,session_id,'labware','None','error',sys.exc_info())484 else:485 self.publish(from_,from_,session_id,'labware','None','error',sys.exc_info())486 print(datetime.datetime.now(),' - send_command_error, name not in drivers: '+sys.exc_info())487 def _make_connection(self, url_protocol='ws', url_domain='0.0.0.0', url_port=8080, url_path='ws', debug=False, debug_wamp=False):488 print(datetime.datetime.now(),' - LabwareClient._make_connection:')489 print('\n\targs: ',locals(),'\n')490 if self.loop.is_running():491 print('self.loop is running. stopping loop now')492 self.loop.stop()493 print(self.transport_factory)494 coro = self.loop.create_connection(self.transport_factory, url_domain, url_port)495 self.transport, self.protocol = self.loop.run_until_complete(coro)496 #protocoler.set_outer(self)497 if not self.loop.is_running():498 print('about to call self.loop.run_forever()')499 self.loop.run_forever()500 def connect(self, url_protocol='ws', url_domain='0.0.0.0', url_port=8080, url_path='ws', debug=False, debug_wamp=False, keep_trying=True, period=5):501 print(datetime.datetime.now(),' - LabwareClient.connect:')502 print('\n\targs: ',locals(),'\n')503 if self.transport_factory is None:504 url = url_protocol+"://"+url_domain+':'+str(url_port)+'/'+url_path505 self.transport_factory = websocket.WampWebSocketClientFactory(self.session_factory,506 url=url,507 debug=debug,508 debug_wamp=debug_wamp)509 self.session_factory._handshake = self.handshake510 self.session_factory._dispatch_message = self.dispatch_message511 if not keep_trying:512 try:513 print('\nLabware attempting crossbar connection\n')514 self._make_connection()515 except:516 print('crossbar connection attempt error:\n',sys.exc_info())517 pass518 else:519 while True:520 while (self.session_factory._crossbar_connected == False):521 try:522 print('\nLabware attempting crossbar connection\n')523 self._make_connection()524 except KeyboardInterrupt:525 self.session_factory._crossbar_connected = True526 except:527 print('crossbar connection attempt error:\n',sys.exc_info())528 pass529 finally:530 print('\nCrossbar connection failed, sleeping for 5 seconds\n')531 time.sleep(period)532 533 def disconnect(self):534 print(datetime.datetime.now(),' - LabwareClient.disconnect:')535 print('\n\targs: ',locals(),'\n')536 self.transport.close()537 self.transport_factory = None538 539 540if __name__ == '__main__':541 try:542 #session_factory = wamp.ApplicationSessionFactory()543 #session_factory.session = WampComponent544 #session_factory._myAppSession = None545 #url = "ws://0.0.0.0:8080/ws"546 #transport_factory = websocket.WampWebSocketClientFactory(session_factory,547 # url=url,548 # debug=False,549 # debug_wamp=False)550 #loop = asyncio.get_event_loop()551 print('\nBEGIN INIT...\n')552 # TRYING THE FOLLOWING IN INSTANTIATE OBJECTS vs here553 # INITIAL SETUP554 print(datetime.datetime.now(),' - INITIAL SETUP - publisher, harness, subscriber ','* * '*10)555 labware_client = LabwareClient()556 # INSTANTIATE DRIVERS557 print(datetime.datetime.now(),' - INSTANTIATE DRIVERS - labbie_driver ','* * '*10)558 labbie_driver = LabwareDriver()559 # ADD DRIVERS560 print(datetime.datetime.now(),' - ADD DRIVERS ','* * '*10) 561 labware_client.add_driver(labware_client.id,'','labware',labbie_driver)562 print(labware_client.drivers(labware_client.id,'',None,None))563 # DEFINE CALLBACKS564 #565 # data_dict format:566 #567 #568 #569 #570 #571 print(datetime.datetime.now(),' - DEFINE CALLBACKS ','* * '*10)572 def frontend(name, from_, session_id, data_dict):573 """574 """575 print(datetime.datetime.now(),' - labware_client.frontend')576 print('\n\targs: ',locals(),'\n')577 dd_name = list(data_dict)[0]578 dd_value = data_dict[dd_name]579 labware_client.publish('frontend',from_,session_id,'labware',name,dd_name,dd_value)580 581 def driver(name, from_, session_id, data_dict):582 """583 """584 print(datetime.datetime.now(),' - labware_client.driver')585 print('\n\targs: ',locals(),'\n')586 dd_name = list(data_dict)[0]587 dd_value = data_dict[dd_name]588 labware_client.publish('driver',from_,session_id,name,dd_name,dd_value)589 def bootstrapper(name, from_, session_id, data_dict):590 """591 """592 print(datetime.datetime.now(),' - labware_client.bootstrapper')593 print('\n\targs: ',locals(),'\n')594 dd_name = list(data_dict)[0]595 dd_value = data_dict[dd_name]596 labware_client.publish('bootstrapper','',session_id,name,dd_name,dd_value)597 def labware(name, from_, session_id, data_dict):598 """599 """600 print(datetime.datetime.now(),' - labware_client.labware')601 print('\n\targs: ',locals(),'\n')602 dd_name = list(data_dict)[0]603 dd_value = data_dict[dd_name]604 labware_client.publish('labware','',session_id,name,dd_name,dd_value)605 def none(name, from_, session_id, data_dict):606 """607 """608 print(datetime.datetime.now(),' - labware_client.none_cb')609 print('\n\targs: ',locals(),'\n')610 dd_name = list(data_dict)[0]611 dd_value = data_dict[dd_name]612 if from_ != session_id:613 labware_client.publish('frontend',from_,session_id,'labware',name,dd_name,dd_value)614 labware_client.publish(from_,from_,session_id,'labware',name,dd_name,dd_value)615 else:616 # next line just for testing617 labware_client.publish('frontend',from_,session_id,'labware',name,dd_name,dd_value)618 619 # ADD CALLBACKS620 labware_client.add_callback('frontend','','labware', {frontend:['frontend']})621 labware_client.add_callback('driver','','labware', {driver:['driver']})622 labware_client.add_callback('bootstrapper','','labware', {bootstrapper:['bootstrapper']})623 labware_client.add_callback('labware','','labware', {labware:['labware']})624 # none is for debugging625 labware_client.add_callback('frontend','','labware', {none:['None']})626 # ADD METACALLBACKS627 print(datetime.datetime.now(),' - DEFINE AND ADD META-CALLBACKS ','* * '*10)628 def on_connect(from_,session_id):629 print(datetime.datetime.now(),' - labware_client.on_connect')630 print('\n\targs: ',locals(),'\n')631 labware_client.publish(from_,from_,session_id,'connect','labware','result','connected')632 def on_disconnect(from_,session_id):633 print(datetime.datetime.now(),' - labware_client.on_disconnect')634 print('\n\targs: ',locals(),'\n')635 labware_client.publish(from_,from_,session_id,'connect','labware','result','disconnected')636 def on_empty_queue(from_,session_id):637 print(datetime.datetime.now(),' - labware_client.on_empty_queue')638 print('\n\targs: ',locals(),'\n')639 labware_client.publish(from_,from_,session_id,'queue','labware','result','empty')640 labware_client.set_meta_callback(labware_client.id,'','labware',{'on_connect':on_connect})641 labware_client.set_meta_callback(labware_client.id,'','labware',{'on_disconnect':on_disconnect})642 labware_client.set_meta_callback(labware_client.id,'','labware',{'on_empty_queue':on_empty_queue})643 # CONNECT TO DRIVERS644 print('\nEND INIT...\n')645 labware_client.connect()646 except KeyboardInterrupt:647 pass648 finally:...
utils.py
Source:utils.py
1import re23import numpy as np4import tensorflow as tf5678# get wordshape of a specific word9def get_wordshape(word):10 wordshape = []11 for char in word:12 if re.match('[0-9]', char):13 wordshape.append(3)14 elif re.match('[A-Z]', char):15 wordshape.append(1)16 elif re.match('[a-z]', char):17 wordshape.append(2)18 else:19 wordshape.append(4)20 return wordshape212223def valid_numeric_list(value):24 try:25 value = list(map(float, value))26 total = sum(value)27 return [v / total for v in value]2829 except ValueError as e:30 raise ValueError("list of numeric values expected : got {}".format(e.args[0]))313233def iob_mask(vocab, start, end, pad=None):34 small = 035 mask = np.ones((len(vocab), len(vocab)), dtype=np.float32)36 for from_ in vocab:37 for to in vocab:38 # Can't move to start39 if to is start:40 mask[vocab[to], vocab[from_]] = small41 # Can't move from end42 if from_ is end:43 mask[vocab[to], vocab[from_]] = small44 # Can only move from pad to pad or end45 if from_ is pad:46 if not (to is pad or to is end):47 mask[vocab[to], vocab[from_]] = small48 elif from_ is start:49 # Can't move from start to a B50 if to.startswith("B-"):51 mask[vocab[to], vocab[from_]] = small52 else:53 if from_.startswith("B-"):54 # Can't move from a B to a B of another type55 if to.startswith("B-"):56 from_type = from_.split("-")[1]57 to_type = to.split("-")[1]58 if from_type != to_type:59 mask[vocab[to], vocab[from_]] = small60 elif from_.startswith("I-"):61 # Can't move from an I to a B of another type62 if to.startswith("B-"):63 from_type = from_.split("-")[1]64 to_type = to.split("-")[1]65 if from_type != to_type:66 mask[vocab[to], vocab[from_]] = small67 elif from_.startswith("O"):68 # Can't move from an O to a B69 if to.startswith("B-"):70 mask[vocab[to], vocab[from_]] = small71 return mask727374def iob2_mask(vocab, start, end, pad=None):75 small = 076 mask = np.ones((len(vocab), len(vocab)), dtype=np.float32)77 for from_ in vocab:78 for to in vocab:79 # Can't move to start80 if to is start:81 mask[vocab[to], vocab[from_]] = small82 # Can't move from end83 if from_ is end:84 mask[vocab[to], vocab[from_]] = small85 # Can only move from pad to pad or end86 if from_ is pad:87 if not (to is pad or to is end):88 mask[vocab[to], vocab[from_]] = small89 elif from_ is start:90 # Can't move from start to a I91 if to.startswith("I-"):92 mask[vocab[to], vocab[from_]] = small93 else:94 if from_.startswith("B-"):95 # Can't move from a B to an I of a different type96 if to.startswith("I-"):97 from_type = from_.split("-")[1]98 to_type = to.split("-")[1]99 if from_type != to_type:100 mask[vocab[to], vocab[from_]] = small101 elif from_.startswith("I-"):102 # Can't move from an I to an I of another type103 if to.startswith("I-"):104 from_type = from_.split("-")[1]105 to_type = to.split("-")[1]106 if from_type != to_type:107 mask[vocab[to], vocab[from_]] = small108 elif from_.startswith("O"):109 # Can't move from an O to an I110 if to.startswith("I-"):111 mask[vocab[to], vocab[from_]] = small112 return (mask)113114115def iobes_mask(vocab, start, end, pad=None):116 small = 0117 mask = np.ones((len(vocab), len(vocab)), dtype=np.float32)118 for from_ in vocab:119 for to in vocab:120 # Can't move to start121 if to is start:122 mask[vocab[to], vocab[from_]] = small123 # Can't move from end124 if from_ is end:125 mask[vocab[to], vocab[from_]] = small126 # Can only move from pad to pad or to end127 if from_ is pad:128 if not (to is pad or to is end):129 mask[vocab[to], vocab[from_]] = small130 elif from_ is start:131 # Can't move from start to I or E132 if to.startswith("I-") or to.startswith("E-"):133 mask[vocab[to], vocab[from_]] = small134 else:135 if from_.startswith("B-"):136 # Can't move from B to B, S, O, End, or Pad137 if (138 to.startswith("B-") or139 to.startswith("S-") or140 to.startswith("O") or141 to is end or142 to is pad143 ):144 mask[vocab[to], vocab[from_]] = small145 # Can only move to matching I or E146 elif to.startswith("I-") or to.startswith("E-"):147 from_type = from_.split("-")[1]148 to_type = to.split("-")[1]149 if from_type != to_type:150 mask[vocab[to], vocab[from_]] = small151 elif from_.startswith("I-"):152 # Can't move from I to B, S, O, End or Pad153 if (154 to.startswith("B-") or155 to.startswith("S-") or156 to.startswith("O") or157 to is end or158 to is pad159 ):160 mask[vocab[to], vocab[from_]] = small161 # Can only move to matching I or E162 elif to.startswith("I-") or to.startswith("E-"):163 from_type = from_.split("-")[1]164 to_type = to.split("-")[1]165 if from_type != to_type:166 mask[vocab[to], vocab[from_]] = small167 elif (168 from_.startswith("E-") or169 from_.startswith("I-") or170 from_.startswith("S-") or171 from_.startswith("O")172 ):173 # Can't move from E to I or E174 # Can't move from I to I or E175 # Can't move from S to I or E176 # Can't move from O to I or E177 if to.startswith("I-") or to.startswith("E-"):178 mask[vocab[to], vocab[from_]] = small179 return mask180181182def transition_mask_np(vocab, span_type, s_idx, e_idx, pad_idx=None):183 """Create a CRF mask.184 Returns a mask with invalid moves as 0 and valid as 1.185 :param vocab: dict, Label vocabulary mapping name to index.186 :param span_type: str, The sequence labeling formalism {IOB, IOB2, BIO, or IOBES}187 :param s_idx: int, What is the index of the GO symbol?188 :param e_idx: int, What is the index of the EOS symbol?189 :param pad_idx: int, What is the index of the PAD symbol?190 Note:191 In this mask the PAD symbol is between the last symbol and EOS, PADS can192 only move to pad and the EOS. Any symbol that can move to an EOS can also193 move to a pad.194 """195 rev_lut = {v: k for k, v in vocab.items()}196 start = rev_lut[s_idx]197 end = rev_lut[e_idx]198 pad = None if pad_idx is None else rev_lut[pad_idx]199 if span_type.upper() == "IOB":200 mask = iob_mask(vocab, start, end, pad)201 if span_type.upper() == "IOB2" or span_type.upper() == "BIO":202 mask = iob2_mask(vocab, start, end, pad)203 if span_type.upper() == "IOBES":204 mask = iobes_mask(vocab, start, end, pad)205 return mask206207208def transition_mask(vocab, span_type, s_idx, e_idx, pad_idx=None):209 """Create a CRF Mask.210 Returns a mask with invalid moves as 0 and valid moves as 1.211 """212 mask = transition_mask_np(vocab, span_type, s_idx, e_idx, pad_idx).T213 inv_mask = (mask == 0).astype(np.float32)214 return tf.constant(mask), tf.constant(inv_mask)215216217if __name__ == '__main__':218 S = '[CLS]'219 E = '[SEP]'220 P = '<PAD>'221 SPAN_TYPE = "IOB2"222 label_vocab = {'<PAD>': 0, 'X': 1, '[CLS]': 2, '[SEP]': 3, 'B-PosReg': 4, 'I-Interaction': 5, 'I-Pathway': 6,223 'B-Var': 7, 'B-NegReg': 8, 'I-Var': 9, 'B-Enzyme': 10, 'I-Reg': 11, 'I-PosReg': 12, 'B-Pathway': 13,224 'I-Enzyme': 14, 'I-MPA': 15, 'B-Disease': 16, 'B-Reg': 17, 'I-Gene': 18, 'I-Disease': 19,225 'B-MPA': 20, 'I-Protein': 21, 'B-Interaction': 22, 'B-CPA': 23, 'B-Gene': 24, 'O': 25,226 'B-Protein': 26, 'I-CPA': 27, 'I-NegReg': 28}227 # label_vocab = {228 # '<PAD>': 0, "B": 1, "I": 2, "O": 3, "X": 4, "[CLS]": 5, "[SEP]": 6229 # }230 mask = transition_mask_np(label_vocab, SPAN_TYPE, label_vocab[S], label_vocab[E], label_vocab[P])
...
xmpp_request_handler.py
Source:xmpp_request_handler.py
1#!/usr/bin/env python2#3# Copyright 2007 Google Inc.4#5# Licensed under the Apache License, Version 2.0 (the "License");6# you may not use this file except in compliance with the License.7# You may obtain a copy of the License at8#9# http://www.apache.org/licenses/LICENSE-2.010#11# Unless required by applicable law or agreed to in writing, software12# distributed under the License is distributed on an "AS IS" BASIS,13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.14# See the License for the specific language governing permissions and15# limitations under the License.16#17"""A handler that allows the user to send XMPP messages to their application."""18import cStringIO19import random20import string21from xml.etree import ElementTree22from google.appengine.tools.devappserver2.admin import admin_request_handler23ElementTree.register_namespace('cli', 'jabber:client')24REMOTE_IP = '0.1.0.10'25class _FormData(object):26 def __init__(self):27 self._data = []28 def __eq__(self, o):29 return isinstance(o, _FormData) and self._data == o._data30 def __str__(self):31 _, content = self.get_boundary_and_content()32 return content33 __repr__ = __str__34 def add_text(self, name, value, sub_type):35 self._data.append((name, value, sub_type))36 def get_boundary_and_content(self):37 """Returns the message boundary and entire content body for the form."""38 boundary = '----=' + ''.join(random.choice(string.letters + string.digits)39 for _ in range(25))40 s = cStringIO.StringIO()41 for name, value, sub_type in self._data:42 s.write('--%s\r\n' % boundary)43 s.write('Content-Type: text/%s; charset="UTF-8"\r\n' % sub_type)44 s.write('Content-Disposition: form-data; name="%s"\r\n' % name)45 s.write('\r\n')46 s.write('%s\r\n' % value.encode('utf-8'))47 s.write('--%s--\r\n' % boundary)48 return boundary, s.getvalue()49class XmppRequestHandler(admin_request_handler.AdminRequestHandler):50 def get(self):51 # TODO: Generate a warning if XMPP is not configured when the52 # configuration is sorted in the new servers world.53 super(XmppRequestHandler, self).get()54 self.response.write(self.render('xmpp.html', {}))55 def post(self):56 super(XmppRequestHandler, self).post()57 message_type = self.request.get('message_type')58 to = self.request.get('to')59 from_ = self.request.get('from')60 if message_type == 'chat':61 self._send_chat(to, from_)62 elif message_type == 'presence':63 if self.request.get('presence') == 'available':64 self._send_presence_available(to, from_)65 elif self.request.get('presence') == 'unavailable':66 self._send_presence_unavailable(to, from_)67 else:68 assert 069 elif message_type == 'subscribe':70 self._send_subscription(to, from_)71 else:72 assert 073 def _generate_chat(self, to, from_, body):74 data = _FormData()75 data.add_text('from', from_, 'plain')76 data.add_text('to', to, 'plain')77 data.add_text('body', body, 'plain')78 message_element = ElementTree.Element(79 ElementTree.QName('jabber:client', 'message'),80 {'from': from_, 'to': to, 'type': 'chat'})81 body_element = ElementTree.SubElement(82 message_element,83 ElementTree.QName('jabber:client', 'body'))84 body_element.text = body85 data.add_text('stanza',86 ElementTree.tostring(message_element, encoding='utf-8'),87 'xml')88 return data89 def _send_chat(self, to, from_):90 body = self.request.get('chat')91 form_data = self._generate_chat(to, from_, body)92 response = self._send('/_ah/xmpp/message/chat/', form_data)93 self.response.status = response.status94 def _generate_presence_available(self, to, from_, show=None):95 data = _FormData()96 data.add_text('from', from_, 'plain')97 data.add_text('to', to, 'plain')98 # If the "presence" attribute is absent, "available" is assumed and it is99 # not sent by Google Talk.100 presence_element = ElementTree.Element(101 ElementTree.QName('jabber:client', 'presence'),102 {'from': from_, 'to': to})103 if show: # This is currently a dead code path.104 # The show element is optional according to RFC 3921, 2.2.2.1.105 data.add_text('show', show, 'plain')106 show_element = ElementTree.SubElement(107 presence_element,108 ElementTree.QName('jabber:client', 'show'))109 show_element.text = show110 data.add_text('stanza',111 ElementTree.tostring(presence_element, 'utf-8'),112 'xml')113 return data114 def _send(self, relative_url, form_data):115 boundary, content = form_data.get_boundary_and_content()116 return self.dispatcher.add_request(117 method='POST',118 relative_url=relative_url,119 headers=[('Content-Type',120 'multipart/form-data; boundary="%s"' % boundary)],121 body=content,122 source_ip=REMOTE_IP,123 fake_login=True)124 def _send_presence_available(self, to, from_):125 # TODO: Support "show" values i.e. "away", "chat" "dnd" and "xa".126 # See RFC 3921, 2.2.2.1.127 form_data = self._generate_presence_available(to, from_)128 response = self._send('/_ah/xmpp/presence/available/', form_data)129 self.response.status = response.status130 def _generate_presence_type(self, to, from_, presence_type):131 data = _FormData()132 data.add_text('from', from_, 'plain')133 data.add_text('to', to, 'plain')134 presence_element = ElementTree.Element(135 ElementTree.QName('jabber:client', 'presence'),136 {'from': from_, 'to': to, 'type': presence_type})137 data.add_text('stanza',138 ElementTree.tostring(presence_element, 'utf-8'),139 'xml')140 return data141 def _send_presence_unavailable(self, to, from_):142 form_data = self._generate_presence_type(to, from_, 'unavailable')143 response = self._send('/_ah/xmpp/presence/unavailable/', form_data)144 self.response.status = response.status145 def _send_subscription(self, to, from_):146 subscription_type = self.request.get('subscription_type')147 # The subscription types defined in RFC 3921, 2.2.1148 assert subscription_type in ['subscribe',149 'subscribed',150 'unsubscribe',151 'unsubscribed']152 form_data = self._generate_presence_type(to, from_, subscription_type)153 response = self._send('/_ah/xmpp/subscription/%s/' % subscription_type,154 form_data)...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run Selene automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
