Spaces:
Runtime error
Runtime error
File size: 6,527 Bytes
82130cb a732fe2 82130cb a732fe2 82130cb a732fe2 82130cb a732fe2 82130cb a732fe2 82130cb |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 |
from __future__ import annotations # For self-referencing annotations
import json
import os
from random import choices
from typing import List, Dict, Optional
from src.architectures import Architecture
from src.common import data_dir
class TestGenerator:
"""
Wrapper class to hold testing questions and serve up examples
"""
questions: List[str] = None
@classmethod
def load_questions(cls, reload=False) -> None:
"""
Load the available questions from the json file.
Default to not re-loading if already done, but allow for the option to do so
"""
if cls.questions is not None and not reload:
return
question_file = os.path.join(data_dir, 'json', 'test_questions.json')
with open(question_file, 'r') as f:
question_json = json.load(f)
cls.questions = question_json['questions']
@classmethod
def question_count(cls) -> int:
cls.load_questions()
return len(cls.questions)
@classmethod
def get_random_questions(cls, n: int):
"""
Return n random questions
"""
cls.load_questions()
return choices(cls.questions, k=n)
class ArchitectureRequestRecord:
"""
Representation of the test data associated with each invocation of an architecture
"""
all: List[ArchitectureRequestRecord] = None
class ArchStep:
"""
Inner class to just hold this data
"""
def __init__(self, name: str, start: int, end: int):
self.name = name
self.start = start
self.end = end
self.elapsed = end - start
def __init__(self, arch: str, response_len: int, start: int, end: int,
elapsed: int, tags: List[str], test_group: Optional[str],
comment: str, steps: List[ArchitectureRequestRecord.ArchStep]):
self.arch = arch
self.response_len = response_len
self.start = start
self.end = end
self.elapsed = elapsed
self.tags = tags
self.test_group = test_group
self.comment = comment
self.steps = steps
@classmethod
def from_dict(cls, test: Dict) -> ArchitectureRequestRecord:
arch = test['architecture']
response_len = len(test['request']['response_evolution'][-1])
start = test['trace']['steps'][0]['start_ms']
end = test['trace']['steps'][-1]['end_ms']
elapsed = end - start
tags = test['test_tags']
test_group = None
for tag in tags:
if tag.startswith("TestGroup"):
test_group = tag
comment = test['test_comment']
steps = []
for s in test['trace']['steps']:
steps.append(ArchitectureRequestRecord.ArchStep(s['name'], s['start_ms'], s['end_ms']))
return ArchitectureRequestRecord( arch, response_len, start, end, elapsed, tags, test_group, comment, steps)
@classmethod
def load_all(cls, reload=False) -> None:
"""
Load all the traces from json trace log
"""
if cls.all is None or reload:
records = []
test_traces = Architecture.get_trace_records()
for trace in test_traces:
records.append(ArchitectureRequestRecord.from_dict(trace))
cls.all = records
class TestGroup:
all: Dict[str, List[ArchitectureRequestRecord]] = None
def __init__(self, test_group:str):
self.arch_request_records: List[ArchitectureRequestRecord] = []
self.test_group = test_group
self.comment = None
self.start = None
self.end = None
self.elapsed = None
self.architectures = set()
@property
def num_archs(self) -> int:
return len(self.architectures)
@property
def num_tests(self) -> int:
return len(self.arch_request_records)
@property
def num_tests_per_arch(self) -> int:
# Should always be an even number but cast to int just in case
return int(self.num_tests / self.num_archs)
def arch_request_records_by_arch(self) -> Dict[List[ArchitectureRequestRecord]]:
grouped = {}
for arr in self.arch_request_records:
if arr.arch not in grouped:
grouped[arr.arch] = []
grouped[arr.arch].append(arr)
return grouped
def summary_stats_by_arch(self) -> List[Dict]:
arch_records = self.arch_request_records_by_arch()
arch_names = list(arch_records.keys())
arch_names.sort()
stats = []
for a in arch_names:
stat_pack = {'arch_name': a, 'elapsed': [rec.elapsed for rec in arch_records[a]],
'response_len': [rec.response_len for rec in arch_records[a]], 'steps': []}
for i in range(len(arch_records[a][0].steps)):
stat_pack['steps'].append({'step_name': arch_records[a][0].steps[i].name})
num_recs = len(arch_records[a])
total_elapsed = 0
for j in range(num_recs):
total_elapsed += arch_records[a][j].steps[i].elapsed
stat_pack['steps'][-1]['mean_elapsed'] = total_elapsed / num_recs
stats.append(stat_pack)
return stats
def add_record(self, arr: ArchitectureRequestRecord) -> None:
if arr.test_group != self.test_group:
raise ValueError("Attempted to group a test record into the wrong group")
self.arch_request_records.append(arr)
self.architectures.add(arr.arch)
if self.comment is None:
self.comment = arr.comment
if self.start is None or self.start > arr.start:
self.start = arr.start
if self.end is None or self.end < arr.end:
self.end = arr.end
self.elapsed = self.end - self.start
@classmethod
def load_all(cls, reload=False):
if cls.all is None or reload:
ArchitectureRequestRecord.load_all(reload=reload)
records = {}
for arr in ArchitectureRequestRecord.all:
if arr.test_group is not None:
if arr.test_group not in records:
records[arr.test_group] = TestGroup(arr.test_group)
records[arr.test_group].add_record(arr)
cls.all = records
@classmethod
def for_test_group_tag(cls, test_group_tag: str) -> TestGroup:
cls.load_all()
return cls.all[test_group_tag]
|