KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use write_list method in stestr
Best Python code snippet using stestr_python
handlers.py
Source:handlers.py 
1"""ÐодÑÐ»Ñ Ñ Ð¾Ð±ÑабоÑÑиками ÑообÑений"""2from string import digits as nums3from utils import *4async def main_window(user):5 """ÐÑкÑÑÐ²Ð°ÐµÑ Ð³Ð»Ð°Ð²Ð½Ð¾Ðµ Ð¼ÐµÐ½Ñ Ð¸ иÑполнÑÐµÑ Ð¶ÐµÐ»Ð°Ð½Ð¸Ðµ полÑзоваÑелÑ"""6 write_list.append((user._id, MAIN_WINDOW_TEXT))7 8 flag = 09 while flag == 0:10 if len(read_list[user._id]) > 0: 11 if read_list[user._id][0] == '1':12 task_list.append([user._id, 13 asyncio.gather(change_time(user))])14 flag = 115 elif read_list[user._id][0] == '2':16 task_list.append([user._id, 17 asyncio.gather(add_reminder(user))])18 flag = 119 elif read_list[user._id][0] == '3':20 task_list.append([user._id, 21 asyncio.gather(view_reminders(user))])22 flag = 123 else:24 await wrong_answer(user)25 del read_list[user._id][0] 26 await asyncio.sleep(TIMEOUT)27async def change_time(user):28 """УзнаеÑ, в каком ÑаÑовом поÑÑе наÑ
одиÑÑÑ Ð¿Ð¾Ð»ÑзоваÑелÑ, и изменÑÐµÑ ÐµÐ³Ð¾"""29 message=CHANGE_TIME_TEXT.split('\n')30 for i in message:31 write_list.append((user._id, i))32 flag = 033 while flag == 0:34 if len(read_list[user._id]) > 0:35 try:36 if 1 <= int(read_list[user._id][0]) <= 36:37 await user.change_timezone(int(read_list[user._id][0]))38 flag = 139 write_list.append((user._id, EDIT_TIME_TEXT))40 elif int(read_list[user._id][0]) == 0:41 flag = 142 write_list.append((user._id, CONTINUE_TEXT))43 else:44 await wrong_answer(user)45 except ValueError:46 await wrong_answer(user)47 del read_list[user._id][0]48 await asyncio.sleep(TIMEOUT)49 await asyncio.sleep(TIMEOUT)50async def add_reminder(user):51 """ÐÐ¾Ð¼Ð¾Ð³Ð°ÐµÑ Ð¿Ð¾Ð»ÑзоваÑÐµÐ»Ñ Ð´Ð¾Ð±Ð°Ð²Ð¸ÑÑ Ð½Ð°Ð¿Ð¾Ð¼Ð¸Ð½Ð°Ð½Ð¸Ðµ"""52 write_list.append((user._id, ADD_REMINDER_ENTER_TEXT))53 flag = 054 stop = 055 while flag == 0:56 if len(read_list[user._id]) > 0:57 if read_list[user._id][0] == '0':58 flag = 159 write_list.append((user._id, CONTINUE_TEXT))60 stop = 161 else:62 if "'" in read_list[user._id][0] or '"' in read_list[user._id][0] or ';' in read_list[user._id][0]:63 write_list.append((user._id, ADD_REMINDER_WRONG_TEXT))64 else:65 remindtext = read_list[user._id][0]66 flag = 167 write_list.append((user._id, ADD_REMINDER_ENTER_TIME))68 del read_list[user._id][0]69 await asyncio.sleep(TIMEOUT)70 flag = 071 while flag == 0 and stop == 0:72 if len(read_list[user._id]) > 0:73 if read_list[user._id][0] == '0':74 flag = 175 stop = 176 write_list.append((user._id, CONTINUE_TEXT))77 else:78 mes = read_list[user._id][0].split(' ')79 if len(mes) != 2:80 write_list.append((user._id, WRONG_DATE_TIME_FORMAT))81 else:82 date = mes[0].split('-')83 if len(date) != 3:84 write_list.append((user._id, WRONG_DATE_TIME_FORMAT))85 else:86 try:87 date[0] = int(date[0])88 date[1] = int(date[1])89 date[2] = int(date[2])90 days = [None, 31, None, 31, 30, 31,91 30, 31, 31, 30, 31, 30, 31]92 if date[2] % 1000 == 0:93 days[2] = 2994 elif date[2] % 100 == 0:95 days[2] = 2896 elif date[2] % 4 == 0:97 days[2] = 2998 else:99 days[2] = 28100 if date[1] < 1 or date[1] > 12 or \101 date[0] < 1 or date[0] > \102 days[date[1]]:103 write_list.append((user._id, WRONG_DAY_MONTH))104 else:105 times = mes[1].split('-')106 if len(times) != 2:107 write_list.append((user._id, 108 WRONG_DATE_TIME_FORMAT))109 else:110 try:111 times[0] = int(times[0])112 times[1] = int(times[1])113 if times[0] < 0 or times[0] > 23 or times[1] < 0 or times[0] > 59:114 write_list.append((user._id, 115 WRONG_TIME))116 else:117 timezone = await user.get_timezone()118 curtime = timegm(time.gmtime())119 timeremind = timegm(120 time.struct_time((date[2], 121 date[1], 122 date[0], 123 times[0], 124 times[1], 125 0, 0, 0, 0))126 ) + TIME_COR[timezone]127 if timeremind <= curtime:128 write_list.append((user._id, 129 TIMEREMIND_LESS_THAN_CURTIME))130 else:131 write_list.append((user._id, 132 ADD_REMINDER_ENTER_FREQUENCY))133 flag = 1134 except ValueError:135 raise136 except ValueError:137 write_list.append((user._id, TEXT_NOT_NUMS))138 del read_list[user._id][0]139 await asyncio.sleep(TIMEOUT)140 141 flag = 0142 while flag == 0 and stop == 0:143 if len(read_list[user._id]) > 0:144 mes = read_list[user._id][0].split(' ')145 if mes[0] == '0':146 flag = 1147 stop = 1148 write_list.append((user._id, CONTINUE_TEXT))149 elif mes[0] != '.' and len(mes) != 2:150 write_list.append((user._id, WRONG_FREQUENCY_FORMAT))151 else:152 if mes[0] == '.':153 repeattime = 0154 flag = 1155 else:156 try:157 mes[0] = int(mes[0])158 if mes[1] != 'м' and mes[1] != 'Ñ' and \159 mes[1] != 'д' and mes[1] != 'н' and\160 mes[1] != 'меÑ' and mes[1] != 'г':161 write_list.append((user._id, 162 WRONG_FREQUENCY_VALUE))163 else:164 repeattime = mes[0]165 if mes[1] == 'м':166 repeattime *= 60167 elif mes[1] == 'Ñ':168 repeattime *= 3600169 elif mes[1] == 'д':170 repeattime *= 3600*24171 elif mes[1] == 'н':172 repeattime *= 3600*24*7173 elif mes[1] == 'меÑ':174 curtime = timegm((date[2],175 date[1],176 date[0],177 times[0],178 times[1],179 0, 0, 0, 0))180 nextmonth = timegm((date[2], date[1] + int(mes[0]),181 date[0], times[0],182 times[1], 0, 0, 0, 0))183 repeattime = nextmonth - curtime184 elif mes[1] == 'г':185 curtime = timegm((date[2], date[1], date[0], 186 times[0], times[1], 0, 0, 0, 0))187 nextmonth = timegm((date[2] + int(mes[0]), date[1], date[0], 188 times[0], times[1], 0, 0, 0, 0))189 repeattime = nextmonth - curtime190 flag = 1191 except ValueError:192 write_list.append((user._id, TEXT_NOT_NUMS))193 del read_list[user._id][0]194 await asyncio.sleep(TIMEOUT)195 if flag == 1:196 await user.add_reminder(remindtext, timeremind, repeattime)197 write_list.append((user._id, REMINDER_ADDED))198 await asyncio.sleep(TIMEOUT)199async def view_reminders(user):200 """ÐоказÑÐ²Ð°ÐµÑ Ð½Ð°Ð¿Ð¾Ð¼Ð¸Ð½Ð°Ð½Ð¸Ñ Ð¿Ð¾Ð»ÑзовÑÐµÐ»Ñ Ð¸ ÑпÑаÑиваеÑ, ÑÑо он Ñ
оÑÐµÑ Ñ Ð½Ð¸Ð¼Ð¸ ÑделаÑÑ"""201 write_list.append((user._id, REMINDERS_LIST))202 prelst = await user.get_reminders()203 ind2 = 1204 for i in prelst:205 write_list.append((user._id, str(ind2)+'. '+i._text))206 ind2 += 1207 208 209 prelst.insert(0, None)210 flag = 0211 stop = 0212 while flag == 0 and stop == 0:213 if len(read_list[user._id]) > 0:214 if read_list[user._id][0] == '0':215 flag = 1216 stop = 1217 write_list.append((user._id, CONTINUE_TEXT))218 else:219 try:220 no = int(read_list[user._id][0])221 if no < 1 or no > len(prelst)-1:222 raise ValueError223 else:224 flag = 1225 write_list.append((user._id, EDIT_REMINDER_TEXT))226 except ValueError:227 write_list.append((user._id, ENTER_REMINDER))228 del read_list[user._id][0]229 await asyncio.sleep(TIMEOUT)230 flag = 0231 while flag == 0 and stop == 0:232 if len(read_list[user._id]) > 0:233 if read_list[user._id][0] == '0':234 flag = 1235 write_list.append((user._id, CONTINUE_TEXT))236 elif read_list[user._id][0] == '1':237 task_list.append([user._id, asyncio.gather(edit_reminder(user,238 prelst.copy(), no))])239 flag = 1240 elif read_list[user._id][0] == '2':241 task_list.append([user._id, asyncio.gather(del_reminder(user, 242 prelst.copy(), no))])243 flag = 1244 else:245 task_list.append([user._id, asyncio.gather(wrong_answer(user))])246 del read_list[user._id][0]247 await asyncio.sleep(TIMEOUT)248async def edit_reminder(user, lst, no):249 """ÐÐ¾Ð¼Ð¾Ð³Ð°ÐµÑ Ð¿Ð¾Ð»ÑзоваÑÐµÐ»Ñ Ð¸Ð·Ð¼ÐµÐ½Ð¸ÑÑ Ð½Ð°Ð¿Ð¾Ð¼Ð¸Ð½Ð°Ð½Ð¸Ðµ"""250 write_list.append((user._id, EDIT_REMINDER_ENTER_TEXT))251 flag = 0252 stop = 0253 while flag == 0 and stop == 0:254 if len(read_list[user._id]) > 0:255 if read_list[user._id][0] == '0':256 flag = 1257 stop = 1258 write_list.append((user._id, CONTINUE_TEXT))259 elif read_list[user._id][0] == '-':260 remindtext = None261 flag = 1262 write_list.append((user._id, EDIT_REMINDER_ENTER_TIME))263 elif "'" in read_list[user._id][0] or '"' in read_list[user._id][0] or ';' in read_list[user._id][0]:264 write_list.append((user._id, EDIT_REMINDER_WRONG_TEXT))265 else:266 remindtext = read_list[user._id][0]267 flag = 1268 write_list.append((user._id, EDIT_REMINDER_ENTER_TIME))269 del read_list[user._id][0]270 await asyncio.sleep(TIMEOUT)271 flag = 0272 while flag == 0 and stop == 0:273 if len(read_list[user._id]) > 0:274 if read_list[user._id][0] == '0':275 flag = 1276 stop = 1277 write_list.append((user._id, CONTINUE_TEXT))278 elif read_list[user._id][0] == '-':279 flag = 1280 timeremind = None281 write_list.append((user._id, EDIT_REMINDER_ENTER_FREQUENCY))282 else:283 mes = read_list[user._id][0]284 mes = mes.split(' ')285 if len(mes) != 2:286 write_list.append((user._id, WRONG_DATE_TIME_FORMAT))287 else:288 date = mes[0].split('-')289 if len(date) != 3:290 write_list.append((user._id, WRONG_DATE_TIME_FORMAT))291 else:292 try:293 date[0] = int(date[0])294 date[1] = int(date[1])295 date[2] = int(date[2])296 days = [None, 31, None, 31, 30, 31, 30,297 31, 31, 30, 31, 30, 31]298 if date[2] % 1000 == 0:299 days[2] = 29300 elif date[2] % 100 == 0:301 days[2] = 28302 elif date[2] % 4 == 0:303 days[2] = 29304 else:305 days[2] = 28306 if date[1] > 12 or date[1] < 1 or \307 date[0] < 1 or date[0] >\308 days[date[1]]:309 write_list.append((user._id, WRONG_DAY_MONTH))310 else:311 try:312 times = mes[1].split('-')313 times[0] = int(times[0])314 times[1] = int(times[1])315 if len(times) != 2:316 write_list.append((user._id, 317 WRONG_DATE_TIME_FORMAT))318 elif times[0] < 0 or times[0] > 23 \319 or times[1] < 0 or times[1]\320 > 59:321 write_list.append((user._id, 322 WRONG_TIME))323 else:324 timezone = await user.get_timezone()325 curtime = timegm(time.gmtime())326 timeremind = timegm(time.struct_time((327 date[2], date[1], date[0], 328 times[0], times[1], 0, 0, 0, 0)329 ))+TIME_COR[timezone]330 if timeremind <= curtime:331 write_list.append((user._id, 332 TIMEREMIND_LESS_THAN_CURTIME))333 else:334 flag = 1335 write_list.append((user._id, 336 EDIT_REMINDER_ENTER_FREQUENCY))337 except IndexError:338 write_list.append((user._id, 339 WRONG_DATE_TIME_FORMAT))340 except ValueError:341 write_list.append((user._id, TEXT_NOT_NUMS))342 del read_list[user._id][0]343 await asyncio.sleep(TIMEOUT)344 flag = 0345 while flag == 0 and stop == 0:346 if len(read_list[user._id]) > 0:347 if read_list[user._id][0] == '0':348 flag = 1349 stop = 1350 elif read_list[user._id][0] == '-':351 repeattime = None352 flag = 1353 elif read_list[user._id][0] == '.':354 repeattime = 0355 flag = 1356 else:357 try:358 mes = read_list[user._id][0].split(' ')359 if mes[0] == '-':360 mes.append('')361 flag = 1362 elif int(mes[0]) <= 0:363 write_list.append((user._id, WRONG_REPEATTIME_VALUE))364 elif mes[1] != 'м' and mes[1] != 'Ñ' and mes[1]\365 != 'д' and mes[1] != 'н' and mes[1] !=\366 'меÑ' and mes[1] != 'г':367 write_list.append((user._id, WRONG_FREQUENCY_VALUE))368 elif mes[1] == 'м':369 repeattime = 60*int(mes[0])370 flag = 1371 elif mes[1] == 'Ñ':372 repeattime = 3600*int(mes[0])373 flag = 1374 elif mes[1] == 'д':375 repeattime = 3600*24*int(mes[0])376 flag = 1377 elif mes[1] == 'н':378 repeattime = 3600*24*7*int(mes[0])379 flag = 1380 elif mes[1] == 'меÑ':381 if timeremind is None:382 a = time.gmtime(lst[no]._time)383 date = (a.tm_mday, a.tm_mon, a.tm_year)384 times = (a.tm_hour, a.tm_min)385 curtime = timegm((date[2], date[1],386 date[0], times[0],387 times[1], 0, 0, 0, 0))388 nextmonth = timegm((date[2], date[1] + int(mes[0]), date[0],389 times[0], times[1], 0, 0, 0, 0))390 repeattime = nextmonth - curtime391 flag = 1392 elif mes[1] == 'г':393 if timeremind is None:394 a = time.gmtime(lst[no]._time)395 date = (a.tm_mday, a.tm_mon, a.tm_year)396 times = (a.tm_hour, a.tm_min)397 curtime = timegm((date[2], date[1],398 date[0], times[0],399 times[1], 0, 0, 0, 0))400 nextmonth = timegm((date[2] + int(mes[0]), date[1], date[0],401 times[0], times[1], 0, 0, 0, 0))402 repeattime = nextmonth - curtime403 flag = 1404 except IndexError:405 write_list.append((user._id, WRONG_FREQUENCY_FORMAT))406 if flag == 1:407 remindid = lst[no]._id408 await user.edit_reminder(remindid, remindtext, timeremind, repeattime)409 write_list.append((user._id, REMINDER_EDITED))410 del read_list[user._id][0]411 await asyncio.sleep(TIMEOUT)412 await asyncio.sleep(TIMEOUT)413async def del_reminder(user, lst, no):414 """ÐÐ¾Ð¼Ð¾Ð³Ð°ÐµÑ Ð¿Ð¾Ð»ÑзоваÑÐµÐ»Ñ ÑдалиÑÑ Ð½Ð°Ð¿Ð¾Ð¼Ð¸Ð½Ð°Ð½Ð¸Ðµ"""415 write_list.append((user._id, ARE_YOU_SURE))416 flag = 0417 stop = 0418 while flag == 0 and stop == 0:419 if len(read_list[user._id]) > 0:420 if read_list[user._id][0] == '0':421 write_list.append((user._id, CONTINUE_TEXT))422 flag = 1423 elif read_list[user._id][0] == '1':424 ind = lst[no]._id425 await user.del_reminder(ind)426 write_list.append((user._id, REMINDER_DELETED))427 flag = 1428 else:429 await wrong_answer(user)430 del read_list[user._id][0]431 await asyncio.sleep(TIMEOUT)432 await asyncio.sleep(TIMEOUT)433async def register(user):434 """ÐÐ¾Ð¼Ð¾Ð³Ð°ÐµÑ Ð¿Ð¾Ð»ÑзоваÑÐµÐ»Ñ Ð·Ð°ÑегиÑÑÑиÑоваÑÑÑÑ"""435 message=REGISTER_TIME.split('\n')436 for i in message:437 write_list.append((user._id, i))438 flag = 0439 while flag == 0:440 if len(read_list[user._id]) > 0:441 if 1 <= int(read_list[user._id][0]) <= 36 :442 await end_register(user, int(read_list[user._id][0]))443 flag = 1444 else:445 await wrong_answer(user)446 del read_list[user._id][0]447 await asyncio.sleep(TIMEOUT)448 await asyncio.sleep(TIMEOUT)449async def end_register(user, tz):450 """ÐавеÑÑÐ°ÐµÑ ÑегиÑÑÑаÑÐ¸Ñ Ð¿Ð¾Ð»ÑзоваÑелÑ"""451 await user.write(tz)452 write_list.append((user._id, USER_REGISTERED))453 await asyncio.sleep(TIMEOUT)454async def wrong_answer(user):455 """СообÑÐ°ÐµÑ Ð¿Ð¾Ð»ÑзоваÑелÑ, ÑÑо он вÑбÑал неÑÑÑеÑÑвÑÑÑий ваÑÐ¸Ð°Ð½Ñ Ð¾ÑвеÑа"""456 write_list.append((user._id, WRONG_ANSWER_TEXT))457 await asyncio.sleep(TIMEOUT)458async def begin(user2, msgtext):459 """ÐпÑеделÑеÑ, ÑÑо нÑжно ÑделаÑÑ, в завиÑимоÑÑи Ð¾Ñ Ñого, заÑегиÑÑÑиÑован ли полÑзоваÑелÑ"""460 try:461 user = await read_db('select * from users where user=(%s)',462 (str(user2._id)))463 except Exception:464 write_list.append((user2._id, DB_ERROR_USER))465 raise466 try:467 if user2._id == user[0]['user']:468 task_list.append([user2._id, asyncio.gather(main_window(user2))])469 except IndexError:470 if msgtext.lower() != 'наÑаÑÑ':471 write_list.append((user2._id, WELCOME))472 else:473 task_list.append([user2._id, asyncio.gather(register(user2))]) ...
create_config.py
Source:create_config.py
1"""2Author: Vinay Bagde3Modifier: Anurag Guda4Maintainer: Andrew Liu, Anurag Guda5Copyright (c) 2018 NVIDIA Corporation. All rights reserved.6"""7import sys8import os9from collections import OrderedDict10tile_map = {1:(1,1), 2: (1,2), 4:(2,2), 6:(2,3), 8:(2,4), 10:(2,5), 12:(3,4), 15:(3,5), 18:(3,6)}11tile_map = OrderedDict(tile_map)12def read_camera_file():13 camera_path = '/etc/config'14 files = os.listdir(camera_path)15 camera_ips = []16 for file in files:17 current_file = os.path.join(camera_path,file)18 if os.path.isfile(current_file):19 camera_file = open(current_file)20 camera_ip = camera_file.readline()21 camera_ip = camera_ip.strip("\n")22 camera_file.close()23 camera_ips.append(camera_ip)24 return camera_ips25def main():26 ips = read_camera_file()27 print(ips)28 print(len(ips))29 n_rows = None30 n_columns = None31 for key,val in tile_map.items():32 if len(ips) < key:33 break34 n_rows = val[0]35 n_columns = val[1]36 write_list = []37 if len(ips) != 0:38 with open(sys.argv[2]) as fp:39 line = fp.readline()40 while line:41 pair = line.split("=")42 if pair[0] == "rows":43 pair[1] = str(n_rows)44 elif pair[0] == "columns":45 pair[1] = str(n_columns)46 elif pair[0] == "batch-size":47 pair[1] = str(len(ips))48 output = line49 if len(pair) > 1:50 output = "=".join(pair)51 output = output.replace('\n','')52 write_list.append(output)53 line = fp.readline()54 fp.close()55 else:56 with open(sys.argv[2]) as fp:57 line = fp.readline()58 while line:59 pair = line.split("=")60 if pair[0] == "rows":61 pair[1] = str("1")62 elif pair[0] == "columns":63 pair[1] = str("1")64 elif pair[0] == "num-sources":65 pair[1] = str("1")66 elif pair[0] == "file-loop":67 pair[1] = str("1")68 elif pair[0] == "batch-size":69 pair[1] = str("1")70 output = line71 if len(pair) > 1:72 output = "=".join(pair)73 output = output.replace('\n','')74 write_list.append(output)75 line = fp.readline()76 fp.close()77 gpulist = os.popen('nvidia-smi -L').read()78 print(gpulist)79 if 'T4' in gpulist:80 if len(ips) != 0:81 for index,ip in enumerate(ips):82 write_list.append("\n")83 write_list.append("[source{}]".format(index))84 write_list.append("enable=1")85 write_list.append("type=4")86 write_list.append("uri={}".format(ip))87 write_list.append("num-sources=1")88 write_list.append("gpu-id=0")89 write_list.append("cudadec-memtype=0")90 write_list.append("\n")91 write_list.append("[sink0]")92 write_list.append("enable=1")93 write_list.append("type=4")94 write_list.append("container=1")95 write_list.append("codec=1")96 write_list.append("sync=0")97 write_list.append("bitrate=2000000")98 write_list.append("profile=0")99 write_list.append("output-file=out.mp4")100 write_list.append("source-id=0")101 if len(ips) == 0:102 write_list.append("\n")103 write_list.append("[sink0]")104 write_list.append("enable=1")105 write_list.append("type=1")106 write_list.append("sync=1")107 write_list.append("codec=1")108 write_list.append("bitrate=4000000")109 write_list.append("rtsp-port=8554")110 write_list.append("udp-port=5400")111 write_list.append("source-id=0")112 write_list.append("gpu-id=0")113 write_list.append("nvbuf-memory-type=0")114 write_list.append("\n")115 write_list.append("[sink2]")116 write_list.append("enable=1")117 write_list.append("type=4")118 write_list.append("container=1")119 write_list.append("codec=1")120 write_list.append("sync=1")121 write_list.append("bitrate=2000000")122 write_list.append("rtsp-port=8554")123 write_list.append("udp-port=5400")124 write_list.append("profile=0")125 write_list.append("output-file=out.mp4")126 write_list.append("source-id=0")127 else:128 if len(ips) != 0:129 for index,ip in enumerate(ips):130 write_list.append("\n")131 write_list.append("[source{}]".format(index))132 write_list.append("enable=1")133 write_list.append("type=4")134 write_list.append("uri={}".format(ip))135 write_list.append("num-sources=1")136 write_list.append("gpu-id=0")137 write_list.append("cudadec-memtype=0")138 write_list.append("\n")139 write_list.append("[sink0]")140 write_list.append("enable=1")141 write_list.append("type=4")142 write_list.append("container=1")143 write_list.append("codec=1")144 write_list.append("enc-type=1")145 write_list.append("sync=0")146 write_list.append("bitrate=2000000")147 write_list.append("profile=0")148 write_list.append("output-file=out.mp4")149 write_list.append("source-id=0")150 if len(ips) == 0:151 write_list.append("\n")152 write_list.append("[sink0]")153 write_list.append("enable=1")154 write_list.append("type=1")155 write_list.append("sync=1")156 write_list.append("codec=1")157 write_list.append("enc-type=1")158 write_list.append("bitrate=4000000")159 write_list.append("rtsp-port=8554")160 write_list.append("udp-port=5400")161 write_list.append("source-id=0")162 write_list.append("gpu-id=0")163 write_list.append("nvbuf-memory-type=0")164 write_list.append("\n")165 write_list.append("[sink2]")166 write_list.append("enable=1")167 write_list.append("type=4")168 write_list.append("container=1")169 write_list.append("codec=1")170 write_list.append("enc-type=1")171 write_list.append("sync=1")172 write_list.append("bitrate=2000000")173 write_list.append("rtsp-port=8554")174 write_list.append("udp-port=5400")175 write_list.append("profile=0")176 write_list.append("output-file=out.mp4")177 write_list.append("source-id=0")178 write_file = os.path.join(os.path.dirname(sys.argv[2]),'run.txt')179 with open(write_file,"w") as file:180 for line in write_list:181 file.write(line)182 file.write("\n")183 file.close()184 print(write_file)185 os.system("{} -c {}".format(sys.argv[1],write_file))186if __name__ == '__main__':...
server.py
Source:server.py
1#coding=utf-823import socket4import time5import datetime6#datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")7import select8import threading910host = '' #support all address11port = 8012s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)13s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)14s.bind((host,port))15s.listen(5)16s.setblocking(False)1718read_list=[s]19write_list=[]20error_list=[s]21#conn,addr = s.accept()22#print 'connected by ',addr2324#def timer_func(arg, kargs):25def timer_func(fd):26 print "fd : ", fd #å¦æå äºéå·,ä¸æ¢è¡27 #print "kargs :", kargs28 try:29 print "sendall start : ", datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")30 fd.sendall("notify")31 print "sendall end : ", datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")32 except socket.error, e:33 err = e.args[0]34 if err == errno.EAGAIN or err == errno.EWOULDBLOCK:35 pass36 else:37 print "sendall error : ", datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")38 print e39 print err40 pass4142while True:43 readable_list, writable_list, exceptional_list = select.select(read_list, write_list, error_list, 1) #1为è¶
æ¶æ¶é´44 for fd in readable_list:45 if fd is s:46 conn, addr = fd.accept()47 print "\n",'connected by ',addr48 conn.setblocking(False)49 read_list.append(conn)50 error_list.append(conn)51 else:52 try:53 data = fd.recv(1024)54 except socket.error, e:55 print "server recv error : ", fd56 if fd in read_list:57 read_list.remove(fd)58 if fd in write_list:59 write_list.remove(fd)60 if fd in error_list:61 error_list.remove(fd)62 fd.close()63 print e64 continue65 if data != '':66 print "received : ", data67 if "subs" == data:68 write_list.append(fd)69 else:70 print datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f"), "server close : ", fd71 if fd in read_list:72 read_list.remove(fd)73 if fd in write_list:74 write_list.remove(fd)75 if fd in error_list:76 error_list.remove(fd)77 fd.close()78 79 for fd in writable_list:80 print "writable_list : ",writable_list81 if fd in write_list:82 write_list.remove(fd)83 #timer = threading.Timer(1, timer_func, (fd,), {"test":"abc"})84 timer = threading.Timer(0, timer_func, (fd,))85 timer.start()86 #raise error(EBADF, 'Bad file descriptor')87 #error: [Errno 9] Bad file descriptor88 89 90 for fd in exceptional_list:91 print "exceptional_list : ",exceptional_list92 print "error conn : ", fd.getpeername()93 if fd in read_list:94 read_list.remove(fd)95 if fd in write_list:96 write_list.remove(fd)97 if fd in error_list:98 error_list.remove(fd)99 fd.close()100 #data = conn.recv(1024)101 #if not data:102 # break103 #if "subs" == data:104 # while True:105 # time.sleep(3)106 # conn.sendall("notify")107 #conn.sendall(data)108s.close()109110111112113114115116
...
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 stestr 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?
