xref: /aosp_15_r20/development/tools/winscope/src/adb/winscope_proxy.py (revision 90c8c64db3049935a07c6143d7fd006e26f8ecca)

#!/usr/bin/python3

# Copyright (C) 2019 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.

#
# This is an ADB proxy for Winscope.
#
# Requirements: python3.10 and ADB installed and in system PATH.
#
# Usage:
#     run: python3 winscope_proxy.py
#

import argparse
import base64
import gzip
import json
import logging
import os
import re
import secrets
import signal
import subprocess
import sys
import threading
import time
from abc import abstractmethod
from enum import Enum
from http import HTTPStatus
from http.server import HTTPServer, BaseHTTPRequestHandler
from logging import DEBUG, INFO
from tempfile import NamedTemporaryFile
from typing import Callable

# GLOBALS #

log = None
secret_token = None

# CONFIG #

def create_argument_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description='Proxy for go/winscope', prog='winscope_proxy')

    parser.add_argument('--info', '-i', dest='loglevel', action='store_const', const=INFO)
    parser.add_argument('--port', '-p', default=5544, action='store')

    parser.set_defaults(loglevel=DEBUG)

    return parser

# Keep in sync with ProxyConnection#VERSION in Winscope
VERSION = '4.0.8'

PERFETTO_TRACE_CONFIG_FILE = '/data/misc/perfetto-configs/winscope-proxy-trace.conf'
PERFETTO_DUMP_CONFIG_FILE = '/data/misc/perfetto-configs/winscope-proxy-dump.conf'
PERFETTO_TRACE_FILE = '/data/misc/perfetto-traces/winscope-proxy-trace.perfetto-trace'
PERFETTO_DUMP_FILE = '/data/misc/perfetto-traces/winscope-proxy-dump.perfetto-trace'
PERFETTO_UNIQUE_SESSION_NAME = 'winscope proxy perfetto tracing'
PERFETTO_TRACING_SESSIONS_QUERY_START = """TRACING SESSIONS:

ID      UID     STATE      BUF (#) KB   DUR (s)   #DS  STARTED  NAME
===     ===     =====      ==========   =======   ===  =======  ====\n"""
PERFETTO_TRACING_SESSIONS_QUERY_END = """\nNOTE: Some tracing sessions are not reported in the list above."""

WINSCOPE_VERSION_HEADER = "Winscope-Proxy-Version"
WINSCOPE_TOKEN_HEADER = "Winscope-Token"

# Location to save the proxy security token
WINSCOPE_TOKEN_LOCATION = os.path.expanduser('~/.config/winscope/.token')

# Winscope traces extensions
WINSCOPE_EXT = ".winscope"
WINSCOPE_EXT_LEGACY = ".pb"
WINSCOPE_EXTS = [WINSCOPE_EXT, WINSCOPE_EXT_LEGACY]

# Winscope traces directories
WINSCOPE_DIR = "/data/misc/wmtrace/"
WINSCOPE_BACKUP_DIR = "/data/local/tmp/last_winscope_tracing_session/"

# Tracing handlers
SIGNAL_HANDLER_LOG = "/data/local/tmp/winscope_signal_handler.log"
WINSCOPE_STATUS = "/data/local/tmp/winscope_status"

# Max interval between the client keep-alive requests in seconds
KEEP_ALIVE_INTERVAL_S = 5

# Perfetto's default timeout for getting an ACK from producer processes is 5s
# We need to be sure that the timeout is longer than that with a good margin.
COMMAND_TIMEOUT_S = 15

class File:
    def __init__(self, file, filetype) -> None:
        self.file = file
        self.type = filetype

    def get_filepaths(self, device_id) -> list[str]:
        return [self.file]

    def get_filetype(self) -> str:
        return self.type


class FileMatcher:
    def __init__(self, path, matcher, filetype) -> None:
        self.path = path
        self.matcher = matcher
        self.type = filetype

    def get_filepaths(self, device_id) -> list[str]:
        if len(self.matcher) > 0:
            matchingFiles = call_adb(
                f"shell su root find {self.path} -name {self.matcher}", device_id)
        else:
            matchingFiles = call_adb(
                f"shell su root find {self.path}", device_id)

        files = matchingFiles.split('\n')[:-1]
        log.debug("Found files %s", files)
        return files

    def get_filetype(self) -> str:
        return self.type


class WinscopeFileMatcher(FileMatcher):
    def __init__(self, path, matcher, filetype) -> None:
        self.path = path
        self.internal_matchers = list(map(lambda ext: FileMatcher(path, f'{matcher}{ext}', filetype),
            WINSCOPE_EXTS))
        self.type = filetype

    def get_filepaths(self, device_id) -> list[str]:
        for matcher in self.internal_matchers:
            files = matcher.get_filepaths(device_id)
            if len(files) > 0:
                return files
        log.debug("No files found")
        return []


def get_shell_args(device_id: str, type: str) -> list[str]:
    shell = ['adb', '-s', device_id, 'shell']
    log.debug(f"Starting {type} shell {' '.join(shell)}")
    return shell

def is_perfetto_data_source_available(name: str, query_result: str) -> bool:
    return name in query_result

def is_any_perfetto_data_source_available(query_result: str) -> bool:
    return is_perfetto_data_source_available('android.inputmethod', query_result) or \
       is_perfetto_data_source_available('android.protolog', query_result) or \
       is_perfetto_data_source_available('android.surfaceflinger.layers', query_result) or \
       is_perfetto_data_source_available('android.surfaceflinger.transactions', query_result) or \
       is_perfetto_data_source_available('com.android.wm.shell.transition', query_result) or \
       is_perfetto_data_source_available('android.viewcapture', query_result) or \
       is_perfetto_data_source_available('android.windowmanager', query_result) or \
       is_perfetto_data_source_available('android.input.inputevent', query_result)


class TraceConfig:
    def __init__(self, is_perfetto: bool) -> None:
        self.is_perfetto = is_perfetto

    @abstractmethod
    def add(self, config: str, value: str | list[str] | None) -> None:
        pass

    @abstractmethod
    def is_valid(self, config: str) -> bool:
        pass

    @abstractmethod
    def execute_command(self, server, device_id):
        pass

    def get_trace_identifiers(self)-> list[str]:
        return [""]

    def get_optional_start_args(self, identifier)-> str:
        return ""

    def execute_optional_config_command(self, server, device_id, shell, command, config_key, config_value):
        process = subprocess.Popen(shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                    stdin=subprocess.PIPE, start_new_session=True)
        out, err = process.communicate(command.encode('utf-8'))
        if process.returncode != 0:
            raise AdbError(
                f"Error executing command:\n {command}\n\n### OUTPUT ###{out.decode('utf-8')}\n{err.decode('utf-8')}")
        log.debug(f"Optional trace config changed on device {device_id} {config_key}:{config_value}")

    def execute_perfetto_config_command(self, server, shell, command, trace_name):
        process = subprocess.Popen(shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                    stdin=subprocess.PIPE, start_new_session=True)
        out, err = process.communicate(command.encode('utf-8'))
        if process.returncode != 0:
            raise AdbError(
                f"Error executing command:\n {command}\n\n### OUTPUT ###{out.decode('utf-8')}\n{err.decode('utf-8')}")
        log.info(f'{trace_name} (perfetto) configured to start along the other perfetto traces.')


class SurfaceFlingerTraceConfig(TraceConfig):
    """Handles optional configuration for Surface Flinger traces.
    """
    LEGACY_FLAGS_MAP = {
        "input": 1 << 1,
        "composition": 1 << 2,
        "metadata": 1 << 3,
        "hwc": 1 << 4,
        "tracebuffers": 1 << 5,
        "virtualdisplays": 1 << 6
    }

    PERFETTO_FLAGS_MAP = {
        "input": "TRACE_FLAG_INPUT",
        "composition": "TRACE_FLAG_COMPOSITION",
        "metadata": "TRACE_FLAG_EXTRA",
        "hwc": "TRACE_FLAG_HWC",
        "tracebuffers": "TRACE_FLAG_BUFFERS",
        "virtualdisplays": "TRACE_FLAG_VIRTUAL_DISPLAYS",
    }

    def __init__(self, is_perfetto: bool) -> None:
        super().__init__(is_perfetto)
        self.flags = []
        self.perfetto_flags = []

        # defaults set for all configs
        self.selected_configs = {
            "sfbuffersize": "16000"
        }

    def add(self, config: str, value: str | None) -> None:
        if config in SurfaceFlingerTraceConfig.LEGACY_FLAGS_MAP:
            self.flags.append(config)
        elif config in self.selected_configs:
            self.selected_configs[config] = value

    def is_valid(self, config: str) -> bool:
        return config in SurfaceFlingerTraceConfig.LEGACY_FLAGS_MAP or config in self.selected_configs

    def execute_command(self, server, device_id):
        shell = get_shell_args(device_id, "sf config")

        if self.is_perfetto:
            self.execute_perfetto_config_command(server, shell, self._perfetto_config_command(), "SurfaceFlinger")
        else:
            self.execute_optional_config_command(server, device_id, shell, self._legacy_flags_command(), "sf flags", self.flags)
            self.execute_optional_config_command(server, device_id, shell, self._legacy_buffer_size_command(), "sf buffer size", self.selected_configs["sfbuffersize"])

    def _perfetto_config_command(self) -> str:
        flags = "\n".join([f"""trace_flags: {SurfaceFlingerTraceConfig.PERFETTO_FLAGS_MAP[flag]}""" for flag in self.flags])

        return f"""
cat << EOF >> {PERFETTO_TRACE_CONFIG_FILE}
data_sources: {{
    config {{
        name: "android.surfaceflinger.layers"
        surfaceflinger_layers_config: {{
            mode: MODE_ACTIVE
            {flags}
        }}
    }}
}}
EOF
"""

    def _legacy_buffer_size_command(self) -> str:
        return f'su root service call SurfaceFlinger 1029 i32 {self.selected_configs["sfbuffersize"]}'

    def _legacy_flags_command(self) -> str:
        flags = 0
        for flag in self.flags:
            flags |= SurfaceFlingerTraceConfig.LEGACY_FLAGS_MAP[flag]

        return f"su root service call SurfaceFlinger 1033 i32 {flags}"


class WindowManagerTraceConfig(TraceConfig):
    PERFETTO_LOG_LEVEL_MAP = {
       "verbose": "LOG_LEVEL_VERBOSE",
       "debug": "LOG_LEVEL_DEBUG",
       "critical": "LOG_LEVEL_CRITICAL",
    }

    PERFETTO_LOG_FREQUENCY_MAP = {
       "frame": "LOG_FREQUENCY_FRAME",
       "transaction": "LOG_FREQUENCY_TRANSACTION",
    }

    """Handles optional selected configuration for Window Manager traces.
    """

    def __init__(self, is_perfetto: bool) -> None:
        super().__init__(is_perfetto)
        # defaults set for all configs
        self.selected_configs = {
            "wmbuffersize": "16000",
            "tracinglevel": "debug",
            "tracingtype": "frame",
        }

    def add(self, config_type: str, config_value: str | None) -> None:
        self.selected_configs[config_type] = config_value

    def is_valid(self, config_type: str) -> bool:
        return config_type in self.selected_configs

    def execute_command(self, server, device_id):
        shell = get_shell_args(device_id, "wm config")

        if self.is_perfetto:
            self.execute_perfetto_config_command(server, shell, self._perfetto_config_command(), "WM")
        else:
            self.execute_optional_config_command(server, device_id, shell, self._legacy_tracing_type_command(), "tracing type", self.selected_configs["tracingtype"])
            self.execute_optional_config_command(server, device_id, shell, self._legacy_tracing_level_command(), "tracing level", self.selected_configs["tracinglevel"])
            # /!\ buffer size must be configured last
            # otherwise the other configurations might override it
            self.execute_optional_config_command(server, device_id, shell, self._legacy_buffer_size_command(), "wm buffer size", self.selected_configs["wmbuffersize"])

    def _perfetto_config_command(self) -> str:
        log_level = WindowManagerTraceConfig.PERFETTO_LOG_LEVEL_MAP[self.selected_configs["tracinglevel"]]
        log_frequency = WindowManagerTraceConfig.PERFETTO_LOG_FREQUENCY_MAP[self.selected_configs["tracingtype"]]
        return f"""
cat << EOF >> {PERFETTO_TRACE_CONFIG_FILE}
data_sources: {{
    config {{
        name: "android.windowmanager"
        windowmanager_config: {{
            log_level: {log_level}
            log_frequency: {log_frequency}
        }}
    }}
}}
EOF
"""

    def _legacy_tracing_type_command(self) -> str:
        return f'su root cmd window tracing {self.selected_configs["tracingtype"]}'

    def _legacy_tracing_level_command(self) -> str:
        return f'su root cmd window tracing level {self.selected_configs["tracinglevel"]}'

    def _legacy_buffer_size_command(self) -> str:
        return f'su root cmd window tracing size {self.selected_configs["wmbuffersize"]}'


class ViewCaptureTraceConfig(TraceConfig):
    """Handles perfetto config for View Capture traces."""

    COMMAND = f"""
cat << EOF >> {PERFETTO_TRACE_CONFIG_FILE}
data_sources: {{
    config {{
        name: "android.viewcapture"
    }}
}}
EOF
    """

    def execute_command(self, server, device_id):
        if (self.is_perfetto):
            shell = get_shell_args(device_id, "perfetto config view capture")
            self.execute_perfetto_config_command(server, shell, ViewCaptureTraceConfig.COMMAND, "View Capture")

class TransactionsConfig(TraceConfig):
    """Handles perfetto config for Transactions traces."""

    COMMAND = f"""
cat << EOF >> {PERFETTO_TRACE_CONFIG_FILE}
data_sources: {{
    config {{
        name: "android.surfaceflinger.transactions"
        surfaceflinger_transactions_config: {{
            mode: MODE_ACTIVE
        }}
    }}
}}
EOF
    """

    def execute_command(self, server, device_id):
        if (self.is_perfetto):
            shell = get_shell_args(device_id, "perfetto config transactions")
            self.execute_perfetto_config_command(server, shell, TransactionsConfig.COMMAND, "SF transactions")


class ProtoLogConfig(TraceConfig):
    """Handles perfetto config for ProtoLog traces."""

    COMMAND = f"""
cat << EOF >> {PERFETTO_TRACE_CONFIG_FILE}
data_sources: {{
    config {{
        name: "android.protolog"
        protolog_config: {{
            tracing_mode: ENABLE_ALL
        }}
    }}
}}
EOF
    """

    def execute_command(self, server, device_id):
        if (self.is_perfetto):
            shell = get_shell_args(device_id, "perfetto config protolog")
            self.execute_perfetto_config_command(server, shell, ProtoLogConfig.COMMAND, "ProtoLog")


class ImeConfig(TraceConfig):
    """Handles perfetto config for IME traces."""

    COMMAND = f"""
cat << EOF >> {PERFETTO_TRACE_CONFIG_FILE}
data_sources: {{
    config {{
        name: "android.inputmethod"
    }}
}}
EOF
    """

    def execute_command(self, server, device_id):
        if (self.is_perfetto):
            shell = get_shell_args(device_id, "perfetto config ime")
            self.execute_perfetto_config_command(server, shell, ImeConfig.COMMAND, "IME tracing")


class TransitionTracesConfig(TraceConfig):
    """Handles perfetto config for Transition traces."""

    COMMAND = f"""
cat << EOF >> {PERFETTO_TRACE_CONFIG_FILE}
data_sources: {{
    config {{
        name: "com.android.wm.shell.transition"
    }}
}}
EOF
    """

    def execute_command(self, server, device_id):
        if (self.is_perfetto):
            shell = get_shell_args(device_id, "perfetto config transitions")
            self.execute_perfetto_config_command(server, shell, TransitionTracesConfig.COMMAND, "Transitions")


class InputConfig(TraceConfig):
    """Handles perfetto config for Input traces."""

    COMMAND = f"""
cat << EOF >> {PERFETTO_TRACE_CONFIG_FILE}
data_sources: {{
    config {{
        name: "android.input.inputevent"
        android_input_event_config {{
            mode: TRACE_MODE_TRACE_ALL
        }}
    }}
}}
EOF
    """

    def execute_command(self, server, device_id):
        if (self.is_perfetto):
            shell = get_shell_args(device_id, "perfetto config input")
            self.execute_perfetto_config_command(server, shell, InputConfig.COMMAND, "Input trace")


class MediaBasedConfig(TraceConfig):
    """Creates trace identifiers for Screen Recording traces and Screenshots."""
    trace_identifiers = ["active"]

    def get_trace_identifiers(self):
        return self.trace_identifiers

    def is_valid(self, config_type: str) -> bool:
        return config_type == "displays"

    def add(self, config_type: str, config_value: str | list[str] | None):
        if config_type != "displays":
           return
        if config_value and len(config_value) > 0:
            if type(config_value) == str:
                self.trace_identifiers = [config_value.split(" ")[0]]
            else:
                self.trace_identifiers = list(map(lambda d: d.split(" ")[0], config_value))

    def execute_command(self, server, device_id):
        pass

class ScreenRecordingConfig(MediaBasedConfig):
    """Creates display args for Screen Recording traces."""
    def get_optional_start_args(self, identifier):
        if identifier == "active":
            return ""
        return f"--display-id {identifier}"

class ScreenshotConfig(MediaBasedConfig):
    """Creates display args for Screenshots."""
    def get_optional_start_args(self, identifier):
        if identifier == "active":
            return ""
        return f"-d {identifier}"


class SurfaceFlingerDumpConfig(TraceConfig):
    """Handles perfetto config for SurfaceFlinger dumps."""
    def __init__(self, is_perfetto: bool) -> None:
        super().__init__(is_perfetto)

    def add(self, config: str, value: str | None) -> None:
        pass

    def is_valid(self, config: str) -> bool:
        return False

    def execute_command(self, server, device_id):
        shell = get_shell_args(device_id, "sf config")

        if self.is_perfetto:
            self.execute_perfetto_config_command(server, shell, self._perfetto_config_command(), "SurfaceFlinger")

    def _perfetto_config_command(self) -> str:
        return f"""
cat << EOF >> {PERFETTO_DUMP_CONFIG_FILE}
data_sources: {{
    config {{
        name: "android.surfaceflinger.layers"
        surfaceflinger_layers_config: {{
            mode: MODE_DUMP
            trace_flags: TRACE_FLAG_INPUT
            trace_flags: TRACE_FLAG_COMPOSITION
            trace_flags: TRACE_FLAG_HWC
            trace_flags: TRACE_FLAG_BUFFERS
            trace_flags: TRACE_FLAG_VIRTUAL_DISPLAYS
        }}
    }}
}}
EOF
"""

    def _legacy_buffer_size_command(self) -> str:
        return f'su root service call SurfaceFlinger 1029 i32 {self.selected_configs["sfbuffersize"]}'

    def _legacy_flags_command(self) -> str:
        flags = 0
        for flag in self.flags:
            flags |= SurfaceFlingerTraceConfig.LEGACY_FLAGS_MAP[flag]

        return f"su root service call SurfaceFlinger 1033 i32 {flags}"


class TraceTarget:
    """Defines a single parameter to trace.

    Attributes:
        trace_name: used as a key to access config and log statements during Start/End endpoints.
        files: the matchers used to identify the paths on the device the trace results are saved to.
        is_perfetto_available: callback to determine if perfetto tracing is available for target.
        trace_start: command to start the trace from adb shell, must not block.
        trace_stop: command to stop the trace, should block until the trace is stopped.
        get_trace_config: getter for optional setup to execute pre-tracing adb commands and define start command arguments.
    """

    def __init__(
            self,
            trace_name: str,
            files: list[File | FileMatcher],
            is_perfetto_available: Callable[[str], bool],
            trace_start: str,
            trace_stop: str,
            get_trace_config: Callable[[bool], TraceConfig] | None = None
        ) -> None:
        self.trace_name = trace_name
        if type(files) is not list:
            files = [files]
        self.files = files
        self.is_perfetto_available = is_perfetto_available
        self.trace_start = trace_start
        self.trace_stop = trace_stop
        self.get_trace_config = get_trace_config
        self.status_filename = WINSCOPE_STATUS + "_" + trace_name


# Order of files matters as they will be expected in that order and decoded in that order
TRACE_TARGETS = {
    "view_capture_trace": TraceTarget(
        "view_capture_trace",
        File('/data/misc/wmtrace/view_capture_trace.zip', "view_capture_trace.zip"),
        lambda res: is_perfetto_data_source_available("android.viewcapture", res),
        """
su root settings put global view_capture_enabled 1
echo 'ViewCapture tracing (legacy) started.'
        """,
        """
su root sh -c 'cmd launcherapps dump-view-hierarchies >/data/misc/wmtrace/view_capture_trace.zip'
su root settings put global view_capture_enabled 0
echo 'ViewCapture tracing (legacy) stopped.'
        """,
        lambda is_perfetto: ViewCaptureTraceConfig(is_perfetto)
    ),
    "window_trace": TraceTarget(
        "window_trace",
        WinscopeFileMatcher(WINSCOPE_DIR, "wm_trace", "window_trace"),
        lambda res: is_perfetto_data_source_available('android.windowmanager', res),
        """
su root cmd window tracing start
echo 'WM trace (legacy) started.'
        """,
        """
su root cmd window tracing stop >/dev/null 2>&1
echo 'WM trace (legacy) stopped.'
        """,
        lambda is_perfetto: WindowManagerTraceConfig(is_perfetto)
    ),
    "layers_trace": TraceTarget(
        "layers_trace",
        WinscopeFileMatcher(WINSCOPE_DIR, "layers_trace", "layers_trace"),
        lambda res: is_perfetto_data_source_available('android.surfaceflinger.layers', res),
        """
su root service call SurfaceFlinger 1025 i32 1
echo 'SF layers trace (legacy) started.'
        """,
        """
su root service call SurfaceFlinger 1025 i32 0 >/dev/null 2>&1
echo 'SF layers trace (legacy) stopped.'
        """,
        lambda is_perfetto: SurfaceFlingerTraceConfig(is_perfetto)
    ),
    "screen_recording": TraceTarget(
        "screen_recording",
        File(
            '/data/local/tmp/screen_{trace_identifier}.mp4',
            "screen_recording_{trace_identifier}"),
        lambda res: False,
        '''
        settings put system show_touches 1 && \
        settings put system pointer_location 1 && \
        screenrecord --bugreport --bit-rate 8M {options} /data/local/tmp/screen_{trace_identifier}.mp4 & \
        echo "ScreenRecorder started."
        ''',
        '''settings put system pointer_location 0 && \
        settings put system show_touches 0 && \
        pkill -l SIGINT screenrecord >/dev/null 2>&1
        '''.strip(),
        lambda is_perfetto: ScreenRecordingConfig(is_perfetto)
    ),
    "transactions": TraceTarget(
        "transactions",
        WinscopeFileMatcher(WINSCOPE_DIR, "transactions_trace", "transactions"),
        lambda res: is_perfetto_data_source_available('android.surfaceflinger.transactions', res),
        """
su root service call SurfaceFlinger 1041 i32 1
echo 'SF transactions trace (legacy) started.'
        """,
        "su root service call SurfaceFlinger 1041 i32 0 >/dev/null 2>&1",
        lambda is_perfetto: TransactionsConfig(is_perfetto)
    ),
    "transactions_legacy": TraceTarget(
        "transactions_legacy",
        [
            WinscopeFileMatcher(WINSCOPE_DIR, "transaction_trace", "transactions_legacy"),
            FileMatcher(WINSCOPE_DIR, f'transaction_merges_*', "transaction_merges"),
        ],
        lambda res: False,
        'su root service call SurfaceFlinger 1020 i32 1\necho "SF transactions recording started."',
        'su root service call SurfaceFlinger 1020 i32 0 >/dev/null 2>&1',
    ),
    "proto_log": TraceTarget(
         "proto_log",
        WinscopeFileMatcher(WINSCOPE_DIR, "wm_log", "proto_log"),
        lambda res: is_perfetto_data_source_available('android.protolog', res),
        """
su root cmd window logging start
echo "ProtoLog (legacy) started."
        """,
        """
su root cmd window logging stop >/dev/null 2>&1
echo "ProtoLog (legacy) stopped."
        """,
        lambda is_perfetto: ProtoLogConfig(is_perfetto)
    ),
    "ime": TraceTarget(
        "ime",
        [WinscopeFileMatcher(WINSCOPE_DIR, "ime_trace_clients", "ime_trace_clients"),
         WinscopeFileMatcher(WINSCOPE_DIR, "ime_trace_service", "ime_trace_service"),
         WinscopeFileMatcher(WINSCOPE_DIR, "ime_trace_managerservice", "ime_trace_managerservice")],
        lambda res: is_perfetto_data_source_available('android.inputmethod', res),
        """
su root ime tracing start
echo "IME tracing (legacy) started."
        """,
        """
su root ime tracing stop >/dev/null 2>&1
echo "IME tracing (legacy) stopped."
        """,
        lambda is_perfetto: ImeConfig(is_perfetto)
    ),
    "wayland_trace": TraceTarget(
        "wayland_trace",
        WinscopeFileMatcher("/data/misc/wltrace", "wl_trace", "wl_trace"),
        lambda res: False,
        'su root service call Wayland 26 i32 1 >/dev/null\necho "Wayland trace started."',
        'su root service call Wayland 26 i32 0 >/dev/null'
    ),
    "eventlog": TraceTarget(
        "eventlog",
        WinscopeFileMatcher("/data/local/tmp", "eventlog", "eventlog"),
        lambda res: False,
        'rm -f /data/local/tmp/eventlog.winscope && EVENT_LOG_TRACING_START_TIME=$EPOCHREALTIME\necho "Event Log trace started."',
        'echo "EventLog\\n" > /data/local/tmp/eventlog.winscope && su root logcat -b events -v threadtime -v printable -v uid -v nsec -v epoch -b events -t $EVENT_LOG_TRACING_START_TIME >> /data/local/tmp/eventlog.winscope',
    ),
    "transition_traces": TraceTarget(
        "transition_traces",
        [WinscopeFileMatcher(WINSCOPE_DIR, "wm_transition_trace", "wm_transition_trace"),
         WinscopeFileMatcher(WINSCOPE_DIR, "shell_transition_trace", "shell_transition_trace")],
        lambda res: is_perfetto_data_source_available('com.android.wm.shell.transition', res),
        """
su root cmd window shell tracing start && su root dumpsys activity service SystemUIService WMShell transitions tracing start
echo "Transition traces (legacy) started."
        """,
        """
su root cmd window shell tracing stop && su root dumpsys activity service SystemUIService WMShell transitions tracing stop >/dev/null 2>&1
echo 'Transition traces (legacy) stopped.'
        """,
        lambda is_perfetto: TransitionTracesConfig(is_perfetto)
    ),
    "input": TraceTarget(
        "input",
        [WinscopeFileMatcher(WINSCOPE_DIR, "input_trace", "input_trace")],
        lambda res: is_perfetto_data_source_available('android.input.inputevent', res),
        "",
        "",
        lambda is_perfetto: InputConfig(is_perfetto)
    ),
    "perfetto_trace": TraceTarget(
         "perfetto_trace",
        File(PERFETTO_TRACE_FILE, "trace.perfetto-trace"),
        lambda res: is_any_perfetto_data_source_available(res),
        f"""
cat << EOF >> {PERFETTO_TRACE_CONFIG_FILE}
buffers: {{
    size_kb: 500000
    fill_policy: RING_BUFFER
}}
duration_ms: 0
file_write_period_ms: 999999999
write_into_file: true
unique_session_name: "{PERFETTO_UNIQUE_SESSION_NAME}"
EOF

rm -f {PERFETTO_TRACE_FILE}
perfetto --out {PERFETTO_TRACE_FILE} --txt --config {PERFETTO_TRACE_CONFIG_FILE} --detach=WINSCOPE-PROXY-TRACING-SESSION
echo 'Started perfetto trace.'
""",
        """
perfetto --attach=WINSCOPE-PROXY-TRACING-SESSION --stop
echo 'Stopped perfetto trace.'
""",
    ),
}


class DumpTarget:
    """Defines a single parameter to dump.

    Attributes:
        trace_name: used as a key to access config and log statements during Dump endpoint.
        files: the matchers used to identify the paths on the device the dump results are saved to.
        dump_command: command to dump state to file.
        get_trace_config: getter for optional setup to execute pre-tracing adb commands and define start command arguments.
    """

    def __init__(
            self,
            trace_name: str,
            files: list[File | FileMatcher],
            is_perfetto_available: Callable[[str], bool],
            dump_command: str,
            get_trace_config: Callable[[bool], TraceConfig] | None = None
        ) -> None:
        self.trace_name = trace_name
        if type(files) is not list:
            files = [files]
        self.files = files
        self.is_perfetto_available = is_perfetto_available
        self.dump_command = dump_command
        self.get_trace_config = get_trace_config


DUMP_TARGETS = {
    "window_dump": DumpTarget(
        "window_dump",
        File(f'/data/local/tmp/wm_dump{WINSCOPE_EXT}', "window_dump"),
        lambda res: False,
        f'su root dumpsys window --proto > /data/local/tmp/wm_dump{WINSCOPE_EXT}'
    ),

    "layers_dump": DumpTarget(
        "layers_dump",
        File(f'/data/local/tmp/sf_dump{WINSCOPE_EXT}', "layers_dump"),
        lambda res: is_perfetto_data_source_available('android.surfaceflinger.layers', res),
        f"""
su root dumpsys SurfaceFlinger --proto > /data/local/tmp/sf_dump{WINSCOPE_EXT}
        """,
        lambda is_perfetto: SurfaceFlingerDumpConfig(is_perfetto)
    ),

    "screenshot": DumpTarget(
        "screenshot",
        File("/data/local/tmp/screenshot_{trace_identifier}.png", "screenshot_{trace_identifier}.png"),
        lambda res: False,
        "screencap -p {options}> /data/local/tmp/screenshot_{trace_identifier}.png",
        lambda is_perfetto: ScreenshotConfig(is_perfetto)
    ),

    "perfetto_dump": DumpTarget(
        "perfetto_dump",
        File(PERFETTO_DUMP_FILE, "dump.perfetto-trace"),
        lambda res: is_any_perfetto_data_source_available(res),
        f"""
cat << EOF >> {PERFETTO_DUMP_CONFIG_FILE}
buffers: {{
    size_kb: 500000
    fill_policy: RING_BUFFER
}}
duration_ms: 1
EOF

rm -f {PERFETTO_DUMP_FILE}
perfetto --out {PERFETTO_DUMP_FILE} --txt --config {PERFETTO_DUMP_CONFIG_FILE}
echo 'Recorded perfetto dump.'
        """
    )
}


# END OF CONFIG #


def get_token() -> str:
    """Returns saved proxy security token or creates new one"""
    try:
        with open(WINSCOPE_TOKEN_LOCATION, 'r') as token_file:
            token = token_file.readline()
            log.debug("Loaded token {} from {}".format(
                token, WINSCOPE_TOKEN_LOCATION))
            return token
    except IOError:
        token = secrets.token_hex(32)
        os.makedirs(os.path.dirname(WINSCOPE_TOKEN_LOCATION), exist_ok=True)
        try:
            with open(WINSCOPE_TOKEN_LOCATION, 'w') as token_file:
                log.debug("Created and saved token {} to {}".format(
                    token, WINSCOPE_TOKEN_LOCATION))
                token_file.write(token)
            os.chmod(WINSCOPE_TOKEN_LOCATION, 0o600)
        except IOError:
            log.error("Unable to save persistent token {} to {}".format(
                token, WINSCOPE_TOKEN_LOCATION))
        return token


class RequestType(Enum):
    GET = 1
    POST = 2
    HEAD = 3


def add_standard_headers(server):
    server.send_header('Cache-Control', 'no-cache, no-store, must-revalidate')
    server.send_header('Access-Control-Allow-Origin', '*')
    server.send_header('Access-Control-Allow-Methods', 'POST, GET, OPTIONS')
    server.send_header('Access-Control-Allow-Headers',
                       WINSCOPE_TOKEN_HEADER + ', Content-Type, Content-Length')
    server.send_header('Access-Control-Expose-Headers',
                       'Winscope-Proxy-Version')
    server.send_header(WINSCOPE_VERSION_HEADER, VERSION)
    server.end_headers()


class RequestEndpoint:
    """Request endpoint to use with the RequestRouter."""

    @abstractmethod
    def process(self, server, path):
        pass


class AdbError(Exception):
    """Unsuccessful ADB operation"""
    pass


class BadRequest(Exception):
    """Invalid client request"""
    pass


class RequestRouter:
    """Handles HTTP request authentication and routing"""

    def __init__(self, handler):
        self.request = handler
        self.endpoints = {}

    def register_endpoint(self, method: RequestType, name: str, endpoint: RequestEndpoint):
        self.endpoints[(method, name)] = endpoint

    def _bad_request(self, error: str):
        log.warning("Bad request: " + error)
        self.request.respond(HTTPStatus.BAD_REQUEST, b"Bad request!\nThis is Winscope ADB proxy.\n\n"
                             + error.encode("utf-8"), 'text/txt')

    def _internal_error(self, error: str):
        log.error("Internal error: " + error)
        self.request.respond(HTTPStatus.INTERNAL_SERVER_ERROR,
                             error.encode("utf-8"), 'text/txt')

    def _bad_token(self):
        log.warning("Bad token")
        self.request.respond(HTTPStatus.FORBIDDEN, b"Bad Winscope authorization token!\nThis is Winscope ADB proxy.\n",
                             'text/txt')

    def process(self, method: RequestType):
        token = self.request.headers[WINSCOPE_TOKEN_HEADER]
        if not token or token != secret_token:
            return self._bad_token()
        path = self.request.path.strip('/').split('/')
        if path and len(path) > 0:
            endpoint_name = path[0]
            try:
                return self.endpoints[(method, endpoint_name)].process(self.request, path[1:])
            except KeyError as ex:
                if "RequestType" in repr(ex):
                    return self._bad_request("Unknown endpoint /{}/".format(endpoint_name))
                return self._internal_error(repr(ex))
            except AdbError as ex:
                return self._internal_error(str(ex))
            except BadRequest as ex:
                return self._bad_request(str(ex))
            except Exception as ex:
                return self._internal_error(repr(ex))
        self._bad_request("No endpoint specified")


def call_adb(params: str, device: str = None, stdin: bytes = None):
    command = ['adb'] + (['-s', device] if device else []) + params.split(' ')
    try:
        log.debug("Call: " + ' '.join(command))
        return subprocess.check_output(command, stderr=subprocess.STDOUT, input=stdin).decode('utf-8')
    except OSError as ex:
        raise AdbError('Error executing adb command: {}\n{}'.format(
            ' '.join(command), repr(ex)))
    except subprocess.CalledProcessError as ex:
        raise AdbError('Error executing adb command: adb {}\n{}'.format(
            params, ex.output.decode("utf-8")))


def call_adb_outfile(params: str, outfile, device: str = None, stdin: bytes = None):
    try:
        process = subprocess.Popen(['adb'] + (['-s', device] if device else []) + params.split(' '), stdout=outfile,
                                   stderr=subprocess.PIPE)
        _, err = process.communicate(stdin)
        outfile.seek(0)
        if process.returncode != 0:
            raise AdbError('Error executing adb command: adb {}\n'.format(params) + err.decode(
                'utf-8') + '\n' + outfile.read().decode('utf-8'))
    except OSError as ex:
        raise AdbError(
            'Error executing adb command: adb {}\n{}'.format(params, repr(ex)))


class ListDevicesEndpoint(RequestEndpoint):
    ADB_INFO_RE = re.compile("^([A-Za-z0-9._:\\-]+)\\s+(\\w+)(.*model:(\\w+))?")
    foundDevices: dict[str | int, dict[str, bool | str]] = {}

    def process(self, server, path):
        lines = list(filter(None, call_adb('devices -l').split('\n')))
        devices = {}
        for m in [ListDevicesEndpoint.ADB_INFO_RE.match(d) for d in lines[1:]]:
            if m:
                authorized = str(m.group(2)) != 'unauthorized'
                try:
                    screen_record_version = call_adb('shell screenrecord --version', m.group(1)) if authorized else '0'
                except AdbError:
                    try:
                        help_text = call_adb('shell screenrecord --help', m.group(1))
                        version_start_index = help_text.find('v') + 1
                        screen_record_version = help_text[version_start_index:version_start_index + 3]
                    except AdbError:
                        screen_record_version = '0'

                try:
                    displays = list(filter(None, call_adb('shell su root dumpsys SurfaceFlinger --display-id',
                                                          m.group(1)).split('\n'))) if authorized else []
                except AdbError:
                    displays = []

                devices[m.group(1)] = {
                    'authorized': authorized,
                    'model': m.group(4).replace('_', ' ') if m.group(4) else '',
                    'displays': displays,
                    'screenrecord_version': screen_record_version,
                }
        self.foundDevices = devices
        j = json.dumps(devices)
        log.info("Detected devices: " + j)
        server.respond(HTTPStatus.OK, j.encode("utf-8"), "text/json")



class CheckWaylandServiceEndpoint(RequestEndpoint):
    def __init__(self, listDevicesEndpoint: ListDevicesEndpoint):
      self._listDevicesEndpoint = listDevicesEndpoint

    def process(self, server, path):
        self._listDevicesEndpoint.process(server, path)
        foundDevices = self._listDevicesEndpoint.foundDevices

        if len(foundDevices) != 1:
            res = 'false'
        else:
            device = list(foundDevices.values())[0]
            if not device.get('authorized') or not device.get('model'):
                res = 'false'
            else:
                raw_res = call_adb('shell service check Wayland')
                res = 'false' if 'not found' in raw_res else 'true'
        server.respond(HTTPStatus.OK, res.encode("utf-8"), "text/json")



class DeviceRequestEndpoint(RequestEndpoint):
    def process(self, server, path):
        if len(path) > 0 and re.fullmatch("[A-Za-z0-9._:\\-]+", path[0]):
            self.process_with_device(server, path[1:], path[0])
        else:
            raise BadRequest("Device id not specified")

    @abstractmethod
    def process_with_device(self, server, path, device_id):
        pass

    def get_request(self, server) -> str:
        try:
            length = int(server.headers["Content-Length"])
        except KeyError as err:
            raise BadRequest("Missing Content-Length header\n" + str(err))
        except ValueError as err:
            raise BadRequest("Content length unreadable\n" + str(err))
        return json.loads(server.rfile.read(length).decode("utf-8"))

    def get_targets_and_prepare_for_tracing(self, server, device_id, perfetto_config_file, targets_map: dict[str, TraceTarget | DumpTarget], perfetto_name):
        warnings: list[str] = []

        call_adb(f"shell su root rm -f {perfetto_config_file}", device_id)
        log.debug("Cleared perfetto config file for previous tracing session")

        trace_requests: list[dict] = self.get_request(server)
        trace_types = [t.get("name") for t in trace_requests]
        log.debug(f"Received client request of {trace_types} for {device_id}")

        perfetto_query_result = call_adb("shell perfetto --query", device_id)

        too_many_perfetto_sessions = self.too_many_perfetto_sessions(perfetto_query_result)
        if (too_many_perfetto_sessions):
            warnings.append("Limit of 5 Perfetto sessions reached on device. Will attempt to collect legacy traces.")
            log.warning(warnings[0])

        trace_targets: list[tuple[DumpTarget | TraceTarget, TraceConfig | None]] = []
        for t in trace_requests:
            try:
                trace_name = t.get("name")
                target = targets_map[trace_name]
                is_perfetto = (not too_many_perfetto_sessions) and target.is_perfetto_available(perfetto_query_result)
                config = None
                if target.get_trace_config is not None:
                    config = target.get_trace_config(is_perfetto)
                    self.apply_config(config, t.get("config"), server, device_id)
                is_valid_perfetto_target = trace_name == perfetto_name and not too_many_perfetto_sessions
                is_valid_trace_target = trace_name != perfetto_name and not is_perfetto
                if is_valid_perfetto_target or is_valid_trace_target:
                    trace_targets.append((target, config))

            except KeyError as err:
                log.warning("Unsupported trace target\n" + str(err))
        trace_targets = self.move_perfetto_target_to_end_of_list(trace_targets)

        self.check_device_and_permissions(server, device_id)
        self.clear_last_tracing_session(device_id)

        log.debug("Trace requested for {} with targets {}".format(
            device_id, ','.join([target.trace_name for target, config in trace_targets])))

        return trace_targets, warnings

    def too_many_perfetto_sessions(self, perfetto_query_result: str):
        concurrent_sessions_start = perfetto_query_result.find(PERFETTO_TRACING_SESSIONS_QUERY_START)
        if concurrent_sessions_start != -1:
            concurrent_sessions = perfetto_query_result[concurrent_sessions_start + len(PERFETTO_TRACING_SESSIONS_QUERY_START):]
            log.info(f"Concurrent sessions:\n{concurrent_sessions}\n")
            concurrent_sessions_end = concurrent_sessions.find(PERFETTO_TRACING_SESSIONS_QUERY_END)
            if concurrent_sessions_end > 0:
                concurrent_sessions_end -= 1
            concurrent_sessions = concurrent_sessions[:concurrent_sessions_end]
            number_of_concurrent_sessions = len(concurrent_sessions.split("\n")) if len(concurrent_sessions) > 0 else 0

            if PERFETTO_UNIQUE_SESSION_NAME in concurrent_sessions:
                call_adb("shell perfetto --attach=WINSCOPE-PROXY-TRACING-SESSION --stop")
                log.debug("Stopped already-running winscope perfetto session.")
                number_of_concurrent_sessions -= 1
            return number_of_concurrent_sessions >= 5
        return False

    def apply_config(self, trace_config: TraceConfig, requested_configs: list[dict], server, device_id):
        for requested_config in requested_configs:
            config_key = requested_config.get("key")
            if not trace_config.is_valid(config_key):
                raise BadRequest(
                    f"Unsupported config {config_key}\n")
            trace_config.add(config_key, requested_config.get("value"))

        if device_id in TRACE_THREADS:
            BadRequest(f"Trace in progress for {device_id}")
        if not self.check_root(device_id):
            raise AdbError(
                f"Unable to acquire root privileges on the device - check the output of 'adb -s {device_id} shell su root id'")
        trace_config.execute_command(server, device_id)

    def check_root(self, device_id):
        log.debug("Checking root access on {}".format(device_id))
        return int(call_adb('shell su root id -u', device_id)) == 0

    def move_perfetto_target_to_end_of_list(self, targets: list[tuple[TraceTarget, TraceConfig | None]]) -> list[tuple[TraceTarget, TraceConfig | None]]:
        # Make sure a perfetto target (if present) comes last in the list of targets, i.e. will
        # be processed last.
        # A perfetto target must be processed last, so that perfetto tracing is started only after
        # the other targets have been processed and have configured the perfetto config file.
        def is_perfetto_target(target: TraceTarget):
            return target == TRACE_TARGETS["perfetto_trace"] or target == DUMP_TARGETS["perfetto_dump"]
        non_perfetto_targets = [t for t in targets if not is_perfetto_target(t[0])]
        perfetto_targets = [t for t in targets if is_perfetto_target(t[0])]
        return non_perfetto_targets + perfetto_targets

    def check_device_and_permissions(self, server, device_id):
        if device_id in TRACE_THREADS:
            log.warning("Trace already in progress for {}", device_id)
            server.respond(HTTPStatus.OK, b'', "text/plain")
        if not self.check_root(device_id):
            raise AdbError(
                "Unable to acquire root privileges on the device - check the output of 'adb -s {} shell su root id'"
                .format( device_id))

    def clear_last_tracing_session(self, device_id):
        call_adb(f"shell su root rm -rf {WINSCOPE_BACKUP_DIR}", device_id)
        call_adb(f"shell su root mkdir {WINSCOPE_BACKUP_DIR}", device_id)
        log.debug("Cleared previous tracing session files from device")


class FetchFilesEndpoint(DeviceRequestEndpoint):
    def process_with_device(self, server, path, device_id):
        file_buffers = self.fetch_existing_files(device_id)

        j = json.dumps(file_buffers)
        server.respond(HTTPStatus.OK, j.encode("utf-8"), "text/json")

    def fetch_existing_files(self, device_id):
        file_buffers = dict()
        file = FileMatcher(f"{WINSCOPE_BACKUP_DIR}*", "", "")
        try:
            file_paths = file.get_filepaths(device_id)
            for file_path in file_paths:
                with NamedTemporaryFile() as tmp:
                    file_name = file_path.split('/')[-1] + ".gz"
                    log.debug(
                        f"Fetching file {file_path} from device to {tmp.name}")
                    try:
                        call_adb_outfile('exec-out su root cat ' +
                                            file_path, tmp, device_id)
                    except AdbError as ex:
                        log.warning(f"Unable to fetch file {file_path} - {repr(ex)}")
                        return
                    log.debug(f"Uploading file {tmp.name}")
                    if file_name not in file_buffers:
                        file_buffers[file_name] = []
                    buf = base64.encodebytes(gzip.compress(tmp.read())).decode("utf-8")
                    file_buffers[file_name].append(buf)
        except:
            self.log_no_files_warning()
        return file_buffers

    def log_no_files_warning(self):
        log.warning("Proxy didn't find any file to fetch")


TRACE_THREADS = {}

class TraceThread(threading.Thread):
    def __init__(self, trace_name: str, device_id: str, command: str, trace_identifier: str, status_filename: str):
        self.trace_command = command
        self.trace_name = trace_name
        self.trace_identifier = trace_identifier
        self.status_filename = status_filename
        self._device_id = device_id
        self._keep_alive_timer = None
        self.out = None,
        self.err = None,
        self._command_timed_out = False
        self._success = False
        try:
            shell = get_shell_args(self._device_id, "trace")
            self.process = subprocess.Popen(shell, stdout=subprocess.PIPE,
                                            stderr=subprocess.PIPE, stdin=subprocess.PIPE, start_new_session=True)
        except OSError as ex:
            raise AdbError(
                'Error executing adb command for trace {}: adb shell\n{}'.format(trace_name, repr(ex)))

        super().__init__()

    def timeout(self):
        if self.is_alive():
            log.warning("Keep-alive timeout for {} trace on {}".format(self.trace_name, self._device_id))
            self.end_trace()

    def reset_timer(self):
        log.info(
            "Resetting keep-alive clock for {} trace on {}".format(self.trace_name, self._device_id))
        if self._keep_alive_timer:
            self._keep_alive_timer.cancel()
        self._keep_alive_timer = threading.Timer(
            KEEP_ALIVE_INTERVAL_S, self.timeout)
        self._keep_alive_timer.start()

    def end_trace(self):
        if self._keep_alive_timer:
            self._keep_alive_timer.cancel()
        log.info("Sending SIGINT to the {} process on {}".format(
            self.trace_name,
            self._device_id))
        self.process.send_signal(signal.SIGINT)
        try:
            log.debug("Waiting for {} trace shell to exit for {}".format(
                self.trace_name,
                self._device_id))
            self.process.wait(timeout=COMMAND_TIMEOUT_S)
        except TimeoutError:
            log.error(
                "TIMEOUT - sending SIGKILL to the {} trace process on {}".format(self.trace_name, self._device_id))
            self.process.kill()
        self.join()

    def run(self):
        retry_interval = 0.1
        log.info("Trace {} started on {}".format(self.trace_name, self._device_id))
        self.reset_timer()
        self.out, self.err = self.process.communicate(self.trace_command)
        log.info("Trace {} ended on {}, waiting for cleanup".format(self.trace_name, self._device_id))
        time.sleep(0.2)
        for i in range(int(COMMAND_TIMEOUT_S / retry_interval)):
            if call_adb(f"shell su root cat {self.status_filename}", device=self._device_id) == 'TRACE_OK\n':
                log.info("Trace {} finished on {}".format(
                    self.trace_name,
                    self._device_id))
                if self.trace_name == "perfetto_trace":
                    self._success = True
                else:
                    self._success = len(self.err) == 0
                return
            log.debug("Still waiting for cleanup on {} for {}".format(self._device_id, self.trace_name))
            time.sleep(retry_interval)

        self._command_timed_out = True

    def success(self):
        return self._success

    def timed_out(self):
        return self._command_timed_out

class StartTraceEndpoint(DeviceRequestEndpoint):
    TRACE_COMMAND = """
set -e

echo "Starting trace..."
echo "TRACE_START" > {winscope_status}

# Do not print anything to stdout/stderr in the handler
function stop_trace() {{
  echo "start" >{signal_handler_log}

  # redirect stdout/stderr to log file
  exec 1>>{signal_handler_log}
  exec 2>>{signal_handler_log}

  set -x
  trap - EXIT HUP INT
  {stop_commands}
  echo "TRACE_OK" > {winscope_status}
}}

trap stop_trace EXIT HUP INT
echo "Signal handler registered."

{start_commands}

# ADB shell does not handle hung up well and does not call HUP handler when a child is active in foreground,
# as a workaround we sleep for short intervals in a loop so the handler is called after a sleep interval.
while true; do sleep 0.1; done
"""

    def process_with_device(self, server, path, device_id):
        trace_targets, warnings = self.get_targets_and_prepare_for_tracing(
            server, device_id, PERFETTO_TRACE_CONFIG_FILE, TRACE_TARGETS, "perfetto_trace")

        for t, config in trace_targets:
            if config:
                trace_identifiers = config.get_trace_identifiers()
            else:
                trace_identifiers = [""]

            for trace_identifier in trace_identifiers:
                if trace_identifier:
                    if config:
                        options = config.get_optional_start_args(trace_identifier)
                    else:
                        options = ""
                    start_cmd = t.trace_start.format(trace_identifier=trace_identifier, options=options)
                else:
                    start_cmd = t.trace_start

                command = StartTraceEndpoint.TRACE_COMMAND.format(
                    winscope_status=t.status_filename,
                    signal_handler_log=SIGNAL_HANDLER_LOG,
                    stop_commands=t.trace_stop,
                    perfetto_config_file=PERFETTO_TRACE_CONFIG_FILE,
                    start_commands=start_cmd,
                )
                log.debug(f"Executing start command for {t.trace_name} on {device_id}...")
                thread = TraceThread(t.trace_name, device_id, command.encode('utf-8'), trace_identifier, t.status_filename)
                if device_id not in TRACE_THREADS:
                    TRACE_THREADS[device_id] = [thread]
                else:
                    TRACE_THREADS[device_id].append(thread)
                thread.start()

        server.respond(HTTPStatus.OK, json.dumps(warnings).encode("utf-8"), "text/json")


def move_collected_files(files: list[File | FileMatcher], device_id, trace_identifier):
    for f in files:
        file_paths = f.get_filepaths(device_id)
        file_type = f.get_filetype().format(trace_identifier=trace_identifier)

        for file_path in file_paths:
            formatted_path = file_path.format(trace_identifier=trace_identifier)
            log.debug(f"Moving file {formatted_path} to {WINSCOPE_BACKUP_DIR}{file_type} on device")
            try:
                call_adb(
                    f"shell su root [ ! -f {formatted_path} ] || su root mv {formatted_path} {WINSCOPE_BACKUP_DIR}{file_type}",
                    device_id)
            except AdbError as ex:
                log.warning(f"Unable to move file {formatted_path} - {repr(ex)}")


class EndTraceEndpoint(DeviceRequestEndpoint):
    def process_with_device(self, server, path, device_id):
        if device_id not in TRACE_THREADS:
            raise BadRequest("No trace in progress for {}".format(device_id))

        errors: list[str] = []

        for thread in TRACE_THREADS[device_id]:
            if thread.is_alive():
                thread.end_trace()
            success = thread.success()
            signal_handler_log = call_adb(f"shell su root cat {SIGNAL_HANDLER_LOG}", device=device_id).encode('utf-8')

            if (thread.timed_out()):
                timeout_message = "Trace {} timed out during cleanup".format(thread.trace_name)
                errors.append(timeout_message)
                log.error(timeout_message)

            if not success:
                log.error("Error ending trace {} on the device".format(thread.trace_name))
                errors.append("Error ending trace {} on the device: {}".format(thread.trace_name, thread.err))

            out = b"### Shell script's stdout ###\n" + \
                (thread.out if thread.out else b'<no stdout>') + \
                b"\n### Shell script's stderr ###\n" + \
                (thread.err if thread.err else b'<no stderr>') + \
                b"\n### Signal handler log ###\n" + \
                (signal_handler_log if signal_handler_log else b'<no signal handler logs>') + \
                b"\n"
            log.debug("### Output ###\n".format(thread.trace_name) + out.decode("utf-8"))
            if thread.trace_name in TRACE_TARGETS:
                files = TRACE_TARGETS[thread.trace_name].files
                move_collected_files(files, device_id, thread.trace_identifier)
            else:
                errors.append(f"File location unknown for {thread.trace_name}")

        call_adb(f"shell su root rm {thread.status_filename}", device=device_id)
        TRACE_THREADS.pop(device_id)
        server.respond(HTTPStatus.OK, json.dumps(errors).encode("utf-8"), "text/plain")


class StatusEndpoint(DeviceRequestEndpoint):
    def process_with_device(self, server, path, device_id):
        if device_id not in TRACE_THREADS:
            raise BadRequest("No trace in progress for {}".format(device_id))
        for thread in TRACE_THREADS[device_id]:
            thread.reset_timer()
        server.respond(HTTPStatus.OK, str(TRACE_THREADS[device_id][0].is_alive()).encode("utf-8"), "text/plain")


class DumpEndpoint(DeviceRequestEndpoint):
    def process_with_device(self, server, path, device_id):
        dump_targets, warnings = self.get_targets_and_prepare_for_tracing(
            server, device_id, PERFETTO_DUMP_CONFIG_FILE, DUMP_TARGETS, "perfetto_dump")

        dump_commands = []
        for t, config in dump_targets:
            if config:
                for trace_identifier in config.get_trace_identifiers():
                    dump_commands.append(
                        t.dump_command.format(trace_identifier=trace_identifier, options=config.get_optional_start_args(trace_identifier)))
            else:
                dump_commands.append(t.dump_command)

        dump_commands = '\n'.join(dump_commands)

        shell = get_shell_args(device_id, "dump")
        process = subprocess.Popen(shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                   stdin=subprocess.PIPE, start_new_session=True)
        log.info("Starting dump on device {}".format(device_id))
        out, err = process.communicate(dump_commands.encode('utf-8'))
        if process.returncode != 0:
            raise AdbError("Error executing dump command." + "\n\n### OUTPUT ###" + out.decode('utf-8') + "\n"
                           + err.decode('utf-8'))
        log.info("Dump finished on device {}".format(device_id))

        for target, config in dump_targets:
            if config:
                trace_identifiers = config.get_trace_identifiers()
                for trace_identifier in trace_identifiers:
                    move_collected_files(target.files, device_id, trace_identifier)
            else:
                move_collected_files(target.files, device_id, "")

        server.respond(HTTPStatus.OK, json.dumps(warnings).encode("utf-8"), "text/plain")


class ADBWinscopeProxy(BaseHTTPRequestHandler):
    def __init__(self, request, client_address, server):
        self.router = RequestRouter(self)
        listDevicesEndpoint = ListDevicesEndpoint()
        self.router.register_endpoint(
            RequestType.GET, "devices", listDevicesEndpoint)
        self.router.register_endpoint(
            RequestType.GET, "status", StatusEndpoint())
        self.router.register_endpoint(
            RequestType.GET, "fetch", FetchFilesEndpoint())
        self.router.register_endpoint(RequestType.POST, "start", StartTraceEndpoint())
        self.router.register_endpoint(RequestType.POST, "end", EndTraceEndpoint())
        self.router.register_endpoint(RequestType.POST, "dump", DumpEndpoint())
        self.router.register_endpoint(
            RequestType.GET, "checkwayland", CheckWaylandServiceEndpoint(listDevicesEndpoint))
        super().__init__(request, client_address, server)

    def respond(self, code: int, data: bytes, mime: str) -> None:
        self.send_response(code)
        self.send_header('Content-type', mime)
        add_standard_headers(self)
        self.wfile.write(data)

    def do_GET(self):
        self.router.process(RequestType.GET)

    def do_POST(self):
        self.router.process(RequestType.POST)

    def do_OPTIONS(self):
        self.send_response(HTTPStatus.OK)
        self.send_header('Allow', 'GET,POST')
        add_standard_headers(self)
        self.end_headers()
        self.wfile.write(b'GET,POST')

    def log_request(self, code='-', size='-'):
        log.info('{} {} {}'.format(self.requestline, str(code), str(size)))


if __name__ == '__main__':
    args = create_argument_parser().parse_args()

    logging.basicConfig(stream=sys.stderr, level=args.loglevel,
                        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

    log = logging.getLogger("ADBProxy")
    secret_token = get_token()

    print("Winscope ADB Connect proxy version: " + VERSION)
    print('Winscope token: ' + secret_token)

    httpd = HTTPServer(('localhost', args.port), ADBWinscopeProxy)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        log.info("Shutting down")