robots-auto-action/tools/atx_sche.py

import subprocess

from tools import loggerKit


# 定时启动ATX-SERVER服务
def start_atx_server(device_list):
    loggerKit.info("atx-server Starting atx-server service...")
    for device in device_list.keys():
        loggerKit.info("Currently linked device: {0}", device)
        try:
            subprocess.run(['python3', '-m', 'uiautomator2', 'init'], check=True)
        except subprocess.CalledProcessError:
            # Error occurred while running the command
            loggerKit.info("Error occurred while checking ATX server status")


# @scheduler.task('interval', id='check_atx_server_status', seconds=30, max_instances=1, misfire_grace_time=60)
def check_atx_server_status(device_list):
    loggerKit.info("atx-server 服务状态检测 ... ")
    for device in device_list.keys():
        loggerKit.info("current linked for device:{0}", device)
        try:
            # Run the command "ps -ef | grep atx-agent server -d" and capture the output
            cmd_ps = f"adb -s {device} shell ps -ef | grep 'atx'"
            ps_output = subprocess.check_output(cmd_ps, shell=True)
            print(f'{ps_output}')

            # Find the process ID (PID) from the output
            pid = None
            for line in ps_output.splitlines():
                if b'atx' in line:
                    pid = int(line.split(None, 1)[1].split()[0])
                    print(f'pid:{pid}')
                    break

            # If a PID is found, kill the process using "kill -9"
            cmd_kill = f"adb -s {device} shell kill -9 {pid}"
            if pid:
                subprocess.run(cmd_kill, shell=True)

            # Run the command "/data/local/tmp/atx-agent server -d"
            cmd_start = f'adb -s {device} shell /data/local/tmp/atx-agent server -d'
            subprocess.run(cmd_start, shell=True)
        except subprocess.CalledProcessError:
            # Error occurred while running the command
            loggerKit.info("Error occurred while checking ATX server status")


def start_atx_agent_all(device):
    try:
        # Run the command to change the permissions of the ATX agent file
        command1 = f'adb -s {device} shell chmod 775 /data/local/tmp/atx-agent'
        subprocess.run(command1, shell=True)

        # Run the command to start the ATX agent service in the foreground
        command2 = f'adb -s {device} shell /data/local/tmp/atx-agent server -d'
        subprocess.run(command2, shell=True)

        loggerKit.info("ATX agent service started successfully")
    except subprocess.CalledProcessError as e:
        loggerKit.info(f"{device} Error occurred while starting ATX agent service: {e}")


# init atx-agent
def init_axt_agent():
    command = "python3 -m uiautomator2 init"
    result = subprocess.run(command, shell=True, capture_output=True, text=True)

    # Access the output and error messages
    output = result.stdout
    error = result.stderr

    print("Output:", output)
    print("Error:", error)