1# Copyright 2022 The Pigweed Authors 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); you may not 4# use this file except in compliance with the License. You may obtain a copy of 5# the License at 6# 7# https://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 12# License for the specific language governing permissions and limitations under 13# the License. 14"""Tools to retrieve and parse metrics.""" 15from collections import defaultdict 16import json 17import logging 18from typing import Any 19from pw_tokenizer import detokenize 20 21_LOG = logging.getLogger(__name__) 22 23 24def _tree(): 25 """Creates a key based on given input.""" 26 return defaultdict(_tree) 27 28 29def _insert(metrics, path_names, value): 30 """Inserts any value in a leaf of the dictionary.""" 31 for index, path_name in enumerate(path_names): 32 if index < len(path_names) - 1: 33 metrics = metrics[path_name] 34 elif path_name in metrics: 35 # the value in this position isn't a float or int, 36 # then collision occurs, throw an error. 37 assert ValueError( 38 'Variable already exists: {p}'.format(p=path_name) 39 ) 40 else: 41 metrics[path_name] = value 42 43 44def parse_metrics( 45 rpcs: Any, 46 detokenizer: detokenize.Detokenizer | None, 47 timeout_s: float | None, 48): 49 """Detokenizes metric names and retrieves their values.""" 50 # Creates a defaultdict that can infinitely have other defaultdicts 51 # without a specified type. 52 metrics: defaultdict = _tree() 53 if not detokenizer: 54 _LOG.error('No metrics token database set.') 55 return metrics 56 stream_response = rpcs.pw.metric.proto.MetricService.Get( 57 pw_rpc_timeout_s=timeout_s 58 ) 59 if not stream_response.status.ok(): 60 _LOG.error('Unexpected status %s', stream_response.status) 61 return metrics 62 for metric_response in stream_response.responses: 63 for metric in metric_response.metrics: 64 path_names = [] 65 for path in metric.token_path: 66 path_name = str( 67 detokenize.DetokenizedString( 68 path, detokenizer.lookup(path), b'', False 69 ) 70 ).strip('"') 71 path_names.append(path_name) 72 value = ( 73 metric.as_float 74 if metric.HasField('as_float') 75 else metric.as_int 76 ) 77 # inserting path_names into metrics. 78 _insert(metrics, path_names, value) 79 # Converts default dict objects into standard dictionaries. 80 return json.loads(json.dumps(metrics)) 81