utils: add source_info utility module

Signed-off-by:  Eric Callahan <arksine.code@gmail.com>

moonraker/components/file_manager/file_manager.py

@@ -18,7 +18,7 @@ import time
 from copy import deepcopy
 from inotify_simple import INotify
 from inotify_simple import flags as iFlags
-from ...utils import MOONRAKER_PATH
+from ...utils import source_info
 
 # Annotation imports
 from typing import (
@@ -70,7 +70,8 @@ class FileManager:
         self.file_paths: Dict[str, str] = {}
         app_args = self.server.get_app_args()
         self.datapath = pathlib.Path(app_args["data_path"])
-        self.add_reserved_path("moonraker", MOONRAKER_PATH, False)
+        srcdir = str(source_info.source_path())
+        self.add_reserved_path("moonraker", srcdir, False)
         db: DBComp = self.server.load_component(config, "database")
         db_path = db.get_database_path()
         self.add_reserved_path("database", db_path, False)

moonraker/components/machine.py

@@ -22,7 +22,7 @@ import tempfile
 import getpass
 import configparser
 from ..confighelper import FileSourceWrapper
-from ..utils import MOONRAKER_PATH
+from ..utils import source_info
 
 # Annotation imports
 from typing import (
@@ -1656,7 +1656,7 @@ class InstallValidator:
         tmp_svc = pathlib.Path(
             tempfile.gettempdir()
         ).joinpath(f"{unit}-tmp.svc")
-        src_path = pathlib.Path(MOONRAKER_PATH)
+        src_path = source_info.source_path()
         # Create local environment file
         sysd_data = self.data_path.joinpath("systemd")
         if not sysd_data.exists():

moonraker/components/update_manager/base_config.py

@@ -8,7 +8,7 @@ from __future__ import annotations
 import os
 import sys
 import copy
-from ...utils import MOONRAKER_PATH
+from ...utils import source_info
 from typing import (
     TYPE_CHECKING,
     Dict
@@ -29,7 +29,7 @@ BASE_CONFIG: Dict[str, Dict[str, str]] = {
         "install_script": "scripts/install-moonraker.sh",
         "host_repo": "arksine/moonraker",
         "env": sys.executable,
-        "path": MOONRAKER_PATH,
+        "path": str(source_info.source_path()),
         "managed_services": "moonraker"
     },
     "klipper": {

moonraker/utils/__init__.py

@@ -19,6 +19,7 @@ import shlex
 import re
 import struct
 import socket
+from . import source_info
 
 # Annotation imports
 from typing import (
@@ -28,13 +29,13 @@ from typing import (
     ClassVar,
     Tuple,
     Dict,
+    Union
 )
 
 if TYPE_CHECKING:
     from types import ModuleType
     from asyncio.trsock import TransportSocket
 
-MOONRAKER_PATH = str(pathlib.Path(__file__).parent.parent.parent.resolve())
 SYS_MOD_PATHS = glob.glob("/usr/lib/python3*/dist-packages")
 SYS_MOD_PATHS += glob.glob("/usr/lib/python3*/site-packages")
 
@@ -87,20 +88,15 @@ def retrieve_git_version(source_path: str) -> str:
     return f"t{tag}-g{ver}-shallow"
 
 def get_software_version() -> str:
+    pkg_ver = source_info.package_version()
+    if pkg_ver is not None:
+        return pkg_ver
     version: str = "?"
+    src_path = source_info.source_path()
     try:
-        import moonraker.__version__ as ver  # type: ignore
+        version = retrieve_git_version(str(src_path))
-        version = ver.__version__
     except Exception:
-        pass
+        vfile = src_path.joinpath("moonraker/.version")
-    else:
-        if version:
-            return version
-    try:
-        version = retrieve_git_version(MOONRAKER_PATH)
-    except Exception:
-        vfile = pathlib.Path(os.path.join(
-            MOONRAKER_PATH, "moonraker/.version"))
         if vfile.exists():
             try:
                 version = vfile.read_text().strip()
@@ -110,12 +106,15 @@ def get_software_version() -> str:
     return version
 
 
-def hash_directory(dir_path: str,
+def hash_directory(
-                   ignore_exts: List[str],
+    dir_path: Union[str, pathlib.Path],
-                   ignore_dirs: List[str]
+    ignore_exts: List[str],
-                   ) -> str:
+    ignore_dirs: List[str]
+) -> str:
+    if isinstance(dir_path, str):
+        dir_path = pathlib.Path(dir_path)
     checksum = hashlib.blake2s()
-    if not os.path.exists(dir_path):
+    if not dir_path.exists():
         return ""
     for dpath, dnames, fnames in os.walk(dir_path):
         valid_dirs: List[str] = []
@@ -135,8 +134,14 @@ def hash_directory(dir_path: str,
                 pass
     return checksum.hexdigest()
 
-def verify_source(path: str = MOONRAKER_PATH) -> Optional[Tuple[str, bool]]:
+def verify_source(
-    rfile = pathlib.Path(os.path.join(path, ".release_info"))
+    path: Optional[Union[str, pathlib.Path]] = None
+) -> Optional[Tuple[str, bool]]:
+    if path is None:
+        path = source_info.source_path()
+    elif isinstance(path, str):
+        path = pathlib.Path(path)
+    rfile = path.joinpath(".release_info")
     if not rfile.exists():
         return None
     try:

moonraker/utils/source_info.py

@@ -0,0 +1,77 @@
+# General Server Utilities
+#
+# Copyright (C) 2023 Eric Callahan <arksine.code@gmail.com>
+#
+# This file may be distributed under the terms of the GNU GPLv3 license
+
+from __future__ import annotations
+import importlib.resources as ilr
+import pathlib
+import sys
+import site
+
+# Annotation imports
+from typing import (
+    Optional,
+)
+
+def package_path() -> pathlib.Path:
+    return pathlib.Path(__file__).parent.parent
+
+def source_path() -> pathlib.Path:
+    return package_path().parent
+
+def is_git_repo(src_path: Optional[pathlib.Path] = None) -> bool:
+    if src_path is None:
+        return source_path().joinpath(".git").is_dir()
+    return src_path.joinpath(".git").is_dir()
+
+def is_dist_package(src_path: Optional[pathlib.Path] = None) -> bool:
+    if src_path is None:
+        # Fetch dist info for moonraker.  This should be the most reliable
+        # method
+        src_path = source_path()
+        site_dirs = site.getsitepackages()
+        return str(src_path) in site_dirs
+    else:
+        # Make an assumption based on the source path.  If its name is
+        # site-packages or dist-packages then presumably it is an
+        # installed package
+        return src_path.name in ["dist-packages", "site-packages"]
+
+def package_version() -> Optional[str]:
+    try:
+        import moonraker.__version__ as ver  # type: ignore
+        version = ver.__version__
+    except Exception:
+        pass
+    else:
+        if version:
+            return version
+    return None
+
+def read_asset(asset_name: str) -> Optional[str]:
+    if sys.version_info < (3, 10):
+        with ilr.path("moonraker.assets", asset_name) as p:
+            if not p.is_file():
+                return None
+            return p.read_text()
+    else:
+        files = ilr.files("moonraker.assets")
+        with ilr.as_file(files.joinpath(asset_name)) as p:
+            if not p.is_file():
+                return None
+            return p.read_text()
+
+def get_asset_path() -> Optional[pathlib.Path]:
+    if sys.version_info < (3, 10):
+        with ilr.path("moonraker.assets", "__init__.py") as p:
+            asset_path = p.parent
+    else:
+        files = ilr.files("moonraker.assets")
+        with ilr.as_file(files.joinpath("__init__.py")) as p:
+            asset_path = p.parent
+    if not asset_path.is_dir():
+        # Somehow running in a zipapp.  This is an error.
+        return None
+    return asset_path