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
|
#!/usr/bin/env python3
# Copyright 2018 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
import base64
import logging
import os
import subprocess
import sys
import tempfile
from distutils.spawn import find_executable
import google.protobuf.text_format as text_format
from importlib import import_module
def compile_proto(proto_path, output_dir):
"""Invoke Protocol Compiler to generate python from given source .proto."""
# Find compiler path
protoc = None
if 'PROTOC' in os.environ and os.path.exists(os.environ['PROTOC']):
protoc = os.environ['PROTOC']
if not protoc:
protoc = find_executable('protoc')
if not protoc:
logging.error(
"Cannot find Protobuf compiler (>=3.0.0), please install"
"protobuf-compiler package. Prefer copying from <top>/prebuilts/tools"
)
logging.error(" prebuilts/tools/linux-x86_64/protoc/bin/protoc")
logging.error("If prebuilts are not available, use apt-get:")
logging.error(" sudo apt-get install protobuf-compiler")
return None
# Validate input proto path
if not os.path.exists(proto_path):
logging.error('Can\'t find required file: %s\n' % proto_path)
return None
# Validate output py-proto path
if not os.path.exists(output_dir):
os.mkdirs(output_dir)
elif not os.path.isdir(output_dir):
logging.error("Output path is not a valid directory: %s" % (output_dir))
return None
input_dir = os.path.dirname(proto_path)
output_filename = os.path.basename(proto_path).replace('.proto', '_pb2.py')
output_path = os.path.join(output_dir, output_filename)
protoc_command = [
protoc,
'-I=%s' % (input_dir),
'--python_out=%s' % (output_dir), proto_path
]
if subprocess.call(protoc_command, stderr=subprocess.STDOUT) != 0:
logging.error("Fail to compile proto")
return None
output_module_name = os.path.splitext(output_filename)[0]
return output_module_name
def compile_import_proto(output_dir, proto_path):
"""
Compile protobuf from PROTO_PATH and put the result in OUTPUT_DIR.
Return the imported module to caller.
:param output_dir: To store generated python proto library
:param proto_path: Path to the .proto file that needs to be compiled
:return: python proto module
"""
output_module_name = compile_proto(proto_path, output_dir)
if not output_module_name:
return None
sys.path.append(output_dir)
output_module = None
try:
output_module = import_module(output_module_name)
except ImportError:
logging.error(
"Cannot import generated py-proto %s" % (output_module_name))
return output_module
def parse_proto_to_ascii(binary_proto_msg):
"""
Parse binary protobuf message to human readable ascii string
:param binary_proto_msg:
:return: ascii string of the message
"""
return text_format.MessageToString(binary_proto_msg)
def dump_metrics():
os.system('adb wait-for-device')
p = subprocess.Popen(
"adb shell dumpsys bluetooth_manager --proto-bin",
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
stdin=subprocess.PIPE)
return p.communicate()
def get_bluetooth_metrics(proto_native_str_64, bluetooth_proto_module):
bluetooth_log = bluetooth_proto_module.BluetoothLog()
proto_native_str = base64.b64decode(proto_native_str_64)
bluetooth_log.MergeFromString(proto_native_str)
return bluetooth_log
def main():
root = logging.getLogger()
root.setLevel(logging.DEBUG)
log_handler = logging.StreamHandler(sys.stderr)
log_handler.setLevel(logging.DEBUG)
formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
log_handler.setFormatter(formatter)
root.addHandler(log_handler)
if len(sys.argv) < 2:
logging.error("Not enough arguments. Need at least 2")
logging.error("Usage: " + sys.argv[0] + " <path_to_metric_proto>")
sys.exit(1)
if sys.argv[1] == "-h":
logging.info("Usage: " + sys.argv[0] + " <path_to_metric_proto>")
logging.info("Requires Protobuf compiler, protoc, version >=3.0.0")
sys.exit(0)
bluetooth_proto_module = compile_import_proto(tempfile.gettempdir(),
sys.argv[1])
if not bluetooth_proto_module:
logging.error("Cannot compile " + sys.argv[1])
sys.exit(1)
stdout, stderr = dump_metrics()
stdout = stdout.strip()
stderr = stderr.strip()
bluetooth_log = get_bluetooth_metrics(stdout, bluetooth_proto_module)
bluetooth_log_ascii = parse_proto_to_ascii(bluetooth_log)
print(bluetooth_log_ascii)
if __name__ == "__main__":
main()
|
