feat: find out the installed mosec path (#624)

Signed-off-by: Keming <[email protected]>

Author: kemingy

.github/workflows/check.yml

@@ -57,7 +57,6 @@ jobs:
     runs-on: ${{ matrix.os }}
     timeout-minutes: 20
     strategy:
-      max-parallel: 18
       matrix:
         python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
         os: [ubuntu-24.04, ubuntu-24.04-arm, macos-13, macos-14]

mosec/runtime.py

@@ -26,6 +26,7 @@
 from mosec.coordinator import Coordinator
 from mosec.env import env_var_context, validate_env, validate_int_ge
 from mosec.log import get_internal_logger
+from mosec.utils import get_mosec_path
 from mosec.worker import Worker
 
 logger = get_internal_logger()
@@ -270,5 +271,5 @@ def start(self, config_path: Path) -> subprocess.Popen:
 
         """
         # pylint: disable=consider-using-with
-        self.process = subprocess.Popen([str("mosec"), config_path])
+        self.process = subprocess.Popen([str(get_mosec_path()), config_path])
         return self.process

mosec/utils/types.py renamed to mosec/utils.py

@@ -1,4 +1,4 @@
-# Copyright 2023 MOSEC Authors
+# Copyright 2025 MOSEC Authors
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -15,9 +15,43 @@
 """Provide useful utils to inspect function type."""
 
 import inspect
+import os
 import sys
+import sysconfig
 from enum import Enum
-from typing import Any, List
+from pathlib import Path
+from typing import Any, List, Optional
+
+
+# adopted from https://github.com/PyO3/maturin/blob/main/maturin/__main__.py
+# License: Apache-2.0 or MIT
+def get_mosec_path() -> Optional[Path]:
+    """Get `mosec` binary path."""
+    SCRIPT_NAME = "mosec"
+
+    def script_dir(scheme: str) -> str:
+        return sysconfig.get_path("scripts", scheme)
+
+    def script_exists(dir: str) -> bool:
+        for _, _, files in os.walk(dir):
+            for f in files:
+                name, *_ = os.path.splitext(f)
+                if name == SCRIPT_NAME:
+                    return True
+
+        return False
+
+    paths = list(
+        filter(
+            script_exists,
+            filter(os.path.exists, map(script_dir, sysconfig.get_scheme_names())),
+        )
+    )
+
+    if paths:
+        return Path(paths[0]) / SCRIPT_NAME
+
+    return None
 
 
 def get_annotations(func) -> dict: