KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use read_tr method in fMBT
Best Python code snippet using fMBT_python
testMQTTplusModbus.py
Source:testMQTTplusModbus.py 
1#!/usr/bin/env python32import paho.mqtt.client as paho3import minimalmodbus4from time import *5#mqtt auth6broker="192.168.1.61"7port=18838#modbus start9minimalmodbus.baudrate = 960010instr = minimalmodbus.Instrument('/dev/ttyACM0', 1)11instr.serial.baudrate=960012instr.debug=False13print(instr)14sleep(2)15#16number_of_triggers=2117topic_to_id={"home/triggers/kotel_to_light_kotel":0,"home/triggers/switch_floor1_to_light_street_floor1":1,"home/triggers/prachka_to_light_hall_floor1":2,"home/triggers/switch2_floor1_to_light_prachka_floor2":3,"home/triggers/switch1_kitchen_to_light_stairs_floor1":4,"home/triggers/switch2_kitchen_to_light1_kitchen":5,"home/triggers/hall_floor1_to_light2_kitchen_tv":6,"home/triggers/switch_input_to_light_hall_floor1":7,"home/triggers/street1_to_light_tambur":8,"home/triggers/switch_room3_floor2_to_light_street_floor1":9,"home/triggers/switch_room1_floor2_to_light_room3_floor2":10,"home/triggers/floor3_to_light_room1_floor2":11,"home/triggers/street2_to_light1_floor3":12,"home/triggers/wardrobe_to_light_street_floor2":13,"home/triggers/hall_floor2_to_light_wardrobe_floor2":14,"home/triggers/sw1_room2_floor2_to_light_on_stairs":15,"home/triggers/sw2_room2_floor2_to_light_room2_floor2":16,"home/triggers/badroom1_to_socket1_floor2":17,"home/triggers/badroom2_to_light_badroom_floor1":18,"home/triggers/room1_floor1_to_light_badroom_floor2":19,"home/triggers/switch2_input_to_light_room1_floor1":20}18id_to_topic = {v: k for k, v in topic_to_id.items()}19topic_to_id_states={"home/states/kotel_to_light_kotel":0,"home/states/switch_floor1_to_light_street_floor1":1,"home/states/prachka_to_light_hall_floor1":2,"home/states/switch2_floor1_to_light_prachka_floor2":3,"home/states/switch1_kitchen_to_light_stairs_floor1":4,"home/states/switch2_kitchen_to_light1_kitchen":5,"home/states/hall_floor1_to_light2_kitchen_tv":6,"home/states/switch_input_to_light_hall_floor1":7,"home/states/street1_to_light_tambur":8,"home/states/switch_room3_floor2_to_light_street_floor1":9,"home/states/switch_room1_floor2_to_light_room3_floor2":10,"home/states/floor3_to_light_room1_floor2":11,"home/states/street2_to_light1_floor3":12,"home/states/wardrobe_to_light_street_floor2":13,"home/states/hall_floor2_to_light_wardrobe_floor2":14,"home/states/sw1_room2_floor2_to_light_on_stairs":15,"home/states/sw2_room2_floor2_to_light_room2_floor2":16,"home/states/badroom1_to_socket1_floor2":17,"home/states/badroom2_to_light_badroom_floor1":18,"home/states/room1_floor1_to_light_badroom_floor2":19,"home/states/switch2_input_to_light_room1_floor1":20}20id_to_topic_states = {v: k for k, v in topic_to_id_states.items()}21changed=[3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3]22def on_publish(client,userdata,result): #create function for callback23 print("data published \n")24 pass25def on_message(client, userdata, message):26 global changed27 print("message received " ,str(message.payload.decode("utf-8")))28 print("message topic=",message.topic)29 print("message qos=",message.qos)30 print("message retain flag=",message.retain)31 if(str(message.payload.decode("utf-8"))=="0"):32 print("zero")33 changed[topic_to_id[message.topic]]=034 elif(str(message.payload.decode("utf-8"))=="1"):35 print("one")36 changed[topic_to_id[message.topic]]=137 #client.publish(id_to_topic_states[topic_to_id[message.topic]],str(changed[topic_to_id[message.topic]]))38 39client1= paho.Client("openhabian1") #create client object40client1.username_pw_set("house","55566678")41client1.on_publish = on_publish #assign function to callback42client1.on_message=on_message43client1.connect(broker,port)44client1.loop_start()45for x in range(0,number_of_triggers):46 client1.subscribe(id_to_topic[x])47#допиÑаÑÑ Ð¿Ñи полÑÑении Ñигнала обÑабоÑÐºÑ Ð²Ñ
одÑÑего Ñопика и запиÑÑ Ð² ÑооÑвеÑÑвÑÑÑий ÑÐ¾Ð¿Ð¸ÐºÑ ÑÐ»ÐµÐ¼ÐµÐ½Ñ Ð¼Ð°ÑÑива48#49read_tr=[3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3]50last_read_tr=[3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3]51last_cb=052while True:53 sleep(0.3)54 cb=instr.read_register(7,0)55 #print(cb)56 for tr in range(0,number_of_triggers):57 if(changed[tr]!=2):58 if(changed[tr]==1):59 instr.write_register(tr,1,0)60 changed[tr]=261 if(changed[tr]==0):62 instr.write_register(tr,0,0)63 changed[tr]=264 if(changed[tr]==3):65 changed[tr]=266 client1.publish(id_to_topic[tr],"begin")67 client1.publish(id_to_topic_states[tr],"begin")68 read_tr[tr]=instr.read_register(tr+21,0)69 if(read_tr[tr]!=last_read_tr[tr]):70 client1.publish(id_to_topic_states[tr],read_tr[tr])71 last_read_tr[tr]=read_tr[tr]...
rpcservice.py
Source:rpcservice.py
1# /usr/bin/python22# -*- coding: utf-8 -*-3import logging, time, struct4from proto import proto5import gevent6from gevent import socket7from gevent.queue import Queue8from gevent.event import AsyncResult9import socket10from google.protobuf import text_format11import command12from .rc4 import RC413from cfg import *14class RpcService(object):15 SESSION_ID = 116 def __init__(self, addr, game):17 self.hub = gevent.get_hub()18 self.addr = addr19 self.sock = None20 self.game = game21 self.time_diff = 022 self.write_queue = Queue()23 self.write_tr = None24 self.read_queue = Queue()25 self.read_tr = None26 self.dispatch_tr = None27 entype = Config.get("encrypt", None)28 if entype == "rc4":29 self.c2s_encrypt = RC4(Config["c2s_key"]).crypt30 self.s2c_encrypt = RC4(Config["s2c_key"]).crypt31 elif entype == None:32 self.c2s_encrypt = None33 self.s2c_encrypt = None34 else:35 raise ValueError("not support %s encrypt"%entype)36 self._sessions = {}37 def _start(self):38 if self.sock:39 return40 # sock = util.RC4Conn(self.addr)41 # sock.connect()42 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)43 sock.connect(self.addr)44 self.sock = sock45 self.read_tr = gevent.spawn(self._read)46 self.write_tr = gevent.spawn(self._write)47 self.dispatch_tr = gevent.spawn(self._dispatch)48 return True49 def set_timestamp(self, timestamp):50 self.time_diff = timestamp - int(time.time())51 def timestamp(self):52 return int(time.time()) + self.time_diff53 def stop(self):54 gevent.spawn(self._stop)55 def _stop(self):56 while True:57 gevent.sleep(1)58 if not self.write_queue.empty():59 continue60 if not self.read_queue.empty():61 continue62 gevent.kill(self.write_tr)63 gevent.kill(self.read_tr)64 gevent.kill(self.dispatch_tr)65 self.sock.close()66 break67 def _write(self):68 while True:69 data = self.write_queue.get()70 if self.c2s_encrypt:71 data = self.c2s_encrypt(data)72 try:73 self.sock.sendall(data)74 except socket.error, e:75 logging.info("write socket failed:%s" % str(e))76 break77 def _read(self):78 left = ""79 while True:80 try:81 buf = self.sock.recv(4*1024)82 if not buf:83 logging.info("client disconnected, %s:%s" % self.addr)84 break85 if self.s2c_encrypt:86 buf = self.s2c_encrypt(buf)87 except socket.error, e:88 logging.info("read socket failed:%s" % str(e))89 break90 left = left + buf91 while True:92 if len(left) < 2:93 break 94 plen, = struct.unpack('!H', left[:2])95 if len(left) < plen + 2:96 break 97 data = left[2:plen+2]98 left = left[plen+2:]99 self.read_queue.put(data)100 def _dispatch(self):101 while True:102 data = self.read_queue.get()103 p = proto.dispatch(data)104 session = p["session"]105 msg = p["msg"]106 if p["type"] == "REQUEST":107 protoname = p["proto"]108 cb = command.get_handle(protoname)109 if not cb:110 print "no handler for proto:", protoname111 continue112 resp = cb(self.game, msg)113 if session:114 # rpc call115 pack = proto.response(protoname, resp, session)116 self._send(pack)117 else:118 # response119 ev = self._sessions[session]120 del self._sessions[session]121 ev.set(msg)122 def _get_session(self):123 cls = type(self)124 if cls.SESSION_ID > 100000000:125 cls.SESSION_ID = 1126 cls.SESSION_ID += 1127 return cls.SESSION_ID128 def _send(self, data):129 self.write_queue.put(struct.pack("!H", len(data)) + data)130 def invoke(self, protoname, msg):131 pack = proto.request(protoname, msg)132 self._send(pack)133 def call(self, protoname, msg):134 session = self._get_session()135 pack = proto.request(protoname, msg, session)136 ev = AsyncResult()137 self._sessions[session] = ev138 self._send(pack)...
114.py
Source:114.py
...4import re5import nltk6import numpy as np7 8def read_tr(dir):9 data1 = []10 way_tr = []11 for file in os.listdir(dir): 12 with codecs.open(os.path.join(dir, file), 'r', 'utf-8') as f:13 data1.append(json.load(f)) 14 way_tr.append(file) 15 return data1, way_tr16def read_mn(dir):17 data2 = []18 way_mn = []19 for file in os.listdir(dir):20 with codecs.open(os.path.join(dir, file), 'r', 'utf-8') as f:21 data2.append(json.load(f))22 way_mn.append(file)23 return data2, way_mn24 25data_tr = read_tr('/home/partizan/classif2/spam_data/train')26data_mn, way_mn = read_mn('/home/partizan/classif2/spam_data/test')27def train(data_tr, way_tr, data_mn, way_mn):28 n = 029 vect_tr = []30 vect_mn = []31 while n < len(data_tr):32 vect_tr.append([0, 0, 0, data_tr[n]['is_spam']])33 tokens = nltk.word_tokenize(data_tr[n]['subject']) 34 for i in tokens:35 if re.search('spam',i) or re.search('SPAM',i):36 vect_tr[n][0] = vect_tr[n][0] +2037 if re.search('Re', i):38 vect_tr[n][1] = vect_tr[n][1] -2039 if re.search('CLICK', i) or re.search('click', i):40 vect_tr[n][2] = vect_tr[n][2] +2041 n = n + 1 42 n = 043 while n < len(data_mn):44 vect_mn.append([0, 0, 0, way_mn[n]])45 tokens = nltk.word_tokenize(data_mn[n]['subject']) 46 for i in tokens:47 if re.search('spam',i) or re.search('SPAM',i):48 vect_mn[n][0] = vect_mn[n][0] +2049 if re.search('Re', i):50 vect_mn[n][1] = vect_mn[n][0] -20 51 if re.search('CLICK', i) or re.search('click', i):52 vect_tr[n][2] = vect_tr[n][2] +20 53 n = n + 1 54 #print way_mn55 return vect_tr, vect_mn 56def classif(vect_tr, vect_mn):57 f = open('test.txt','w')58 n = 059 m = 060 num_mass = []61 num = 062 sorted(vect_tr)63 while (n < len(data_mn)):64 koaf = 0 65 while (m < len(data_tr)):66 iter = np.sqrt((vect_tr[m][0] - vect_mn[n][0])**2 + (vect_tr[m][1] - vect_mn[n][1])**2 + (vect_tr[m][2] - vect_mn[n][2])**2)67 if (m == 0):68 koaf = iter69 num = m70 else:71 if iter < koaf:72 koaf = iter73 num = m74 m = m + 175 col_tr = 076 col_fl = 077 try:78 if vect_tr[num - 1][3] == True: 79 col_tr = col_tr + 180 else:81 col_fl = col_fl + 182 except: print 183 try:84 if vect_tr[num + 1][3] == True: 85 col_tr = col_tr + 186 else:87 col_fl = col_fl + 188 except: print 189 if vect_tr[num][3] == True:90 col_tr = col_tr + 191 else:92 col_fl = col_fl + 193 if col_fl > col_tr:94 f.write(u'%s\t%s\n' % (way_mn[n], False))95 else:96 f.write(u'%s\t%s\n' % (way_mn[n], True))97 print col_fl, col_tr98 num_mass = []99 m = 0100 n = n + 1101 f.close()102data_tr, way_tr = read_tr('/home/partizan/classif2/spam_data/train')103data_mn, way_mn = read_mn('/home/partizan/classif2/spam_data/test')104print len(data_tr), " ", len(way_tr)105print len(data_mn), " ", len(way_mn)106vect_tr, vect_mn = train(data_tr, way_tr, data_mn, way_mn)...
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 fMBT 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?
