You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
140 lines
5.2 KiB
140 lines
5.2 KiB
#!/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()
|
|
|