KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use set_idx method in Airtest
Best Python code snippet using Airtest
matchdata.py
Source:matchdata.py 
1"""Matchdata."""2import difflib3import json4import logging5import re6from PyQt5.QtCore import QObject, pyqtSignal7import hwctool.settings8# create logger9module_logger = logging.getLogger('hwctool.matchdata')10class matchData(QObject):11 """Matchdata."""12 dataChanged = pyqtSignal(str, object)13 metaChangedSignal = pyqtSignal()14 def __init__(self, controller):15 """Init and define custom providers."""16 super().__init__()17 self.__rawData = None18 self.__controller = controller19 self.__initData()20 self.emitLock = EmitLock()21 def __emitSignal(self, scope, name='', data=None):22 if not self.emitLock.locked():23 if scope == 'data':24 self.dataChanged.emit(name, data)25 elif scope == 'meta':26 self.metaChangedSignal.emit()27 elif scope == 'outcome':28 self.dataChanged.emit('outcome', self.getWinner())29 for idx in range(self.getNoSets()):30 if self.getMapScore(idx) == 0:31 colorData = self.getColorData(idx)32 self.__emitSignal(33 'data', 'color', {34 'set_idx': idx,35 'score_color': colorData["score_color"],36 'border_color': colorData["border_color"],37 'hide': colorData["hide"],38 'opacity': colorData["opacity"]})39 def readJsonFile(self):40 """Read json data from file."""41 try:42 with open(hwctool.settings.getJsonFile('matchdata'),43 'r', encoding='utf-8-sig') as json_file:44 self.__data = json.load(json_file)45 except Exception as e:46 # module_logger.exception("message")47 self.setCustom(5)48 finally:49 self.__emitSignal('meta')50 def writeJsonFile(self):51 """Write json data to file."""52 try:53 with open(hwctool.settings.getJsonFile('matchdata'),54 'w', encoding='utf-8-sig') as outfile:55 json.dump(self.__data, outfile)56 except Exception as e:57 module_logger.exception("message")58 def __str__(self):59 """Return match data as string."""60 return str(self.__data)61 def isValid(self):62 """Check if data is valid."""63 return self.__data is not None64 def __initData(self):65 self.__data = {}66 self.__data['league'] = "TBD"67 self.__data['id'] = 068 self.__data['matchlink'] = ""69 self.__data['no_sets'] = 070 self.__data['best_of'] = 071 self.__data['min_sets'] = 072 self.__data['allkill'] = False73 self.__data['solo'] = True74 self.__data['my_team'] = 075 self.__data['swapped'] = False76 self.__data['teams'] = []77 self.__data['teams'].append({'name': 'TBD', 'tag': None})78 self.__data['teams'].append({'name': 'TBD', 'tag': None})79 self.__data['sets'] = []80 self.__data['players'] = [[], []]81 def swapTeams(self):82 module_logger.info("Swapping teams")83 self.__data['swapped'] = not self.__data.get('swapped', False)84 self.__data['my_team'] = -self.__data['my_team']85 self.__data['teams'][1], self.__data['teams'][0] = \86 self.__data['teams'][0], self.__data['teams'][1]87 self.__data['players'][1], self.__data['players'][0] = \88 self.__data['players'][0], self.__data['players'][1]89 for set_idx in range(len(self.__data['sets'])):90 self.__data['sets'][set_idx]['score'] = - \91 self.__data['sets'][set_idx]['score']92 self.__emitSignal('meta')93 def getSwappedIdx(self, idx):94 if self.isSwapped():95 return 1 - idx96 else:97 return idx98 def isSwapped(self):99 return bool(self.__data.get('swapped', False))100 def resetSwap(self):101 self.__data['swapped'] = False102 def setMinSets(self, minSets):103 """Set minium number of sets that are played."""104 if(minSets > 0):105 if(minSets > self.getBestOfRaw()):106 self.__data['min_sets'] = self.getBestOfRaw()107 else:108 self.__data['min_sets'] = int(minSets)109 else:110 self.__data['min_sets'] = 0111 def getMinSets(self):112 """Get the minium number of sets that are played."""113 try:114 return int(self.__data['min_sets'])115 except Exception:116 return 0117 def setSolo(self, solo):118 """Set allkill format."""119 self.__data['solo'] = bool(solo)120 if self.__data['solo']:121 for set_idx in range(self.getNoSets()):122 for team_idx in range(2):123 self.setPlayer(team_idx, set_idx,124 self.getPlayer(team_idx, 0))125 def getSolo(self):126 """Check if format is solo (or team)."""127 return bool(self.__data['solo'])128 def setAllKill(self, allkill):129 """Set allkill format."""130 self.__data['allkill'] = bool(allkill)131 def getAllKill(self):132 """Check if format is allkill."""133 return bool(self.__data['allkill'])134 def allkillUpdate(self):135 """Move the winner to the next set in case of allkill format."""136 if(not self.getAllKill()):137 return False138 for set_idx in range(self.getNoSets()):139 if self.getMapScore(set_idx) == 0:140 if(set_idx == 0):141 continue142 previous_score = self.getMapScore(set_idx - 1)143 if previous_score == 0:144 continue145 team_idx = int((previous_score + 1) / 2)146 player = self.getPlayer(team_idx, set_idx).strip().lower()147 if(player != "tbd" and player != ""):148 continue149 self.setPlayer(team_idx, set_idx, self.getPlayer(team_idx,150 set_idx - 1),151 self.getRace(team_idx, set_idx - 1))152 return True153 return False154 def setCustom(self, bestof, allkill=False, solo=True):155 """Set a custom match format."""156 bestof = int(bestof)157 allkill = bool(allkill)158 if(bestof == 2):159 no_sets = 2160 else:161 no_sets = bestof + 1 - bestof % 2162 self.setNoSets(no_sets, bestof)163 self.resetLabels()164 self.setAllKill(allkill)165 self.setID(0)166 self.setURL("")167 self.setSolo(solo)168 def resetData(self, reset_options=True):169 """Reset all data to default values."""170 with self.emitLock():171 for team_idx in range(2):172 for set_idx in range(self.getNoSets()):173 self.setPlayer(team_idx, set_idx, "TBD", "Random")174 self.setTeam(team_idx, "TBD", "TBD")175 for set_idx in range(self.getNoSets()):176 self.setMapScore(set_idx, 0, overwrite=True)177 self.setMap(set_idx)178 self.setAce(set_idx, False)179 self.setLeague("TBD")180 self.setMyTeam(0)181 if reset_options:182 self.setAllKill(False)183 self.setSolo(True)184 self.__emitSignal('meta')185 def resetLabels(self):186 """Reset the map labels."""187 best_of = self.__data['best_of']188 no_sets = self.getNoSets()189 if(best_of == 2):190 for set_idx in range(no_sets):191 self.setLabel(set_idx, "Map " + str(set_idx + 1))192 return193 ace_start = no_sets - 3 + 2 * (best_of % 2)194 skip_one = (ace_start + 1 == no_sets)195 ace = (best_of % 2) == 0196 for set_idx in range(no_sets):197 self.setLabel(set_idx, "Map " + str(set_idx + 1))198 self.setAce(set_idx, False)199 if ace:200 for set_idx in range(ace_start, no_sets):201 self.setAce(set_idx, True)202 if(skip_one):203 self.setLabel(set_idx, "Ace Map")204 else:205 self.setLabel(set_idx, "Ace Map " +206 str(set_idx - ace_start + 1))207 def setNoSets(self, no_sets=5, bestof=False, resetPlayers=False):208 """Set the number of sets/maps."""209 try:210 no_sets = int(no_sets)211 if(no_sets < 0):212 no_sets = 0213 elif(no_sets > hwctool.settings.max_no_sets):214 no_sets = hwctool.settings.max_no_sets215 if((not bestof) or bestof <= 0 or bestof > no_sets):216 self.__data['best_of'] = no_sets217 else:218 self.__data['best_of'] = int(bestof)219 sets = []220 players = [[], []]221 for i in range(no_sets):222 try:223 map = self.__data['sets'][i]['map']224 except Exception:225 map = "TBD"226 try:227 score = self.__data['sets'][i]['score']228 except Exception:229 score = 0230 try:231 label = self.__data['sets'][i]['label']232 except Exception:233 label = 'Map ' + str(i + 1)234 try:235 ace = bool(self.__data['sets'][i].get('ace', False))236 except Exception:237 ace = False238 for j in range(2):239 if(not resetPlayers):240 try:241 player_name = self.__data['players'][j][i]['name']242 except Exception:243 player_name = 'TBD'244 try:245 player_race = getRace(246 self.__data['players'][j][i]['race'])247 except Exception:248 player_race = 'Random'249 else:250 player_name = 'TBD'251 player_race = 'Random'252 players[j].append(253 {'name': player_name, 'race': player_race})254 sets.append({'label': label, 'map': map,255 'score': score, 'ace': ace})256 self.__data['no_sets'] = no_sets257 self.__data['min_sets'] = 0258 self.__data['sets'] = sets259 self.__data['players'] = players260 except Exception as e:261 module_logger.exception("message")262 def setMyTeam(self, myteam, swap=False):263 """Set "my team"."""264 if(isinstance(myteam, str)):265 new = self.__selectMyTeam(myteam)266 elif(myteam in [-1, 0, 1]):267 new = myteam268 else:269 return False270 if(new != self.__data['my_team']):271 self.__data['my_team'] = new272 for i in range(self.getNoSets()):273 score = self.getMapScore(i)274 colorData = self.getColorData(i)275 self.__emitSignal(276 'data',277 'color-data', {278 'set_idx': i,279 'score': score,280 'score_color': colorData['score_color'],281 'border_color': colorData['border_color'],282 'hide': colorData['hide'],283 'opacity': colorData['opacity']})284 if swap and int(self.__data['my_team']) > 0:285 self.swapTeams()286 return True287 return False288 def getMyTeam(self):289 """Return my team: (-1,0,1)."""290 try:291 return int(self.__data['my_team'])292 except Exception:293 return 0294 def __selectMyTeam(self, string):295 teams = [self.getTeam(0).lower(), self.getTeam(1).lower()]296 matches = difflib.get_close_matches(string.lower(), teams, 1)297 if(len(matches) != 1):298 return 0299 elif(matches[0] == teams[0]):300 return -1301 else:302 return 1303 def getNoSets(self):304 """Get number of sets."""305 try:306 return int(self.__data['no_sets'])307 except Exception:308 return 0309 def setMap(self, set_idx, map="TBD"):310 """Set the map of a set."""311 try:312 if(not (set_idx >= 0 and set_idx < self.__data['no_sets'])):313 return False314 map, _ = autoCorrectMap(map)315 if(self.__data['sets'][set_idx]['map'] != map):316 self.__data['sets'][set_idx]['map'] = map317 self.__emitSignal(318 'data', 'map', {'set_idx': set_idx, 'value': map})319 return True320 except Exception:321 return False322 def getMap(self, set_idx):323 """Get the map of a set."""324 try:325 if(not (set_idx >= 0 and set_idx < self.__data['no_sets'])):326 return False327 return str(self.__data['sets'][set_idx]['map'])328 except Exception:329 return False330 def yieldMaps(self):331 yielded = set()332 for idx in range(self.getNoSets()):333 map = self.getMap(idx)334 if map and map.lower() != "TBD" and map not in yielded:335 yield map336 yielded.add(map)337 def getScoreString(self, middleStr=':'):338 """Get the score as a string."""339 score = self.getScore()340 return str(score[0]) + middleStr + str(score[1])341 def getScore(self):342 """Get the score as an list."""343 score = [0, 0]344 for set_idx in range(self.getNoSets()):345 map_score = self.getMapScore(set_idx)346 if(map_score < 0):347 score[0] += 1348 elif(map_score > 0):349 score[1] += 1350 else:351 continue352 return score353 def getBestOfRaw(self):354 """Get raw BestOf number."""355 try:356 return int(self.__data['best_of'])357 except Exception:358 return False359 def getBestOf(self):360 """Get flitered BestOf number (only odd)."""361 try:362 best_of = self.__data['best_of']363 if(best_of == 2):364 return 3365 if(best_of % 2): # odd, okay366 return best_of367 else: # even368 score = self.getScore()369 if(min(score) < best_of / 2 - 1):370 return best_of - 1371 else:372 return best_of + 1373 return374 except Exception:375 return False376 def isDecided(self):377 """Check if match is decided."""378 return max(self.getScore()) > int(self.getBestOf() / 2)379 def getWinner(self):380 score = self.getScore()381 if not self.isDecided() or score[0] == score[1]:382 return 0383 elif score[0] > score[1]:384 return -1385 else:386 return 1387 def setMapScore(self, set_idx, score, overwrite=False, applySwap=False):388 """Set the score of a set."""389 try:390 if(not (set_idx >= 0 and set_idx < self.__data['no_sets'])):391 return False392 if(score in [-1, 0, 1]):393 if applySwap and self.isSwapped():394 score = -score395 if(overwrite or self.__data['sets'][set_idx]['score'] == 0):396 if(self.__data['sets'][set_idx]['score'] != score):397 was_decided = self.isDecided()398 self.__data['sets'][set_idx]['score'] = score399 outcome_changed = self.isDecided() != was_decided400 if outcome_changed:401 self.__emitSignal('outcome')402 self.__emitSignal('data', 'score',403 {'set_idx': set_idx,404 'value': score})405 return True406 else:407 return False408 except Exception:409 return False410 def getMapScore(self, set_idx):411 """Get the score of a set."""412 try:413 if(not (set_idx >= 0 and set_idx < self.__data['no_sets'])):414 return False415 return int(self.__data['sets'][set_idx]['score'])416 except Exception:417 return False418 def getNextSet(self, force=False):419 for set_idx in range(self.getNoSets()):420 if self.getMapScore(set_idx) == 0:421 return set_idx422 if force:423 return self.getNoSets() - 1424 else:425 return -1426 def getNextMap(self, next_idx=-1):427 set_idx = self.getNextSet()428 if set_idx != -1 and (next_idx == -1 or set_idx == next_idx):429 return self.getMap(set_idx)430 else:431 return "TBD"432 def wasMapPlayed(self, map):433 for set_idx in range(self.getNoSets()):434 if (map.lower() == self.getMap(set_idx).lower() and435 self.getMapScore(set_idx) != 0):436 return True437 return False438 def getNextPlayer(self, team_idx):439 """Get the player of the next undecided set."""440 player = "TBD"441 for set_idx in range(self.getNoSets()):442 if self.getMapScore(set_idx) == 0:443 player = self.getPlayer(team_idx, set_idx)444 break445 return player446 def getNextRace(self, team_idx):447 """Get the players race of the next undecided set."""448 player = "Random"449 for set_idx in range(self.getNoSets()):450 if self.getMapScore(set_idx) == 0:451 player = self.getRace(team_idx, set_idx)452 break453 return player454 def setPlayer(self, team_idx, set_idx, name="TBD", race=False):455 """Set the player of a set."""456 try:457 if(not (set_idx >= 0 and set_idx < self.__data['no_sets'] and458 team_idx in range(2))):459 return False460 if(self.__data['players'][team_idx][set_idx]['name'] != name):461 self.__data['players'][team_idx][set_idx]['name'] = name462 self.__emitSignal('data', 'player', {463 'team_idx': team_idx,464 'set_idx': set_idx,465 'value': name})466 if(race):467 self.setRace(team_idx, set_idx, race)468 return True469 except Exception:470 return False471 def getPlayerList(self, team_idx):472 """Get complete player list of a team."""473 list = []474 try:475 for set_idx in range(self.getNoSets()):476 list.append(self.getPlayer(team_idx, set_idx))477 return list478 except Exception:479 return []480 def getPlayer(self, team_idx, set_idx):481 """Get the player (name) of a set."""482 try:483 if(not (set_idx >= 0 and set_idx < self.__data['no_sets'] and484 team_idx in range(2))):485 return False486 return self.__data['players'][team_idx][set_idx]['name'].strip()487 except Exception:488 return False489 def setRace(self, team_idx, set_idx, race="Random"):490 """Set a players race."""491 try:492 if(not (set_idx >= 0 and set_idx < self.__data['no_sets'] and493 team_idx in range(2))):494 return False495 race = getRace(race)496 if(self.__data['players'][team_idx][set_idx]['race'] != race):497 self.__data['players'][team_idx][set_idx]['race'] = race498 self.__emitSignal(499 'data', 'race', {'team_idx': team_idx,500 'set_idx': set_idx,501 'value': race})502 return True503 except Exception:504 return False505 def getRace(self, team_idx, set_idx):506 """Get a players race."""507 try:508 if(not (set_idx >= 0 and set_idx < self.__data['no_sets'] and509 team_idx in range(2))):510 return False511 return getRace(self.__data['players'][team_idx][set_idx]['race'])512 except Exception:513 return False514 def setAce(self, set_idx, ace):515 """Label set as ace."""516 ace = bool(ace)517 try:518 if(not (set_idx >= 0 and set_idx < self.__data['no_sets'])):519 return False520 if(self.__data['sets'][set_idx]['ace'] != ace):521 self.__data['sets'][set_idx]['ace'] = ace522 return True523 except Exception:524 return False525 def isAce(self, set_idx):526 """Return if set is labeld as ace."""527 try:528 return bool(self.__data['sets'][set_idx].get('ace', False))529 except Exception:530 return False531 def setLabel(self, set_idx, label):532 """Set a map label."""533 try:534 if(not (set_idx >= 0 and set_idx < self.__data['no_sets'])):535 return False536 if(self.__data['sets'][set_idx]['label'] != label):537 self.__data['sets'][set_idx]['label'] = label538 self.__emitSignal('data', 'map_label', {539 'set_idx': set_idx, 'value': label})540 return True541 except Exception:542 return False543 def getLabel(self, set_idx):544 """Get a map label."""545 try:546 if(not (set_idx >= 0 and set_idx < self.__data['no_sets'])):547 return False548 return str(self.__data['sets'][set_idx]['label'])549 except Exception:550 return False551 def setTeam(self, team_idx, name, tag=False):552 """Set a team name."""553 if team_idx not in range(2):554 return False555 new = str(name.strip())556 if(self.__data['teams'][team_idx]['name'] != new):557 self.__data['teams'][team_idx]['name'] = new558 self.__emitSignal('data', 'team', {'idx': team_idx, 'value': new})559 if(tag):560 self.setTeamTag(team_idx, tag)561 return True562 def getTeam(self, team_idx):563 """Get team name."""564 if team_idx not in range(2):565 return False566 return str(self.__data['teams'][team_idx]['name'])567 def getTeamOrPlayer(self, team_idx):568 """Get team name or player name depending on mode."""569 if self.getSolo():570 return self.getPlayer(team_idx, 0)571 else:572 return self.getTeam(team_idx)573 def setTeamTag(self, team_idx, tag):574 """Set team tag."""575 if team_idx not in range(2):576 return False577 new = str(tag)578 if(self.__data['teams'][team_idx]['tag'] != new):579 self.__data['teams'][team_idx]['tag'] = new580 return True581 def getTeamTag(self, team_idx):582 """Get team tag."""583 if team_idx not in range(2):584 return False585 name = str(self.__data['teams'][team_idx]['tag'])586 if(name):587 return str(name)588 else:589 return self.getTeam(team_idx)590 def setID(self, id):591 """Set match id."""592 self.__data['id'] = int(id)593 return True594 def getID(self):595 """Get match id."""596 return int(self.__data['id'])597 def setLeague(self, league):598 """Set league."""599 league = str(league)600 if(self.__data['league'] != league):601 self.__data['league'] = league602 self.__emitSignal('data', 'league', league)603 return True604 def getLeague(self):605 """Get league."""606 return self.__data['league']607 def setURL(self, url):608 """Set URL."""609 self.__data['matchlink'] = str(url)610 return True611 def getURL(self):612 """Get league."""613 return str(self.__data['matchlink'])614 def createStreamingTextFiles(self):615 """Create OBS txt files."""616 try:617 if(self.hasAnySetChanged()):618 f = open(hwctool.settings.getAbsPath(619 hwctool.settings.casting_data_dir +620 "/lineup.txt"),621 mode='w', encoding='utf-8-sig')622 f2 = open(hwctool.settings.getAbsPath(623 hwctool.settings.casting_data_dir +624 "/maps.txt"),625 mode='w', encoding='utf-8-sig')626 for idx in range(self.getNoSets()):627 map = self.getMap(idx)628 f.write(map + "\n")629 f2.write(map + "\n")630 try:631 string = self.getPlayer(632 0, idx) + ' vs ' + self.getPlayer(1, idx)633 f.write(string + "\n\n")634 except IndexError:635 f.write("\n\n")636 pass637 f.close()638 f2.close()639 try:640 score = self.getScore()641 score_str = str(score[0]) + " - " + str(score[1])642 except Exception:643 score_str = "0 - 0"644 f = open(hwctool.settings.getAbsPath(645 hwctool.settings.casting_data_dir +646 "/score.txt"), mode='w',647 encoding='utf-8-sig')648 f.write(score_str)649 f.close()650 f = open(hwctool.settings.getAbsPath(651 hwctool.settings.casting_data_dir +652 "/nextplayer1.txt"), mode='w',653 encoding='utf-8-sig')654 f.write(self.getNextPlayer(0))655 f.close()656 f = open(hwctool.settings.getAbsPath(657 hwctool.settings.casting_data_dir +658 "/nextplayer2.txt"), mode='w',659 encoding='utf-8-sig')660 f.write(self.getNextPlayer(1))661 f.close()662 f = open(hwctool.settings.getAbsPath(663 hwctool.settings.casting_data_dir +664 "/nextrace1.txt"), mode='w',665 encoding='utf-8-sig')666 f.write(self.getNextRace(0))667 f.close()668 f = open(hwctool.settings.getAbsPath(669 hwctool.settings.casting_data_dir +670 "/nextrace2.txt"), mode='w',671 encoding='utf-8-sig')672 f.write(self.getNextRace(1))673 f.close()674 if(self.hasMetaChanged()):675 f = open(hwctool.settings.getAbsPath(676 hwctool.settings.casting_data_dir +677 "/teams_vs_long.txt"),678 mode='w', encoding='utf-8-sig')679 f.write(self.getTeam(0) + ' vs ' + self.getTeam(1) + "\n")680 f.close()681 f = open(hwctool.settings.getAbsPath(682 hwctool.settings.casting_data_dir +683 "/teams_vs_short.txt"),684 mode='w', encoding='utf-8-sig')685 f.write(self.getTeamTag(0) + ' vs ' +686 self.getTeamTag(1) + "\n")687 f.close()688 f = open(hwctool.settings.getAbsPath(689 hwctool.settings.casting_data_dir +690 "/team1.txt"),691 mode='w', encoding='utf-8-sig')692 f.write(self.getTeam(0))693 f.close()694 f = open(hwctool.settings.getAbsPath(695 hwctool.settings.casting_data_dir +696 "/team2.txt"),697 mode='w', encoding='utf-8-sig')698 f.write(self.getTeam(1))699 f.close()700 f = open(hwctool.settings.getAbsPath(701 hwctool.settings.casting_data_dir +702 "/tournament.txt"),703 mode='w', encoding='utf-8-sig')704 f.write(self.getLeague())705 f.close()706 except Exception as e:707 module_logger.exception("message")708 def getScoreData(self):709 data = dict()710 if self.getSolo():711 data['team1'] = self.getPlayer(0, 0)712 data['team2'] = self.getPlayer(1, 0)713 else:714 data['team1'] = self.getTeam(0)715 data['team2'] = self.getTeam(1)716 score = [0, 0]717 winner = [False, False]718 sets = []719 threshold = int(self.getBestOf() / 2)720 for i in range(self.getNoSets()):721 sets.insert(i, ["", ""])722 if(max(score) > threshold and i >= self.getMinSets()):723 sets[i][0] = hwctool.settings.config.parser.get(724 "MapIcons", "notplayed_color")725 sets[i][1] = hwctool.settings.config.parser.get(726 "MapIcons", "notplayed_color")727 elif(self.getMapScore(i) == -1):728 sets[i][0] = hwctool.settings.config.parser.get(729 "MapIcons", "win_color")730 sets[i][1] = hwctool.settings.config.parser.get(731 "MapIcons", "lose_color")732 score[0] += 1733 elif(self.getMapScore(i) == 1):734 sets[i][0] = hwctool.settings.config.parser.get(735 "MapIcons", "lose_color")736 sets[i][1] = hwctool.settings.config.parser.get(737 "MapIcons", "win_color")738 score[1] += 1739 else:740 sets[i][0] = hwctool.settings.config.parser.get(741 "MapIcons", "undecided_color")742 sets[i][1] = hwctool.settings.config.parser.get(743 "MapIcons", "undecided_color")744 winner[0] = score[0] > threshold745 winner[1] = score[1] > threshold746 data['sets'] = sets747 data['winner'] = winner748 data['score1'] = score[0]749 data['score2'] = score[1]750 idx = self.getNextSet()751 if idx == -1:752 idx = self.getNoSets() - 1753 file = 'src/img/races/{}.png'754 data['logo1'] = file.format(self.getRace(0, idx).replace(' ', '_'))755 data['logo2'] = file.format(self.getRace(1, idx).replace(' ', '_'))756 return data757 def getScoreIconColor(self, team_idx, set_idx):758 score = self.getMapScore(set_idx)759 team = 2 * team_idx - 1760 if score == 0:761 if (set_idx >= self.getMinSets() and762 max(self.getScore()) > int(self.getBestOf() / 2)):763 return hwctool.settings.config.parser.get(764 "MapIcons",765 "notplayed_color")766 else:767 return hwctool.settings.config.parser.get(768 "MapIcons",769 "undecided_color")770 elif score == team:771 return hwctool.settings.config.parser.get(772 "MapIcons",773 "win_color")774 else:775 return hwctool.settings.config.parser.get(776 "MapIcons",777 "lose_color")778 def getColorData(self, set_idx):779 score = self.getMapScore(set_idx)780 team = self.getMyTeam()781 won = score * team782 hide = team == 0783 opacity = hwctool.settings.config.parser.get(784 "MapIcons",785 "notplayed_opacity")786 if won == 1:787 border_color = hwctool.settings.config.parser.get(788 "MapIcons",789 "win_color")790 score_color = border_color791 opacity = 0.0792 elif won == -1:793 border_color = hwctool.settings.config.parser.get(794 "MapIcons",795 "lose_color")796 score_color = border_color797 opacity = 0.0798 else:799 if (score == 0 and800 set_idx >= self.getMinSets() and801 max(self.getScore()) > int(self.getBestOf() / 2)):802 border_color = hwctool.settings.config.parser.get(803 "MapIcons",804 "notplayed_color")805 score_color = border_color806 else:807 border_color = hwctool.settings.config.parser.get(808 "MapIcons",809 "default_border_color")810 score_color = hwctool.settings.config.parser.get(811 "MapIcons",812 "undecided_color")813 opacity = 0.0814 return {'score_color': score_color,815 'border_color': border_color,816 'opacity': opacity,817 'hide': hide}818 def getMapIconsData(self):819 """Update map icons."""820 websocket_data = dict()821 try:822 team = self.getMyTeam()823 score = [0, 0]824 hide_scoreicon = team == 0825 for i in range(self.getNoSets()):826 winner = self.getMapScore(i)827 won = winner * team828 opacity = 0.0829 threshold = int(self.getBestOf() / 2)830 if(max(score) > threshold and i >= self.getMinSets()):831 border_color = hwctool.settings.config.parser.get(832 "MapIcons", "notplayed_color")833 score_color = hwctool.settings.config.parser.get(834 "MapIcons", "notplayed_color")835 opacity = hwctool.settings.config.parser.get(836 "MapIcons", "notplayed_opacity")837 winner = 0838 elif(won == 1):839 border_color = hwctool.settings.config.parser.get(840 "MapIcons", "win_color")841 score_color = hwctool.settings.config.parser.get(842 "MapIcons", "win_color")843 elif(won == -1):844 border_color = hwctool.settings.config.parser.get(845 "MapIcons", "lose_color")846 score_color = hwctool.settings.config.parser.get(847 "MapIcons", "lose_color")848 else:849 border_color = hwctool.settings.config.parser.get(850 "MapIcons", "default_border_color")851 score_color = hwctool.settings.config.parser.get(852 "MapIcons", "undecided_color")853 if(winner == -1):854 player1status = 'winner'855 player2status = 'loser'856 score[0] += 1857 elif(winner == 1):858 player1status = 'loser'859 player2status = 'winner'860 score[1] += 1861 else:862 player1status = ''863 player2status = ''864 data = dict()865 data['player1'] = self.getPlayer(0, i)866 data['player2'] = self.getPlayer(1, i)867 data['race1'] = self.getRace(0, i).lower()868 data['race2'] = self.getRace(1, i).lower()869 data['map_img'] = self.__controller.getMapImg(self.getMap(i))870 data['mapname'] = self.getMap(i)871 data['maplabel'] = self.getLabel(i)872 data['score_color'] = score_color873 data['border_color'] = border_color874 data['hide_scoreicon'] = hide_scoreicon875 data['opacity'] = opacity876 data['status1'] = player1status877 data['status2'] = player2status878 websocket_data[i + 1] = data879 except Exception as e:880 module_logger.exception("message")881 return websocket_data882 def autoSetMyTeam(self, swap=False):883 """Try to set team via fav teams."""884 try:885 team_matches = []886 for team_idx in range(2):887 team = self.__data['teams'][team_idx]['name']888 if not team or team == "TBD":889 continue890 matches = difflib.get_close_matches(891 team.lower(), hwctool.settings.config.getMyTeams(), 1)892 if(len(matches) > 0):893 team_matches.append(team_idx)894 if len(team_matches) == 1:895 self.setMyTeam(team_matches.pop() * 2 - 1, swap)896 return True897 else:898 self.setMyTeam(0)899 return False900 except Exception as e:901 module_logger.exception("message")902 return False903 def _useTemplate(self, filein, fileout, replacements):904 filein = hwctool.settings.getAbsPath(filein)905 fileout = hwctool.settings.getAbsPath(fileout)906 regex = re.compile(r"%[\w_-]+%")907 compiled_replacements = dict()908 for placeholder, value in replacements.items():909 compiled_replacements['%{}%'.format(910 placeholder.upper())] = str(value)911 with open(filein, "rt", encoding='utf-8-sig') as fin:912 with open(fileout, "wt", encoding='utf-8-sig') as fout:913 for line in fin:914 for placeholder, value in compiled_replacements.items():915 line = line.replace(placeholder, value)916 line = regex.sub("", line)917 fout.write(line)918def autoCorrectMap(map):919 """Corrects map using list in config."""920 return "TBD", True921def getRace(str):922 """Get race by using the first three letter."""923 if str in hwctool.settings.races:924 return str925 else:926 return "Random"927class EmitLock():928 def __init__(self):929 self.__locked = False930 self.__useLock = True931 self.__signal = None932 def __call__(self, useLock=True, signal=None):933 self.__useLock = bool(useLock)934 self.__signal = signal935 return self936 def __enter__(self):937 self.__locked = self.__useLock938 return self939 def __exit__(self, type, value, traceback):940 self.__locked = False941 if self.__useLock and self.__signal:942 self.__signal.emit()943 def locked(self):...
main.py
Source:main.py
1#!/usr/bin/env python32# -*- coding: utf-8 -*-3"""4Created on Wed Oct 7 13:53:06 20205@author: lnguyen6"""7from sequence_data import sequence_data8from joblib import Parallel,delayed9from tqdm import tqdm10from sklearn.metrics import confusion_matrix11from sklearn.svm import SVC12from sklearn.metrics.pairwise import chi2_kernel13from clustering import clustering14from mining_motifs import mining_motif15from train_val import histogram_dyntex16from motifs_filtering import filtering_motifs17import argparse18import torch19import numpy as np20import os21parser = argparse.ArgumentParser()22parser.add_argument('--patch_size',type=int,default=8, 23 help="Patch size, default: 16.")24parser.add_argument('--num_clusters', type = int, default=50,25 help="Number of clusters in a codebook, default: 20")26parser.add_argument('--set_idx', type = int, default=0,27 help="The index of the subset, default: 0")28parser.add_argument('--gap', type = int, default=5,29 help="The maximum gap constraint, default: 5")30parser.add_argument('--loop', type = int, default=1,31 help="The loop number iteration, default: 1")32parser.add_argument('--coef', type = int, default=1,33 help="The subdataset index of the DynTex dataset. This is used for crossvalidation. Default: 0")34parser.add_argument('--gamma', type = int, default=1,35 help="The subdataset index of the DynTex dataset. This is used for crossvalidation. Default: 0")36args = parser.parse_args()37dataset_path = '../../datasets/UCLA dataset/'38print(args)39def main():40 result = one_iter(args.set_idx)41 if not os.path.exists('results_gap{}_loop{}/gamma_{}_coef_{}'.format(args.gap,args.loop,args.gamma,42 args.coef)):43 os.makedirs('results_gap{}_loop{}/gamma_{}_coef_{}'.format(args.gap,args.loop,args.gamma,44 args.coef))45 torch.save(result,'results_gap{}_loop{}/gamma_{}_coef_{}/c{}_p{}_setidx_{}_result.pt'.format(args.gap,args.loop,args.gamma,46 args.coef,47 args.num_clusters,48 args.patch_size,49 args.set_idx))50def one_iter(set_idx):51 train_hist_path = 'histogram_data/train_diff_clip_gap{}_loop{}/c{}_p{}_setidx_{}.pt'.format(args.gap,args.loop,args.num_clusters,52 args.patch_size,53 set_idx)54 55 test_hist_path = 'histogram_data/test_diff_clip_gap{}_loop{}/c{}_p{}_setidx_{}.pt'.format(args.gap,args.loop,args.num_clusters,56 args.patch_size,57 set_idx)58 # Constructing codebook if necessary59 codebook_path = './codebooks/codebook_{}_p{}_setidx_{}_loop{}.joblib'.format(args.num_clusters,60 args.patch_size,61 set_idx,args.loop)62 63 if not os.path.exists(codebook_path):64 clustering(dataset_path = dataset_path,65 set_idx = set_idx,66 patch_size = args.patch_size,67 num_clusters = args.num_clusters,68 loop = args.loop)69 70 # Computing probabilistic sequences and deterministic sequences71 data_path = 'data/train_{}_clusters_p{}_setidx_{}/'.format(args.num_clusters,72 args.patch_size,73 set_idx)74 if not os.path.exists(data_path):75 computing_ps = sequence_data(path = dataset_path,76 patch_size=args.patch_size,77 num_clusters=args.num_clusters,78 set_idx=set_idx,79 sequence_length=15,80 loop = args.loop)81 if not os.path.exists(train_hist_path):82 # Parallel(n_jobs=12,backend="multiprocessing")(delayed(computing_ps.each_item)(idx) for idx in tqdm(range(computing_ps.__len__())))83 for idx in tqdm(range(computing_ps.__len__())):84 computing_ps.each_item(idx)85 86 # Mining key motifs from the Dyntex dataset87 motif_path = 'mined_motif/{}_clusters_p{}_setidx_{}_gap{}_loop{}/'.format(args.num_clusters,88 args.patch_size,89 set_idx,args.gap,args.loop)90 if not os.path.exists(motif_path):91 motifs_mining = mining_motif(patch_size = args.patch_size,92 num_clusters = args.num_clusters,93 set_idx = set_idx,gap=args.gap,epsilon=0.05,94 loop=args.loop)95 motifs_mining.mining()96 97 # Getting vectors of supports using the mined key motifs98 99 histogram_construction = histogram_dyntex(dataset_path = dataset_path,100 patch_size = args.patch_size,101 num_clusters = args.num_clusters,102 set_idx = set_idx,103 sequence_length = 15,104 gap=args.gap,105 loop=args.loop)106 107 classes = histogram_construction.classes108 if not os.path.exists(train_hist_path):109 print('Computing train histogram for the set index {} dataset!'.format(args.set_idx))110 histogram_construction.getting_hist_data()111 112 if not os.path.exists(test_hist_path):113 print('Computing test histogram for the set index {} dataset!'.format(args.set_idx))114 histogram_construction.get_test_hist()115 116 train_hists_data = np.array(torch.load(train_hist_path)[0])117 train_labels_data = np.array(torch.load(train_hist_path)[1])118 supports = np.load('histogram_data/supports_c{}_p{}_setidx_{}_gap{}_loop{}.npy'.format(args.num_clusters,119 args.patch_size,120 args.set_idx,121 args.gap,args.loop))122 123 epsilon_list = np.arange(0.05,0.16,0.02).tolist()124 results = {}125 index2keep = filtering_motifs(num_clusters=args.num_clusters,126 patch_size=args.patch_size,127 set_idx=args.set_idx,128 gap=args.gap,129 loop=args.loop,130 percentage=0.9)131 132 train_hists_data = train_hists_data[:,index2keep]133 supports = supports[index2keep]134 135 for epsilon in epsilon_list:136 train_hists = train_hists_data[:,supports>=epsilon]137 random_idx = np.random.permutation(train_hists.shape[0])138 train_hists = train_hists[random_idx,:]139 train_labels = train_labels_data[random_idx]140 train_chi_distance = chi2_kernel(X = train_hists,gamma=args.gamma/100)141 clf = SVC(C=args.coef, kernel='precomputed').fit(X=train_chi_distance,142 y=train_labels)143 144 test_data = torch.load(test_hist_path)145 predicted_vid_labels = []146 ground_truth = []147 for item in test_data:148 video_hists = test_data[item][0]149 video_label = test_data[item][1]150 ground_truth.append(video_label)151 test_vid_hists = np.array(video_hists)152 test_vid_hists = test_vid_hists[:,index2keep]153 test_vid_hists = test_vid_hists[:,supports>=epsilon]154 test_video_distances = chi2_kernel(X = train_hists, Y = test_vid_hists, gamma=args.gamma/100)155 output_res = clf.predict(test_video_distances.transpose())156 output_res = list(output_res)157 predicted_vid_labels.append(classes[max(output_res,key=output_res.count)])158 cm = confusion_matrix(ground_truth,predicted_vid_labels)159 print(cm)160 print('The classification rate of this set is:',cm.trace()/cm.sum())161 results[str(epsilon)] = (predicted_vid_labels,ground_truth)162 return results163if __name__ == "__main__":...
dataloader.py
Source:dataloader.py
1import os2import torch3from torch.utils.data import Dataset 4import numpy as np5from PIL import Image6import pickle7import gzip8def pkl_load(dir):9 with gzip.open(dir, 'rb') as f:10 return pickle.load(f)11def scaling(arr):12 arr2 = np.zeros_like(arr)13 for i in range(arr.shape[0]):14 arr2[i] = arr[i] / (np.max(arr[i]) + 0.0001)15 return arr216class KITTIDataset(Dataset):17 def __init__(self, lidar_dir, camera_dir, label_dir, set_idx, base_dir="/home/jaekyungcho/2021_2/SLAM/finalProject/KITTI"):18 self.set_idx = set_idx19 self.lidar_dir = lidar_dir20 self.camera_dir = camera_dir21 # self.label_file = np.loadtxt(label_dir,delimiter=" ", dtype=float)22 self.label_dir = label_dir23 self.base_dir = base_dir24 self.set_count = [len(os.listdir(os.path.join(self.base_dir,idx,self.lidar_dir)))-1 for idx in self.set_idx]25 for idx in self.set_idx:26 if len(os.listdir(os.path.join(self.base_dir,idx,self.lidar_dir))) != len(os.listdir(os.path.join(self.base_dir,idx,self.camera_dir))):27 assert "The number of lidar and camera files are not matched"28 if len(os.listdir(os.path.join(self.base_dir,idx,self.lidar_dir))) != np.loadtxt(os.path.join(self.base_dir,idx,self.label_dir)).shape[0]+1:29 assert "The number of files and labels are not matched"30 31 def __len__(self):32 total_len = 033 for set_idx in self.set_idx:34 total_len += len(os.listdir(os.path.join(self.base_dir,set_idx,self.lidar_dir)))-135 return total_len36 def __getitem__(self, idx):37 set_idx = 038 for count in self.set_count:39 if idx < count:40 break41 idx -= count42 set_idx += 143 x_lidar1_name = os.path.join(self.base_dir, self.set_idx[set_idx], self.lidar_dir,"%06d.pickle"%idx)44 x_lidar1 = pkl_load(x_lidar1_name).transpose((2,0,1))45 x_lidar1 = scaling(x_lidar1)46 x_lidar1 = torch.Tensor(x_lidar1.copy())47 x_lidar2_name = os.path.join(self.base_dir, self.set_idx[set_idx], self.lidar_dir,"%06d.pickle"%(idx+1))48 x_lidar2 = pkl_load(x_lidar2_name).transpose((2,0,1))49 x_lidar2 = scaling(x_lidar2)50 x_lidar2 = torch.Tensor(x_lidar2.copy())51 x_camera1_name = os.path.join(self.base_dir, self.set_idx[set_idx], self.camera_dir,"%06d.png"%idx)52 x_camera1 = np.asarray(Image.open(x_camera1_name).resize((416,128))).transpose((2,1,0))53 x_camera1 = scaling(x_camera1)54 x_camera1 = torch.Tensor(x_camera1.copy())55 x_camera2_name = os.path.join(self.base_dir, self.set_idx[set_idx], self.camera_dir,"%06d.png"%(idx+1))56 x_camera2 = np.asarray(Image.open(x_camera2_name).resize((416,128))).transpose((2,1,0))57 x_camera2 = scaling(x_camera2)58 x_camera2 = torch.Tensor(x_camera2.copy())59 label = np.loadtxt(os.path.join(self.base_dir, self.set_idx[set_idx], self.label_dir),delimiter=" ", dtype=float)[idx]60 label = torch.Tensor(label.copy())61 return (x_lidar1, x_lidar2), (x_camera1, x_camera2), label62class KITTIDatasetRNN(Dataset):63 def __init__(self, lidar_dir, camera_dir, label_dir, set_idx, seq_length=10, base_dir="/home/jaekyungcho/2021_2/SLAM/finalProject/KITTI"):64 self.set_idx = set_idx65 self.lidar_dir = lidar_dir66 self.camera_dir = camera_dir67 # self.label_file = np.loadtxt(label_dir,delimiter=" ", dtype=float)68 self.label_dir = label_dir69 self.base_dir = base_dir70 self.seq_length = seq_length71 self.set_count = [len(os.listdir(os.path.join(self.base_dir,idx,self.lidar_dir)))-(seq_length-1) for idx in self.set_idx]72 for idx in self.set_idx:73 if len(os.listdir(os.path.join(self.base_dir,idx,self.lidar_dir))) != len(os.listdir(os.path.join(self.base_dir,idx,self.camera_dir))):74 assert "The number of lidar and camera files are not matched"75 if len(os.listdir(os.path.join(self.base_dir,idx,self.lidar_dir))) != np.loadtxt(os.path.join(self.base_dir,idx,self.label_dir)).shape[0]+seq_length-1:76 assert "The number of files and labels are not matched"77 78 def __len__(self):79 return sum(self.set_count)80 def __getitem__(self, idx):81 set_idx = 082 for count in self.set_count:83 if idx < count:84 break85 idx -= count86 set_idx += 187 x_lidar, x_camera = None, None88 for i in range(self.seq_length):89 x_lidar_name = os.path.join(self.base_dir, self.set_idx[set_idx], self.lidar_dir,"%06d.pickle"%(idx+i))90 x_lidar_tmp = pkl_load(x_lidar_name).transpose((2,0,1))91 x_lidar_tmp = torch.Tensor(x_lidar_tmp.copy()).unsqueeze(0)92 x_camera_name = os.path.join(self.base_dir, self.set_idx[set_idx], self.camera_dir,"%06d.png"%(idx+i))93 x_camera_tmp = np.asarray(Image.open(x_camera_name).resize((416,128))).transpose((2,1,0))94 x_camera_tmp = torch.Tensor(x_camera_tmp.copy()).unsqueeze(0)95 if x_lidar is None:96 x_lidar = x_lidar_tmp97 x_camera = x_camera_tmp98 else:99 x_lidar = torch.vstack([x_lidar,x_lidar_tmp])100 x_camera = torch.vstack([x_camera, x_camera_tmp])101 102 label = np.loadtxt(os.path.join(self.base_dir, self.set_idx[set_idx], self.label_dir),delimiter=" ", dtype=float)[idx:idx+self.seq_length-1]103 label = torch.Tensor(label.copy())...
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 Airtest 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?
