blob: 0d2ec57f885233c5cba2afb2e8fc059f899a6dfa [file] [log] [blame] [edit]
# !/usr/bin/env vpython3
# Copyright 2023 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Flask app main file for frontend."""
import os
from pathlib import Path
import socket
from typing import List, Optional
# pylint: disable=import-error
import flask
from google.protobuf import json_format
from chromite.api.controller import controller_util
from chromite.contrib.sdk_server.grpc_server import client
from chromite.contrib.sdk_server.grpc_server import sdk_server_pb2
from chromite.contrib.sdk_server.grpc_server.chromite.api import image_pb2
from chromite.contrib.sdk_server.grpc_server.chromite.api import sdk_pb2
from chromite.contrib.sdk_server.grpc_server.chromite.api import sysroot_pb2
from chromite.contrib.sdk_server.grpc_server.chromiumos import (
common_pb2 as common,
)
from chromite.lib.parser import package_info
UI_DIR = Path(__file__).parent
app = flask.Flask(
"SDK Server",
template_folder=UI_DIR / "templates",
static_folder=UI_DIR / "static",
)
index_data = {}
# Code 204 "No content" for endpoints which just execute a function.
NO_RESPONSE_OK = ("", 204)
COMMANDS_TO_IGNORE = (
"cros_workon_info",
"current_boards",
"cros_workon_list",
"cros_workon_start",
"cros_workon_stop",
)
def logGenerator(endpoint, req):
"""Utility generator for endpoints with long logs."""
for response in endpoint(req):
if response.logging_info:
yield response.logging_info
@app.route("/get-logs", methods=["GET", "POST"])
def get_logs():
if flask.request.method != "POST":
return flask.redirect(flask.url_for("index"))
logsResponse = client.get_logs(sdk_server_pb2.LogsRequest())
resp = []
for log in logsResponse.logs:
if log.command not in COMMANDS_TO_IGNORE:
second_line = log.logs.index("\n\n") + 2
# Internal logger adds a second newline to output which
# already has them. Remove them to make it more readable.
log.logs = log.logs[second_line:].replace("\n\n", "\n")
resp += [json_format.MessageToDict(log)]
return flask.jsonify(resp[-1::-1])
@app.route("/clear-logs", methods=["GET", "POST"])
def clear_logs():
"""Clears log files."""
if flask.request.method != "POST":
return flask.redirect(flask.url_for("index"))
req = sdk_server_pb2.ClearLogsRequest()
client.clear_logs(req)
return NO_RESPONSE_OK
@app.route("/workon-start", methods=["GET", "POST"])
def workon_start():
if flask.request.method != "POST":
# Handles user just going to /workon-start, rather than via the button.
return flask.redirect(flask.url_for("index"))
board = flask.request.json["board"]
package = flask.request.json["package"]
parsed_pkg = package_info.parse(package)
package = common.PackageInfo()
controller_util.serialize_package_info(parsed_pkg, package)
req = sdk_server_pb2.WorkonStartRequest(
build_target=common.BuildTarget(name=board),
package_info=package,
)
client.cros_workon_start(req)
return NO_RESPONSE_OK
@app.route("/workon-stop", methods=["GET", "POST"])
def workon_stop():
if flask.request.method != "POST":
# Handles user just going to /workon-stop, rather than via the button.
return flask.redirect(flask.url_for("index"))
board = flask.request.json["board"]
package = flask.request.json["package"]
parsed_pkg = package_info.parse(package)
package = common.PackageInfo()
controller_util.serialize_package_info(parsed_pkg, package)
req = sdk_server_pb2.WorkonStopRequest(
build_target=common.BuildTarget(name=board),
package_info=package,
)
client.cros_workon_stop(req)
return NO_RESPONSE_OK
@app.route("/repo-refresh", methods=["GET", "POST"])
def repo_refresh():
"""App route to run gRPC repo status endpoint and parse."""
if flask.request.method != "POST":
# GET (user visiting URL) redirects to homepage.
return flask.redirect(flask.url_for("index"))
# POST only sent by script.
response = client.repo_status(sdk_server_pb2.RepoStatusRequest())
project = ""
branch = ""
files = []
for line in response.info:
text = line.split()
if text[0] == "project":
project = text[1]
branch = text[3]
else:
files += [
{"file": text[1], "head": text[0][0], "working": text[0][1]}
]
# Returns dict-like format which JS parses into HTML.
return flask.jsonify(
{
"project": project,
"branch": branch,
"files": files,
}
)
@app.route("/repo-sync", methods=["GET", "POST"])
def repo_sync():
"""Enacts repo sync and streams logging data."""
if flask.request.method != "POST":
return flask.redirect(flask.url_for("index"))
req = sdk_server_pb2.RepoSyncRequest()
return flask.Response(logGenerator(client.repo_sync, req))
@app.route("/get-packages", methods=["GET", "POST"])
def get_packages():
"""App route to get packages from gRPC server."""
if flask.request.method != "POST":
# GET (user visiting URL) redirects to homepage.
return flask.redirect(flask.url_for("index"))
# POST only sent by script.
packages_json = {}
boards_to_get = []
# May request just for a single board (when updating from stop/start),
# but default should be all boards (for page loading).
if flask.request.json["board"]:
boards_to_get = [
sdk_server_pb2.BoardImages(
build_target=common.BuildTarget(
name=flask.request.json["board"]
)
)
]
else:
boards_to_get = client.current_boards(
sdk_server_pb2.CurrentBoardsRequest()
).board_images
for board in boards_to_get:
req = sdk_server_pb2.WorkonListRequest(build_target=board.build_target)
board_packages = (client.cros_workon_list(req)).package_info
board_packages_data = []
for pkg_msg in board_packages:
req = sdk_server_pb2.WorkonInfoRequest(
build_target=board.build_target, package_info=pkg_msg
)
# Parse info for modals.
_, repo, source = client.cros_workon_info(req).info.split(" ")
repo = repo.split(",")
source = source.split(",")
pkg = controller_util.deserialize_package_info(pkg_msg)
board_packages_data += [
{
"name": pkg.atom.strip(),
"repo": repo,
"source": source,
}
]
board_images_data = []
for image in board.images:
board_images_data += [
{
"path": image.path,
"type": common.ImageType.Name(image.type),
"latest": (image == board.latest),
}
]
packages_json[board.build_target.name] = {
"images": board_images_data,
"packages": board_packages_data,
}
return flask.jsonify(packages_json)
@app.route("/update-chroot", methods=["GET", "POST"])
def update_chroot():
"""App route to call BAPI Update."""
if flask.request.method != "POST":
return flask.redirect(flask.url_for("index"))
flags = sdk_pb2.UpdateRequest.Flags(
build_source=flask.request.json["buildSource"],
toolchain_changed=flask.request.json["toolchainChanged"],
)
targets = flask.request.json["toolchainTargets"]
toolchain_targets = [common.BuildTarget(name=b) for b in targets]
req = sdk_server_pb2.UpdateChrootRequest(
request=sdk_pb2.UpdateRequest(
flags=flags, toolchain_targets=toolchain_targets
)
)
return flask.Response(logGenerator(client.update_chroot, req))
@app.route("/replace-chroot", methods=["GET", "POST"])
def replace_chroot():
"""Forwards requests for replace and create chroot endpoints."""
if flask.request.method != "POST":
return flask.redirect(flask.url_for("index"))
flags = sdk_pb2.CreateRequest.Flags(
no_replace=False,
bootstrap=flask.request.json["bootstrap"],
no_use_image=flask.request.json["noUseImage"],
)
req = sdk_server_pb2.ReplaceSdkRequest(
request=sdk_pb2.CreateRequest(
flags=flags, sdk_version=flask.request.json["version"]
)
)
return flask.Response(logGenerator(client.replace_sdk, req))
@app.route("/delete-chroot", methods=["GET", "POST"])
def delete_sdk():
"""Calls delete SDK endpoint."""
if flask.request.method != "POST":
return flask.redirect(flask.url_for("index"))
req = sdk_server_pb2.DeleteSdkRequest(request=sdk_pb2.DeleteRequest())
return flask.Response(logGenerator(client.delete_sdk, req))
@app.route("/build-packages", methods=["GET", "POST"])
def build_packages():
"""Formulates/forwards all 3 requests for build packages endpoint."""
if flask.request.method != "POST":
return flask.redirect(flask.url_for("index"))
package = ""
if "package" in flask.request.json:
package = flask.request.json["package"]
parsed_pkg = package_info.parse(package)
package = common.PackageInfo()
controller_util.serialize_package_info(parsed_pkg, package)
create_sysroot = sysroot_pb2.SysrootCreateRequest(
flags=sysroot_pb2.SysrootCreateRequest.Flags(
chroot_current=flask.request.json["chrootCurrent"],
replace=flask.request.json["replace"],
toolchain_changed=flask.request.json["toolchainChanged"],
use_cq_prebuilts=flask.request.json["CQPrebuilts"],
),
build_target=common.BuildTarget(name=flask.request.json["buildTarget"]),
)
install_toolchain = sysroot_pb2.InstallToolchainRequest(
flags=sysroot_pb2.InstallToolchainRequest.Flags(
compile_source=flask.request.json["compileSource"],
toolchain_changed=flask.request.json["toolchainChanged"],
)
)
install_packages = sysroot_pb2.InstallPackagesRequest(
flags=sysroot_pb2.InstallPackagesRequest.Flags(
compile_source=flask.request.json["compileSource"],
toolchain_changed=flask.request.json["toolchainChanged"],
dryrun=flask.request.json["dryrun"],
workon=flask.request.json["workon"],
),
)
if package:
install_packages.packages.append(package)
req = sdk_server_pb2.BuildPackagesRequest(
create_req=create_sysroot,
toolchain_req=install_toolchain,
packages_req=install_packages,
)
return flask.Response(logGenerator(client.build_packages, req))
@app.route("/build-image", methods=["GET", "POST"])
def build_image():
"""Forwards build image request to gRPC server."""
if flask.request.method != "POST":
return flask.redirect(flask.url_for("index"))
# Converts strings to proto enum values.
toEnum = common.ImageType.Value
req = sdk_server_pb2.BuildImageRequest(
request=image_pb2.CreateImageRequest(
build_target=common.BuildTarget(
name=flask.request.json["buildTarget"]
),
image_types=[toEnum(im) for im in flask.request.json["imageTypes"]],
disable_rootfs_verification=flask.request.json[
"disableRootfsVerification"
],
version=flask.request.json["version"],
disk_layout=flask.request.json["diskLayout"],
builder_path=flask.request.json["builderPath"],
base_is_recovery=flask.request.json["baseIsRecovery"],
)
)
return flask.Response(logGenerator(client.build_image, req))
@app.route("/chroot-info", methods=["GET", "POST"])
def chroot_info():
"""Fetches basic Chroot Info for panel."""
if flask.request.method != "POST":
return flask.redirect(flask.url_for("index"))
return flask.jsonify(
json_format.MessageToDict(
client.chroot_info(sdk_server_pb2.ChrootInfoRequest())
)
)
@app.route("/custom", methods=["GET", "POST"])
def custom_endpoint():
"""Forwards custom endpoint/JSON request."""
if flask.request.method != "POST":
return flask.redirect(flask.url_for("index"))
req = sdk_server_pb2.CustomRequest(
endpoint=flask.request.json["endpoint"],
request=flask.request.json["request"],
)
return flask.Response(logGenerator(client.custom_endpoint, req))
@app.route("/all-packages", methods=["GET", "POST"])
def all_packages():
if flask.request.method != "POST":
return flask.redirect(flask.url_for("index"))
req = sdk_server_pb2.AllPackagesRequest(
build_target=common.BuildTarget(name=flask.request.json["board"])
)
resp = client.all_packages(req)
return flask.jsonify([p.package_name for p in resp.package_info])
@app.route("/", methods=["GET", "POST"])
def index():
"""Home page route. Renders index.html file with templating data."""
return flask.render_template("index.html", data=index_data)
def setup():
"""Populates initial templating data from gRPC requests."""
index_data["user"] = os.getlogin()
index_data["hostname"] = socket.gethostname()
all_boards = client.query_boards(sdk_server_pb2.QueryBoardsRequest())
all_boards = sorted([b.name for b in all_boards.build_target])
index_data["all_boards"] = all_boards
# List of boards with active sysroots for packages display.
current_boards = client.current_boards(
sdk_server_pb2.CurrentBoardsRequest()
)
current_boards = sorted(
[b.build_target.name for b in current_boards.board_images]
)
index_data["current_boards"] = current_boards
all_endpoints = client.get_methods(sdk_server_pb2.MethodsRequest()).response
all_endpoints = [m.method for m in all_endpoints.methods]
index_data["all_endpoints"] = all_endpoints
# pylint: disable=unused-argument
def main(argv: Optional[List[str]]) -> Optional[int]:
"""Runs setup and hosts the Flask app."""
setup()
app.jinja_env.auto_reload = True
app.run(debug=True, host="0.0.0.0")