Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.