How to use match_in method in Airtest

Best Python code snippet using Airtest

rok.py

Source:rok.py Github

copy

Full Screen

...32 start_time = time.time()33 for i in range(threshold):34 # 查找确认按钮35 screen = dev.snapshot()36 pos = Template(r"item/sure.png").match_in(screen)37 if pos is not None:38 x, y = pos39 logger.info('检测到人机<<确认>>(%d, %d)按钮' % (x, y))40 logger.info('开始从全屏帧中截取验证码区域')41 img = screen42 if DEBUG:43 xs, ys = 200, 73044 img = img[200:, 730:1810]45 else:46 xs, ys = 134, 73047 img = img[134:, 730:1540]48 st = time.time()49 logger.info('开始向第三方打码平台发送打码请求')50 data = get_vcode_click_pos(uname='okiyar', pwd='okiyar',51 img=np.array(cv2.imencode('.jpg', img)[1]).tobytes())52 logger.info('收到返回结果: (%s) 耗时: %.5s 秒' % (data, (time.time() - st)))53 if data['success']:54 result = data["data"]["result"]55 result = result.split('|')56 logger.info('开始模拟人工点击验证码')57 for i in range(len(result)):58 r = result[i].split(',')59 x = int(r[0]) + ys60 y = int(r[1]) + xs61 dev.touch((int(r[0]) + ys, int(r[1]) + xs))62 logger.info('点击打点%d坐标: (%d, %d)' % (i + 1, x, y))63 sleep(0.5)64 # 保存打码验证码65 save_touch_screen(pos, dev.snapshot(), filename='screenshot/vcode/%s.png' % datetime.datetime.now())66 dev.touch(pos)67 sleep(3)68 # 检测是否通过验证69 if not Template("item/sure.png").match_in(dev.snapshot()):70 logger.info("已通过验证")71 logger.info('本次验证共计耗时: %.5s 秒' % (time.time() - start_time))72 break73 else:74 logger.info("未通过验证")75 continue76 else:77 logger.info("打码失败，重新打码，说明：%s" % data["message"])78 else:79 # 查找验证按钮80 screen = dev.snapshot()81 pos = Template(r"item/verify.png").match_in(screen)82 if pos is not None:83 x, y = pos84 logger.info('找到人机<<验证>>(%d, %d)按钮，开始点击，并等待5秒加载验证码，进行下一步骤查找<<确认>>按钮' % (x, y))85 dev.touch(pos)86 # save_touch_screen(pos, screen)87 logger.info("人机验证标志循环检测中... %d/%d" % (i + 1, threshold))88 else:89 logger.info("检测次数超出阈值仍未检测到相应标志，退出人机验证任务")90def switch_role(threshold=10):91 """92 大小号切换93 :return:94 """95 # 保证当前界面含有个人头像96 logger.info("开始检测当前界面是否含有个人头像界面...")97 for i in range(10):98 pos = Template("item/home-map.png").match_in(dev.snapshot()) or Template("item/home-tower.png").match_in(99 dev.snapshot())100 if pos:101 logger.info("检测到当前界面含有个人头像界面")102 dev.touch((100, 70))103 logger.info("已点击个人头像")104 break105 dev.keyevent("BACK")106 sleep(1)107 logger.info("个人头像界面循环检测中... (%d/%d)" % (i + 1, threshold))108 else:109 logger.info("检测次数超出阈值仍未检测到相应标志，退出切换角色任务")110 return111 for i in range(threshold):112 pos = Template(r"item/settings.png").match_in(dev.snapshot())113 if pos:114 logger.info("检测到<<设置>>图标，期望<<角色管理>>图标")115 dev.touch(pos)116 sleep(1)117 pos = Template(r"item/role.png").match_in(dev.snapshot())118 if pos:119 logger.info("检测到<<角色管理>>图标，期望<<✅>>图标")120 dev.touch(pos)121 sleep(1)122 pos = Template(r"item/green-flag.png").match_in(dev.snapshot())123 if pos:124 logger.info("检测到<<✅>>图标，期望<<是>>按钮")125 x, y = pos126 if x < 1000:127 x += 1600128 else:129 x -= 500130 dev.touch((x, y))131 sleep(1)132 pos = Template(r"item/yes.png").match_in(dev.snapshot())133 if pos:134 logger.info("检测到<<是>>图标")135 dev.touch(pos)136 break137 sleep(1)138 logger.info("操作标志循环检测中... (%d/%d)" % (i + 1, threshold))139 else:140 logger.info("检测次数超出阈值仍未检测到相应标志，退出切换角色任务")141 return142 logger.info("等待角色切换进入游戏内堡界面中...")143 while not Template(r"item/home-map.png").match_in(dev.snapshot()):144 try:145 pkg_name = dev.get_top_activity()[0]146 if pkg_name != ROK_PACKAGE_NAME:147 dev.start_app(ROK_PACKAGE_NAME)148 except:149 pass150 sleep(1)151 else:152 logger.info("角色切换已进入游戏内堡界面")153 logger.info("已完成角色切换任务")154def home(threshold=5):155 """156 返回家157 :return: None158 """159 logger.info("开始检测当前界面，返回执政官内堡界面中...")160 for i in range(threshold):161 pos = Template("item/exit-game-tip.png").match_in(dev.snapshot())162 if pos:163 keyevent('BACK')164 sleep(2)165 pos = Template("item/home-tower.png").match_in(dev.snapshot())166 if pos is not None:167 dev.touch(pos)168 sleep(1)169 break170 pos = Template(r"item/home-map.png").match_in(dev.snapshot())171 if pos is not None:172 dev.touch(pos)173 sleep(1)174 dev.touch(pos)175 sleep(1)176 break177 keyevent('BACK')178 sleep(1)179 logger.info("界面循环检测中... (%d/%d)" % (i + 1, threshold))180 else:181 logger.info("检测次数超出阈值仍未检测到相应标志，开始重启应用实现进入内堡界面")182 dev.start_app_timing(ROK_PACKAGE_NAME, ROK_MAIN_ACTIVITY)183 while not Template(r'item/home-map.png').match_in(dev.snapshot()):184 try:185 pkg_name = dev.get_top_activity()[0]186 if pkg_name != ROK_PACKAGE_NAME:187 dev.start_app(ROK_PACKAGE_NAME)188 except:189 pass190 sleep(1)191 logger.info("当前界面已恢复至执政官内堡初始界面")192def walker():193 if Template("item/earth.png").match_in(dev.snapshot()):194 # print("阶段三")195 pass196 elif not Template("item/menu.png").match_in(dev.snapshot()):197 # print("阶段二")198 pass199 else:200 dev.pinch(steps=50)201 info = dev.get_display_info()202 x = int(info['width'] / 2)203 y = int(info['height'] / 2)204 dx = random.randint(-500, 500)205 dy = random.randint(-500, 500)206 dev.swipe([x, y], [x + dx, y + dy])207def scout(threshold=3):208 logger.info("==================================================")209 logger.info("开始斥候任务")210 for i in range(threshold):211 pos = Template("item/home-tower.png").match_in(dev.snapshot())212 if pos:213 dev.touch(pos)214 sleep(1)215 else:216 pos = Template("item/home-map.png").match_in(dev.snapshot())217 if pos is None:218 logger.info("当前界面未知，退出本次斥候任务")219 return220 info = dev.get_display_info()221 w = info['width']222 h = info['height']223 dev.touch([int(w / 2), int(h / 2)])224 sleep(1)225 pos = Template("item/scout.png").match_in(dev.snapshot())226 if pos:227 dev.touch(pos)228 sleep(1)229 pos = Template("item/scout-manage-string.png").match_in(dev.snapshot())230 if pos:231 flag = False232 for j in range(2 * threshold):233 pos = Template("item/explore-string.png").match_in(dev.snapshot())234 if pos:235 save_touch_screen(pos, dev.snapshot())236 dev.touch(pos)237 sleep(1)238 pos = Template("item/send-string.png").match_in(dev.snapshot())239 if pos:240 if not flag:241 flag = True242 screen = dev.snapshot()243 tpos = Template("item/go-back.png").match_in(screen) \244 or Template("item/station.png").match_in(screen)245 save_touch_screen(pos, screen)246 if tpos:247 dev.touch(tpos)248 sleep(1)249 else:250 dev.touch(pos)251 save_touch_screen(pos, screen)252 sleep(1)253 logger.info("已派遣斥候")254 break255 else:256 dev.touch(pos)257 save_touch_screen(pos, screen)258 sleep(1)259 logger.info("已派遣斥候")260 break261 logger.info("循环检测标志中... (%d/%d)" % (j + 1, 2 * threshold))262 else:263 logger.info("检测次数超出阈值仍未检测到相应标志，退出斥候任务")264 return265 else:266 logger.info("检测次数超出阈值仍未检测到相应标志，退出斥候任务")267def farm(times=None, threshold=10, index=None):268 # 保证当前界面为堡外界面269 logger.info("开始检测当前界面是否为堡外界面...")270 for i in range(10):271 pos = Template("item/home-map.png").match_in(dev.snapshot())272 if pos:273 logger.info("检测到当前界面在执政官城堡内部，开始点击<<地图>>小图标显示堡外界面 (%d/%d)" % (i + 1, threshold))274 dev.touch(pos)275 logger.info("已点击<<地图>>小图标按钮")276 break277 pos = Template("item/home-tower.png").match_in(dev.snapshot())278 if pos:279 logger.info("检测到当前界面为堡外界面，退出堡外界面保证循环检测，进入下一步骤... (%d/%d)" % (i + 1, threshold))280 break281 dev.keyevent("BACK")282 sleep(1)283 logger.info("堡外界面循环检测中... (%d/%d)" % (i + 1, threshold))284 else:285 logger.info("检测次数超出阈值仍未检测到相应标志，退出采集任务")286 return287 info = dev.get_display_info()288 cx = int(info['width'] / 2)289 cy = int(info['height'] / 2)290 img_list = ("farm-string.png", "wood-string.png", "stone-string.png")291 img_string = ("农田", "木材", "石头", "金币")292 sc_counter = 0293 # key: r_index -> value: list294 has_searched_pos = dict()295 index_black = [False for i in range(len(img_list))]296 allow_remnant_ore = False297 for i in range(threshold):298 flag = False299 # 根据多次搜索反馈重新选择合理采集资源点种类300 logger.info("开始初始化本次循环采集的资源矿种类")301 while True:302 if index is None:303 r_index = random.randint(0, len(img_list) - 1)304 else:305 if isinstance(index, int):306 r_index = index307 if index_black[r_index]:308 allow_remnant_ore = True309 else:310 r_index = index[random.randint(0, len(index) - 1)]311 if all(index_black):312 logger.info("检测到所有种类的资源矿都含有残矿，忽略残矿检测，本次允许采集残矿")313 allow_remnant_ore = True314 break315 if not index_black[r_index]:316 if r_index not in has_searched_pos:317 has_searched_pos[r_index] = []318 logger.info("已选择本次循环采集的资源矿种类：<<%s>>" % img_string[r_index])319 break320 logger.info("开始检测堡外界面采集入口标志")321 screen = dev.snapshot()322 pos = Template("item/search.png").match_in(screen)323 if pos:324 logger.info("检测到堡外界面采集入口标志<<搜索>>小图标")325 dev.touch(pos)326 logger.info("点击<<搜索>>小图标，进入资源搜索界面")327 # save_touch_screen(pos, screen)328 sleep(1)329 logger.info("开始检测资源搜索界面<<搜索>>文字蓝色按钮")330 screen = dev.snapshot()331 pos = Template("item/search-string.png").match_in(screen)332 if pos:333 logger.info("检测到资源搜索界面<<搜索>>文字蓝色按钮")334 logger.info("开始检测资源矿等级递增标志<<+>>号蓝色图标")335 pos = Template("item/jia.png").match_in(dev.snapshot())336 if pos:337 logger.info("检测到资源矿等级递增标志<<+>>号蓝色图标")338 logger.info("开始连续点击<<＋>>号图标，使其递增至最高级等级矿")339 for i in range(5):340 dev.touch(pos)341 logger.info("点击<<＋>>号图标 %d/%d" % (i + 1, 5))342 sleep(0.2)343 sleep(1)344 logger.info("开始检测搜索前资源矿<<%s>>文字图标" % img_string[r_index])345 screen = dev.snapshot()346 pos = Template('item/' + img_list[r_index]).match_in(screen)347 if pos:348 logger.info("检测到搜索前资源矿<<%s>>文字图标" % img_string[r_index])349 # Todo 后续优化智能均衡选矿350 dev.touch(pos)351 logger.info("点击资源矿<<%s>>文字图标，进入下一步骤..." % img_string[r_index])352 save_touch_screen(pos, screen)353 sleep(1)354 tlc = 0355 r_index_plist = has_searched_pos[r_index]356 is_continue = False357 is_down = False358 search_click_counter = 0359 logger.info("开始%s资源矿采集保证循环检测..." % img_string[r_index])360 same_pos_times = 0361 for j in range(2 * threshold):362 logger.info("开始检测%s资源矿<<搜索>>文字按钮..." % img_string[r_index])363 screen = dev.snapshot()364 pos = Template('item/search-string.png').match_in(screen)365 if pos is None:366 logger.info("检测到%s资源矿<<搜索>>文字按钮消失，开始检测当前搜索的资源点是否符合采集要求..." % img_string[r_index])367 # 检查坐标是否搜索过368 logger.info("开始检测%s资源点坐标" % img_string[r_index])369 # pos = Template("item/cpos-flag.png").match_in(dev.snapshot())370 # pos = Template("item/c2.png").match_in(dev.snapshot())371 # if pos:372 # logger.info("检测到%s资源点坐标位置参考标志<<收藏>>图标" % img_string[r_index])373 # save_touch_screen(pos, screen)374 # logger.info("开始以参考标志截取资源点坐标")375 logger.info("开始向第三方平台发送 OCR 识别请求资源点坐标...")376 st = time.time()377 result = img_to_str(378 np.array(cv2.imencode('.jpg',379 dev.snapshot()[15:65, 530: 850]380 # screen[pos[1] - 30:pos[1] + 30, pos[0] - 330:pos[0] - 100]381 )[1]).tobytes())382 logger.info("收到请求结果：%s, 耗时：%.2f 秒" % ("".join(result), (time.time() - st)))383 if result is not None:384 # 判断当前坐标是否检测过385 logger.info("开始检测坐标是否已经检测过...")386 result = ",".join(re.findall(r'(?:\d){1,4}', result))387 if result not in r_index_plist:388 r_index_plist.append(result)389 print(r_index_plist)390 logger.info("检测到本次选矿是全新的位置")391 # 检测残矿392 if not allow_remnant_ore:393 logger.info("开始检测残矿...")394 for k in range(threshold):395 dev.touch((cx, cy))396 sleep(1)397 # 查找资源点详情界面398 screen = dev.snapshot()399 rs_pos = Template('item/res-sum.png').match_in(screen)400 if rs_pos:401 # 查找<<采集>>蓝色按钮402 screen = dev.snapshot()403 save_touch_screen((cx, cy), screen)404 pos = Template('item/gather-string.png').match_in(screen)405 if pos:406 dev.touch(pos)407 sleep(1)408 logger.info("开始检测当前搜索矿资源储量...")409 # 农田和木材 6|1008000 5|756000 4|504000 3|378000 2|252000 1|126000410 # 石头 6|756000 5|567000 4|378000 3|283500 2|189000 1|94500411 # 金币 6|33600 5|252000 4|168000 3|126000 2|84000 1|42000412 tvs = (413 1008000, 756000, 567000, 504000, 378000, 336000, 283500, 252000,414 189000, 126000,415 94500,416 84000, 42000)417 logger.info("开始向第三方平台发送 OCR 识别请求资源点资源储量...")418 st = time.time()419 result = img_to_str(420 np.array(cv2.imencode('.jpg',421 screen[rs_pos[1] - 25:rs_pos[1] + 25,422 rs_pos[0] + 300:rs_pos[0] + 550]423 )[1]).tobytes())424 logger.info(425 "收到请求结果：%s, 耗时：%.2f 秒" % ("".join(result), (time.time() - st)))426 if result is not None:427 if isinstance(result, str):428 v = int(result.replace(",", ""))429 elif isinstance(result, list):430 v = int("".join(result))431 if v in tvs:432 is_continue = False433 logger.info("检测到当前资源矿储量完整(%s)，开始采集..." % v)434 else:435 is_continue = True436 logger.info("检测到当前资源点储量 %s， 判定为残矿，开始重新选矿" % v)437 dev.touch((cx, cy))438 else:439 is_continue = True440 search_click_counter = 0441 logger.info("检测到当前资源点已经被其他执政官占领，忽略当前矿，进行再选矿...")442 dev.touch((cx, cy))443 break444 else:445 logger.info("检测次数超出阈值仍未检测到相应标志，退出采集任务")446 return447 if not is_continue:448 break449 else:450 logger.info("忽略残矿检查，开始采集...")451 break452 else:453 is_continue = True454 if same_pos_times > 2:455 is_down = True456 else:457 is_down = False458 same_pos_times += 1459 dev.touch((cx, cy))460 logger.info("当前采集坐标(%s)，已被检测过判定为不适合采集坐标，忽略当前矿，进行再选矿..." % result)461 else:462 logger.info("检测到%s资源矿<<搜索>>文字按钮..." % img_string[r_index])463 search_click_counter += 1464 dev.touch(pos)465 logger.info("点击%s资源矿<<搜索>>文字按钮..." % img_string[r_index])466 save_touch_screen(pos, screen)467 sleep(1)468 if is_continue:469 search_click_counter = 0470 logger.info("开始检测资源采集界面入口标志<<搜索>>图标")471 pos = Template("item/search.png").match_in(dev.snapshot())472 if pos:473 logger.info("检测到资源采集界面入口标志<<搜索>>图标")474 dev.touch(pos)475 sleep(1)476 if is_down or search_click_counter > 1:477 search_click_counter = 0478 screen = dev.snapshot()479 logger.info("多次搜索当前等级资源矿，仍无适合采集矿，开始递减一次矿等级")480 logger.info("开始检测资源矿递减标志<<->>图标")481 pos = Template("item/jian.png").match_in(screen)482 if pos:483 same_pos_times = 0484 logger.info("检测到资源矿递减标志<<->>图标")485 if tlc == 5:486 dev.touch((cx, cy))487 flag = True488 logger.info("检测到遍历所有等级都未搜索到可采集点，进行重新选取其他种类矿")489 break490 tlc += 1491 dev.touch(pos)492 save_touch_screen(pos, screen)493 sleep(0.5)494 logger.info("资源矿搜索等级当前等级%d" % (6 - tlc))495 is_down = False496 is_continue = False497 logger.info("资源矿采集保证循环检测... (%d/%d)" % (j + 1, 2 * threshold))498 else:499 logger.info("检测次数超出阈值仍未检测到相应标志，退出采集任务")500 return501 if flag:502 index_black[r_index] = True503 logger.info("添加%s资源矿至本次采集黑名单" % img_string[r_index])504 continue505 for i in range(threshold):506 pos = Template('item/create-troop-string.png').match_in(dev.snapshot())507 if pos:508 logger.info("开始创建采集部队")509 dev.touch(pos)510 sleep(1)511 break512 pos = Template('item/march-string.png').match_in(dev.snapshot())513 if pos:514 logger.info("检测到队列已满，取消当前采集行军操作！")515 dev.touch((cx, cy))516 return517 else:518 logger.info("检测次数超出阈值仍未检测到相应标志，退出采集任务")519 return520 for i in range(threshold):521 pos = Template('item/march-string.png').match_in(dev.snapshot())522 if pos:523 dev.touch(pos)524 logger.info("已派遣采集行军部队")525 sc_counter += 1526 sleep(1)527 break528 else:529 logger.info("检测次数超出阈值仍未检测到相应标志，退出采集任务")530 return531 if times is not None and times == sc_counter:532 break533 logger.info("操作标志循环检测中... (%d/%d)" % (i + 1, threshold))534 else:535 logger.info("检测次数超出阈值仍未检测到相应标志，退出采集任务")536if __name__ == "__main__":537 # if not cli_setup():538 logger = init_logging()539 logger.info("连接安卓模拟器中...")540 is_ok = False541 os.system("open /Applications/NemuPlayer.app")542 while not is_ok:543 try:544 auto_setup(__file__, logdir=True, devices=[545 "android:///%s" % (VM if DEBUG else WIRE_VIVO),546 ])547 dev = device()548 is_ok = True549 except:550 os.system("adb devices")551 sleep(1)552 logger.info("已连接上安卓模拟器")553 if not dev.check_app(ROK_PACKAGE_NAME):554 logger.error("检测到当前设备未安装 ROK 应用，自动化脚本执行结束")555 sys.exit(1)556 has_locked_rs_index = [0, 1, 2]557 alliance_name = '[11TC]'558 help_counter = 0559 loop_counter = 1560 forced_logout_counter = 1561 open_offline_sleep = True562 running_time = random.randint(60 * 30, 60 * 50)563 logger.info("ROK 自动化脚本开始运行...")564 logger.info("防封号保护：游戏将在运行%d分钟后进行下线休眠" % int(running_time / 60))565 start_time = time.time()566 while True:567 logger.info('守屏卫士第%d次巡逻中...' % loop_counter)568 flag = False569 # 检测应用是否正常运行中570 pkg_name = dev.get_top_activity()[0]571 if pkg_name != ROK_PACKAGE_NAME:572 logger.info("检测到当前 Top Activity 非 ROK 应用")573 logger.info("开始启动 ROK 应用，等待游戏进入执政官城堡主界面...")574 dev.start_app_timing(ROK_PACKAGE_NAME, ROK_MAIN_ACTIVITY)575 while not Template(r'item/home-map.png').match_in(dev.snapshot()):576 sleep(1)577 # 查找帮助小图标578 screen = dev.snapshot()579 pos = Template(r"item/help.png").match_in(screen)580 if pos is not None:581 flag = True582 x, y = pos583 help_counter += 1584 logger.info('检测到<<帮助>>图标(%d, %d)，小手第%d次点了起来' % (x, y, help_counter))585 dev.touch(pos)586 save_touch_screen(pos, screen)587 # 检查队列588 screen = dev.snapshot()589 pos = Template("item/troop-flag2.png").match_in(screen)590 try:591 if pos:592 save_touch_screen(pos, screen)593 # screen = screen[pos[1] - 25:pos[1] + 15, pos[0] + 60:pos[0] + 130]594 screen = screen[pos[1] - 25:pos[1] + 30, pos[0] + 60:pos[0] + 130]595 save_touch_screen([0, 0], screen)596 result = img_to_str(np.array(cv2.imencode('.jpg', screen)[1]).tobytes())597 if result is not None:598 result = re.findall('(?:^\d{1}|\d{1}$)', result)599 ut = int(result[0])600 st = int(result[1])601 if ut < st:602 flag = True603 logger.info("检测到空闲队列(%d/%d), 开始种田..." % (ut, st))604 farm(times=st - ut, index=has_locked_rs_index)605 else:606 pos = Template("item/home-map.png").match_in(screen) or Template("item/home-tower.png").match_in(screen)607 if pos:608 flag = True609 logger.info("检测到所有队列空闲，开始种田...")610 farm(index=has_locked_rs_index)611 except:612 logger.error("种田出现异常：%s" % traceback.format_exc())613 # 断开连接614 pos = Template('item/disconnect_tip_strings.png').match_in(dev.snapshot())615 if pos:616 pos = Template('item/blue_sure_string.png').match_in(dev.snapshot())617 if pos:618 dev.touch(pos)619 stop_app(ROK_PACKAGE_NAME)620 logger.info('检测到网络中断与游戏服务器断开连接，终止游戏应用进行重新启动操作')621 # 检测顶号622 pos = Template('item/logout_tip_strings.png').match_in(dev.snapshot())623 if pos:624 stop_app(ROK_PACKAGE_NAME)625 logger.info('检测到被顶号，十分钟后重新启动游戏应用')626 # 五分钟后重启627 st = time.time()628 while time.time() - st <= 30 * 60:629 sleep(1)630 start_time = time.time()631 logger.info("由于被顶号原因，游戏运行时长重新计时")632 logger.info("游戏已进入执政官城堡主界面，开始自动化任务...")633 continue634 else:635 home()636 # 查找人机验证奖励箱子637 screen = dev.snapshot()638 pos = Template(r"item/vcase.png").match_in(screen)639 if pos is not None:640 x, y = pos641 logger.info('检测到人机验证<<箱子>>图标(%d, %d)，进入验证步骤' % (x, y))642 dev.touch(pos)643 save_touch_screen(pos, screen)644 pass_geetest_vcode()645 # 漫游者646 # walker()647 sleep(30)648 loop_counter += 1649 end_time = time.time()650 logger.info("轮号倒计时：%d 秒" % (900-int(end_time-start_time)))651 if end_time - start_time >= 900:652 switch_role()...

Full Screen

Full Screen

tafor.py

Source:tafor.py Github

copy

Full Screen

1#!/usr/bin/env python2"""3This module defines the tafor class. A tafor object represents the weather report encoded by a single tafor code.4"""5__author__ = "Pablo Rozas-Larraondo"6__email__ = "p.rozas.larraondo@gmail.com"7__version__ = "0.1"8__LICENSE__ = """9Copyright (c) 2014, %s10All rights reserved.11Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:12Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.13Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.14THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.15""" % __author__16import re17import datetime18#from Datatypes import *19## regular expressions to decode various groups of the tafor code20MISSING_RE = re.compile(r"^[M/]+$")21TYPE_RE = re.compile(r"^(?P<type>TAF)\s+")22MODIFIER_RE = re.compile(r"^(?P<mod>AUTO|AMD|COR|NIL)\s+")23STATION_RE = re.compile(r"^(?P<station>[A-Z]{4})\s+")24TIME_RE = re.compile(r"""^(?P<day>[\d]{2})25 (?P<hour>[\d]{2})26 (?P<min>[\d]{2})Z?\s+""",27 re.VERBOSE)28TIMEVALID_RE = re.compile(r"""^(?P<dayfrom>[\d]{2})29 (?P<hourfrom>[\d]{2})/30 (?P<dayto>[\d]{2})31 (?P<hourto>[\d]{2})\s+""",32 re.VERBOSE)33WIND_RE = re.compile(r"""^(?P<dir>[\d]{3}|///|VRB)34 (?P<speed>[\d]{2}|//)35 (G(?P<gust>(\d{1,3}|[/M]{1,3})))?36 (?P<units>KT|KMH|MPS)?\s+""",37 re.VERBOSE)38VISIBILITY_RE = re.compile(r"""^(?P<vis>(?P<dist>\d\d\d\d|////)39 (?P<dir>[NSEW][EW]? | NDV)? | CAVOK )\s+""", 40 re.VERBOSE)41WEATHER_RE = re.compile(r"""^(?P<int>(-|\+|VC)*)42 (?P<desc>(MI|PR|BC|DR|BL|SH|TS|FZ)+)?43 (?P<prec>(DZ|RA|SN|SG|IC|PL|GR|GS|UP|/)*)44 (?P<obsc>BR|FG|FU|VA|DU|SA|HZ|PY)?45 (?P<other>PO|SQ|FC|SS|DS|NSW|/+)?46 (?P<int2>[-+])?\s+""",47 re.VERBOSE)48SKY_RE= re.compile(r"""^(?P<cover>VV|CLR|SKC|SCK|NSC|NCD|BKN|SCT|FEW|OVC|///)49 (?P<height>[\d]{3}|///)?50 (?P<cloud>([A-Z][A-Z]+|///))?\s+""",51 re.VERBOSE)52MAXTEMP_RE = re.compile(r"""^TX(?P<temp>(M|-)?[\d]{2}|//)/53 (?P<day>[\d]{2})54 (?P<hour>[\d]{2})Z?\s+""",55 re.VERBOSE)56MINTEMP_RE = re.compile(r"""^TN(?P<temp>(M|-)?[\d]{2}|//)/57 (?P<day>[\d]{2})58 (?P<hour>[\d]{2})Z?\s+""",59 re.VERBOSE)60CHANGES_RE_OLD = re.compile(r"""^(?P<prob>PROB[\d]{2})?61 (?P<change>(FM[\d]{6}|BECMG|TEMPO|INTER)*)?\s+""",62 re.VERBOSE)63CHANGES_RE = re.compile(r"""^(?P<prob>PROB[\d]{2})?\s*64 (?P<change>FM|BECMG|TEMPO|INTER)?65 (?P<from>[\d]{6})?\s+""",66 re.VERBOSE)67class Tafor(object):68 """TAFOR (terminal aerodrome forecast)"""69 70 def __init__( self, tafor_code):71 """Parse raw tafor code."""72 self.tafor = dict() # Tafor object containing all information73 self.tafor['raw'] = tafor_code74 tafor_code = tafor_code + ' '75 try:76 # Report type77 match = TYPE_RE.match(tafor_code)78 if match:79 self.tafor['type'] = match.group('type')80 tafor_code = tafor_code[match.end():]81 # Modifier82 match = MODIFIER_RE.match(tafor_code)83 if match:84 self.tafor['mod'] = match.group('mod')85 tafor_code = tafor_code[match.end():]86 # Airport code87 match = STATION_RE.match(tafor_code)88 if match:89 self.tafor['code'] = match.group('station')90 tafor_code = tafor_code[match.end():]91 # Datetime92 match = TIME_RE.match(tafor_code)93 if match:94 now = datetime.datetime.now()95 tafor_date = datetime.datetime(now.year, now.month, int(match.group('day')), int(match.group('hour')), int(match.group('min')))96 self.tafor['datetime'] = tafor_date97 tafor_code = tafor_code[match.end():]98 # Time valid99 match = TIMEVALID_RE.match(tafor_code)100 if match:101 validity = {}102 now = datetime.datetime.now()103 if match.group('hourto') == '24':104 valid_from = datetime.datetime(now.year, now.month, int(match.group('dayfrom')), int(match.group('hourfrom')), 0)105 validity["from"] = valid_from106 valid_to = datetime.datetime(now.year, now.month, int(match.group('dayto')), 0, 0)107 valid_to = valid_to + datetime.timedelta(days=1)108 validity["to"] = valid_to109 else:110 valid_from = datetime.datetime(now.year, now.month, int(match.group('dayfrom')), int(match.group('hourfrom')), 0)111 validity["from"] = valid_from112 valid_to = datetime.datetime(now.year, now.month, int(match.group('dayto')), int(match.group('hourto')), 0)113 validity["to"] = valid_to114 self.tafor['validity'] = validity115 tafor_code = tafor_code[match.end():]116 # Wind117 match = WIND_RE.match(tafor_code)118 if match:119 self.tafor['wind'] = match.groupdict()120 tafor_code = tafor_code[match.end():]121 # Visibility122 match = VISIBILITY_RE.match(tafor_code)123 if match:124 self.tafor['visibility'] = []125 while match:126 self.tafor['visibility'].append(match.group('vis'))127 tafor_code = tafor_code[match.end():]128 match = VISIBILITY_RE.match(tafor_code)129 # Weather130 match = WEATHER_RE.match(tafor_code)131 if match:132 self.tafor['weather'] = []133 while match:134 self.tafor['weather'].append(match.groupdict())135 tafor_code = tafor_code[match.end():]136 match = WEATHER_RE.match(tafor_code)137 # Sky138 match = SKY_RE.match(tafor_code)139 if match:140 self.tafor['sky'] = []141 while match:142 self.tafor['sky'].append(match.groupdict())143 tafor_code = tafor_code[match.end():]144 match = SKY_RE.match(tafor_code)145 # Changes146 match = CHANGES_RE.match(tafor_code)147 if match:148 self.tafor['changes'] = []149 while match:150 tafor_code = tafor_code[match.end():]151 change = {}152 change["prob"] = match.group('prob')153 change["modifier"] = match.group('change')154 validity = {}155 now = datetime.datetime.now()156 if match.group('from') != None:157 valid_from = datetime.datetime(now.year, now.month, int(match.group('from')[:2]), int(match.group('from')[2:-2]), int(match.group('from')[-2:]))158 validity["from"] = valid_from159 change["validity"] = validity160 # Validity161 match_in = TIMEVALID_RE.match(tafor_code)162 if match_in:163 print(match.group('change'))164 validity = {}165 now = datetime.datetime.now()166 if match_in.group('hourto') == '24':167 valid_from = datetime.datetime(now.year, now.month, int(match_in.group('dayfrom')), int(match_in.group('hourfrom')), 0)168 validity["from"] = valid_from169 valid_to = datetime.datetime(now.year, now.month, int(match_in.group('dayto')), 0, 0)170 valid_to = valid_to + datetime.timedelta(days=1)171 validity["to"] = valid_to172 else:173 valid_from = datetime.datetime(now.year, now.month, int(match_in.group('dayfrom')), int(match_in.group('hourfrom')), 0)174 validity["from"] = valid_from175 valid_to = datetime.datetime(now.year, now.month, int(match_in.group('dayto')), int(match_in.group('hourto')), 0)176 validity["to"] = valid_to177 change["validity"] = validity178 tafor_code = tafor_code[match_in.end():]179 # Wind180 match_in = WIND_RE.match(tafor_code)181 if match_in:182 change['wind'] = match_in.groupdict()183 tafor_code = tafor_code[match_in.end():]184 # Visibility185 match_in = VISIBILITY_RE.match(tafor_code)186 if match_in:187 change['visibility'] = []188 while match_in:189 change['visibility'].append(match_in.group('vis'))190 tafor_code = tafor_code[match_in.end():]191 match_in = VISIBILITY_RE.match(tafor_code)192 # Weather193 match_in = WEATHER_RE.match(tafor_code)194 if match_in:195 change['weather'] = []196 while match_in:197 change['weather'].append(match_in.groupdict())198 tafor_code = tafor_code[match_in.end():]199 match_in = WEATHER_RE.match(tafor_code)200 # Sky201 match_in = SKY_RE.match(tafor_code)202 if match_in:203 change['sky'] = []204 while match_in:205 change['sky'].append(match_in.groupdict())206 tafor_code = tafor_code[match_in.end():]207 match_in = SKY_RE.match(tafor_code)208 self.tafor['changes'].append(change)209 match = CHANGES_RE.match(tafor_code)210 # Quotient211 self.tafor['quotient'] = tafor_code212 except Exception, err:...

Full Screen

Full Screen

utils.py

Source:utils.py Github

copy

Full Screen

1import uuid2from typing import Callable3import pandas4import re5from cpgintegrate import ProcessingException6def match_indices(match_from: pandas.DataFrame, match_in: pandas.DataFrame,7 index_transform: Callable = lambda x: int(re.compile(r'[^\d]+').sub("", x))) -> pandas.DataFrame:8 """9 Replace index values in a frame by searching using transformed index in another10 Replace indices of match_from with first instance found in match_in when both are transformed using index_transform.11 KeyError if any index of match_from is missing.12 :param match_from: DataFrame whose index values I should check and replace13 :param match_in: DataFrame which I should search in14 :param index_transform: Callable that acts on index value, default takes numeric parts15 :return:16 """17 match_in_reindexed = (match_in18 .drop(columns=match_in.columns)19 .assign(orig_index=lambda df: df.index)20 .pipe(lambda df: df.set_index(df.index.map(index_transform)))21 # Only want one row for each transformed index to avoid multiplying rows in match_from frame22 .groupby(level=0)23 .first())24 matched = (25 match_from26 .assign(**{'transformed_index': lambda df: df.index.map(index_transform)})27 .join(match_in_reindexed, on='transformed_index')28 )29 assert not matched.orig_index.isnull().any(), "Bad IDs {0}".format(matched.loc[matched.orig_index.isnull(),30 'transformed_index'])31 return match_from.set_index(matched.set_index('orig_index').rename_axis(match_from.index.name).index)32def edit_using(frame_to_edit: pandas.DataFrame, edits: pandas.DataFrame) -> pandas.DataFrame:33 """34 Edit DataFrame using list of edits in another DataFrame.35 Alters each target_field value in frame_to_edit to target_value where match_field == match_value, or removes the row36 if remove_row is > 037 :param frame_to_edit: DataFrame to edit38 :param edits: DataFrame with edits in columns match_field, match_value, target_field, target_value39 optionally remove_row40 :return:41 """42 frame_to_edit = frame_to_edit.copy(deep=True)43 # remove_rows44 if "remove_row" in edits.columns:45 for _, remove_spec in edits[edits.remove_row > 0].iterrows():46 frame_to_edit = frame_to_edit[~(frame_to_edit[remove_spec.match_field] == remove_spec.match_value)]47 # Use temporary name to avoid clobbering any existing column48 temp_col_name = None49 orig_index_name = None50 if frame_to_edit.index.name and (frame_to_edit.index.name in edits.target_field.values):51 temp_col_name = str(uuid.uuid4())52 orig_index_name = frame_to_edit.index.name53 frame_to_edit[temp_col_name] = frame_to_edit.index54 edits.loc[edits.target_field == orig_index_name, 'target_field'] = temp_col_name55 for _, row in edits.loc[edits.target_field.fillna("") != ""].iterrows():56 try:57 frame_to_edit.loc[frame_to_edit[row.match_field] == row.match_value, row.target_field] \58 = row.get("target_value", None)59 except KeyError:60 raise ProcessingException("Edits error on %s , %s, %s, %s"61 % (row.match_field, row.match_value,62 row.target_field, row.get("target_value", None)))63 if temp_col_name:64 return frame_to_edit.set_index(temp_col_name).rename_axis(orig_index_name)65 return frame_to_edit66def replace_indices(target: pandas.DataFrame, index_source: pandas.DataFrame) -> pandas.DataFrame:67 return target.set_index(68 target.index.to_series().replace(to_replace=list(index_source.iloc[:, 0]), value=list(index_source.index))...

Full Screen

Full Screen

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:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run Airtest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful