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
|
#!/usr/bin/env vpython3
# Copyright 2024 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
""" The module to create and manage measures using in the process. """
import functools
import json
import os
import sys
from google.protobuf import any_pb2
from google.protobuf.json_format import MessageToDict
# Add to sys.path so that this module can be imported by other modules that
# have different path setup, e.g. android test runner, and ios test runner.
sys.path.append(os.path.abspath(os.path.dirname(__file__)))
from average import Average
from count import Count
from data_points import DataPoints
from measure import Measure
from metric import Metric
from time_consumption import TimeConsumption
# This is used as the key when being uploaded to ResultDB via result_sink
# and shouldn't be changed
TEST_SCRIPT_METRICS_KEY = 'test_script_metrics'
# The file name is used as the key when being loaded into the ResultDB and
# shouldn't be changed.
TEST_SCRIPT_METRICS_JSONPB_FILENAME = f'{TEST_SCRIPT_METRICS_KEY}.jsonpb'
_metric = Metric()
def _create_name(*name_pieces: str) -> str:
if len(name_pieces) == 0:
raise ValueError('Need at least one name piece.')
return '/'.join(list(name_pieces))
def _register(m: Measure) -> Measure:
_metric.register(m)
return m
def average(*name_pieces: str) -> Average:
return _register(Average(_create_name(*name_pieces)))
def count(*name_pieces: str) -> Count:
return _register(Count(_create_name(*name_pieces)))
def data_points(*name_pieces: str) -> DataPoints:
return _register(DataPoints(_create_name(*name_pieces)))
def time_consumption(*name_pieces: str) -> TimeConsumption:
return _register(TimeConsumption(_create_name(*name_pieces)))
def timed_func(*name_pieces: str):
"""time_consumption() as a @decorator."""
def decorator(func):
@functools.wraps(func)
def wrapped(*args, **kwargs):
with time_consumption(*name_pieces):
func(*args, **kwargs)
return wrapped
return decorator
def tag(*args: str) -> None:
"""Adds a tag to the Metric to tag the final results; see Metric for details.
"""
_metric.tag(*args)
def clear() -> None:
"""Clears all the registered Measures."""
_metric.clear()
def size() -> int:
"""Gets the current size of registered Measures."""
return _metric.size()
def to_dict() -> dict:
"""Converts all the registered Measures to a dict.
The records are wrapped in protobuf Any message before exported as dict
so that an additional key "@type" is included.
"""
any_msg = any_pb2.Any()
any_msg.Pack(_metric.dump())
return MessageToDict(any_msg, preserving_proto_field_name=True)
def to_json() -> str:
"""Converts all the registered Measures to a json str."""
return json.dumps(to_dict(), sort_keys=True, indent=2)
# TODO(crbug.com/343242386): May need to implement a lock and reset logic to
# clear in-memory data and lock the instance to block further operations and
# avoid accidentally accumulating data which won't be published at all.
def dump(dir_path: str) -> None:
"""Dumps the metric data into test_script_metrics.jsonpb in the |path|."""
os.makedirs(dir_path, exist_ok=True)
with open(os.path.join(dir_path, TEST_SCRIPT_METRICS_JSONPB_FILENAME),
'w',
encoding='utf-8') as wf:
wf.write(to_json())
|
