import functools
import hashlib
import warnings
import json
from logging import getLogger
import os
import random
import re
import time
import uuid
from pathlib import Path
from typing import Annotated, Union, Literal
import pandas as pd
from pydantic import (
BaseModel,
Field,
PrivateAttr,
ValidationError,
field_serializer,
field_validator,
Discriminator,
ConfigDict,
)
from pydantic_settings import (
BaseSettings,
PydanticBaseSettingsSource,
SettingsConfigDict,
TomlConfigSettingsSource,
)
try: # pydantic-settings >= 2.4 exports it at the top level
from pydantic_settings import SettingsError
except ImportError: # older versions keep it in .sources
from pydantic_settings.sources import SettingsError
import requests
from platformdirs import PlatformDirs
import toml
from ._errors import (
PywaspError,
LicenseError,
LicenseConfigError,
LicenseServerUnreachableError,
PointsLimitError,
)
from ._warnings import _find_user_stacklevel
logger = getLogger(__name__)
# Constants
APP_NAME = "pywasp"
APP_AUTHOR = "DTU Wind Energy"
DEFAULT_PORT = "34523"
DEFAULT_DTU_HOST = "130.226.48.178"
ACCOUNT = "0244d026-758b-442b-9668-811ce125a7e3"
PRODUCT_ID = "f0c12bfb-c1f5-4fc2-bfb1-bd20fbdae727"
MAX_PTS = {
"b32d02b9-3d7e-43a4-80a3-4b8aa13083c6": 25_000, # Tier 1
"87fcb9e4-09df-4673-bb32-e2748c49a183": 125_000, # Tier 2
"819d348f-b49e-4ddd-aeac-c1074de9c971": 625_000, # Tier 3
}
# Must use roaming=True to be consistent with rusts app_dirs2 package.
app_dir = PlatformDirs(appname=APP_NAME, appauthor=APP_AUTHOR, roaming=True)
_CONFIG_FILE = app_dir.user_config_path / "pywasp_config.toml"
_POINTS_LIMIT_CACHE_FILE = app_dir.user_cache_path / "points_limit_cache.json"
_POINTS_LIMIT_CACHE_TTL = 24 * 60 * 60 # seconds
class DownloadCfg(BaseModel):
download_prompt: bool = True
download_global_nc_files: bool = True
class _UnknownPolicyError(PywaspError):
"""The license server returned a policy id with no known point limit.
Internal: callers treat it like an unreachable server (warn once and
skip the points check for the session) so that a valid license on a
tier newer than this pywasp version keeps working.
"""
def _retry_after_detail(response) -> str:
"""Human-readable wait hint from a rate-limited response.
Keygen includes a ``Retry-After`` header (seconds) on HTTP 429
responses; fall back to generic advice when it is absent or unparseable.
"""
try:
seconds = int(response.headers.get("Retry-After"))
return f"retry after {seconds} seconds"
except (AttributeError, TypeError, ValueError):
return "try again later"
def _check_server_status(test_url):
"""Check whether a license server is reachable.
This is a pure reachability probe against an unauthenticated endpoint;
it says nothing about credentials. A 429 counts as reachable: it proves
a working, rate-limiting license server. Only 2xx responses (and 429)
count as "ok" — anything else, including a redirect (requests are made
with ``allow_redirects=False``, so a 3xx usually means a captive portal
or a wrong host), means the configured address does not behave like a
license server.
Parameters
----------
test_url : str
The URL to test connectivity against.
Returns
-------
tuple of (str, str)
Status ("ok" or "unreachable") and a human-readable detail message.
Raises
------
LicenseError
If the server certificate cannot be verified (code 123).
PywaspError
On other SSL problems.
"""
logger.debug(f"Attempting to connect to: {test_url}")
try:
response = requests.get(test_url, timeout=10, allow_redirects=False)
except requests.exceptions.SSLError as e:
if "SSL: CERTIFICATE_VERIFY_FAILED" in str(e):
raise LicenseError(123)
raise PywaspError(
f"Problem with ssl verification, try adding a root_ca_filepath to your Licensing configuration: {e}"
)
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
return (
"unreachable",
f"Failed to connect to license server at {test_url}. "
"You may be offline, or the host/port in your configuration may "
f"be incorrect (`pywasp configure` updates them). Error: {e}",
)
except requests.exceptions.RequestException as e:
return (
"unreachable",
f"An error occurred while checking license server at {test_url}. Error: {e}",
)
if response.status_code == 429:
return (
"ok",
f"The license server at {test_url} is reachable but rate limiting "
f"requests (HTTP 429); {_retry_after_detail(response)}.",
)
if 200 <= response.status_code < 300:
return (
"ok",
f"Successfully connected to {test_url} (Status: {response.status_code})",
)
return (
"unreachable",
f"The server at {test_url} returned an unexpected response "
f"(HTTP {response.status_code}); it does not appear to be a "
"working license server.",
)
def _points_cache_key(license_id: str) -> str:
"""Cache key for a license id: hashed so the id (a secret) is not stored."""
return hashlib.sha256(license_id.encode()).hexdigest()[:16]
def _read_points_limit_cache(license_id: str) -> int | None:
"""Return the cached points limit for ``license_id``, or None on miss.
A missing, corrupt, or expired cache entry is a miss, never an error.
"""
try:
data = json.loads(_POINTS_LIMIT_CACHE_FILE.read_text())
entry = data[_points_cache_key(license_id)]
# A future fetched_at (clock skew, tampering) is invalid, not fresh
if 0 <= time.time() - entry["fetched_at"] < _POINTS_LIMIT_CACHE_TTL:
return int(entry["points_limit"])
except (OSError, ValueError, KeyError, TypeError):
pass
return None
def _write_points_limit_cache(license_id: str, points_limit: int) -> None:
"""Store the points limit for ``license_id``; failures only log."""
try:
try:
data = json.loads(_POINTS_LIMIT_CACHE_FILE.read_text())
if not isinstance(data, dict):
data = {}
except (OSError, ValueError):
data = {}
data[_points_cache_key(license_id)] = {
"points_limit": points_limit,
"fetched_at": time.time(),
}
_POINTS_LIMIT_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
# Atomic write: parallel workers starting together (the scenario the
# cache exists for) must not tear the file — a torn cache reads as a
# miss and would silently re-create the per-run server traffic
tmp_file = _POINTS_LIMIT_CACHE_FILE.with_name(
f"{_POINTS_LIMIT_CACHE_FILE.name}.tmp-{os.getpid()}-{uuid.uuid4().hex[:8]}"
)
try:
tmp_file.write_text(json.dumps(data))
os.replace(tmp_file, _POINTS_LIMIT_CACHE_FILE)
except OSError:
tmp_file.unlink(missing_ok=True)
raise
except OSError:
logger.debug("Could not write the points-limit cache", exc_info=True)
class LicensingLocalCfg(BaseModel):
license_type: Literal["local"] = "local"
host: str
port: int = Field(34523, ge=0, le=65535)
@property
def _url(self) -> str:
return f"http://{self.host}:{self.port}"
def check_server_status(self):
"""Check whether the local license server is reachable.
Returns
-------
tuple of (str, str)
Status ("ok" or "unreachable") and a detail message.
"""
return _check_server_status(self._url)
class LicensingRemoteCfg(BaseModel):
license_type: Literal["keygen_cloud"] = "keygen_cloud"
redirect_host: str | None = None
license_id: uuid.UUID
activation_token: str
root_ca_filepath: Path | None = None
# Use computed_field instead of private attribute
model_config = ConfigDict(arbitrary_types_allowed=True)
@property
def _url(self) -> str:
"""Construct the base URL for the Keygen API. Handles redirection if specified."""
if self.redirect_host is None:
host = "api.keygen.sh"
else:
host = self.redirect_host
return f"https://{host}"
@property
def _verify(self) -> bool | str:
"""TLS verification argument for requests: the configured CA bundle or True."""
return True if self.root_ca_filepath is None else str(self.root_ca_filepath)
def _redact(self, text: str | None) -> str:
"""Remove the license id and activation token (secrets) from message text.
Server-derived text and low-level request errors can echo request
headers (e.g. ``requests.InvalidHeader`` embeds the full
``Authorization`` value), so both secrets are stripped.
"""
if text is None:
return "unknown"
# Case-insensitive: str(uuid.UUID) is lowercase, but a server or
# library may echo the id uppercased. The token stays exact-match
# (tokens are case-sensitive).
text = re.sub(
re.escape(str(self.license_id)),
"<license-id>",
text,
flags=re.IGNORECASE,
)
if self.activation_token:
text = text.replace(self.activation_token, "<activation-token>")
return text
@field_serializer("license_id")
def serialize_license_id(self, license_id: uuid.UUID) -> str:
"""Serialize the UUID license ID to a string for API requests."""
return str(license_id)
@field_validator("root_ca_filepath")
@classmethod
def validate_root_ca_filepath(cls, root_ca_filepath: Path | None) -> Path | None:
"""The configured root CA file must exist (``~`` is expanded)."""
if root_ca_filepath is None:
return None
root_ca_filepath = root_ca_filepath.expanduser()
if not root_ca_filepath.exists():
raise ValueError(
f"Root CA file does not exist at specified path: {root_ca_filepath}"
)
return root_ca_filepath
@field_serializer("root_ca_filepath")
def serialize_root_ca_filepath(self, root_ca_filepath: Path | None) -> str | None:
"""Serialize the root CA file path to a string for API requests."""
if root_ca_filepath is None:
return None
return str(root_ca_filepath)
def _request_validation(self) -> dict:
"""Send a request to the license server to validate the license. Returns the response as a dictionary.
Including information about the license status, such as whether it is active, the number of runs used,
the runs limit, and the expiry date.
Only transient failures (network errors, malformed responses, and 5xx
responses) are retried, with jittered exponential backoff. A rejection
(4xx response or an invalid license) is raised immediately: retrying
it cannot succeed and repeated attempts can get the client
rate-limited by the license server. A 429 (rate limited) is also not
retried, and is reported as the server being unavailable rather than
as a credentials problem.
Raises
------
LicenseConfigError
If the server rejects the credentials, the license is not valid,
or the request cannot be built (e.g. a malformed `redirect_host`).
LicenseServerUnreachableError
If the server cannot be reached after all retries, or is rate
limiting requests.
LicenseError
If the server certificate cannot be verified (code 123).
PywaspError
On other SSL problems.
"""
uri_validate = f"{self._url}/v1/accounts/{ACCOUNT}/licenses/{self.license_id}/actions/validate"
headers_validate = {
"Content-Type": "application/vnd.api+json",
"Accept": "application/vnd.api+json",
"Authorization": f"Bearer {self.activation_token}",
}
data_validate = json.dumps({"meta": {"scope": {"product": f"{PRODUCT_ID}"}}})
sleep_s = 0.5
last_error = None
for attempt in range(5):
if attempt > 0:
# Jittered exponential backoff so simultaneously started
# workers do not retry in synchronized waves
time.sleep(sleep_s * random.uniform(0.75, 1.25))
sleep_s *= 2
try:
res = requests.post(
uri_validate,
headers=headers_validate,
data=data_validate,
timeout=10,
verify=self._verify,
)
except requests.exceptions.SSLError as e:
if "SSL: CERTIFICATE_VERIFY_FAILED" in str(e):
raise LicenseError(123)
raise PywaspError(f"Problem with connection: {self._redact(str(e))}")
except (
requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
requests.exceptions.ChunkedEncodingError,
requests.exceptions.ContentDecodingError,
) as e:
last_error = str(e)
continue
except requests.exceptions.RequestException as e:
# Not transient: the request itself cannot be built/sent
# (e.g. a malformed redirect_host giving InvalidURL)
raise LicenseConfigError(
f"Could not send a valid request to the license server: "
f"{self._redact(str(e))}. Check the `redirect_host` in "
"your configuration."
)
if res.status_code == 200:
try:
res_json = res.json()
constant = res_json["meta"]["constant"]
except (ValueError, KeyError, TypeError) as e:
# e.g. an intercepting proxy returning an HTML page
last_error = f"Malformed response from license server: {e}"
continue
if constant == "VALID":
return res_json
raise LicenseConfigError(
f"The license server rejected the license: "
f"{self._redact(str(constant))[:200]}. "
"Check your license status with `pywasp status` or contact "
"DTU if you believe this is wrong."
)
if res.status_code == 429:
raise LicenseServerUnreachableError(
"The license server is rate limiting requests (HTTP 429); "
f"{_retry_after_detail(res)}."
)
if 400 <= res.status_code < 500:
raise LicenseConfigError(
f"The license server is reachable but rejected your credentials "
f"(HTTP {res.status_code}). Check the `license_id` and "
"`activation_token` in your configuration; run `pywasp configure` "
"in a terminal to update them."
)
last_error = f"HTTP {res.status_code}: {res.text[:500]}"
raise LicenseServerUnreachableError(
f"Could not reach the license server at {self._url} after 5 attempts. "
f"Last error: {self._redact(last_error)}"
)
def get_points_limit(self) -> int:
"""Get the maximum points allowed by the license.
The limit is fixed for the lifetime of a license (one tier per
license), so it is cached on disk for 24 hours, keyed by the license
id. This keeps repeated runs and parallel workers from hammering the
license server.
"""
points_limit = _read_points_limit_cache(str(self.license_id))
if points_limit is not None:
return points_limit
response_dict = self._request_validation()
try:
policy_id = response_dict["data"]["relationships"]["policy"]["data"]["id"]
points_limit = MAX_PTS[policy_id]
except KeyError as e:
# e.g. a newly introduced license tier this pywasp version does
# not know about: a valid license must not fail hard on this
raise _UnknownPolicyError(
f"Unknown policy ID in license response: {self._redact(str(e))[:200]}"
)
_write_points_limit_cache(str(self.license_id), points_limit)
return points_limit
def get_license_status(self) -> dict:
"""Get the status of the license.
Returns
-------
dict:
A dictionary containing the license status, including whether it is active,
the number of runs used, the runs limit, and the expiry date.
Raises
------
LicenseConfigError
If the license server rejects the configured credentials.
LicenseServerUnreachableError
If the license server cannot be reached.
"""
# Typed errors carry actionable messages (bad credentials, server
# down) and must reach the user, e.g. via `pywasp status`.
response_dict = self._request_validation()
try:
return {
"active": True,
"runs_used": int(response_dict["data"]["attributes"]["uses"]),
"runs_limit": int(response_dict["data"]["attributes"]["maxUses"]),
"expiry": pd.to_datetime(
response_dict["data"]["attributes"]["expiry"]
).date(),
}
except Exception as e:
logger.error(f"Error getting license status: {e}")
return {
"active": None,
"runs_used": None,
"runs_limit": None,
"expiry": None,
}
class _PywaspSettings(BaseSettings):
"""Settings from ``pywasp_*`` environment variables, ``.env`` and the config file.
Environment variables take precedence over the file.
"""
model_config = SettingsConfigDict(
toml_file=_CONFIG_FILE,
env_prefix="pywasp_",
extra="ignore",
env_nested_delimiter="__",
env_file=".env",
)
# Add toml settings to end of default 4 sources
@classmethod
def settings_customise_sources(
cls,
settings_cls: type[BaseSettings],
init_settings: PydanticBaseSettingsSource,
env_settings: PydanticBaseSettingsSource,
dotenv_settings: PydanticBaseSettingsSource,
file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]:
return (
init_settings,
env_settings,
dotenv_settings,
file_secret_settings,
TomlConfigSettingsSource(settings_cls),
)
class _DownloadSettings(_PywaspSettings):
"""The ``[download]`` section alone, readable without a licensing section."""
download: DownloadCfg = DownloadCfg()
class Settings(_PywaspSettings):
download: DownloadCfg
licensing: Annotated[
Union[LicensingLocalCfg, LicensingRemoteCfg],
Discriminator("license_type"),
]
# Session-level state for the point-limit check: the limit is fetched at
# most once per process, and a check disabled by an unreachable server
# stays disabled for the rest of the session.
_points_limit: int | None = PrivateAttr(default=None)
_points_check_disabled: bool = PrivateAttr(default=False)
def save(self):
# Atomic write: sibling workers (processes and threads) and the
# licensing library, which reads this file directly, must not see a
# half-written file. The temp name is unique per writer (pid alone
# would collide across threads), and it is only unlinked on the
# failure path so one writer can never delete another's live temp
# file. Created 0600 from the start (not chmod'ed after) because it
# contains the activation token. Written as UTF-8 explicitly: TOML
# must be UTF-8, and without an explicit encoding Windows writes the
# locale code page (e.g. cp1252), which the Rust licensing library
# cannot read back when paths contain non-ASCII characters.
_CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
tmp_file = _CONFIG_FILE.with_name(
f"{_CONFIG_FILE.name}.tmp-{os.getpid()}-{uuid.uuid4().hex[:8]}"
)
try:
fd = os.open(tmp_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(fd, "w", encoding="utf-8") as f:
toml.dump(self.model_dump(exclude_none=True), f)
os.replace(tmp_file, _CONFIG_FILE)
except BaseException:
tmp_file.unlink(missing_ok=True)
raise
logger.info(f"Configuration saved to {_CONFIG_FILE}.")
def get_license_status(self) -> dict | None:
"""Get the status of the license.
Returns
-------
dict:
A dictionary containing the license status, including whether it is active,
the number of runs used, the runs limit, and the expiry date.
"""
if isinstance(self.licensing, LicensingLocalCfg):
return None
elif isinstance(self.licensing, LicensingRemoteCfg):
return self.licensing.get_license_status()
else:
raise PywaspError(
f"Unknown license type: {self.licensing.license_type}. Cannot get license status."
)
def check_points_within_limit(self, n_points: int) -> None:
"""Check that the number of points is within the license limit.
Local licenses have no limit. For remote licenses the limit is
fetched at most once per session (backed by a 24-hour on-disk cache).
If the license server cannot be reached, or reports a policy this
pywasp version does not know, a single warning is emitted and the
check is skipped for the rest of the session; the WAsP core still
enforces the license itself.
Parameters
----------
n_points : int
The number of points to check.
Raises
------
PointsLimitError
If the number of points exceeds the license limit.
LicenseConfigError
If the license server rejects the configured credentials.
LicenseError
If the server certificate cannot be verified (code 123).
"""
if isinstance(self.licensing, LicensingLocalCfg):
return # No limit for local licenses
if not isinstance(self.licensing, LicensingRemoteCfg):
raise PywaspError(
f"Unknown license type: {self.licensing.license_type}. Cannot check point limit."
)
if self._points_check_disabled:
return
if self._points_limit is None:
try:
self._points_limit = self.licensing.get_points_limit()
except (LicenseServerUnreachableError, _UnknownPolicyError) as e:
self._points_check_disabled = True
warnings.warn(
"Could not determine the license point limit; the check "
"is skipped for the rest of this session. The license "
f"itself is still enforced. Details: {e}",
UserWarning,
stacklevel=_find_user_stacklevel(),
)
return
if n_points > self._points_limit:
raise PointsLimitError(n_points, self._points_limit)
# The configuration is resolved lazily: importing pywasp never contacts the
# license server and never warns about a missing/invalid configuration.
# `_config` is populated on the first call to `get_config()` (or replaced by
# `create_config_interactively`).
_config = None
def get_config() -> Settings:
"""Return the PyWAsP configuration, loading it on first use.
The configuration is read from the configuration file and the
``pywasp_*`` environment variables. No connection to the license server
is made.
Returns
-------
Settings
The loaded configuration.
Raises
------
LicenseConfigError
If no configuration exists or it fails validation. The message
explains how to fix it.
"""
global _config
if _config is None:
try:
_config = Settings.model_validate({})
# SettingsError is defensive: pydantic-settings currently wraps a
# malformed TOML file in a ValidationError, but that is its wrapping
# behavior, not a contract — a raw parser error here must never
# reach the user as a bare traceback
except (ValidationError, OSError, SettingsError) as e:
if _CONFIG_FILE.exists():
sources = (
f"the configuration file '{_CONFIG_FILE}' and any "
"`pywasp_*` environment variables (environment variables "
"take precedence over the file)"
)
else:
sources = (
"`pywasp_*` environment variables (no configuration file "
f"was found at '{_CONFIG_FILE}')"
)
if isinstance(e, ValidationError):
# Format without input values: str(e) embeds input_value=...,
# which would leak the activation token into the message
details = "\n".join(
f" {'.'.join(str(loc) for loc in err['loc'])}: {err['msg']}"
for err in e.errors(include_input=False, include_url=False)
)
else:
details = f" Could not read the configuration file: {e}"
raise LicenseConfigError(
f"The PyWAsP configuration is missing or invalid. Settings are "
f"read from {sources}. Run `pywasp configure` in a terminal to "
"create a configuration file, or fix the failing fields:\n"
f"{details}"
) from e
return _config
def _get_download_cfg() -> DownloadCfg:
"""Return the ``[download]`` settings, or their defaults.
Read from the ``pywasp_download__*`` environment variables and the
``[download]`` section of the configuration file only, so they apply
even when there is no (valid) licensing configuration. Defaults are used
when neither source sets them; an invalid section is reported with a
warning and also falls back to the defaults.
"""
try:
return _DownloadSettings.model_validate({}).download
except (ValidationError, OSError, SettingsError) as e:
logger.warning(
"Ignoring invalid [download] settings and using the defaults "
"(download allowed, prompt in a terminal): %s",
e,
)
return DownloadCfg()
def _license_disabled() -> bool:
"""Whether license checks are disabled via the PW_LICENSE environment variable.
``PW_LICENSE=False`` disables license validation in the WAsP core (used
for testing); the Python-side checks mirror it.
"""
return os.environ.get("PW_LICENSE", "True").lower() in ("false", "0")
def _read_config_file() -> dict | None:
"""Return the parsed configuration file, or None when missing/unreadable."""
if not _CONFIG_FILE.exists():
return None
try:
return toml.load(_CONFIG_FILE)
except (OSError, toml.TomlDecodeError):
# Unreadable (e.g. permissions) or unparseable: treat as differing
return None
def _sync_config_to_disk(config: Settings) -> bool:
"""Sync the PyWAsP settings to disk for the WAsP core license check.
The core reads ``pywasp_config.toml`` only (not the ``pywasp_*``
environment variables), so the file is written whenever the effective
settings differ from what is on disk.
Returns
-------
bool
True when the file matches the effective settings on return (it
already did, or it was written now); False when it could not be
written.
"""
global _config
on_disk = _read_config_file()
if on_disk == config.model_dump(exclude_none=True):
return True
# The file differs from the in-memory settings. It may have been fixed
# or updated externally (e.g. `pywasp configure` run in a terminal while
# this session was open), so re-resolve from the sources instead of
# trusting the cached object — otherwise a stale session would overwrite
# the user's fix with old settings. Environment variables still take
# precedence over the file, by pydantic source order.
try:
fresh = Settings.model_validate({})
if fresh.licensing == config.licensing:
# Same licensing config: carry the session points-check state
# over, so a disabled check stays disabled for the session and
# the limit is not re-fetched. A changed licensing config gets
# a fresh fetch instead — the old state may not apply to it.
fresh._points_limit = config._points_limit
fresh._points_check_disabled = config._points_check_disabled
config = fresh
_config = config
except (ValidationError, OSError, SettingsError):
# The external change is invalid; keep the known-good session
# config and restore the file from it below
pass
if on_disk != config.model_dump(exclude_none=True):
try:
config.save()
except OSError as e:
# e.g. a read-only home directory in a container: proceed — the
# write only matters if the on-disk file must differ from the
# effective settings, and the core reports its own error if so.
warnings.warn(
f"Could not write the configuration file at '{_CONFIG_FILE}' "
f"({e}). If calculations fail with a license error, ensure the "
"file exists and matches your environment settings.",
UserWarning,
stacklevel=_find_user_stacklevel(),
)
return False
return True
def _requires_license(func):
"""Mark a function as requiring a valid PyWAsP license.
Performs no network I/O itself. On call, two purely local steps run:
the configuration is resolved (raising
:class:`~pywasp._errors.LicenseConfigError` with instructions when it is
missing or invalid) and synced to disk. The sync is required because the
licensing library reads only ``pywasp_config.toml`` — it does not read
the ``pywasp_*`` environment variables, so overrides supplied through
them must land in the file. The WAsP core performs the actual license
validation during the calculation.
"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
if not _license_disabled():
_sync_config_to_disk(get_config())
return func(*args, **kwargs)
# Introspectable marker so tests can cross-check the gated set
wrapper.__pywasp_requires_license__ = True
return wrapper
def _check_points_within_limit(n_points):
"""Check ``n_points`` against the license point limit.
No-op when licensing is disabled via ``PW_LICENSE=False``.
"""
if _license_disabled():
return
get_config().check_points_within_limit(n_points)
def _prompt_download_settings() -> DownloadCfg:
"""Prompt for download preferences."""
print(
"Some functions need the global mesoclimate files (ERA5/CFSR mean "
"atmospheric fields), fetched on first use (or with "
"`pywasp mesoclimate download`)."
)
download_prompt = (
input("Ask before downloading, when a terminal is available? (y/N): ")
.strip()
.lower()
)
download_global = input("Allow PyWAsP to download them? (Y/n): ").strip().lower()
return DownloadCfg(
download_prompt=download_prompt in ["y", "yes", "true"],
download_global_nc_files=download_global not in ["n", "no", "false"],
)
def _prompt_save_anyway() -> bool:
"""Ask whether to save the configuration despite a failed check."""
save = input("Save configuration anyway? (y/N): ").strip().lower()
if save in ["y", "yes"]:
return True
print("Configuration creation cancelled.")
return False
def _prompt_local_licensing() -> LicensingLocalCfg | None:
"""Prompt for local license server settings; None if the user cancels."""
print("\nConfiguring local license server...")
while True:
host = input("License server host: ").strip()
if host:
break
print("Please enter a valid host.")
while True:
port_input = input("License server port: ").strip()
if port_input:
break
print("Please enter a valid port.")
try:
port = int(port_input)
if not (0 <= port <= 65535):
raise ValueError("Port must be between 0 and 65535")
except ValueError as e:
print(f"Invalid port: {e}")
return None
licensing_cfg = LicensingLocalCfg(host=host, port=port)
print(f"Testing connection to {host}:{port}...")
status, detail = licensing_cfg.check_server_status()
if status == "ok":
print("Connection successful!")
else:
print(f"Connection failed: {detail}")
if not _prompt_save_anyway():
return None
return licensing_cfg
def _prompt_remote_licensing() -> LicensingRemoteCfg | None:
"""Prompt for remote (Keygen cloud) license settings; None if the user cancels."""
print("\nConfiguring remote license (Keygen cloud)...")
license_id_input = input("License ID (UUID): ").strip()
try:
license_id = uuid.UUID(license_id_input)
except ValueError:
print("Invalid UUID format for license ID.")
return None
activation_token = input("Activation token: ").strip()
if not activation_token:
print("Activation token is required.")
return None
redirect = input("Redirect host through license.windenergy.dtu.dk? (y/N): ").strip()
redirect_host = (
"license.windenergy.dtu.dk"
if redirect.lower() in ["y", "yes", "true"]
else None
)
# Re-prompt on a nonexistent path: constructing the config below
# validates existence and would abort the whole session otherwise
while True:
root_ca_path = input(
"Root CA file path (optional, press Enter to skip): "
).strip()
if not root_ca_path:
root_ca_filepath = None
break
root_ca_filepath = Path(root_ca_path).expanduser()
if root_ca_filepath.exists():
break
print(
f"CA file not found at {root_ca_filepath}. "
"Enter a valid path, or press Enter to skip."
)
licensing_cfg = LicensingRemoteCfg(
license_id=license_id,
activation_token=activation_token,
redirect_host=redirect_host,
root_ca_filepath=root_ca_filepath,
)
print("Testing license validation...")
try:
licensing_cfg._request_validation()
print("License validated!")
except Exception as e:
print(f"License validation failed: {e}")
if not _prompt_save_anyway():
return None
return licensing_cfg
[docs]
def create_config_interactively():
"""
Interactively prompt the user to create a pywasp_config.toml file.
This function guides the user through setting up their PyWAsP configuration
by asking for licensing and download preferences. The configuration is saved
to the default config file location.
Returns
-------
Settings or None
The created configuration settings object, or None if the user
cancelled the process or the provided input was invalid.
"""
global _config
print(f"Creating PyWAsP configuration file at: {_CONFIG_FILE}")
print("=" * 60)
try:
if _CONFIG_FILE.exists():
overwrite = (
input(
f"Configuration file already exists at {_CONFIG_FILE}. Overwrite? (y/N): "
)
.strip()
.lower()
)
if overwrite not in ["y", "yes"]:
print("Configuration creation cancelled.")
return None
print("\n1. Download Settings")
print("-" * 20)
download_cfg = _prompt_download_settings()
print("\n2. Licensing Settings")
print("-" * 20)
print("Choose license type:")
print("1. Remote license (Keygen cloud)")
print("2. Local license server")
while True:
license_choice = input("Enter choice (1 or 2): ").strip()
if license_choice in ["1", "2"]:
break
print("Please enter 1 or 2.")
if license_choice == "2":
licensing_cfg = _prompt_local_licensing()
else:
licensing_cfg = _prompt_remote_licensing()
if licensing_cfg is None:
return None
config = Settings(download=download_cfg, licensing=licensing_cfg)
config.save()
_config = config
print(f"\nConfiguration saved successfully to: {_CONFIG_FILE}")
print("\nConfiguration summary:")
print(f" Download prompts: {download_cfg.download_prompt}")
print(f" Auto-download global files: {download_cfg.download_global_nc_files}")
print(f" License type: {licensing_cfg.license_type}")
if isinstance(licensing_cfg, LicensingLocalCfg):
print(f" License server: {licensing_cfg.host}:{licensing_cfg.port}")
else:
print(f" License ID: {licensing_cfg.license_id}")
print(f" Redirect host: {licensing_cfg.redirect_host or 'None'}")
return config
except (KeyboardInterrupt, EOFError):
# EOFError covers Ctrl+Z on Windows and closed/non-interactive stdin.
print("\n\nConfiguration creation cancelled by user.")
return None
except Exception as e:
print(f"\nError creating configuration: {e}")
return None
