diff --git a/src/init.py b/src/init.py deleted file mode 100644 index f31608c..0000000 --- a/src/init.py +++ /dev/null @@ -1,81 +0,0 @@ -import logging -import re - -from ppadb.client import Client as AdbClient - -REGEX_USER = re.compile(r"UserInfo\{([0-9]*):([a-zA-Z ]*):.*") -REGEX_FOCUS = re.compile(r"mCurrentFocus.*com.Psyonix.RL2D.*") - -PSYONIX_PACKAGE_NAME = "com.Psyonix.RL2D" -PSYONIX_ACTIVITY_NAME = "com.epicgames.ue4.SplashActivity" - -ADB_HOST = "127.0.0.1" -ADB_PORT = 5037 - - -def connect_adb(): - client = AdbClient(host=ADB_HOST, port=ADB_PORT) - devices = client.devices() - devices_names = list(map(lambda d: d.serial, devices)) - - logging.debug(f"devices detected: {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 :") - # TODO - - logging.debug(f"device selected: {device.serial}") - - -def get_users(device): - device_users = device.shell("pm list users") - users = REGEX_USER.findall(device_users) - - logging.debug(f"users detected: {users}") - return users - - -def detect_game(device, users): - playable_users = [] - 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, users): - 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 :") - # TODO - - logging.debug(f"user selected: {user_name}") - device.shell(f"am start --user {user_id} -n {PSYONIX_PACKAGE_NAME}/{PSYONIX_ACTIVITY_NAME}") - logging.debug("game activity started") - - -def set_notifications(device, enabled: bool): - device.shell(f"settings put global heads_up_notifications_enabled {int(enabled)}") - - if enabled: - logging.debug("enabled heads up notificaiton") - else: - logging.debug("disabled heads up notificaiton") - - -def detect_focus(device) -> bool: - activity_dump = device.shell("dumpsys activity activities") - result = REGEX_FOCUS.search(activity_dump) - return result is not None diff --git a/src/utils.py b/src/utils.py new file mode 100644 index 0000000..151b962 --- /dev/null +++ b/src/utils.py @@ -0,0 +1,122 @@ +import logging +import re +import subprocess +import time + +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 + +DETECT_DELAY: int = 1 + +UserList = list[tuple[int, str]] + + +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}") + + 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}") + + logging.debug("game activity started") + device.shell(f"am start --user {user_id} -n {PSYONIX_PACKAGE_NAME}/{PSYONIX_ACTIVITY_NAME}") + logging.debug("game activity started") + + +def set_notifications(device: ppadb.device.Device, enabled: bool) -> None: + device.shell(f"settings put global heads_up_notifications_enabled {int(enabled)}") + + if enabled: + logging.debug("enabled heads up notifications") + else: + logging.debug("disabled heads up notifications") + + +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) + set_notifications(device, False) + + while not (detect_zen_mode(device) and detect_focus(device)): + time.sleep(DETECT_DELAY) + + time.sleep(DETECT_DELAY) + + start_scrpy()