projet-reseaux-profond/src/utils.py

import logging
import re
import sqlite3
import subprocess
import time
import uuid

import ppadb.client
import ppadb.command.serial
import ppadb.device

REGEX_USER: re.Pattern = re.compile(r"UserInfo\{([0-9]*):([a-zA-Z ]*):.*")
REGEX_FOCUS: re.Pattern = re.compile(r"mCurrentFocus.*com.Psyonix.RL2D.*")

PSYONIX_PACKAGE_NAME: str = "com.Psyonix.RL2D"
PSYONIX_ACTIVITY_NAME: str = "com.epicgames.ue4.SplashActivity"

ADB_HOST: str = "127.0.0.1"
ADB_PORT: int = 5037

DELAY_DETECT: int = 1

DATA_FOLDER = "data/"

UserList = list[tuple[int, str]]


def tap(pos):
    device.shell(f"input tap {pos[0]} {pos[1]}")


def motion(pos, type):
    device.shell(f"input motionevent {type} {pos[0]} {pos[1]}")


def connect_adb() -> ppadb.device.Device:
    client: ppadb.client.Client = ppadb.client.Client(host=ADB_HOST, port=ADB_PORT)
    devices: list[ppadb.device.Device] = client.devices()

    devices_names: list[ppadb.command.serial.Serial] = list(map(lambda d: d.serial, devices))
    logging.debug(f"detected devices: {devices_names}")

    global device
    if len(devices) == 1:
        device = devices[0]
    elif len(devices) == 0:
        print("No device connected")
        exit(1)
    else:
        print("Select a device :")
        raise NotImplementedError  # TODO

    logging.debug(f"selected device: {device.serial}")

    return device


def get_users(device: ppadb.device.Device) -> UserList:
    device_users: str = device.shell("pm list users")
    users: UserList = REGEX_USER.findall(device_users)

    logging.debug(f"detected users: {users}")

    return users


def detect_game(device: ppadb.device.Device, users: UserList) -> UserList:
    playable_users: UserList = list()
    for (id, name) in users:
        if "package:" in device.shell(f"pm path --user {id} {PSYONIX_PACKAGE_NAME}"):
            playable_users.append((id, name))

    logging.debug(f"playable users: {playable_users}")

    return playable_users


def start_game(device: ppadb.device.Device, users: UserList) -> None:
    if len(users) == 1:
        (user_id, user_name) = users[0]
    elif len(users) == 0:
        print("No Playable users detected")
        exit(1)
    else:
        print("Select a user to game with :")
        raise NotImplementedError  # TODO

    logging.debug(f"selected user: {user_name}")

    device.shell(f"am start --user {user_id} -n {PSYONIX_PACKAGE_NAME}/{PSYONIX_ACTIVITY_NAME}")
    logging.debug("game activity started")


def detect_zen_mode(device: ppadb.device.Device) -> bool:
    return device.shell("settings get global zen_mode") != "1\n"


def detect_focus(device: ppadb.device.Device) -> bool:
    activity_dump = device.shell("dumpsys activity activities")
    result = REGEX_FOCUS.search(activity_dump)

    return result is not None


def start_scrpy() -> None:
    subprocess.Popen(
        "scrcpy --crop=700:1830:160:200 --v4l2-sink=/dev/video2 --disable-screensaver -N -m 250".split(),
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )


def startup() -> None:
    device = connect_adb()
    users = get_users(device)
    users = detect_game(device, users)
    start_game(device, users)

    while not (detect_zen_mode(device) or detect_focus(device)):
        time.sleep(DELAY_DETECT)

    time.sleep(DELAY_DETECT)

    start_scrpy()


def screenshot() -> uuid.UUID:
    filename = uuid.uuid1()

    subprocess.Popen(
        f"ffmpeg -f video4linux2 -i /dev/video2 -frames:v 1 {DATA_FOLDER}/{filename}.jpg".split(),
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )

    return filename


def insert(database: sqlite3.Connection, uuid: uuid.UUID, loadout: tuple, x_rotation: int, y_rotation: int) -> None:
    ((model, sticker), wheel, hat, team, primary_color, secondary_color) = loadout
    database.execute(
        f"INSERT INTO data VALUES ('{uuid}',{model},{team},{primary_color},{secondary_color},{hat},{sticker},{wheel},{x_rotation},{y_rotation})"
    )
    database.commit()