from logging import Logger import os class GitOperations: logger: Logger pat: str def __init__(self, pat: str, logger: Logger): self.pat = pat self.logger = logger def _get_credential_callbacks(self): import pygit2 creds = pygit2.credentials.UserPass("sulfilogger", self.pat) callbacks = pygit2.RemoteCallbacks(credentials=creds) return callbacks def _safe_checkout_strategy(self, pygit2_module): # Support both newer and older pygit2 APIs. checkout_strategy = getattr(pygit2_module, "CheckoutStrategy", None) if checkout_strategy is not None: return checkout_strategy.SAFE | checkout_strategy.RECREATE_MISSING return pygit2_module.GIT_CHECKOUT_SAFE | pygit2_module.GIT_CHECKOUT_RECREATE_MISSING def install_or_update(self, product_name="unknown", repo_path=".", repo_url=None, remote_name="origin", branch_name=None): self.logger.info(f"Installing or updating {product_name} from git repository at {repo_path}...") if os.path.isdir(os.path.join(repo_path, ".git")): self.logger.info(f"{product_name} repository already exists. Attempting to update...") self.pull_ff_only(repo_path=repo_path, remote_name=remote_name, branch_name=branch_name) elif os.path.exists(repo_path): raise RuntimeError(f"Cannot install {product_name}: path exists but is not a git repository: {repo_path}") else: self.logger.info(f"{product_name} repository does not exist. Cloning...") self.clone(repo_url=repo_url, repo_path=repo_path, branch_name=branch_name) def clone(self, repo_url, repo_path=".", branch_name=None): if branch_name is None: branch_name = "main" import pygit2 self.logger.info(f"Cloning repository from {repo_url} to {repo_path}...") try: pygit2.clone_repository(repo_url, repo_path, callbacks=self._get_credential_callbacks(), checkout_branch=branch_name) self.logger.info(f"Successfully cloned repository from {repo_url} to {repo_path}.") except Exception as e: self.logger.error(f"Failed to clone repository from {repo_url} to {repo_path}: {str(e)}") raise RuntimeError(f"Failed to clone repository from {repo_url} to {repo_path}: {str(e)}") def pull_ff_only(self, repo_path=".", remote_name="origin", branch_name=None): import pygit2 repo = pygit2.Repository(repo_path) checkout_strategy = self._safe_checkout_strategy(pygit2) if repo.head_is_detached: raise RuntimeError("Cannot ff-only pull: HEAD is detached") if repo.head_is_unborn: raise RuntimeError("Cannot ff-only pull: HEAD is unborn") # Default to the currently checked-out branch. if branch_name is None: branch_name = repo.head.shorthand local_ref_name = f"refs/heads/{branch_name}" remote_ref_name = f"refs/remotes/{remote_name}/{branch_name}" remote = repo.remotes[remote_name] # Fetch updates refs/remotes/origin/, but does not touch the worktree. remote.fetch(callbacks=self._get_credential_callbacks()) remote_ref = repo.references.get(remote_ref_name) if remote_ref is None: raise RuntimeError(f"Remote ref not found after fetch: {remote_ref_name}") their_oid = remote_ref.target local_ref = repo.references.get(local_ref_name) # If the local branch does not exist yet, adopt the remote branch directly. # This avoids merge_analysis KeyError on refs/heads/ for first-time setup. if local_ref is None: their_commit = repo[their_oid] repo.checkout_tree( their_commit, strategy=checkout_strategy, ) repo.create_branch(branch_name, their_commit) repo.set_head(local_ref_name) self.logger.info(f"Created local branch {branch_name} from {remote_name}/{branch_name} at {their_oid}") return # Ensure HEAD is on the requested branch so ff analysis applies to that branch. try: current_head_name = repo.head.name except Exception: current_head_name = None if current_head_name != local_ref_name: current_commit = repo[local_ref.target] repo.checkout_tree( current_commit, strategy=checkout_strategy, ) repo.set_head(local_ref_name) analysis, _preference = repo.merge_analysis(their_oid) if analysis & pygit2.GIT_MERGE_ANALYSIS_UP_TO_DATE: self.logger.info("Already up to date") return if not (analysis & pygit2.GIT_MERGE_ANALYSIS_FASTFORWARD): raise RuntimeError( f"Cannot fast-forward {branch_name}; local and remote histories diverged" ) their_commit = repo[their_oid] # Safe checkout: refuses to overwrite conflicting local changes. repo.checkout_tree( their_commit, strategy=checkout_strategy, ) # Move the checked-out branch ref forward. local_ref = repo.references[local_ref_name] local_ref.set_target(their_oid, f"pull --ff-only: {remote_name}/{branch_name}") self.logger.info(f"Fast-forwarded {branch_name} to {their_oid}") import configparser from enum import Enum from logging import Logger from pathlib import Path import shutil import sys import os import platform import subprocess import time import urllib.error import urllib.request branch = "release" # should be changed to release in the future class ExternalDependency(Enum): SERVICEMANAGER = "servicemanager" INFLUXDB = "influxdb" MSSQLDRIVER = "mssql_driver" class PipDependency(Enum): VENV = "venv" PYGIT2 = "pygit2" PYODBC = "pyodbc" PSUTIL = "psutil" class MissingSecretException(Exception): """Raised when a required secret is missing from the secrets.ini file.""" def __init__(self, secret_name: str, instructions: str): super().__init__(f"Missing required secret: {secret_name}\n\t{instructions}") self.secret_name = secret_name self.instructions = instructions class Machine: logger: Logger secrets_path: str secrets = configparser.ConfigParser() def __init__(self, logger: Logger) -> None: self.logger = logger self.secrets_path = self._secrets_path() self.secrets.read(self.secrets_path) #region General helpers def download_file(self, url: str, destination: str) -> None: destination = Path(destination) destination.parent.mkdir(parents=True, exist_ok=True) request = urllib.request.Request( url, headers={ "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/125.0 Safari/537.36" ), "Accept": "application/zip,application/octet-stream,*/*", }, ) try: with urllib.request.urlopen(request, timeout=120) as response: with destination.open("wb") as out_file: shutil.copyfileobj(response, out_file) except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") raise RuntimeError( f"Download failed: HTTP {exc.code} {exc.reason}\n" f"URL: {url}\n" f"Response body:\n{body[:2000]}" ) from exc #endregion #region External dependency helpers def initialize_influxdb(self) -> None: if not self.check_external_dependency(ExternalDependency.INFLUXDB): raise RuntimeError("InfluxDB is not installed. Cannot initialize InfluxDB.") # Initialization assumes we control the bootstrap InfluxDB instance. # If something is already listening on 8181, token/db operations may hit # the wrong server and return misleading auth errors. if self.influxdb_wait_for_startup(timeout=1): raise RuntimeError( "Cannot initialize InfluxDB: port 8181 is already in use. " "Stop the running InfluxDB/xSensrKernel instance and retry." ) if os.path.isdir(os.path.join(self.influxdb_dir(), "data")): self.logger.error("InfluxDB data directory already exists. Cannot initialize InfluxDB.") confirm_delete = input("Are you sure you want to delete the existing InfluxDB data directory and re-initialize InfluxDB? (yes/no): ").strip().lower() if confirm_delete != "yes": self.logger.info("Aborting InfluxDB initialization.") return shutil.rmtree(os.path.join(self.influxdb_dir(), "data")) # create admin token admin_token_cmd = [self.influxdb_binary_path(), "create", "token", "--admin", "--offline", "--output-file", self.influxdb_admin_token_path()] result = subprocess.run(admin_token_cmd, capture_output=True, text=True) if result.returncode != 0: self.logger.error(f"Failed to create InfluxDB admin token.") self.logger.error(f"Ran command: {' '.join(admin_token_cmd)}.\n\tReturn code: {result.returncode}.\n\n\tStandard output: {result.stdout}\n\n\tError output: {result.stderr}.") raise RuntimeError("Failed to create InfluxDB admin token.") # load admin token from json file and store it in secrets.ini admin_token = "" with open(self.influxdb_admin_token_path(), "r") as token_file: import json admin_token_data = json.load(token_file) admin_token = admin_token_data.get("token", "").strip() if not admin_token or len(admin_token) < 10: raise RuntimeError("Failed to read InfluxDB admin token from file.") if not self.secrets.has_section("influxdb"): self.secrets.add_section("influxdb") self.secrets.set("influxdb", "admin_token", admin_token) self.write_secrets() serve_cmd = self.influxdb_serve_cmd() influx = subprocess.Popen(serve_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) started = self.influxdb_wait_for_startup(timeout=30) if influx.poll() is not None: stdout, stderr = influx.communicate(timeout=5) self.logger.error("InfluxDB process exited during initialization startup check.") self.logger.error( f"Ran command: {' '.join(serve_cmd)}.\n\n\tStandard output: {stdout.decode(errors='replace')}\n\n\tError output: {stderr.decode(errors='replace')}." ) raise RuntimeError("Failed to start InfluxDB server for initialization.") if not started: influx.terminate() stdout, stderr = influx.communicate(timeout=5) self.logger.error(f"Failed to start InfluxDB server for initialization.") self.logger.error(f"Ran command: {' '.join(serve_cmd)}.\n\n\tStandard output: {stdout.decode()}\n\n\tError output: {stderr.decode()}.") raise RuntimeError("Failed to start InfluxDB server for initialization.") db_name = "xSensrKernel" create_database_cmd = [self.influxdb_binary_path(), "create", "database", "--retention-period", "90d", "--token", admin_token, db_name] result = subprocess.run(create_database_cmd, capture_output=True, text=True) if result.returncode != 0: influx.terminate() stdout, stderr = influx.communicate(timeout=5) self.logger.error(f"Failed to create InfluxDB database.") self.logger.error(f"Ran command: {' '.join(create_database_cmd)}.\n\tReturn code: {result.returncode}.\n\n\tStandard output: {result.stdout}\n\n\tError output: {result.stderr}.") raise RuntimeError("Failed to create InfluxDB database.") def influxdb_admin_token(self) -> str: if not self.secrets.has_option("influxdb", "admin_token"): raise MissingSecretException("influxdb:admin_token", "InfluxDB admin token is missing from secrets file. Please ensure that InfluxDB is initialized and the admin token is stored in the secrets file.") return self.secrets.get("influxdb", "admin_token").strip() def influxdb_wait_for_startup(self, timeout: int = 30) -> bool: for _ in range(timeout): try: import socket with socket.create_connection(("localhost", 8181), timeout=1): return True break except Exception: time.sleep(1) return False def influxdb_serve_cmd(self) -> list[str]: """Returns the command to start the InfluxDB server.""" return [ self.influxdb_binary_path(), "serve", "--node-id", f"xSK-{self.get_hostname()}", "--object-store", "file", "--data-dir", os.path.join(self.influxdb_dir(), "data"), "--admin-token-file", self.influxdb_admin_token_path(), "--http-bind", "0.0.0.0:8181", "--log-filter", "warn", ] def check_external_dependency(self, dependency: ExternalDependency) -> bool: raise NotImplementedError("check_external_dependency method must be implemented by subclasses") def install_external_dependency(self, dependency: ExternalDependency) -> bool: raise NotImplementedError("install_external_dependency method must be implemented by subclasses") def ensure_external_dependency(self, dependency: ExternalDependency) -> None: if not self.check_external_dependency(dependency): self.logger.info(f"{dependency.value} is not installed. Attempting to install {dependency.value}...") try: self.install_external_dependency(dependency) self.logger.info(f"{dependency.value} installation successful.") except Exception as e: self.logger.error(f"{dependency.value} installation failed: {e}") raise else: self.logger.info(f"{dependency.value} is already installed.") if not self.check_external_dependency(dependency): raise Exception(f"Could not verify external dependency {dependency.value} installation after successful installation.") #endregion #region Application environment helpers def has_system_level_permissions(self) -> bool: if platform.system() == "Windows": try: import ctypes return ctypes.windll.shell32.IsUserAnAdmin() != 0 except Exception as e: return False elif platform.system() in ["Linux", "Darwin"]: return os.geteuid() == 0 else: raise NotImplementedError(f"Unsupported platform: {platform.system()}") def check_xsensr_kernel_running(self) -> bool: import socket with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: result = sock.connect_ex(('localhost', 5000)) if result == 0: self.logger.info("xSensrKernel service is now running and listening on port 5000.") return True else: return False def ensure_virtualenv(self) -> bool: result = sys.prefix != getattr(sys, "base_prefix", sys.prefix) or hasattr(sys, "real_prefix") if not result: self.logger.error("Not currently in a virtual environment.") exit(1) def check_running_as_service(self) -> bool: raise NotImplementedError("check_running_as_service method must be implemented by subclasses") def get_hostname(self) -> str: return platform.node() #endregion #region Pip environment helpers def ensure_pip(self) -> None: # will python always resolve to the currently running python? should we use sys.executable instead? python_executable = sys.executable result = subprocess.run([python_executable, "-m", "pip", "--version"], capture_output=True, text=True) if result.returncode != 0: raise Exception(f"Failed to ensure pip is installed: {result.stderr}") #only first line of stdout self.logger.info(f"pip is available: {result.stdout.splitlines()[0]}") def check_pip_dependency(self, dependency: PipDependency) -> bool: # check if the pip dependency is installed by trying to import it try: __import__(dependency.value) return True except ImportError: return False def install_pip_dependency(self, dependency: PipDependency) -> None: # get python executable for the virtual environment python_executable = sys.executable result = subprocess.run([python_executable, "-m", "pip", "install", dependency.value], capture_output=True, text=True) self.logger.debug(f"pip install output: {result.stdout}") if result.returncode != 0: raise Exception(f"Failed to install pip package {dependency.value}: {result.stderr}") if not self.check_pip_dependency(dependency): raise Exception(f"Could not verify pip package {dependency.value} installation after installation attempt: {result.stderr}") def ensure_pip_dependency(self, dependency: PipDependency) -> None: if not self.check_pip_dependency(dependency): self.logger.info(f"pip package {dependency.value} is not installed. Attempting to install {dependency.value}...") try: self.install_pip_dependency(dependency) self.logger.info(f"pip package {dependency.value} installation successful.") except Exception as e: self.logger.error(f"pip package {dependency.value} installation failed: {e}") raise else: self.logger.info(f"pip package {dependency.value} is already installed.") #endregion #region Filesystem paths def app_data_dir(self) -> str: """Returns the application data directory for the current user, depending on the platform. Windows: %APPDATA%\\xSensrKernel Linux/macOS: ~/.local/share/xSensrKernel""" data_dir = "" if platform.system() == "Windows": data_dir = os.getenv("APPDATA") elif platform.system() in ["Linux", "Darwin"]: # if running as sudo, use the original user's home directory instead of root's home directory if os.getenv("SUDO_USER"): sudo_user_home = os.path.expanduser(f"~{os.getenv('SUDO_USER')}") data_dir = os.path.join(sudo_user_home, ".local", "share") else: data_dir = os.path.expanduser("~/.local/share") else: raise NotImplementedError(f"Unsupported platform: {platform.system()}") return os.path.join(data_dir, "xSensrKernel") def repo_dir(self) -> str: """Returns the directory where the xSensrKernel repository is cloned.""" return os.path.join(self.app_data_dir(), "repos", "xSensrKernel") def bootstrap_script_path(self) -> str: """Returns the path to the bootstrap script inside the xSensrKernel repository.""" return os.path.join(self.repo_dir(), "bootstrap", "bootstrap.py") def venv_dir(self) -> str: """Returns the directory where the Python virtual environment is located.""" return os.path.join(self.app_data_dir(), "tools", "python", "venv") def influxdb_dir(self) -> str: """Returns the directory where InfluxDB is installed.""" return os.path.join(self.app_data_dir(), "tools", "influxdb") def influxdb_binary_path(self) -> str: """Returns the path to the InfluxDB binary.""" system_name = platform.system() if system_name == "Windows": return os.path.join(self.influxdb_dir(), "influxdb3.exe") elif system_name in ["Linux", "Darwin"]: return os.path.join(self.influxdb_dir(), "influxdb3") else: raise NotImplementedError(f"Unsupported platform: {system_name}") def influxdb_zip_path(self) -> str: """Returns the path to the InfluxDB zip archive.""" return os.path.join(self.app_data_dir(), "tools", "influxdb.zip") def influxdb_admin_token_path(self) -> str: """Returns the path to the InfluxDB admin token file.""" return os.path.join(self.influxdb_dir(), "admin_token.json") def _secrets_path(self) -> str: """Returns the path to the secrets.ini file, preferring a system-wide location if available, otherwise falling back to a user-level location. Windows: C:\\ProgramData\\xsensr\\secrets.ini (system-wide) or %APPDATA%\\xsensr\\secrets.ini (user-level) Linux/macOS: /etc/xsensr/secrets.ini (system-wide) or ~/.local/share/xsensr/secrets.ini (user-level)""" system_wide_path = "" # should be system-level, not user-level if platform.system() == "Windows": system_wide_path = os.path.join(os.getenv("ProgramData"), "xsensr", "secrets.ini") elif platform.system() in ["Linux", "Darwin"]: system_wide_path = "/etc/xsensr/secrets.ini" else: raise NotImplementedError(f"Unsupported platform: {platform.system()}") if os.path.isfile(system_wide_path): # can read and write to the system-wide secrets file, use it if os.access(system_wide_path, os.R_OK | os.W_OK): self.logger.info(f"Using system-wide secrets file at {system_wide_path}.") return system_wide_path else: self.logger.error(f"System-wide secrets file found at {system_wide_path} but is not readable/writable. Falling back to user-level secrets file.") user_level_path = os.path.join(self.app_data_dir(), "secrets.ini") if not os.path.isfile(user_level_path): if not os.path.exists(os.path.dirname(user_level_path)): os.makedirs(os.path.dirname(user_level_path)) # create an empty secrets file if it doesn't exist at either location Path(user_level_path).touch() if not os.access(user_level_path, os.R_OK | os.W_OK): raise Exception(f"User-level secrets file at {user_level_path} is not readable/writable. Please check permissions.") self.logger.info(f"Using user-level secrets file at {user_level_path}.") return user_level_path #endregion #region Installation and update helpers def install_as_service(self) -> None: raise NotImplementedError("install_as_service method must be implemented by subclasses") def restart(self) -> None: raise NotImplementedError("restart method must be implemented by subclasses") def ensure_xsensr_kernel(self) -> None: target_dir = self.repo_dir() cls = globals().get("GitOperations") if cls is None: from machine.git_operations import GitOperations as _GitOperations cls = _GitOperations git_operations = cls(pat=self.get_git_pat(), logger=self.logger) git_operations.install_or_update( product_name="xSensrKernel", repo_path=target_dir, repo_url="https://dev.azure.com/sulfilogger/Production/_git/xSensrKernel", branch_name=branch) self.install_xsensr_requirements() def install_xsensr_requirements(self) -> None: requirements_path = os.path.join(self.repo_dir(), "requirements.txt") if not os.path.isfile(requirements_path): self.logger.error(f"requirements.txt not found at {requirements_path}. Cannot install pip requirements.") return self.logger.info(f"Installing xSensrKernel pip requirements from {requirements_path}...") python_executable = sys.executable result = subprocess.run([python_executable, "-m", "pip", "install", "-r", requirements_path], capture_output=True, text=True) self.logger.debug(f"pip install output: {result.stdout}") if result.returncode != 0: raise Exception(f"Failed to install xSensrKernel pip requirements: {result.stderr}") self.logger.info(f"Successfully installed xSensrKernel pip requirements from {requirements_path}.") #endregion def get_git_pat(self) -> None: pat: str | None = None if self.secrets.has_option("azure_devops", "personal_access_token"): pat = self.secrets.get("azure_devops", "personal_access_token") if not pat or len(pat) < 20: self.logger.warning("No personal access token found for Azure DevOps in secrets file.") self.logger.warning("A personal access token with read access is needed for downloading the xSensrKernel repository.") self.logger.warning("You can find the personal access token in Bitwarden with the name 'Azure DevOps Production Read PAT'.") while not pat or len(pat) < 20: if self.check_running_as_service(): self.logger.error("Running as a service, but no valid personal access token found in secrets file. Cannot prompt for input.") raise MissingSecretException("azure_devops:personal_access_token", "A personal access token with read access is needed for downloading the xSensrKernel repository. You can find the personal access token in Bitwarden with the name 'Azure DevOps Production Read PAT'.") pat = input("Provide the personal access token from Bitwarden: ").strip() self.secrets["azure_devops"] = {"personal_access_token": pat} self.write_secrets() return pat def write_secrets(self) -> None: with open(self.secrets_path, "w") as secrets_file: self.secrets.write(secrets_file) def find_processes_by_names(self, process_names: list[str]) -> list: """Find all processes with names in the given list.""" try: self.ensure_pip_dependency(PipDependency.PSUTIL) import psutil process_names_lower = [name.lower() for name in process_names] return [proc for proc in psutil.process_iter(['name']) if proc.info['name'] and proc.info['name'].lower() in process_names_lower] except ImportError: self.logger.warning("psutil module not found. Cannot find processes by name.") return [] except Exception as e: self.logger.error(f"Error finding processes by name: {e}") return [] def get_machine(logger: Logger) -> Machine: """Returns an instance of the appropriate Machine subclass based on the current platform.""" system_name = platform.system() if system_name == "Windows": cls = globals().get("WindowsMachine") if cls is None: from machine.windows_machine import WindowsMachine as _WindowsMachine cls = _WindowsMachine machine = cls(logger) elif system_name == "Linux": cls = globals().get("LinuxMachine") if cls is None: from machine.linux_machine import LinuxMachine as _LinuxMachine cls = _LinuxMachine machine = cls(logger) else: logger.error(f"Unsupported platform: {system_name}") exit(1) return machine import subprocess import os import signal import sys import threading if __name__ != "__main__": from machine.machine import ExternalDependency, Machine, PipDependency class WindowsMachine(Machine): def _run_nssm(self, nssm_path: str, args: list[str], action: str, *, allow_nonzero: bool = False) -> subprocess.CompletedProcess: cmd = [nssm_path] + args result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0 and not allow_nonzero: raise RuntimeError( f"Failed to {action}.\n" f"Command: {' '.join(cmd)}\n" f"Return code: {result.returncode}\n" f"Stdout: {result.stdout}\n" f"Stderr: {result.stderr}" ) return result def _nssm_service_exists(self, nssm_path: str, service_name: str) -> bool: result = self._run_nssm( nssm_path, ["status", service_name], f"query status for {service_name}", allow_nonzero=True, ) return result.returncode == 0 def check_external_dependency(self, dependency: ExternalDependency) -> bool: if dependency == ExternalDependency.SERVICEMANAGER: return self._check_nssm() elif dependency == ExternalDependency.INFLUXDB: return os.path.isfile(self.influxdb_binary_path()) elif dependency == ExternalDependency.MSSQLDRIVER: return self._check_mssql_driver() else: raise NotImplementedError(f"Dependency check not implemented for {dependency}") def install_external_dependency(self, dependency: ExternalDependency) -> None: if dependency == ExternalDependency.SERVICEMANAGER: self._install_nssm() elif dependency == ExternalDependency.INFLUXDB: self._install_influxdb() elif dependency == ExternalDependency.MSSQLDRIVER: self._install_mssql_driver() else: raise NotImplementedError(f"Dependency installation not implemented for {dependency}") def check_running_as_service(self) -> bool: try: self.ensure_pip_dependency(PipDependency.PSUTIL) import psutil parent = psutil.Process(os.getppid()) return parent.name().lower() == "services.exe" except ImportError: self.logger.warning("psutil module not found. Cannot check if running as a service.") return False except Exception as e: self.logger.error(f"Error checking if running as a service: {e}") return False def install_as_service(self) -> None: if not self.has_system_level_permissions(): self.logger.warning("Cannot install as service: not running as administrator.") return self.ensure_external_dependency(ExternalDependency.SERVICEMANAGER) # kill any existing xSensrKernel, service or not, to avoid port conflicts self.logger.info("Stopping any existing xSensrKernel processes...") for proc in self.find_processes_by_names(["python.exe", "python3.exe"]): try: cmdline = " ".join(proc.cmdline()) if "main.py" in cmdline and "xSensrKernel" in cmdline: self.logger.info(f"Terminating existing xSensrKernel process (PID {proc.pid})...") proc.terminate() proc.wait(timeout=10) except Exception as e: self.logger.warning(f"Failed to terminate process (PID {proc.pid}): {e}") nssm_path = os.path.join(self.app_data_dir(), "tools", "nssm.exe") if not os.path.isfile(nssm_path): raise RuntimeError(f"Cannot install as service: nssm.exe not found at {nssm_path}. Please ensure that the Service Manager dependency is installed.") service_name = "xSensrKernel" python_executable = sys.executable app_directory = self.repo_dir() script_path = os.path.join(self.repo_dir(), "main.py") if self._nssm_service_exists(nssm_path, service_name): self.logger.info(f"Service {service_name} already exists. Updating configuration...") else: self._run_nssm( nssm_path, ["install", service_name, python_executable, script_path], f"install service {service_name}", ) self.logger.info(f"Successfully installed {service_name} as a service.") # Ensure service points to the current Python and main.py every run. self._run_nssm(nssm_path, ["set", service_name, "Application", python_executable], f"set Application for {service_name}") self._run_nssm(nssm_path, ["set", service_name, "AppParameters", script_path], f"set AppParameters for {service_name}") self._run_nssm(nssm_path, ["set", service_name, "AppDirectory", app_directory], f"set AppDirectory for {service_name}") self._run_nssm(nssm_path, ["set", service_name, "Start", "SERVICE_AUTO_START"], f"set Start mode for {service_name}") # Restart service to apply updated config; ignore stop failure if not running. self._run_nssm(nssm_path, ["stop", service_name], f"stop {service_name}", allow_nonzero=True) self._run_nssm(nssm_path, ["start", service_name], f"start {service_name}") # wait for port 5000 to be open (the port that xSensrKernel listens on) before returning import time self.logger.info("Waiting for xSensrKernel service to start and listen on port 5000...") for _ in range(30): # wait up to 30 seconds if self.check_xsensr_kernel_running(): return time.sleep(1) self.logger.error("xSensrKernel service failed to start and listen on port 5000 within 30 seconds.") raise RuntimeError("xSensrKernel service failed to start and listen on port 5000 within 30 seconds.") def restart(self) -> None: repo_dir = self.repo_dir() main_py = os.path.join(repo_dir, "main.py") if not os.path.isfile(main_py): raise RuntimeError(f"Cannot restart: main.py not found at {main_py}") pid = os.getpid() if self.check_running_as_service(): self.logger.info("Restart requested while running as service. Terminating process for service manager restart.") def _kill_self() -> None: try: os.kill(pid, signal.SIGTERM) except Exception: os._exit(0) threading.Timer(0.5, _kill_self).start() return self.logger.info("Restart requested while running as executable. Spawning replacement process.") creationflags = subprocess.CREATE_NEW_PROCESS_GROUP try: subprocess.Popen( [sys.executable, main_py], cwd=repo_dir, creationflags=creationflags, ) except Exception as e: raise RuntimeError(f"Failed to restart executable process: {e}") threading.Timer(0.5, lambda: os._exit(0)).start() #region External dependency installation helpers def _check_nssm(self) -> bool: nssm_path = os.path.join(self.app_data_dir(), "tools", "nssm.exe") return os.path.isfile(nssm_path) def _install_nssm(self) -> None: nssm_url = "https://nssm.cc/release/nssm-2.24.zip" nssm_zip_path = os.path.join(self.app_data_dir(), "tools", "nssm.zip") nssm_extract_path = os.path.join(self.app_data_dir(), "tools") self.download_file(nssm_url, nssm_zip_path) # extract the zip file using a built in python library to avoid adding a dependency on something like zipfile import zipfile with zipfile.ZipFile(nssm_zip_path, 'r') as zip_ref: zip_ref.extractall(nssm_extract_path) # delete the zip file after extraction os.remove(nssm_zip_path) def _install_influxdb(self) -> None: zip_url = "https://dl.influxdata.com/influxdb/releases/influxdb3-core-3.10.0-windows_amd64.zip" zip_path = self.influxdb_zip_path() extract_path = self.influxdb_dir() self.download_file(zip_url, zip_path) # extract the zip file using a built in python library to avoid adding a dependency on something like zipfile import zipfile with zipfile.ZipFile(zip_path, 'r') as zip_ref: zip_ref.extractall(extract_path) # delete the zip file after extraction os.remove(zip_path) self.initialize_influxdb() def _check_mssql_driver(self) -> bool: ODBC_DRIVER_NAME = "ODBC Driver 17 for SQL Server" powershell_cmd = [ "powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", ( f"$d = Get-OdbcDriver -ErrorAction SilentlyContinue " f"| Where-Object {{ $_.Name -eq '{ODBC_DRIVER_NAME}' }}; " f"if ($d) {{ exit 0 }} else {{ exit 1 }}" ), ] result = subprocess.run( powershell_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) return result.returncode == 0 def _install_mssql_driver(self) -> None: # Microsoft fwlink for ODBC Driver 17 MSI MSI_URL = "https://go.microsoft.com/fwlink/?linkid=2361646" MSI_PATH = os.path.join(self.app_data_dir(), "tools", "msodbcsql17.msi") import urllib.request with urllib.request.urlopen(MSI_URL) as response: with open(MSI_PATH, "wb") as file: file.write(response.read()) self.logger.warning(f"Downloaded MSSQL ODBC Driver installer to {MSI_PATH}.") self.logger.warning(f"You must manually run the installer and follow the prompts\nto complete installation of the MSSQL ODBC Driver.") self.logger.warning(f"After installation, press Enter to continue...") input() os.remove(MSI_PATH) #endregion import os import platform import shutil import signal import stat import subprocess import sys import tarfile import tempfile import threading import time if __name__ != "__main__": from machine.machine import ExternalDependency, Machine class LinuxMachine(Machine): def check_external_dependency(self, dependency: ExternalDependency) -> bool: if dependency == ExternalDependency.SERVICEMANAGER: return self._check_systemd() elif dependency == ExternalDependency.INFLUXDB: return os.path.isfile(self.influxdb_binary_path()) elif dependency == ExternalDependency.MSSQLDRIVER: return self._check_mssql_driver() else: raise NotImplementedError(f"Dependency check not implemented for {dependency}") def install_external_dependency(self, dependency: ExternalDependency) -> None: if dependency == ExternalDependency.SERVICEMANAGER: # systemd is expected to be provided by the host OS. raise RuntimeError( "Linux service manager dependency requires systemd. " "systemd was not detected on this host." ) elif dependency == ExternalDependency.INFLUXDB: self._install_influxdb() elif dependency == ExternalDependency.MSSQLDRIVER: self._install_mssql_driver() else: raise NotImplementedError(f"Dependency installation not implemented for {dependency}") def check_running_as_service(self) -> bool: if not self._check_systemd(): return False # systemd injects INVOCATION_ID for service processes. return bool((os.getenv("INVOCATION_ID") or "").strip()) def install_as_service(self) -> None: if not self.has_system_level_permissions(): self.logger.warning("Cannot install as service: not running as root.") self.logger.warning("Run this command with sudo to install xSensrKernel as a service:") self.logger.warning(f"\tsudo {sys.executable} {self.bootstrap_script_path()} --install-service") return self.ensure_external_dependency(ExternalDependency.SERVICEMANAGER) # kill any existing xSensrKernel, service or not, to avoid port conflicts self.logger.info("Stopping any existing xSensrKernel processes...") for proc in self.find_processes_by_names(["python3", "python"]): try: cmdline = " ".join(proc.cmdline()) if "main.py" in cmdline and "xSensrKernel" in cmdline: self.logger.info(f"Terminating existing xSensrKernel process (PID {proc.pid})...") proc.terminate() proc.wait(timeout=10) except Exception as e: self.logger.warning(f"Failed to terminate process {proc.pid}: {e}") repo_dir = self.repo_dir() main_py = os.path.join(repo_dir, "main.py") if not os.path.isfile(main_py): raise RuntimeError(f"Cannot install service: main.py not found at {main_py}") service_name = "xsensrkernel.service" service_path = os.path.join("/etc/systemd/system", service_name) # If invoked via sudo, prefer running the service as the invoking user. run_user = (os.getenv("SUDO_USER") or os.getenv("USER") or "").strip() user_line = f"User={run_user}\n" if run_user else "" unit_content = ( "[Unit]\n" "Description=xSensrKernel\n" "After=network.target\n" "\n" "[Service]\n" "Type=simple\n" f"WorkingDirectory={repo_dir}\n" f"ExecStart={sys.executable} {main_py}\n" f"{user_line}" "Restart=always\n" "RestartSec=2\n" "\n" "[Install]\n" "WantedBy=multi-user.target\n" ) with open(service_path, "w", encoding="utf-8") as f: f.write(unit_content) self._run_checked(["systemctl", "daemon-reload"], "reload systemd daemon") self._run_checked(["systemctl", "enable", "--now", service_name], f"enable/start {service_name}") self.logger.info("Waiting for xSensrKernel service to start and listen on port 5000...") for _ in range(30): if self.check_xsensr_kernel_running(): return time.sleep(1) raise RuntimeError("xSensrKernel service failed to start and listen on port 5000 within 30 seconds.") def restart(self) -> None: repo_dir = self.repo_dir() main_py = os.path.join(repo_dir, "main.py") if not os.path.isfile(main_py): raise RuntimeError(f"Cannot restart: main.py not found at {main_py}") pid = os.getpid() if self.check_running_as_service(): self.logger.info("Restart requested while running as service. Terminating process for service manager restart.") def _kill_self() -> None: try: os.kill(pid, signal.SIGTERM) except Exception: os._exit(0) threading.Timer(0.5, _kill_self).start() return self.logger.info("Restart requested while running as executable. Spawning replacement process.") try: subprocess.Popen( [sys.executable, main_py], cwd=repo_dir, start_new_session=True, ) except Exception as e: raise RuntimeError(f"Failed to restart executable process: {e}") threading.Timer(0.5, lambda: os._exit(0)).start() # region helpers def _check_systemd(self) -> bool: systemctl = shutil.which("systemctl") if not systemctl: return False # /run/systemd/system is the standard marker that PID 1 is systemd. if not os.path.isdir("/run/systemd/system"): return False return True def _run_checked(self, cmd: list[str], action: str, env: dict[str, str] | None = None) -> None: local_env = os.environ.copy() if env is not None: local_env.update(env) result = subprocess.run(cmd, capture_output=True, text=True, env=local_env) if result.returncode != 0: raise RuntimeError( f"Failed to {action}.\n" f"Command: {' '.join(cmd)}\n" f"Return code: {result.returncode}\n" f"Stdout: {result.stdout}\n" f"Stderr: {result.stderr}" ) def _install_influxdb(self) -> None: arch = platform.machine().lower() arch_map = { "x86_64": "amd64", "amd64": "amd64", "aarch64": "arm64", "arm64": "arm64", } mapped_arch = arch_map.get(arch) if not mapped_arch: raise RuntimeError(f"Unsupported Linux architecture for InfluxDB install: {arch}") url = f"https://dl.influxdata.com/influxdb/releases/influxdb3-core-3.10.0_linux_{mapped_arch}.tar.gz" archive_path = os.path.join(self.app_data_dir(), "tools", "influxdb.tar.gz") self.download_file(url, archive_path) extract_dir = tempfile.mkdtemp(prefix="xsensr_influxdb_") try: with tarfile.open(archive_path, "r:gz") as tar: tar.extractall(extract_dir) src_binary = None for root, _dirs, files in os.walk(extract_dir): if "influxdb3" in files: src_binary = os.path.join(root, "influxdb3") break if not src_binary or not os.path.isfile(src_binary): raise RuntimeError("InfluxDB archive did not contain an influxdb3 binary") os.makedirs(self.influxdb_dir(), exist_ok=True) dst_binary = self.influxdb_binary_path() shutil.copy2(src_binary, dst_binary) os.chmod(dst_binary, os.stat(dst_binary).st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) finally: try: if os.path.isfile(archive_path): os.remove(archive_path) except Exception: pass shutil.rmtree(extract_dir, ignore_errors=True) self.initialize_influxdb() def _check_mssql_driver(self) -> bool: cmd = ["odbcinst", "-q", "-d"] try: result = subprocess.run(cmd, capture_output=True, text=True, check=False) except FileNotFoundError: return False if result.returncode != 0: return False stdout = result.stdout or "" names = ( "ODBC Driver 17 for SQL Server", "ODBC Driver 18 for SQL Server", ) return any(name in stdout for name in names) def _install_mssql_driver(self) -> None: if not self.has_system_level_permissions(): raise RuntimeError( "Installing MSSQL ODBC driver repository requires root permissions. " f"Run with sudo:\n\tsudo {sys.executable} {sys.argv[0]} --install-mssql\n\n" ) for cmd in ("apt-get", "dpkg", "gpg"): if not shutil.which(cmd): raise RuntimeError(f"Required command '{cmd}' not found. Cannot configure Microsoft ODBC repository.") self.logger.info("[xSensrKernel] Installing Microsoft ODBC 17...") keyring_path = "/usr/share/keyrings/microsoft-prod.gpg" if not os.path.isfile(keyring_path): with tempfile.NamedTemporaryFile(prefix="microsoft_", suffix=".asc", delete=False) as tmp: tmp_asc = tmp.name try: self.download_file("https://packages.microsoft.com/keys/microsoft.asc", tmp_asc) self._run_checked( ["gpg", "--dearmor", "-o", keyring_path, tmp_asc], "install Microsoft apt signing key", ) finally: try: os.remove(tmp_asc) except Exception: pass # Remove legacy list file that can conflict with apt source parsing. legacy_repo_path = "/etc/apt/sources.list.d/mssql-release.list" try: if os.path.isfile(legacy_repo_path): os.remove(legacy_repo_path) except Exception as e: raise RuntimeError(f"Failed removing legacy repo file {legacy_repo_path}: {e}") repo_path = "/etc/apt/sources.list.d/microsoft-prod.list" repo_line = ( "deb [arch=amd64,arm64,armhf signed-by=/usr/share/keyrings/microsoft-prod.gpg] " "https://packages.microsoft.com/debian/12/prod bookworm main\n" ) with open(repo_path, "w", encoding="utf-8") as f: f.write(repo_line) self._run_checked(["apt-get", "update", "-y"], "update apt package lists") dpkg_res = subprocess.run(["dpkg", "-s", "msodbcsql17"], capture_output=True, text=True, check=False) if dpkg_res.returncode == 0: self.logger.info("[xSensrKernel] ODBC 17 already installed") else: self.logger.info("[xSensrKernel] ODBC 17 repository configured") self.logger.info("[xSensrKernel] Installing ODBC 17...") self._run_checked(["apt-get", "install", "-y", "msodbcsql17"], "install Microsoft ODBC 17", env={"ACCEPT_EULA": "Y"}) self.logger.info("[xSensrKernel] ODBC 17 installed successfully") # endregion import logging import os import platform import sys import time BOOTSTRAP_VERSION = "1" # Ensure repository root is importable when running bootstrap/bootstrap.py directly. REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if REPO_ROOT not in sys.path: sys.path.insert(0, REPO_ROOT) try: from machine.machine import ExternalDependency, Machine, PipDependency, get_machine except: pass logging.basicConfig( level=logging.INFO, format="%(name)s %(levelname)s: %(message)s", ) logger = logging.getLogger("[bootstrap python]") if __name__ == "__main__": logger.info(f"Python version: {sys.version}") if sys.version_info < (3, 11): logger.error("Python 3.11 or higher is required. 3.12 is recommended.") exit(1) machine = get_machine(logger) machine.ensure_pip() machine.ensure_virtualenv() # Check for command line arguments # These are primarily for system-level tasks that require admin privileges if len(sys.argv) > 1: arg = sys.argv[1] if arg == "--install-service": logger.info("Install service flag detected. Will install xSensrKernel as a service.") machine.ensure_external_dependency(ExternalDependency.SERVICEMANAGER) machine.install_as_service() exit(0) elif arg == "--initialize-influxdb": logger.info("Initialize InfluxDB flag detected. Will initialize InfluxDB.") machine.install_external_dependency(ExternalDependency.INFLUXDB) exit(0) elif arg == "--install-mssql": logger.info("Install MSSQL driver flag detected. Will install MSSQL ODBC driver.") machine.install_external_dependency(ExternalDependency.MSSQLDRIVER) exit(0) else: logger.error(f"Unknown command line argument: {arg}") exit(1) machine.ensure_pip_dependency(PipDependency.PYGIT2) machine.ensure_external_dependency(ExternalDependency.MSSQLDRIVER) machine.ensure_external_dependency(ExternalDependency.INFLUXDB) machine.ensure_xsensr_kernel() logger.info("Bootstrap complete.") if machine.check_xsensr_kernel_running(): logger.info("xSensrKernel is already running, or port 5000 is in use. Exiting bootstrap process.") exit(0) # Start main.py using the current Python executable so we keep the same environment. main_script = os.path.join(machine.repo_dir(), "main.py") if platform.system() == "Windows": logger.info("Windows detected: bootstrap will not launch xSensrKernel automatically.") logger.info(f"Start xSensrKernel manually with:\n\tcd \"{machine.repo_dir()}\"\n\t\"{sys.executable}\" \"{main_script}\"") logger.info(f"Or, to install xSensrKernel as a service that autoruns on boot, run the following with an elevated Powershell terminal (Run as Administrator):\n\t\"{sys.executable}\" \"{machine.bootstrap_script_path()}\" --install-service") exit(0) logger.info("Running xSensrKernel in 5 seconds...") time.sleep(1) logger.info("Running xSensrKernel in 4 seconds...") time.sleep(1) logger.info("Running xSensrKernel in 3 seconds...") time.sleep(1) logger.info("Running xSensrKernel in 2 seconds...") time.sleep(1) logger.info("Running xSensrKernel in 1 second...") time.sleep(1) logger.info("Starting xSensrKernel in the current terminal (foreground mode)...") os.chdir(machine.repo_dir()) os.execv(sys.executable, [sys.executable, main_script])