xref: /aosp_15_r20/external/perfetto/python/perfetto/trace_processor/platform.py (revision 6dbdd20afdafa5e3ca9b8809fa73465d530080dc) 
1#!/usr/bin/env python3
2# Copyright (C) 2022 The Android Open Source Project
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8#      http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16import contextlib
17import datetime
18from datetime import timezone
19import os
20import socket
21import stat
22import subprocess
23import tempfile
24from typing import Tuple
25from urllib import request
26
27from perfetto.trace_uri_resolver.path import PathUriResolver
28from perfetto.trace_uri_resolver.registry import ResolverRegistry
29
30# URL to download script to run trace_processor
31SHELL_URL = 'https://get.perfetto.dev/trace_processor'
32
33
34class PlatformDelegate:
35  """Abstracts operations which can vary based on platform."""
36
37  def get_resource(self, file: str) -> bytes:
38    ws = os.path.dirname(__file__)
39    with open(os.path.join(ws, file), 'rb') as x:
40      return x.read()
41
42  def get_shell_path(self, bin_path: str) -> str:
43    if bin_path is not None:
44      if not os.path.isfile(bin_path):
45        raise Exception('Path to binary is not valid')
46      return bin_path
47
48    tp_path = os.path.join(tempfile.gettempdir(), 'trace_processor_python_api')
49    if self._should_download_tp(tp_path):
50      with contextlib.ExitStack() as stack:
51        req = stack.enter_context(request.urlopen(request.Request(SHELL_URL)))
52        file = stack.enter_context(open(tp_path, 'wb'))
53        file.write(req.read())
54    st = os.stat(tp_path)
55    os.chmod(tp_path, st.st_mode | stat.S_IEXEC)
56    return tp_path
57
58  def _should_download_tp(self, tp_path):
59    try:
60      st = os.stat(tp_path)
61
62      # If the file was empty (i.e. failed to be written properly last time),
63      # download it.
64      if st.st_size == 0:
65        return True
66
67      # Try and redownload if we last modified this file more than 7 days
68      # ago.
69      mod_time = datetime.datetime.fromtimestamp(st.st_mtime, tz=timezone.utc)
70      cutoff = datetime.datetime.now().astimezone() - datetime.timedelta(days=7)
71      return mod_time < cutoff
72    except OSError:
73      # Should happen if the file does not exist (i.e. this function has not
74      # been run before or tmp was cleared).
75      return True
76
77  def get_bind_addr(self, port: int) -> Tuple[str, int]:
78    if port:
79      return 'localhost', port
80
81    free_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
82    free_socket.bind(('', 0))
83    free_socket.listen(5)
84    port = free_socket.getsockname()[1]
85    free_socket.close()
86    return 'localhost', port
87
88  def default_resolver_registry(self) -> ResolverRegistry:
89    return ResolverRegistry(resolvers=[PathUriResolver])
90