KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _clear_dict method in autotest
Best Python code snippet using autotest_python
preprocess_convlab_multiwoz.py
Source:preprocess_convlab_multiwoz.py 
1#!/usr/bin/env python2import re3import sys4import json5import shutil6import types7import logging8import subprocess9import zipfile10from io import BytesIO11from itertools import chain12from tqdm import tqdm13from collections import OrderedDict, defaultdict, Counter14import requests15import os16from copy import deepcopy17import inspect18sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))19from utils import setup_logging # noqa: E40220setup_logging()21logger = logging.getLogger()22DATASETS_PATH = os.path.join(os.path.expanduser(os.environ.get('DATASETS_PATH', '~/datasets')), 'augpt')23MW_DOMAINS = ['restaurant', 'hotel', 'attraction', 'train', 'taxi', 'hospital', 'police']24DEFAULT_IGNORE_VALUES = ['not mentioned', 'dont care', 'don\'t care', 'dontcare', 'do n\'t care', 'none']25class Lexicalizer:26 def __init__(self, zipf):27 self.path = zipf.filename28 placeholder_re = re.compile(r'\[(\s*[\w_\s]+)\s*\]')29 number_re = re.compile(r'.*(\d+|one|two|three|four|five|six|seven|eight|nine|ten|eleven|twelve)\s$')30 time_re = re.compile(r'((?:\d{1,2}[:]\d{2,3})|(?:\d{1,2} (?:am|pm)))', re.IGNORECASE)31 @staticmethod32 def ends_with_number(s):33 return bool(Lexicalizer.number_re.match(s))34 @staticmethod35 def extend_database_results(database_results, belief):36 # Augment database results from the belief state37 database_results = OrderedDict(database_results)38 if belief is not None:39 for i, (domain, (num_results, results)) in enumerate(database_results.items()):40 if domain not in belief:41 continue42 elif "ruk" in belief[domain].keys():43 belief[domain].pop("ruk")44 elif "topic" in belief[domain].keys():45 belief[domain].pop("topic")46 if num_results == 0:47 database_results[domain] = (1, [belief[domain]])48 else:49 new_results = []50 for r in results:51 r = dict(**r)52 for k, val in belief[domain].items():53 if k not in r:54 r[k] = val55 new_results.append(r)56 database_results[domain] = (num_results, new_results)57 return database_results58 def __call__(self, text, database_results, belief=None, context=None):59 database_results = Lexicalizer.extend_database_results(database_results, belief)60 result_index = 061 last_assignment = defaultdict(set)62 def trans(label, span, force=False, loop=100):63 nonlocal result_index64 nonlocal last_assignment65 result_str = None66 for domain, (count, results) in database_results.items():67 if count == 0:68 continue69 result = results[result_index % len(results)]70 if label in result:71 result_str = result[label]72 if result_str == '?':73 result_str = 'unknown'74 if label == 'price range' and result_str == 'moderate' and \75 not text[span[1]:].startswith(' price range') and \76 not text[span[1]:].startswith(' in price'):77 result_str = 'moderately priced'78 if label == 'type':79 if text[:span[0]].endswith('no ') or text[:span[0]].endswith('any ') or \80 text[:span[0]].endswith('some ') or Lexicalizer.ends_with_number(text[:span[0]]):81 if not result_str.endswith('s'):82 result_str += 's'83 if label == 'time' and ('[leave at]' in text or '[arrive by]' in text) and \84 belief is not None and 'train' in belief and \85 any([k in belief['train'] for k in ('leave at', 'arrive by')]):86 # this is a specific case in which additional [time] slot needs to be lexicalised87 # directly from the belief state88 # "The earliest train after [time] leaves at ... and arrives by ..."89 if 'leave at' in belief['train']:90 result_str = belief['train']['leave at']91 else:92 result_str = belief['train']['arrive by']93 elif force:94 if label == 'time':95 if 'leave at' in result or 'arrive by' in result:96 if 'arrive' in text and 'arrive by' in result:97 result_str = result['arrive by'].lstrip('0')98 elif 'leave at' in result:99 result_str = result['leave at'].lstrip('0')100 elif context is not None and len(context) > 0:101 last_utt = context[-1]102 mtch = Lexicalizer.time_re.search(last_utt)103 if mtch is not None:104 result_str = mtch.group(1).lstrip('0')105 if result_str is not None:106 break107 if force and result_str is None:108 if label == 'reference':109 result_str = 'YF86GE4J'110 elif label == 'phone':111 result_str = '01223358966'112 elif label == 'postcode':113 result_str = 'CB11JG'114 elif label == 'agent':115 result_str = 'Cambridge Towninfo Centre'116 elif label == 'stars':117 result_str = '4'118 if result_str is not None and result_str.lower() in last_assignment[label] and loop > 0:119 result_index += 1120 return trans(label, force=force, loop=loop - 1, span=span)121 if result_str is not None:122 last_assignment[label].add(result_str.lower())123 return result_str or f'[{label}]'124 text = Lexicalizer.placeholder_re.sub(lambda m: trans(m.group(1), span=m.span()), text)125 text = Lexicalizer.placeholder_re.sub(lambda m: trans(m.group(1), force=True, span=m.span()), text)126 return text127 def save(self, path):128 shutil.copy(self.path, os.path.join(path, os.path.split(self.path)[-1]))129DB_ONTOLOGY = True130class Database:131 def __init__(self, zipf):132 self.path = zipf.filename133 module = types.ModuleType('convlab_dbquery')134 exec(zipf.read('convlab_dbquery.py').decode('utf-8'), module.__dict__)135 convlab_database = getattr(module, 'Database')136 self.ignore_values = DEFAULT_IGNORE_VALUES137 self.supported_domains = MW_DOMAINS138 self._name_map = None139 self._ontology = None140 self._regexp = None141 # Load database files142 def hacked_init(self):143 self.dbs = {}144 for domain in MW_DOMAINS:145 with zipf.open(f'db/{domain}_db.json') as f:146 self.dbs[domain] = json.load(f)147 setattr(convlab_database, '__init__', hacked_init)148 self.inner = getattr(module, 'Database')()149 # Load ontology150 if globals().get('DB_ONTOLOGY', True):151 with zipf.open('db_ontology.json') as f:152 self._ontology = {tuple(k.split('-')): set(v) for k, v in json.load(f).items()}153 self._build_replace_dict()154 price_re = re.compile(r'\d+\.\d+')155 @staticmethod156 def hack_query(belief):157 new_belief = OrderedDict()158 for domain, bs in belief.items():159 new_bs = OrderedDict()160 new_belief[domain] = new_bs161 for key, val in bs.items():162 val = bs[key]163 if domain == 'restaurant' and key == 'name' and val.lower() == 'charlie':164 val = 'charlie chan'165 if domain == 'restaurant' and key == 'name' and val.lower() == 'good luck':166 val = 'the good luck chinese food takeaway'167 # if domain == 'hotel' and key == 'name' and val.lower() == 'el shaddai guesthouse':168 # val = 'el shaddai'169 new_bs[key] = val170 return new_belief171 @staticmethod172 def capitalize(val):173 def _mk(v):174 i, v = v175 if i == 0 or v not in {'the', 'an', 'a', 'of', 'in', 'for', 'as', 'these', 'at', 'up', 'on', 'and', 'or'}:176 return v[:1].upper() + v[1:]177 else:178 return v179 return ' '.join(map(_mk, enumerate(val.split())))180 @staticmethod181 def map_database_key(key):182 if key == 'trainID':183 key = 'id'184 key = ''.join([' '+i.lower() if i.isupper()185 else i for i in key]).lstrip(' ')186 key = key.replace('_', ' ')187 if key == 'pricerange':188 key = 'price range'189 if key == 'taxi phone' or key == 'phone':190 key = 'phone'191 if key == 'taxi colors':192 key = 'color'193 if key == 'taxi types':194 key = 'brand'195 if key == 'ref':196 key = 'reference'197 if key == 'leaveAt':198 key = 'leave at'199 if key == 'arriveBy':200 key = 'arrive by'201 if key == 'entrance fee':202 key = 'fee'203 return key204 @staticmethod205 def map_database_row(domain, row, query):206 results = dict()207 for k, val in row.items():208 k2 = Database.map_database_key(k)209 if k == 'location':210 continue211 elif k == 'post code' or k == 'postcode':212 val = val.upper()213 elif k == 'name':214 val = Database.capitalize(val)215 elif k == 'type' and val == 'concerthall':216 val = 'concert hall'217 elif k == 'price' and domain == 'hotel' and isinstance(val, dict):218 val = val.get('single', val.get('double', next(iter(val.values()))))219 val = f'{val} pounds'220 if k2 == 'people':221 # BUG in MW2.0222 val = val.lstrip('`')223 results[k2] = val224 if 'color' in results and 'brand' in results:225 results['car'] = f"{results['color']} {results['brand']}"226 if domain == 'train' and 'price' in row and 'people' in query:227 people = int(query['people'])228 def multiply_people(m):229 price = float(m.group(0))230 price *= people231 return format(price, '.2f')232 if people != 1:233 results['price'] = Database.price_re.sub(multiply_people, row['price'])234 return results235 @staticmethod236 def normalize_for_db(s):237 s = ','.join(s.split(' ,'))238 s = s.replace('swimming pool', 'swimmingpool')239 s = s.replace('night club', 'nightclub')240 s = s.replace('concert hall', 'concerthall')241 return s242 @staticmethod243 def translate_to_db_col(s):244 if s == 'leave at':245 return 'leaveAt'246 elif s == 'arrive by':247 return 'arriveBy'248 elif s == 'price range':249 return 'pricerange'250 else:251 return s252 def domain_not_empty(self, domain_bs):253 return any(len(val) > 0 and val not in self.ignore_values for val in domain_bs.values())254 def _build_replace_dict(self):255 if self._regexp is not None:256 return257 clear_values = {'the', 'a', 'an', 'food'}258 clear_values.update(self._ontology[('hotel', 'type')])259 clear_values.update(self._ontology[('hotel', 'price range')])260 clear_values.update(self._ontology[('hotel', 'area')])261 clear_values.update(self._ontology[('restaurant', 'price range')])262 clear_values.update(self._ontology[('restaurant', 'food')])263 clear_values.update(self._ontology[('restaurant', 'area')])264 clear_values.update(self._ontology[('attraction', 'type')])265 clear_values = (f' {x} ' for x in clear_values)266 self._regexp = re.compile('|'.join(map(re.escape, clear_values)))267 db_entities = chain(self.inner.dbs['attraction'], self.inner.dbs['hotel'], self.inner.dbs['restaurant'])268 self._name_map = {self._clear_name(r): r['name'].lower() for r in db_entities}269 def _clear_name(self, domain_bs):270 name = ' ' + domain_bs['name'].lower() + ' '271 name = self._regexp.sub(' ', name)272 name = re.sub(r'\s+', ' ', name)273 name = name.strip()274 return name275 @staticmethod276 def _to_minutes(time):277 hour, minutes = tuple(map(int, time.split(':')))278 return minutes + 60 * hour279 def __call__(self, belief, return_results=False):280 belief = Database.hack_query(belief)281 all_results = OrderedDict()282 for domain, domain_bs in belief.items():283 if domain not in self.supported_domains:284 continue # skip unsupported domains285 if self.domain_not_empty(domain_bs) or \286 domain in [d.lower() for d in {'Police', 'Hospital'}]:287 def query_single(domain_bs):288 blocked_slots = {'people', 'booked', 'stay', 'ruk', 'topic'}289 if domain != 'train' and domain != 'bus':290 blocked_slots.add('day')291 query_bs = [(Database.translate_to_db_col(slot), Database.normalize_for_db(val))292 for slot, val in domain_bs.items() if slot not in blocked_slots]293 result = self.inner.query(domain, query_bs)294 result = [Database.map_database_row(domain, k, domain_bs) for k in result]295 # Implement sorting missing in convlab296 if domain == 'train' and 'arrive by' in domain_bs:297 result.sort(key=lambda x: self._to_minutes(x['arrive by']), reverse=True)298 elif domain == 'train' and 'leave at' in domain_bs:299 result.sort(key=lambda x: self._to_minutes(x['leave at']))300 return result301 result = query_single(domain_bs)302 if len(result) == 0 and 'name' in domain_bs and self._clear_name(domain_bs) in self._name_map:303 domain_bs = dict(**domain_bs)304 domain_bs['name'] = self._name_map[self._clear_name(domain_bs)]305 result = query_single(domain_bs)306 if return_results:307 all_results[domain] = (len(result), result)308 else:309 all_results[domain] = len(result)310 return all_results311 def save(self, path):312 shutil.copy(self.path, os.path.join(path, os.path.split(self.path)[-1]))313class BeliefStateTransformation:314 def __init__(self):315 self.ignore_values = DEFAULT_IGNORE_VALUES316 def _process_domain(self, domain_bs):317 return {self._map_slot(slot): self._clear_value(val) for slot, val in domain_bs.items()318 if (len(val) > 0 and val not in self.ignore_values)}319 def _map_slot(self, slot):320 if slot == 'arriveBy':321 return 'arrive by'322 if slot == 'leaveAt':323 return 'leave at'324 if slot == 'pricerange':325 slot = 'price range'326 return slot327 def _clear_value(self, value):328 value = value.replace('>', ' ')329 if value == 'el shaddia guesthouse':330 value = 'el shaddai'331 if value == 'concerthall':332 value = 'concert hall'333 if value == 'nightclub':334 value = 'night club'335 # BUG in MW2.0336 value = value.lstrip('`')337 return value338 @staticmethod339 def domain_not_empty(domain_bs, ignore_values):340 return any(len(val) > 0 and val not in ignore_values for val in domain_bs.values())341 def __call__(self, belief_state, dialogue_act, active_domain):342 clean_belief = dict()343 for domain, domain_bs in belief_state.items():344 new_domain_bs = {}345 if 'semi' in domain_bs:346 new_domain_bs.update(domain_bs['semi'])347 if 'book' in domain_bs:348 new_domain_bs.update({k: v for k, v in domain_bs['book'].items() if k != 'booked'})349 '''350 if 'ruk' in domain_bs: # extend belief state in TAHOE351 new_domain_bs.update({"ruk": domain_bs['ruk']})352 new_domain_bs.update({"topic": domain_bs['topic']})353 '''354 if not BeliefStateTransformation.domain_not_empty(domain_bs, self.ignore_values):355 continue356 new_domain_bs = self._process_domain(new_domain_bs)357 # TODO: uncomment for new iteration358 # if len(new_domain_bs) == 0: # TODO: remove this condition in next iteration359 # continue360 if 'internet' in new_domain_bs and new_domain_bs['internet'] == 'no':361 del new_domain_bs['internet'] # no internet by default362 if 'parking' in new_domain_bs and new_domain_bs['parking'] == 'no':363 del new_domain_bs['parking'] # no parking by default364 # TODO: comment for new iteration365 if len(new_domain_bs) > 0: # TODO: remove this condition in next iteration366 clean_belief[domain] = new_domain_bs367 for domain in {'Police', 'Hospital'}:368 if any([da[1] == domain for da in dialogue_act]):369 clean_belief[domain.lower()] = {}370 # Sort belief371 clean_belief = {k: OrderedDict(sorted(v.items(), key=lambda x: x[0])) for k, v in clean_belief.items()}372 active_bs = None373 if active_domain is not None:374 active_domain = active_domain.lower()375 active_bs = clean_belief.pop(active_domain, None)376 items = [(active_domain, active_bs)] if active_bs is not None else [] # put active domain at first377 items += [(k, v) for k, v in sorted(clean_belief.items(), key=lambda x: x[0])]378 result = OrderedDict(items)379 return result380def is_booked(raw_belief, domain):381 return domain in raw_belief and 'book' in raw_belief[domain] and \382 'booked' in raw_belief[domain]['book'] and \383 any('reference' in x for x in raw_belief[domain]['book']['booked'])384def get_booked_domains(raw_belief):385 for domain in raw_belief.keys():386 if is_booked(raw_belief, domain):387 yield domain388def parse_goal(dialog_goal):389 belief_transformation = BeliefStateTransformation()390 """Parses user goal into dictionary format."""391 goal = {}392 for domain in MW_DOMAINS:393 if not dialog_goal[domain]:394 continue395 goal[domain] = {}396 goal[domain] = {'informable': [], 'requestable': [], 'booking': {}}397 if 'info' in dialog_goal[domain]:398 # if d['goal'][domain].has_key('info'):399 if domain == 'train':400 # we consider dialogues only where train had to be booked!401 if 'book' in dialog_goal[domain]:402 # if d['goal'][domain].has_key('book'):403 goal[domain]['requestable'].append('reference')404 if 'reqt' in dialog_goal[domain]:405 # if d['goal'][domain].has_key('reqt'):406 if 'trainID' in dialog_goal[domain]['reqt']:407 goal[domain]['requestable'].append('id')408 else:409 if 'reqt' in dialog_goal[domain]:410 # if d['goal'][domain].has_key('reqt'):411 for s in dialog_goal[domain]['reqt']: # addtional requests:412 if s in ['phone', 'address', 'postcode', 'reference', 'id']:413 # ones that can be easily delexicalised414 goal[domain]['requestable'].append(s)415 if 'book' in dialog_goal[domain]:416 # if d['goal'][domain].has_key('book'):417 goal[domain]['requestable'].append("reference")418 goal[domain]["informable"] = dialog_goal[domain]['info']419 if 'book' in dialog_goal[domain]:420 # if d['goal'][domain].has_key('book'):421 goal[domain]["booking"] = dialog_goal[domain]['book']422 if 'invalid' in goal[domain]['booking']:423 del goal[domain]['booking']['invalid']424 if 'pre_invalid' in goal[domain]['booking']:425 del goal[domain]['booking']['pre_invalid']426 belief = {domain: {'semi': goal[domain]['informable'], 'book': goal[domain]['booking']}}427 belief = belief_transformation(belief, [], domain).get(domain, dict())428 goal[domain]['informable'] = belief429 del goal[domain]['booking']430 return goal431def clear_whitespaces(text):432 text = re.sub(r'[\s\n\r]+', ' ', text)433 text = ' ' + text + ' '434 text = re.sub(r'\s([,\.:\?\!\']+)', lambda m: m.group(1), text)435 return text.strip()436def build_fix_belief_from_database(database_engine):437 _clear_dict = None438 def clear_dict():439 nonlocal _clear_dict440 if _clear_dict is None:441 # Build clear dict442 _clear_dict = dict()443 db_values = set()444 for x in database_engine.inner.dbs['attraction']:445 db_values.add(x['name'])446 _clear_dict = OrderedDict((x.replace("'", ''), x) for x in db_values)447 return _clear_dict448 def call(belief):449 # fix belief state, put back apostrophs450 for domain, bs in belief.items():451 for key, value in bs.items():452 bs[key] = clear_dict().get(value, value)453 return belief454 return call455def normalize(text):456 text = text.replace('swimmingpool', 'swimming pool')457 text = text.replace('nigthclub', 'night club')458 text = text.replace('Shanghi', 'Shanghai')459 return text460DELEX_LABEL_MAP = {461 'Price': 'price range',462 'Fee': None, # 'fee',463 'Addr': 'address',464 'Area': 'area',465 'Stars': 'stars',466 'Department': None, # 'department',467 'Stay': None, # 'stay',468 'Ref': 'reference',469 'Food': 'food',470 'Type': 'type',471 'Choice': None, # ignore472 'Phone': 'phone',473 'Ticket': 'price',474 'Day': None, # 'day',475 'Name': 'name',476 'Car': 'car',477 'Leave': 'leave at',478 'Time': 'time',479 'Arrive': 'arrive by',480 'Post': 'postcode',481 'Depart': None, # 'departure',482 'People': None, # 'people',483 'Dest': None, # 'destination',484 'Open': None, # ignore485 'Id': 'id',486}487def delexicalise_spans(response, spans, allowed_slots=None):488 allowed_slots = set(allowed_slots or [])489 # First, we clear the spans490 new_spans = []491 for i, span in enumerate(spans):492 if span[1] == 'Fee' and 'vary' in span[-3]:493 pass494 elif DELEX_LABEL_MAP[span[1]] not in allowed_slots:495 pass496 else:497 new_spans.append(span)498 spans = new_spans499 delex = []500 assignment = []501 textlen = 0502 for i, original in enumerate(response.split()):503 for span in spans:504 label = DELEX_LABEL_MAP[span[1]]505 textlen += 1 + len(original)506 if label is None:507 continue # causes this token to be added508 if label == 'time' and ('minute' in span[-3] or 'hour' in span[-3] or 'day' in span[-3]):509 label = 'duration'510 if original in {',', '.', ':'}:511 if i == span[3]:512 delex.append(original)513 delex.append(f'[{label}]')514 assignment.append((label, None, textlen))515 else:516 continue517 if i == span[3]:518 if label == 'stars' and '-star' in original:519 number, ext = original.split('-')520 delex.append(f'[{label}]-{ext}')521 original = number522 assignment.append((label, original, textlen - len(original)))523 elif label == 'area' and original == 'the':524 delex.append('the')525 delex.append(f'[{label}]')526 original = None527 assignment.append((label, original, textlen))528 elif label == 'area' and original == 'in' and span[-3].startswith('in the '):529 delex.extend(['in', 'the'])530 delex.append(f'[{label}]')531 original = None532 assignment.append((label, original, textlen))533 elif label == 'time' and original == 'a':534 delex.append('a')535 delex.append(f'[{label}]')536 original = None537 assignment.append((label, original, textlen))538 elif label == 'stay' and 'day' in original:539 delex.append(f'[{label}]')540 delex.append('days' if 'days' in original else 'day')541 assignment.append((label, original, textlen - len(original)))542 elif label == 'address' and len(delex) >= 2 and delex[-1] == ',' and delex[-2] == '[address]':543 delex.pop()544 label, text, index = assignment[-1]545 assignment[-1] = (label, f'{text} , {original}', index)546 else:547 delex.append(f'[{label}]')548 assignment.append((label, original, textlen - len(original)))549 break550 elif span[3] < i <= span[4]:551 # already added the label552 label, text, index = assignment[-1]553 if text is None:554 text = original555 else:556 text = f'{text} {original}'557 if i == span[4] and label == 'area' and text.endswith(' of town'):558 delex.extend(['of', 'town'])559 text = text[:-len(' of town')]560 if i == span[4] and label == 'time' and text.endswith(' ride'):561 delex.append('ride')562 text = text[:-len(' ride')]563 assignment[-1] = (label, text, index)564 break565 else:566 delex.append(original)567 return ' '.join(delex), assignment568def delexicalise(utt, return_replacements=False, database_results=None, belief=None, spans=None):569 database_results = Lexicalizer.extend_database_results(database_results, belief)570 # Delexicalise only the stuff that we can put back571 allowed_keys = {'reference'} # always delex reference572 for domain, (count, results) in database_results.items():573 if count > 0:574 allowed_keys.update(results[0].keys())575 if 'arrive by' in allowed_keys or 'leave at' in allowed_keys:576 allowed_keys.add('time')577 # First we use the span_info annotations578 spans = sorted(spans, key=lambda x: x[-2])579 utt, replacements = delexicalise_spans(utt, spans, allowed_keys)580 if return_replacements:581 replacements = [x[:2] for x in replacements]582 # replacements.sort(key=lambda k: k[0])583 return utt, replacements584 return utt585def export_data(path, zipf):586 global DB_ONTOLOGY587 def da2tuples(dialog_act):588 tuples = []589 for domain_intent, svs in dialog_act.items():590 for slot, value in sorted(svs, key=lambda x: x[0]):591 domain, intent = domain_intent.split('-')592 tuples.append([intent, domain, slot, value])593 return tuples594 transform_belief = BeliefStateTransformation()595 DB_ONTOLOGY = False596 with zipfile.ZipFile(os.path.join(path, 'database.zip')) as dbzipf:597 db = Database(dbzipf)598 DB_ONTOLOGY = True599 fix_belief_from_database = build_fix_belief_from_database(db)600 ontology = defaultdict(lambda: set())601 splits = []602 for split in ['train', 'val', 'test']:603 ignored_dialogues = 0604 dialogues = []605 splits.append((split, dialogues))606 with zipfile.ZipFile(BytesIO(zipf.read(f'{split}.json.zip'))) as zsplitf:607 data = json.load(zsplitf.open(f'{split}.json'))608 logger.info('loaded {}, size {}'.format(split, len(data)))609 for sess_id, sess in data.items():610 goal = parse_goal(sess['goal'])611 dialogue = dict(name=sess_id, items=[], goal=goal)612 active_domain = None613 ignore_dialogue = False614 for i, turn in enumerate(sess['log']):615 text = turn['text']616 da = da2tuples(turn['dialog_act'])617 item = dict(618 speaker='user' if i % 2 == 0 else 'system',619 text=text,620 dialogue_act=da621 )622 if item['speaker'] == 'system':623 belief = turn['metadata']624 item['span_info'] = turn['span_info']625 # Detect active domain and Judge turn type626 ruk = False627 for domain, meta in belief.items():628 if "ruk" in meta.keys():629 item['active_domain'] = domain630 ruk = True631 item['document'] = turn['knowledge']['snippet']['body']632 belief[domain]['topic'] = " ".join(turn['knowledge']['keywords'])633 break634 item['uk_based'] = ruk635 if not ruk:636 item['document'] = ""637 domain_counter = Counter({x[1].lower() for x in da}.intersection(MW_DOMAINS))638 if domain_counter:639 active_domain = domain_counter.most_common(1)[0][0]640 item['active_domain'] = active_domain641 '''In TAHOE, this needs modification to get extended belief'''642 belief = transform_belief(belief, da, active_domain)643 belief = fix_belief_from_database(belief) # normalize value (entity name)644 item['belief'] = belief645 if 'bus' in belief:646 # We need to ignore this dialogue647 # There is no data for the bus domain648 ignore_dialogue = True649 break650 for k, bs in belief.items():651 for k2, val in bs.items():652 ontology[(k, k2)].add(val)653 # Add booked property654 item['booked_domains'] = sorted(get_booked_domains(turn['metadata']))655 dialogue['items'].append(item)656 if not ignore_dialogue:657 dialogues.append(dialogue)658 else:659 ignored_dialogues += 1660 if ignored_dialogues > 0:661 logger.warning(f'dialogues were ignored {ignored_dialogues * 100 / (ignored_dialogues + len(dialogues)):.2f}% due to a missing domain "bus"') # noqa: E501662 with zipfile.ZipFile(os.path.join(path, 'database.zip'), 'a') as dbzipf:663 # with dbzipf.open('db_ontology.json', 'w') as f:664 # f.write(json.dumps({'-'.join(k): list(v) for k, v in ontology.items()}).encode('utf-8'))665 db = Database(dbzipf)666 # Delexicalize loaded data667 for split, dialogues in splits:668 for dialogue in tqdm(dialogues, desc=f'delexicalising {split}'):669 for item in dialogue['items']:670 text = item['text']671 if item['speaker'] == 'system':672 belief = item['belief']673 span_info = item['span_info']674 del item['span_info']675 database_results = db(belief, return_results=True)676 '''677 if item['uk_based']: # do not delex newly inserted turns678 delexicalised_text = normalize(text)679 '''680 delexicalised_text = delexicalise(text, return_replacements=False,681 database_results=database_results,682 belief=deepcopy(belief), spans=span_info)683 item['delexicalised_text'] = clear_whitespaces(delexicalised_text)684 database_results = OrderedDict((domain, count) for domain, (count, results)685 in database_results.items())686 item['database'] = database_results687 text = normalize(text)688 item['text'] = clear_whitespaces(text)689 with open(os.path.join(path, f'{split}.json'), 'w+') as f:690 json.dump(dict(dialogues=dialogues, domains=MW_DOMAINS), f)691def preprocess():692 path = os.path.join(DATASETS_PATH, 'multiwoz-2.1')693 os.makedirs(path, exist_ok=True)694 # Download the dataset695 # commit_sha = 'e368deeb3d405caf19236fb768360a6517a24fcd'696 with zipfile.ZipFile(os.path.join(path, 'data_aug.zip')) as zipf:697 export_data(path, zipf)698 # Generating blacklist699 logger.info('Generating blacklist')700 cwd = os.path.dirname(os.path.abspath(__file__))701 subprocess.run(['python', os.path.join(cwd, 'build_multiwoz_blacklist.py'), '--dataset', 'multiwoz-2.1'], cwd=cwd)702if __name__ == "__main__":...
object_detection.py
Source:object_detection.py
...94 return resp195 '''96 Clear the dictionary that keep tracks of the detected objects97 '''98 def _clear_dict(self):99 self.dict_obj[HeadMovement.CENTRO].clear()100 self.dict_obj[HeadMovement.SINISTRA].clear()101 self.dict_obj[HeadMovement.DESTRA].clear()102 '''103 Callback called after receiving an update on the topic in which I'm subscribed104 105 @param: data Message containing head position as integer and data stream of image.106 '''107 def callback(self, data: ImagePos):108 #Retrieve postion from the image109 pos = data.pos110 # convert Image into numpy array111 img = ros_numpy.numpify(data.image)112 # image preprocessing113 img = img[:, :, ::-1]114 input_tensor = tf.convert_to_tensor(img)115 input_tensor = input_tensor[tf.newaxis, ...]116 # detect classes into the image117 start = time.time()118 detections = detect_fn(input_tensor)119 end = time.time()120 elapse = str(round(end - start, 2))121 print("Detection completed in", elapse, "seconds")122 num_above_thresh = np.sum(detections['detection_scores'] > 0.5)123 detections.pop('num_detections')124 detections = {key: value[0, :num_above_thresh].numpy() for key, value in detections.items()}125 detections['detection_classes'] = detections['detection_classes'].astype(np.int64)126 #For every detected object update or initialize the counting of that object in that direction127 for c, s in zip(detections['detection_classes'], detections['detection_scores']):128 if self.dict_obj[pos].get(classmap[c]) is None:129 self.dict_obj[pos][classmap[c]] = 1130 else:131 self.dict_obj[pos][classmap[c]] = self.dict_obj[pos][classmap[c]] + 1132 print(pos, self.dict_obj[pos])133 #Update the counter variable "count"134 self.sum_count()135 '''136 Callback is called when the dictionary is ready to be sent.137 138 @param: req Client request 139 '''140 def handleService(self, req):141 #First case, Pepper didn't do all 3 movements, so wait to be notified after every detection142 if(self.count != 3):143 scheduler.acquire()144 scheduler.wait()145 scheduler.release()146 147 self.count = 0148 try:149 #Call the method talk to make Pepper perform the speech150 resp = self.talk()151 #Clear the dictionary with found and detected objects152 self._clear_dict()153 #Return the result of 'capture_ended' service154 return capture_endedResponse(resp.result)155 except rospy.ServiceException as e:156 rospy.logwarn("Service call failed: %s" % e)157 self._clear_dict()158 return capture_endedResponse(False)159def init_model():160 img = np.full((512, 512, 3), 0, dtype=np.uint8)161 img = img[:, :, ::-1]162 input_tensor = tf.convert_to_tensor(img)163 input_tensor = input_tensor[tf.newaxis, ...]164 detect_fn(input_tensor)165if __name__ == '__main__':166 #Wait the service animatedSay167 rospy.wait_for_service('animatedSay')168 #Call the service Say169 call = rospy.ServiceProxy('animatedSay', Say)170 #call("One moment please. I'm loading the model into my brain")171 print('Loading model...', end='')...
diction.py
Source:diction.py
...10 capital = False11 12 def __init__(self, letter_dict = None):13 if letter_dict is None:14 self._clear_dict()15 else:16 self.diction = letter_dict17 self._clear_buff()18 19 def _clear_buff(self):20 self.syllables = {}21 self.buffer = ''22 self.index = 023 24 def _clear_dict(self):25 self.diction = {**self._vowelGrab(cm.cDict), **cm.cDictVow}26 if self.capital:27 self._set_capital()28 29 @staticmethod30 def _UpperDict(cdict):31 return {k.capitalize():v for (k,v) in cdict.items()}32 33 @staticmethod34 def _doubleDict(cdict):35 return {k*2:v for (k,v) in cdict.items()}36 37 @staticmethod38 def _extDict(cdict):...
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 autotest 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?
