Function run_task

src-tauri/src-python/adb_auto_player/__main__.py:194–232

Wrapper to run task in a separate process.

(
    command: str,
    log_queue: Queue,
    summary_queue: Queue,
    app_config_dir: Path,
    resource_dir: Path,
)

def run_task(
    command: str,
    log_queue: Queue,
    summary_queue: Queue,
    app_config_dir: Path,
    resource_dir: Path,
) -> None:
    """Wrapper to run task in a separate process."""
    queue_handler = QueueHandler(log_queue)
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    logger.addHandler(queue_handler)

    def summary_callback(msg: str | None):
        # We are catching all exceptions here regardless
        # because we never want the summary to actually stop the process via error
        if summary_queue.full():
            try:
                summary_queue.get_nowait()
            except (queue.Empty, Exception):
                pass
        try:
            summary_queue.put_nowait(msg)
        except (queue.Full, Exception):
            # queue.Full should really never happen here but leaving as is
            pass

    SummaryGenerator.set_callback(summary_callback)
    SettingsLoader.set_app_config_dir(app_config_dir)
    SettingsLoader.set_resource_dir(resource_dir)

    try:
        e = Execute.find_command_and_execute(command, get_game_tasks())
        if isinstance(e, BaseException):
            logging.error(e)
            sys.exit(1)
    except Exception as exc:
        logging.error(exc)
        sys.exit(1)


class StartTaskBody(ProfileContext):