Merge pull request #9 from google-ai-edge:feat-mediatek-support

PiperOrigin-RevId: 913747130

Author: copybara-github

litert_cli/commands/benchmark/android.py

@@ -128,6 +128,15 @@ def run_android(*, model_path: pathlib.Path, accelerator: str) -> None:
           f"--compiler_plugin_library_path={shlex.quote(cli_android_root)}"
       )
 
+      if soc_vendor == "mediatek":
+        recommend_version = constants.MEDIATEK_SOC_VERSION_MAP.get(
+            target_model, ""
+        )
+        if "v9" in recommend_version:
+          bench_args.append("--mediatek_nerun_pilot_version=version9")
+        elif "v8" in recommend_version:
+          bench_args.append("--mediatek_nerun_pilot_version=version8")
+
     env_vars = ""
     if remote_dispatch_dir:
       quoted_dispatch_dir = shlex.quote(remote_dispatch_dir)
@@ -137,6 +146,32 @@ def run_android(*, model_path: pathlib.Path, accelerator: str) -> None:
       )
 
     full_command = env_vars + " ".join(bench_args)
-    subprocess.run(["adb", "shell", full_command], check=True)
-  except subprocess.CalledProcessError as e:
-    click.secho(f"Execution failed on device: {repr(e)}", fg="red")
+    process = subprocess.Popen(
+        ["adb", "shell", full_command],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+    )
+
+    from litert_cli.core.log_filters import BenchmarkLogFilter
+
+    output_lines = []
+    log_filter = BenchmarkLogFilter(constants.DEFAULT_QUIET)
+
+    for line in process.stdout:
+      output_lines.append(line)
+      if log_filter.should_show(line):
+        click.echo(line, nl=False)
+
+    process.wait()
+    if process.returncode != 0:
+      click.secho(
+          f"Execution failed on device with exit code {process.returncode}",
+          fg="red",
+      )
+      click.echo("Full output for debugging:")
+      for line in output_lines:
+        click.echo(line, nl=False)
+      raise click.ClickException("Benchmark failed on device.")
+  except Exception as e:
+    raise click.ClickException(f"Failed to execute benchmark on device: {e}")

litert_cli/commands/compile.py

@@ -21,6 +21,7 @@
 from __future__ import annotations
 
 from collections.abc import Sequence
+import importlib
 import pathlib
 import shutil
 import textwrap
@@ -30,6 +31,7 @@
 from litert_cli.core import deps
 from litert_cli.core import npu_utils
 from litert_cli.core import utils
+from litert_cli.core.targets_manager import TargetsManager
 
 
 @click.command(
@@ -52,6 +54,16 @@
         """),
 )
 @click.argument("model_path", type=str)
+@click.option(
+    "--update-targets",
+    type=str,
+    required=False,
+    default=None,
+    help=(
+        "Update SoC target lists from GitHub. Pass 'main' for latest, or a"
+        " version tag like 'v2.1.4'."
+    ),
+)
 @click.option(
     "--target",
     type=str,
@@ -93,6 +105,7 @@
 def compile_cmd(
     model_path: str,
     target: Sequence[str],
+    update_targets: str | None,
     export_aipack: pathlib.Path | None,
     output_dir: pathlib.Path | None,
 ) -> None:
@@ -115,6 +128,24 @@ def compile_cmd(
   if constants.DEFAULT_QUIET:
     utils.enable_quiet_mode()
 
+  # Initialize targets
+  manager = TargetsManager()
+
+  # Handle update or first-run download
+  if update_targets:
+    manager.download_targets(version=update_targets)
+    importlib.reload(constants)
+  else:
+    # Check if cache exists
+    if not manager.load_targets():
+      click.echo("No target cache found. Downloading default target lists...")
+      try:
+        manager.download_targets(version="main")
+        importlib.reload(constants)
+      except Exception as e:
+        click.echo(f"Warning: Failed to download default targets: {e}")
+        click.echo("Falling back to built-in static target lists.")
+
   resolved_model_path, _ = core_models.resolve_model_reference(model_path)
   if str(resolved_model_path) != str(model_path):
     click.echo(f"Resolved model '{model_path}' to '{resolved_model_path}'")

litert_cli/commands/run/android.py

@@ -352,16 +352,35 @@ def run_android(
       env_vars = ""
     cmd_str = f"{env_vars} " if env_vars else ""
     cmd_str += " ".join(shlex.quote(arg) for arg in run_cmd_args)
-    subprocess.run(
-        [
-            "adb",
-            "shell",
-            cmd_str,
-        ],
-        check=True,
+    process = subprocess.Popen(
+        ["adb", "shell", cmd_str],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
     )
-  except subprocess.CalledProcessError as e:
-    raise click.ClickException(f"Execution failed on device: {e!r}") from e
+
+    from litert_cli.core.log_filters import RunLogFilter
+
+    output_lines = []
+    log_filter = RunLogFilter(constants.DEFAULT_QUIET, print_tensors)
+
+    for line in process.stdout:
+      output_lines.append(line)
+      if log_filter.should_show(line):
+        click.echo(line, nl=False)
+
+    process.wait()
+    if process.returncode != 0:
+      click.secho(
+          f"Execution failed on device with exit code {process.returncode}",
+          fg="red",
+      )
+      click.echo("Full output for debugging:")
+      for line in output_lines:
+        click.echo(line, nl=False)
+      raise click.ClickException("Execution failed on device.")
+  except Exception as e:
+    raise click.ClickException(f"Failed to execute on device: {e}")
   finally:
     # Cleanup remote paths
     click.echo("Clearing remote files...")

litert_cli/core/constants.py

@@ -34,7 +34,7 @@
 ENV_LITERT_CLI_FORCE_OSS: str = "LITERT_CLI_FORCE_OSS"
 ENV_LITERT_VERBOSE: str = "LITERT_VERBOSE"
 
-DEFAULT_QUIET: bool = os.environ.get(ENV_LITERT_VERBOSE, "1") != "1"
+DEFAULT_QUIET: bool = os.environ.get(ENV_LITERT_VERBOSE, "0") != "1"
 
 _FORCE_OSS = os.environ.get(ENV_LITERT_CLI_FORCE_OSS, "").lower() in (
     "1",
@@ -80,61 +80,46 @@
     "https://softwarecenter.qualcomm.com/api/download/software/sdks/"
     f"Qualcomm_AI_Runtime_Community/All/{QAIRT_SDK_VERSION}/v{QAIRT_SDK_VERSION}.zip"
 )
+MEDIATEK_SDK_URL: str = (
+    "https://s3.ap-southeast-1.amazonaws.com/mediatek.neuropilot.com/"
+    "66f2c33a-2005-4f0b-afef-2053c8654e4f.gz"
+)
+MEDIATEK_V8_VERSION: str = "v8_0_10"
+MEDIATEK_V9_VERSION: str = "v9_0_3"
+
+from litert_cli.core.targets_manager import TargetsManager
+
+_manager = TargetsManager()
+_loaded_targets = _manager.load_targets()
+
+_qnn_map = {}
+_mtk_map = {}
+_aot_map = {}
+
+if _loaded_targets:
+  # Reconstruct maps from loaded targets
+  _qnn_map = {
+      k: v.properties.get("qnn_version", "")
+      for k, v in _loaded_targets.items()
+      if v.vendor == "qualcomm"
+  }
 
-# TODO: switch to read from litert/vendors/qualcomm/supported_soc.csv
-QNN_SOC_VERSION_MAP: types.MappingProxyType[str, str] = types.MappingProxyType({
-    "sm8350": "68",
-    "sm8450": "69",
-    "sm8550": "73",
-    "sm8650": "75",
-    "sm8750": "79",
-    "sm8850": "81",
-    # Sub-flagship & Mid-range Mobile SoCs
-    "sm8635": "73",
-    "sm7675": "73",
-    "sm7550": "73",
-    "sm7475": "69",
-    # Automotive Cockpit SoCs
-    "sa8255": "73",
-    "sa8295": "68",
-    "qnn_all": "81",
-})
+  _mtk_map = {
+      k: v.properties.get("recommend_version", "")
+      for k, v in _loaded_targets.items()
+      if v.vendor == "mediatek"
+  }
+
+  _aot_map = {k: (v.vendor, v.vendor_id) for k, v in _loaded_targets.items()}
+
+QNN_SOC_VERSION_MAP: types.MappingProxyType[str, str] = types.MappingProxyType(
+    _qnn_map
+)
+
+MEDIATEK_SOC_VERSION_MAP: types.MappingProxyType[str, str] = (
+    types.MappingProxyType(_mtk_map)
+)
 
 AOT_SUPPORTED_TARGETS: types.MappingProxyType[str, tuple[str, str]] = (
-    types.MappingProxyType({
-        # Qualcomm
-        "sm8350": ("qualcomm", "SM8350"),
-        "sm8450": ("qualcomm", "SM8450"),
-        "sm8550": ("qualcomm", "SM8550"),
-        "sm8650": ("qualcomm", "SM8650"),
-        "sm8750": ("qualcomm", "SM8750"),
-        "sm8850": ("qualcomm", "SM8850"),
-        # Sub-flagship & Mid-range Mobile SoCs
-        "sm8635": ("qualcomm", "SM8635"),
-        "sm7675": ("qualcomm", "SM7675"),
-        "sm7550": ("qualcomm", "SM7550"),
-        "sm7475": ("qualcomm", "SM7475"),
-        # Automotive Cockpit SoCs
-        "sa8255": ("qualcomm", "SA8255"),
-        "sa8295": ("qualcomm", "SA8295"),
-        "qnn_all": ("qualcomm", "ALL"),
-        # MediaTek
-        "mt6853": ("mediatek", "MT6853"),
-        "mt6877": ("mediatek", "MT6877"),
-        "mt6878": ("mediatek", "MT6878"),
-        "mt6879": ("mediatek", "MT6879"),
-        "mt6886": ("mediatek", "MT6886"),
-        "mt6893": ("mediatek", "MT6893"),
-        "mt6895": ("mediatek", "MT6895"),
-        "mt6897": ("mediatek", "MT6897"),
-        "mt6983": ("mediatek", "MT6983"),
-        "mt6985": ("mediatek", "MT6985"),
-        "mt6989": ("mediatek", "MT6989"),
-        "mt6991": ("mediatek", "MT6991"),
-        "mt6993": ("mediatek", "MT6993"),
-        "mt8171": ("mediatek", "MT8171"),
-        "mt8188": ("mediatek", "MT8188"),
-        "mt8189": ("mediatek", "MT8189"),
-        "mtk_all": ("mediatek", "ALL"),
-    })
+    types.MappingProxyType(_aot_map)
 )

litert_cli/core/log_filters.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+# Copyright 2026 The LiteRT CLI Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==============================================================================
+
+"""Log filters for LiteRT CLI commands to prevent terminal noise."""
+
+
+class BenchmarkLogFilter:
+  """Filters output of litert benchmark command."""
+
+  def __init__(self, default_quiet: bool):
+    self.default_quiet = default_quiet
+
+  def should_show(self, line: str) -> bool:
+    """Determines if a line should be shown in the output."""
+    is_core_info = (
+        "benchmark_litert_model" in line
+        or "compiler_plugin.cc" in line
+        or ("Replacing" in line and "node(s) with delegate" in line)
+    )
+    return not self.default_quiet or is_core_info
+
+
+class RunLogFilter:
+  """Filters output of litert run command."""
+
+  def __init__(self, default_quiet: bool, print_tensors: bool):
+    self.default_quiet = default_quiet
+    self.print_tensors = print_tensors
+
+  def should_show(self, line: str) -> bool:
+    """Determines if a line should be shown in the output."""
+    is_core_info = (
+        "run_model.cc" in line
+        or "compiler_plugin.cc" in line
+        or ("Replacing" in line and "node(s) with delegate" in line)
+    )
+    return not self.default_quiet or is_core_info or self.print_tensors