Rooted android phone cpu topology

androp.py

#!/usr/bin/env python3
#
# MIT License
# alekum(aka rojaster)
# 19/12/2019 ver 0.0.1
#

import os
import sys
from ppadb.client import Client as ADBClient
import pprint
from typing import List, Dict

ADB_CONFIG = {'host': os.getenv('ADB_HOST', '127.0.0.1'), 'port': os.getenv('ADB_PORT', 5037)}
DEVICE_SERIAL = os.getenv('DEVICE_SERIAL', '')
SYSTEM_CPU_DEVICE_PATH = '/sys/devices/system/cpu'

adb = ADBClient(**ADB_CONFIG)
device = adb.device(DEVICE_SERIAL)


class CPU:
    def __init__(self, _cpu_id, _policy):
        super().__init__()
        self.cpu_id: int = _cpu_id
        self.cpu_capacity: int = device.shell(f'cat {SYSTEM_CPU_DEVICE_PATH}/cpu{_cpu_id}/cpu_capacity')
        self.policy: Policy = _policy
        self.online: int = device.shell(f'cat {SYSTEM_CPU_DEVICE_PATH}/cpu{_cpu_id}/online')

    @property
    def cpu_min_freq(self) -> int:
        return self.policy.scaling_min_freq

    @property
    def cpu_max_freq(self) -> int:
        return self.policy.scaling_max_freq

    @property
    def cpu_cur_freq(self) -> int:
        return self.policy.scaling_cur_freq

    def __repr__(self):
        return f'CPU{self.cpu_id}{{\
        online: {self.online} \
        cpu_capacity: {self.cpu_capacity} \
        cpu_cur_freq: {self.cpu_cur_freq}}}\n'


class Policy:
    def __init__(self, _policy_id):
        super().__init__()
        self.policy_id: str = _policy_id
        self.affected_cpus: List[int] = list()
        self.scaling_max_freq: int = 0
        self.scaling_min_freq: int = 0
        self.scaling_cur_freq: int = 0
        self.scaling_governor: str = None
        self.scaling_available_governors: List[str] = list()
        self.scaling_available_frequencies: List[int] = list()

    def __repr__(self):
        return "{}{{\n{}}}\n".format(
            self.policy_id,
            "\n".join(["\t{}: {}".format(k, v) for k, v in self.__dict__.items() if '__' not in k and 'id' not in k]))


class Topology:
    def __init__(self, _device_name):
        super().__init__()
        self.device: str = _device_name
        self.total_cpus: int = 0
        self.cpus: List[CPU] = []
        self.polices: [Policy] = []

    def __repr__(self):
        return "Topology {}\n".format("\n".join(
            ["{}: {}".format(k, pprint.pformat(v)) for k, v in self.__dict__.items() if '__' not in k]))


def make_policy(policy_device_path) -> Policy:
    p = Policy(os.path.basename(policy_device_path))
    p.affected_cpus = [int(cpu) for cpu in device.shell(f'cat {policy_device_path}/affected_cpus').split()]
    p.scaling_governor = device.shell(f'cat {policy_device_path}/scaling_governor').strip()
    p.scaling_available_governors = device.shell(
        f'cat {policy_device_path}/scaling_available_governors').strip().split()
    p.scaling_available_frequencies = [
        int(freq) for freq in device.shell(f'cat {policy_device_path}/scaling_available_frequencies').strip().split()
    ]
    p.scaling_cur_freq, p.scaling_min_freq, p.scaling_max_freq = device.shell(
        f'cat {policy_device_path}/scaling_{{cur,min,max}}_freq').strip().split('\n')

    return p


def get_device_policies_dict() -> List[Policy]:
    policies_device_path = f'{SYSTEM_CPU_DEVICE_PATH}/cpufreq'
    return [
        make_policy(f'{policies_device_path}/{policy}')
        for policy in device.shell(f'ls {policies_device_path}').strip().split()
    ]


def get_device_cpus_list(policies: List[Policy]) -> List[CPU]:
    cpu_list: List[CPU] = []
    for p in policies:
        cpu_list.extend([CPU(cpu_id, p) for cpu_id in p.affected_cpus])
    return cpu_list


def get_device_cpus_topology() -> Topology:
    topology = Topology(device.serial)
    topology.total_cpus = device.cpu_count()
    topology.polices = get_device_policies_dict()
    topology.cpus = get_device_cpus_list(topology.polices)
    return topology


if __name__ == '__main__':
    topology = get_device_cpus_topology()
    pprint.pprint(topology)