18 from typing
import List, Any, Tuple, Optional, Callable
19 from pathlib
import Path
20 from functools
import wraps
21 from importlib.metadata
import version
as getLibraryVersion
29 from py3nvml
import py3nvml
35 from ...configuration
import DEFAULT_VENV_PATH
36 from ...utils.process
import command
39 def formatCliVersion(version: Tuple[int, int, int]) -> str:
40 return ".".join(map(str, version))
43 def fetchCtxSource() -> Optional[str]:
44 _, output, _ = command([sys.executable,
"-m",
"pip",
"freeze"], ignoreStdout =
True, ignoreStderr =
True)
45 packages = output.splitlines()
47 for package
in packages:
48 if "coretex" in package:
49 return package.replace(
" ",
"")
54 def createEnvironment(venvPython: Path) ->
None:
55 if DEFAULT_VENV_PATH.exists():
56 shutil.rmtree(DEFAULT_VENV_PATH)
58 venv.create(DEFAULT_VENV_PATH, with_pip =
True)
60 if platform.system() ==
"Windows":
61 venvPython = DEFAULT_VENV_PATH /
"Scripts" /
"python.exe"
63 ctxSource = fetchCtxSource()
64 if ctxSource
is not None:
65 command([str(venvPython),
"-m",
"pip",
"install", ctxSource], ignoreStdout =
True, ignoreStderr =
True)
68 def checkEnvironment() -> None:
69 venvPython = DEFAULT_VENV_PATH /
"bin" /
"python"
70 venvActivate = DEFAULT_VENV_PATH /
"bin" /
"activate"
71 venvCoretex = DEFAULT_VENV_PATH /
"bin" /
"coretex"
73 if not venvActivate.exists()
or not venvPython.exists():
74 createEnvironment(venvPython)
78 command([str(venvCoretex),
"version"], check =
True, ignoreStderr =
True, ignoreStdout =
True)
80 createEnvironment(venvPython)
84 def updateLib() -> None:
85 command([sys.executable,
"-m",
"pip",
"install",
"--no-cache-dir",
"--upgrade",
"coretex"], ignoreStdout =
True, ignoreStderr =
True)
88 def parseLibraryVersion(version: str) -> Optional[Tuple[int, int, int]]:
89 parts = version.split(
".")
94 if all(part.isdigit()
for part
in parts):
95 major, minor, patch = map(int, version.split(
'.'))
96 return major, minor, patch
101 def fetchCurrentVersion() -> Optional[Tuple[int, int, int]]:
102 version = parseLibraryVersion(getLibraryVersion(
"coretex"))
104 logging.getLogger(
"cli").debug(f
"Couldn't parse current version from string: {version}")
110 def fetchLatestVersion() -> Optional[Tuple[int, int, int]]:
111 url =
"https://pypi.org/pypi/coretex/json"
112 response = requests.get(url)
115 logging.getLogger(
"cli").debug(f
"Failed to fetch version of coretex library. Response code: {response.status_code}")
118 data = response.json()
119 infoDict = data.get(
"info")
120 if not isinstance(infoDict, dict):
121 logging.getLogger(
"cli").debug(
"Value of json field of key \"info\" in \"data\" dictionary is not of expected type (dict).")
124 version = infoDict.get(
"version")
125 if not isinstance(version, str):
126 logging.getLogger(
"cli").debug(
"Value of json field of key \"version\" in \"info\" dictionary is not of expected type (str).")
129 parsedVersion = parseLibraryVersion(version)
130 if parsedVersion
is None:
131 logging.getLogger(
"cli").debug(f
"Couldn't parse latest version from string: {version}")
137 def checkLibVersion() -> None:
138 currentVersion = fetchCurrentVersion()
139 latestVersion = fetchLatestVersion()
141 if currentVersion
is None or latestVersion
is None:
144 if latestVersion > currentVersion:
146 f
"Newer version of Coretex library is available. "
147 f
"Current: {formatCliVersion(currentVersion)}, Latest: {formatCliVersion(latestVersion)}."
149 ui.stdEcho(
"Use \"coretex update\" command to update library to latest version.")
152 def getExecPath(executable: str) -> str:
153 _, path, _ = command([
"which", executable], ignoreStdout =
True, ignoreStderr =
True)
154 pathParts = path.strip().split(
'/')
155 execPath =
'/'.join(pathParts[:-1])
160 def isGPUAvailable() -> bool:
163 py3nvml.nvmlShutdown()
169 def onBeforeCommandExecute(
170 fun: Callable[..., Any],
171 excludeOptions: Optional[List[str]] =
None,
172 excludeSubcommands: Optional[List[str]] =
None
175 if excludeOptions
is None:
178 if excludeSubcommands
is None:
179 excludeSubcommands = []
181 def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
183 def wrapper(*args: Any, **kwargs: Any) -> Any:
184 if click.get_current_context().invoked_subcommand
in excludeSubcommands:
185 return f(*args, **kwargs)
187 for key, value
in click.get_current_context().params.items():
188 if key
in excludeOptions
and value:
189 return f(*args, **kwargs)
192 return f(*args, **kwargs)