Skip to content
This repository was archived by the owner on May 3, 2026. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 11 additions & 53 deletions examples/EmergencyManagement/Server/server_emergency.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,65 +8,23 @@
from pathlib import Path
from typing import Any, Dict, Optional, Sequence

if __package__ in (None, ""):
project_root = Path(__file__).resolve().parents[3]
project_root_str = str(project_root)
if project_root_str not in sys.path:
sys.path.insert(0, project_root_str)

def _ensure_standard_library_on_path() -> None:
"""Ensure CPython standard library directories are available.

Some execution environments replace ``sys.path`` with only the script
directory before running this module. That removes the standard library
entries the interpreter normally injects, which breaks imports performed by
the service during initialization. This helper reconstructs a minimal set of
default directories based on the active interpreter configuration so that
modules such as ``pkgutil`` remain importable.

Returns:
None: The function mutates ``sys.path`` in place.
"""

version = f"python{sys.version_info.major}.{sys.version_info.minor}"
zipped = f"python{sys.version_info.major}{sys.version_info.minor}.zip"
base_dirs = {sys.base_prefix, sys.exec_prefix, sys.prefix}
lib_dir_names = ["lib", "Lib"]

for base_dir in base_dirs:
if not base_dir:
continue

for lib_dir_name in lib_dir_names:
lib_dir = f"{base_dir}/{lib_dir_name}"
candidates = [
f"{lib_dir}/{zipped}",
f"{lib_dir}/{version}",
f"{lib_dir}/{version}/lib-dynload",
f"{lib_dir}/{version}/site-packages",
f"{lib_dir}/{version}/dist-packages",
f"{lib_dir}/site-packages",
f"{lib_dir}/dist-packages",
]

for candidate in candidates:
if candidate and candidate not in sys.path:
sys.path.append(candidate)


def _ensure_project_root_on_path() -> None:
"""Ensure the repository root is importable when run as a script."""

# Reason: Allow running the server example from its directory by ensuring
# project-level imports resolve when executed as a script.
package_name = __package__ or ""
if not package_name or "." not in package_name:
project_root = Path(__file__).resolve().parents[3]
project_root_str = str(project_root)
if project_root_str not in sys.path:
sys.path.insert(0, project_root_str)
from examples.EmergencyManagement.utils.bootstrap import (
ensure_project_root,
ensure_standard_library,
)


def _configure_environment() -> None:
"""Prepare import paths required for the example service."""

_ensure_standard_library_on_path()
_ensure_project_root_on_path()
ensure_standard_library()
ensure_project_root(package_name=__package__, file_path=__file__)


EmergencyService = object()
Expand Down
54 changes: 13 additions & 41 deletions examples/EmergencyManagement/client/client_emergency.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,44 +5,16 @@
from pathlib import Path
from typing import Optional


def _ensure_standard_library_on_path() -> None:
"""Ensure CPython standard library directories are available."""

version = f"python{sys.version_info.major}.{sys.version_info.minor}"
zipped = f"python{sys.version_info.major}{sys.version_info.minor}.zip"
base_dirs = {sys.base_prefix, sys.exec_prefix, sys.prefix}
lib_dir_names = ["lib", "Lib"]

for base_dir in base_dirs:
if not base_dir:
continue

for lib_dir_name in lib_dir_names:
lib_dir = f"{base_dir}/{lib_dir_name}"
candidates = [
f"{lib_dir}/{zipped}",
f"{lib_dir}/{version}",
f"{lib_dir}/{version}/lib-dynload",
f"{lib_dir}/{version}/site-packages",
f"{lib_dir}/{version}/dist-packages",
f"{lib_dir}/site-packages",
f"{lib_dir}/dist-packages",
]

for candidate in candidates:
if candidate and candidate not in sys.path:
sys.path.append(candidate)


def _ensure_project_root_on_path() -> None:
"""Allow running the example as a script from the client directory."""

if __package__ in (None, ""):
project_root = Path(__file__).resolve().parents[3]
project_root_str = str(project_root)
if project_root_str not in sys.path:
sys.path.insert(0, project_root_str)
if __package__ in (None, ""):
project_root = Path(__file__).resolve().parents[3]
project_root_str = str(project_root)
if project_root_str not in sys.path:
sys.path.insert(0, project_root_str)

from examples.EmergencyManagement.utils.bootstrap import (
ensure_project_root,
ensure_standard_library,
)


CONFIG_FILENAME = "client_config.json"
Expand Down Expand Up @@ -72,8 +44,8 @@ def _ensure_project_root_on_path() -> None:
CONFIG_PATH = Path(__file__).with_name(CONFIG_FILENAME)


_ensure_standard_library_on_path()
_ensure_project_root_on_path()
ensure_standard_library()
ensure_project_root(package_name=__package__, file_path=__file__)


try:
Expand Down Expand Up @@ -226,7 +198,7 @@ async def main():
before printing.
"""

_ensure_project_root_on_path()
ensure_project_root(package_name=__package__, file_path=__file__)

from examples.EmergencyManagement.client.client import (
LXMFClient,
Expand Down
77 changes: 77 additions & 0 deletions examples/EmergencyManagement/utils/bootstrap.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
"""Bootstrap utilities shared by Emergency Management examples."""

from __future__ import annotations

import inspect
import sys
from pathlib import Path
from types import FrameType
from typing import Optional


def _caller_context() -> tuple[Optional[str], Optional[str]]:
"""Return the ``__package__`` and ``__file__`` for the calling module."""

frame: Optional[FrameType]
frame = inspect.currentframe()
package_name: Optional[str] = None
file_path: Optional[str] = None
try:
caller = frame.f_back if frame is not None else None
if caller is not None:
package_name = caller.f_globals.get("__package__")
file_path = caller.f_globals.get("__file__")
finally:
del frame
return package_name, file_path


def ensure_standard_library() -> None:
"""Ensure CPython standard library directories are available."""

version = f"python{sys.version_info.major}.{sys.version_info.minor}"
zipped = f"python{sys.version_info.major}{sys.version_info.minor}.zip"
base_dirs = {sys.base_prefix, sys.exec_prefix, sys.prefix}
lib_dir_names = ["lib", "Lib"]

for base_dir in base_dirs:
if not base_dir:
continue

for lib_dir_name in lib_dir_names:
lib_dir = f"{base_dir}/{lib_dir_name}"
candidates = [
f"{lib_dir}/{zipped}",
f"{lib_dir}/{version}",
f"{lib_dir}/{version}/lib-dynload",
f"{lib_dir}/{version}/site-packages",
f"{lib_dir}/{version}/dist-packages",
f"{lib_dir}/site-packages",
f"{lib_dir}/dist-packages",
]

for candidate in candidates:
if candidate and candidate not in sys.path:
sys.path.append(candidate)


def ensure_project_root(package_name: Optional[str] = None, file_path: Optional[str] = None) -> None:
"""Ensure the repository root is importable when run as a script."""

if package_name is None and file_path is None:
detected_package, detected_file = _caller_context()
package_name = detected_package if package_name is None else package_name
file_path = detected_file if file_path is None else file_path

package_value = package_name or ""
if "." in package_value:
return

target_file = file_path or __file__
project_root = Path(target_file).resolve().parents[3]
project_root_str = str(project_root)
if project_root_str not in sys.path:
sys.path.insert(0, project_root_str)


__all__ = ["ensure_standard_library", "ensure_project_root"]