Quelle  __init__.py   Sprache: unbekannt

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

import multiprocessing
import os
import time
import webbrowser
from threading import Timer

from gecko_taskgraph.target_tasks import filter_by_uncommon_try_tasks

from tryselect.cli import BaseTryParser
from tryselect.push import (
    check_working_directory,
    generate_try_task_config,
    push_to_try,
)
from tryselect.tasks import generate_tasks

here = os.path.abspath(os.path.dirname(__file__))


class ChooserParser(BaseTryParser):
    name = "chooser"
    arguments = []
    common_groups = ["push", "task"]
    task_configs = [
        "artifact",
        "browsertime",
        "chemspill-prio",
        "disable-pgo",
        "env",
        "existing-tasks",
        "gecko-profile",
        "path",
        "pernosco",
        "rebuild",
        "worker-overrides",
    ]


def run(
    update=False,
    query=None,
    try_config_params=None,
    full=False,
    parameters=None,
    save=False,
    preset=None,
    mod_presets=False,
    stage_changes=False,
    dry_run=False,
    message="{msg}",
    closed_tree=False,
    push_to_lando=False,
    push_to_vcs=False,
):
    from .app import create_application

    push = not stage_changes and not dry_run
    check_working_directory(push)

    tg = generate_tasks(parameters, full)

    # Remove tasks that are not to be shown unless `--full` is specified.
    if not full:
        excluded_tasks = [
            label
            for label in tg.tasks.keys()
            if not filter_by_uncommon_try_tasks(label)
        ]
        for task in excluded_tasks:
            tg.tasks.pop(task)

    queue = multiprocessing.Queue()

    if os.environ.get("WERKZEUG_RUN_MAIN") == "true":
        # we are in the reloader process, don't open the browser or do any try stuff
        app = create_application(tg, queue)
        app.run()
        return

    # give app a second to start before opening the browser
    url = "http://127.0.0.1:5000"
    Timer(1, lambda: webbrowser.open(url)).start()
    print("Starting trychooser on {}".format(url))
    process = multiprocessing.Process(
        target=create_and_run_application, args=(tg, queue)
    )
    process.start()

    selected = queue.get()

    # Allow the close page to render before terminating the process.
    time.sleep(1)
    process.terminate()
    if not selected:
        print("no tasks selected")
        return

    msg = "Try Chooser Enhanced ({} tasks selected)".format(len(selected))
    return push_to_try(
        "chooser",
        message.format(msg=msg),
        try_task_config=generate_try_task_config(
            "chooser", selected, params=try_config_params
        ),
        stage_changes=stage_changes,
        dry_run=dry_run,
        closed_tree=closed_tree,
        push_to_lando=push_to_lando,
        push_to_vcs=push_to_vcs,
    )


def create_and_run_application(tg, queue: multiprocessing.Queue):
    from .app import create_application

    app = create_application(tg, queue)

    app.run()