import functools
import logging
import os
import warnings
from collections.abc import Callable, Iterator, Mapping, Sequence
from contextlib import contextmanager
from dataclasses import replace
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, ParamSpec, TypeVar
from embedl_hub._internal.compat import UTC
from embedl_hub._internal.core.types import RemotePath
if TYPE_CHECKING:
from embedl_hub._internal.core.context import HubContext
from embedl_hub._internal.core.config import (
API_KEY_ENV_VAR_NAME,
SettingSource,
resolve_api_key,
resolve_base_url,
)
from embedl_hub._internal.tracking.artifact_upload import upload_artifact_file
from embedl_hub._internal.tracking.dashboard import (
Dashboard,
DashboardSchemaError,
)
from embedl_hub._internal.tracking.errors import (
TrackingFailureWarning,
TrackingProjectChangedWarning,
TrackingUsageError,
UnsupportedDeviceError,
raise_if_run_error,
)
from embedl_hub._internal.tracking.rest_api import (
ApiConfig,
ApiError,
CompletedRunStatus,
Device,
ExternalLink,
Metric,
Parameter,
Project,
Run,
RunDashboard,
RunListPage,
RunStatus,
RunType,
Tag,
_RemoveParent,
coerce_run_type,
create_project,
create_run,
create_run_dashboard,
get_devices,
get_project_by_name,
get_run,
js_trim,
list_runs,
log_external_link,
log_metric,
log_param,
log_tag,
resolve_run,
start_pending_run,
truncate_param_value,
update_run,
)
from embedl_hub._internal.tracking.run_log import (
LoggedArtifact,
LoggedExternalLink,
LoggedMetric,
LoggedParam,
RunLog,
RunLogHistory,
)
# The run and project an invocation was handed, which `embedl-hub exec`
# exports for the command it wraps. Defined here, with the other
# environment names, and re-exported by `cli.target` so the CLI and the
# client cannot disagree about what they are called.
RUN_ENV_VAR_NAME = "EMBEDL_HUB_RUN"
PROJECT_ENV_VAR_NAME = "EMBEDL_HUB_PROJECT"
# Set by `embedl-hub exec --safe` so that adopting the run it handed over
# tolerates an unreachable Hub, as creating that run did. It covers the
# handover only: whether the command's own logging tolerates failure is
# the command's decision, made with `safe_log`.
SAFE_ENV_VAR_NAME = "EMBEDL_HUB_SAFE"
# Values that turn it off. Anything else set means on, so that a
# variable someone exported as "yes" or "on" is not read as a refusal.
_FALSE_ENV_VALUES = frozenset({"", "0", "false", "no", "off"})
# Statuses a run does not come back from, so one cannot be adopted.
_SETTLED_RUN_STATUSES = (
RunStatus.FINISHED,
RunStatus.FAILED,
RunStatus.KILLED,
)
logger = logging.getLogger(__name__)
_P = ParamSpec("_P")
_R = TypeVar("_R")
# Frames between the warning helper and the caller it should blame.
# Logging calls arrive through one decorator frame; the run lifecycle
# adds a `contextlib` frame because it is a generator context manager.
_LOG_STACKLEVEL = 3
_RUN_STACKLEVEL = 4
# Completing a run happens inside `_governing`, one more generator context
# manager nested in the run's own, so two frames further from the caller.
_FINISH_STACKLEVEL = _RUN_STACKLEVEL + 2
def _safe_by_environment() -> bool:
"""Whether the environment asks for a tolerant handover.
``embedl-hub exec --safe`` exports
:data:`SAFE_ENV_VAR_NAME` for the command it wraps, having already
tolerated a failure to create the run. A script that then adopts
that run should not raise where the wrapper warned: the flag exists
so the workload survives an unreliable Hub, and the adoption is part
of the same handover.
:return: Whether the variable is set to anything but an off value.
"""
value = os.getenv(SAFE_ENV_VAR_NAME)
return value is not None and value.strip().lower() not in _FALSE_ENV_VALUES
def _warn_tracking_failure(
operation: str,
exc: BaseException,
*,
stacklevel: int = _LOG_STACKLEVEL,
) -> None:
"""Warn that `operation` failed and was skipped.
:param operation: Human-readable name of the failed operation, used
verbatim in the message (e.g. ``"log a metric"``).
:param exc: The exception being downgraded. Its type and message
are included so the cause is not lost.
:param stacklevel: Frames to skip so the warning is attributed to
the user's call site rather than to this module.
"""
# "may not have arrived" rather than "did not": a request that
# failed while reading the response may already have been committed,
# and telling a caller the data was lost invites a retry that
# duplicates it.
#
# The cause goes last: exception messages often end in a period,
# and trailing punctuation reads fine there but not mid-sentence.
warnings.warn(
f"Embedl Hub could not {operation}; continuing without it. "
f"This data may not have reached the Hub. "
f"Leave safe mode to raise instead of warn. "
f"Cause: {type(exc).__name__}: {exc}",
TrackingFailureWarning,
stacklevel=stacklevel,
)
def _degrades_when_safe(
operation: str,
) -> Callable[[Callable[_P, _R]], Callable[_P, _R | None]]:
"""Make a logging method warn instead of raise inside a safe scope.
Wraps a :class:`Client` logging method so that, when the client is
inside a :meth:`Client.safe_log` scope, a failure is reported as a
:class:`TrackingFailureWarning` and ``None`` is returned instead of
propagating. Outside a safe scope the method behaves exactly as if
it were undecorated.
:param operation: Human-readable name of the operation, used in the
warning message.
:return: A decorator for a ``Client`` logging method.
"""
def decorator(fn: Callable[_P, _R]) -> Callable[_P, _R | None]:
@functools.wraps(fn)
def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R | None:
client = args[0]
try:
return fn(*args, **kwargs)
except TrackingUsageError:
# A malformed call, not an unreliable Hub. Safe mode
# never hides these; see TrackingUsageError.
raise
except Exception as exc: # noqa: BLE001 — re-raised below
if not client.safe_logging:
raise
_warn_tracking_failure(operation, exc)
return None
return wrapper
return decorator
# Field name, expected tuple length, and the shape to quote back.
_BATCH_SHAPES = (
("metrics", 3, "(name, value, step)"),
("params", 2, "(name, value)"),
("external_links", 2, "(label, url)"),
)
def _validate_batch_shapes(
metrics: list[tuple] | None,
params: list[tuple] | None,
external_links: list[tuple] | None,
) -> None:
"""Check the tuple shapes `log_batch` was handed.
Without this the mistake surfaces as "not enough values to unpack"
from a comprehension, which names neither the argument nor the entry.
:raises TrackingUsageError: If any entry has the wrong length.
"""
for entries, (field, arity, shape) in zip(
(metrics, params, external_links), _BATCH_SHAPES
):
for index, entry in enumerate(entries or []):
if len(entry) != arity:
raise TrackingUsageError(
f"{field}[{index}] has {len(entry)} values; expected "
f"{arity}. Each entry is {shape}."
)
def _resolve_run_type(
type: RunType | str,
custom_type: str | None = None,
) -> tuple[RunType, str | None]:
"""Resolve a run type and optional custom type label.
When ``type`` is a string it is first matched case-insensitively
(after stripping whitespace) against the :class:`RunType` enum.
Matching strings are coerced to the corresponding enum member.
Strings that do not match any known member are treated as custom
type labels and ``RunType.CUSTOM`` is returned.
:param type: A ``RunType`` enum member or a string label. Known
type names are coerced to their enum member regardless of case
(e.g. ``"compile"`` → :attr:`RunType.COMPILE`). Unknown strings
become custom type labels.
:param custom_type: An explicit custom type label (only used when
``type`` is :attr:`RunType.CUSTOM`).
:return: A ``(RunType, custom_type_label)`` tuple.
:raises ValueError: If ``type`` (or its coerced value) is
:attr:`RunType.CUSTOM` but no label is supplied.
"""
if isinstance(type, str):
coerced = coerce_run_type(type)
if isinstance(coerced, RunType):
type = coerced
else:
return RunType.CUSTOM, type
if type is RunType.CUSTOM and not custom_type:
raise ValueError(
"A custom_type label is required when type=RunType.CUSTOM."
)
return type, custom_type if type is RunType.CUSTOM else None
def _normalize_param_filters(params: Mapping[str, str]) -> dict[str, str]:
"""Normalize parameter filters the way :func:`log_param` stores them.
:raises ValueError: If a name is blank, a name or value contains a
NUL character, or two names are the same once trimmed.
"""
given_as: dict[str, str] = {}
normalized: dict[str, str] = {}
for name, value in params.items():
trimmed = js_trim(name)
if not trimmed:
raise ValueError(
f"The parameter name {name!r} is blank. Pass the name the "
"parameter was logged under."
)
stored = truncate_param_value(trimmed, value)
if "\0" in trimmed or "\0" in stored:
raise ValueError(
f"The parameter {name!r} contains a NUL character, which "
"the hub cannot store, so no run can have logged it."
)
if trimmed in normalized:
raise ValueError(
f"The parameter names {given_as[trimmed]!r} and {name!r} "
f"are the same once trimmed ({trimmed!r}). Pass each "
"parameter once."
)
given_as[trimmed] = name
normalized[trimmed] = stored
return normalized
def _split_run_types(
types: Sequence[RunType | str] | None,
) -> tuple[list[RunType] | None, list[str] | None]:
"""Split run types and custom type labels for the hub's two filters.
Strings are read as :func:`_resolve_run_type` reads them: a built-in
type name selects that type, any other label selects custom runs.
The hub applies both filters at once, so built-in types and labels
cannot be combined in one listing.
:raises ValueError: If a label is blank or contains a comma, or if
built-in types are mixed with labels.
"""
if not types:
return None, None
run_types: list[RunType] = []
labels: list[str] = []
for entry in types:
if isinstance(entry, str) and not js_trim(entry):
raise ValueError(
"A run type label is blank. Pass a built-in type such as "
'RunType.EVAL or "eval", or a custom type label such as '
'"training".'
)
resolved = coerce_run_type(entry) if isinstance(entry, str) else entry
if isinstance(resolved, RunType):
if resolved not in run_types:
run_types.append(resolved)
continue
if "," in resolved:
raise ValueError(
f"The run type label {resolved!r} contains a comma, which "
"the hub's filter cannot express. List the custom runs with "
"types=[RunType.CUSTOM] and pick by custom_type instead."
)
labels.append(resolved)
if labels and any(
run_type is not RunType.CUSTOM for run_type in run_types
):
raise ValueError(
"Built-in run types and custom type labels cannot be combined "
"in one listing: the hub matches both filters at once, so no "
"run satisfies them. Make one call per kind, e.g. "
'types=[RunType.EVAL] and then types=["training"].'
)
if labels and RunType.CUSTOM not in run_types:
run_types.append(RunType.CUSTOM)
return run_types, labels or None
[docs]
class Client:
"""Tracks projects and runs for the Embedl Hub web app."""
_api_config: ApiConfig | None
_project: Project | None
_active_run: Run | None
_current_run_log: RunLog | None
_run_log_history: list[RunLog]
_log_remote_artifacts: bool
_safe_log_depth: int
def __init__(
self,
api_config: ApiConfig | None = None,
*,
log_remote_artifacts: bool = True,
) -> None:
"""Create a new tracking client.
:param api_config: API configuration. When ``None``, the
configuration is loaded from the environment on first use.
:param log_remote_artifacts: Whether to fetch remote artifacts
from devices and upload them to the Hub when
:meth:`log_artifact` is called with a remote path or a
remote :class:`~embedl_hub._internal.tracking.run_log.LoggedArtifact`.
Defaults to ``True``. Set to ``False`` to skip the
download-and-upload step and speed up execution.
"""
self._api_config = api_config
# Which layer supplied a resolved key, so `api_key_source` can
# describe this client rather than re-reading the machine. An
# injected config came from the caller and has no layer.
self._api_key_source: SettingSource | None = None
self._project = None
self._active_run = None
self._current_run_log = None
self._run_log_history = []
self._log_remote_artifacts = log_remote_artifacts
self._safe_log_depth = 0
@property
def log_remote_artifacts(self) -> bool:
"""Whether remote artifacts are fetched and uploaded to the Hub.
When ``True`` (the default), calling :meth:`log_artifact` with
a remote path or a remote
:class:`~embedl_hub._internal.tracking.run_log.LoggedArtifact` will
download the file from the device and upload it to the
Embedl Hub.
Set to ``False`` to skip the download-and-upload step and
speed up execution. The artifact is still recorded in the
run log, but its contents are not transferred.
"""
return self._log_remote_artifacts
@log_remote_artifacts.setter
def log_remote_artifacts(self, value: bool) -> None:
self._log_remote_artifacts = value
@property
def safe_logging(self) -> bool:
"""Whether logging calls currently warn instead of raising.
``True`` while execution is inside a :meth:`safe_log` scope.
Read-only — safe mode is entered by scope, never by assignment,
so a client handed to library code cannot be permanently
disarmed.
"""
return self._safe_log_depth > 0
[docs]
@contextmanager
def safe_log(self) -> Iterator["Client"]:
"""Downgrade logging failures to warnings inside this scope.
Within the scope, :meth:`log_param`, :meth:`log_metric`,
:meth:`log_link`, :meth:`log_batch`, :meth:`log_tag`,
:meth:`log_artifact` and :meth:`log_dashboard` emit a
:class:`TrackingFailureWarning` and
return ``None`` instead of raising, so a Hub outage cannot abort
the surrounding script::
with client.safe_log():
client.log_metric("accuracy", 0.92)
Scopes nest; leaving one restores the previous state. Only the
client's own logging calls are affected — exceptions raised by
your code inside the scope propagate untouched, and calls made
outside it raise as usual.
This does not cover starting or ending a run. Pass
``safe=True`` to :meth:`start_run` for that.
:return: A context manager yielding this client.
"""
self._safe_log_depth += 1
try:
yield self
finally:
self._safe_log_depth -= 1
[docs]
def set_project(self, name: str, *, safe: bool = False) -> Project | None:
"""Set or create the current project by name.
:param name: The project name. Created on the Hub when it does
not already exist.
:param safe: When ``True``, a failure emits a
:class:`TrackingFailureWarning` and returns ``None`` instead
of raising.
:return: The project, or ``None`` if it could not be set and
`safe` is ``True``. Either way, a failure leaves the client
with **no** project selected rather than the one selected
before, so a later run cannot be filed against the wrong
project.
:raises TrackingUsageError: If a run is active. Moving the
project out from under it would leave the client addressing
that run inside a project it is not in, which every log call
composes independently and so cannot detect: the requests go
to a run that is not there. Raised whatever `safe` is set
to, since this is a caller mistake rather than a Hub
failure.
:raises ApiError: If the Hub rejects the request and `safe` is
``False``.
"""
self._refuse_project_change_during_run()
try:
project = get_project_by_name(self.api_config, name)
if not project:
project = create_project(self.api_config, name)
except Exception as exc: # noqa: BLE001 — re-raised below
# A failed switch leaves no project selected, rather than the
# one selected before. Keeping the old one would let safe
# mode carry on and file the run against the wrong project —
# silently wrong data is worse than none.
self._project = None
if not safe:
raise
_warn_tracking_failure("set the project", exc)
return None
self._project = project
return project
[docs]
def use_project(self, project: Project) -> Project:
"""Select a project already looked up, without a Hub call.
For a :class:`Project` in hand, such as the one
:meth:`get_run_project` returns. It selects exactly that
project, where :meth:`set_project` looks one up by name and
cannot tell apart two projects that share one.
:param project: The project to select.
:return: `project`, now the current project.
:raises TrackingUsageError: If a run is active; see
:meth:`set_project`.
"""
self._refuse_project_change_during_run()
self._project = project
return project
def _refuse_project_change_during_run(self) -> None:
if self._active_run is None:
return
raise TrackingUsageError(
f"Run '{self._active_run.id}' is still active, so the "
f"project cannot be changed: the run would no longer be "
f"in it, and everything logged would be filed against a "
f"run the project does not have. Leave the 'start_run' "
f"block first, or select the other project on its own "
f"client."
)
[docs]
def use_run(
self,
run_id: str,
project: str | None = None,
*,
safe: bool | None = None,
) -> Run | None:
"""Log to a run that already exists, started somewhere else.
For picking up work another process began — most often the run
``embedl-hub exec`` created for the command it is wrapping. Use
:meth:`use_inherited_run` for that case, which reads the run and
project out of the environment rather than being told them.
Adopting a run does not take responsibility for it. Nothing
here completes it, and leaving the process does not either:
whoever started the run still decides its outcome. Under
``exec`` that is the wrapper, which completes the run from the
command's exit status.
:param run_id: The run to log to.
:param project: The project the run is in. ``None`` means the
project already selected, which the run must then be in;
with no project selected, the run is looked up by ID alone
and its own project becomes the selection. A name switches
to that project first and warns with
:class:`TrackingProjectChangedWarning`, because a run
belongs to one project and adopting it has to move the
client with it.
:param safe: When ``True``, a failure to reach the Hub emits a
:class:`TrackingFailureWarning` and returns ``None`` instead
of raising, leaving nothing adopted. As elsewhere, this
covers an unreliable Hub and not a malformed call.
``None``, the default, takes the answer from the
environment: ``embedl-hub exec --safe`` sets
``EMBEDL_HUB_SAFE`` for the command it wraps, so a script
adopting the run that wrapper handed it does not raise where
the wrapper itself warned. Pass ``False`` to insist on
raising even there.
:return: The adopted run, or ``None`` if it could not be
adopted and `safe` is ``True``.
:raises TrackingUsageError: If `run_id` is blank, if a run is
already active, or if the run has already finished. Raised
whatever `safe` is set to.
:raises ApiError: If the Hub rejects the request and `safe` is
``False``.
"""
self._refuse_unadoptable(run_id)
return self._adopt(
run_id,
project,
safe=_safe_by_environment() if safe is None else safe,
operation="use a run",
stacklevel=3,
)
[docs]
def use_inherited_run(self, *, safe: bool | None = None) -> Run | None:
"""Log to the run this process was handed, if it was handed one.
Reads ``EMBEDL_HUB_RUN`` and ``EMBEDL_HUB_PROJECT``, which
``embedl-hub exec`` exports for the command it wraps, so a
script can record into the run the wrapper already created::
embedl-hub exec --type training ./train.py
.. code-block:: python
client = Client()
client.use_inherited_run()
client.log_metric("loss", 0.1)
A missing ``EMBEDL_HUB_RUN`` is not an error. It is what
``exec --safe`` leaves behind when the Hub could not be reached:
the project is set, no run exists, and the command is meant to
carry on untracked. So this warns and returns ``None`` rather
than raising, and a script that treats the return value as
optional works under both.
:param safe: As for :meth:`use_run`, including that the default
defers to ``EMBEDL_HUB_SAFE``, which is set for exactly this
case by ``exec --safe``.
:return: The adopted run, or ``None`` if the environment named
none, or if it could not be adopted and `safe` is ``True``.
:raises TrackingUsageError: If a run is already active.
:raises ApiError: If the Hub rejects the request and `safe` is
``False``.
"""
run_id = (os.getenv(RUN_ENV_VAR_NAME) or "").strip()
project = (os.getenv(PROJECT_ENV_VAR_NAME) or "").strip() or None
if not run_id:
warnings.warn(
f"No run to adopt: {RUN_ENV_VAR_NAME} is not set. "
f"Nothing will be recorded unless this client starts a "
f"run of its own. Running under 'embedl-hub exec' sets "
f"it, except when tracking was skipped.",
TrackingFailureWarning,
stacklevel=2,
)
return None
return self.use_run(run_id, project, safe=safe)
[docs]
@contextmanager
def use_pending_run(
self,
run_id: str,
project: str | None = None,
*,
safe: bool | None = None,
) -> Iterator[Run | None]:
"""Start a pending run and finish it when the context exits.
For the job a pending run was created for: something else made
the run with :meth:`create_run` and ``pending=True``, annotated
it, and handed over its ID. This starts it as the work begins
and, unlike :meth:`use_run`, takes responsibility for it — the
run is completed from how the block ends, exactly as
:meth:`start_run` completes a run it created::
with client.use_pending_run(run_id):
client.log_metric("loss", 0.1)
:param run_id: The pending run to start.
:param project: As for :meth:`use_run`: ``None`` means the
selected project, or, with none selected, that the run is
looked up by ID alone and its own project becomes the
selection; a name switches to that project first.
:param safe: As for :meth:`use_run`, including that ``None``
defers to ``EMBEDL_HUB_SAFE``. When tolerated, a failure to
find or start the run warns, leaves nothing selected, and
yields ``None``; the body still runs.
:return: An iterator yielding the run, now running, or ``None``
if it could not be started and `safe` is ``True``.
:raises TrackingUsageError: If `run_id` is blank, if a run is
already active, or if the run is not pending — it is running
or finished, or something else started it while it was being
adopted here — since whoever did that owns its outcome; use
:meth:`use_run` to log to a run something else is running.
Raised whatever `safe` is set to.
:raises ApiError: If the Hub rejects a request and `safe` is
``False``.
"""
self._refuse_unadoptable(run_id)
tolerant = _safe_by_environment() if safe is None else safe
previous = self._project
run = self._adopt(
run_id,
project,
safe=tolerant,
operation="start a pending run",
pending=True,
stacklevel=4,
)
if run is None:
yield None
return
project_id = self.project.id
try:
start_pending_run(self.api_config, project_id, run.id)
except Exception as exc: # noqa: BLE001 — sorted out below
if isinstance(exc, ApiError) and exc.status_code == 409:
# Something else started it between the lookup and here.
# Their run now, whatever `safe` says: the selection goes
# back to what it was, as when `_adopt` refuses a run.
self._project = previous
self._active_run = None
raise TrackingUsageError(
f"Run '{run_id}' was started by something else while "
f"it was being adopted here, so it is no longer "
f"pending and whoever started it owns its outcome."
) from exc
# A refusal means nothing started. A lost answer does not:
# the start may well have landed, and a run left running with
# nobody to complete it is the one outcome to avoid, so ask.
if isinstance(exc, ApiError) or not self._running_now(
project_id, run.id
):
self._project = None
self._active_run = None
if not tolerant:
raise
_warn_tracking_failure(
"start a pending run", exc, stacklevel=_RUN_STACKLEVEL
)
yield None
return
# Not read back on success: a read failing after the start would
# leave a running run that nobody completes. The Hub stamped the
# start with its own clock; this one is close enough locally.
run = run.model_copy(
update={
"status": RunStatus.RUNNING,
"started_at": datetime.now(UTC),
}
)
self._active_run = run
self._current_run_log = RunLog(
run_id=run.id,
run_name=run.name,
started_at=run.started_at,
project_id=project_id,
hub_url=self.api_config.base_url,
)
with self._governing(safe=tolerant):
yield run
def _running_now(self, project_id: str, run_id: str) -> bool:
"""Whether a start whose answer was lost landed anyway.
Unknown counts as not started: with the Hub out of reach there is
nothing to reconcile against, and the warning that follows already
says the request may have reached it.
"""
try:
run = get_run(self.api_config, project_id, run_id)
except Exception: # noqa: BLE001 — unknown is not started
return False
return run.status is RunStatus.RUNNING
def _refuse_unadoptable(self, run_id: str) -> None:
if not run_id or not run_id.strip():
raise TrackingUsageError(
"A run id is required to adopt a run; got an empty "
"value. A shell variable that expanded to nothing is "
"the usual cause."
)
# Same rule as `start_run`: runs do not nest, and silently
# replacing one would move later logging somewhere the caller
# did not ask for.
if self._active_run is not None:
raise TrackingUsageError(
f"Run '{self._active_run.id}' is still active, so "
f"'{run_id}' cannot be adopted. Runs do not nest: leave "
f"the block of the active run first."
)
def _adopt(
self,
run_id: str,
project: str | None,
*,
safe: bool,
operation: str,
pending: bool = False,
stacklevel: int,
) -> Run | None:
"""Select `project` and `run_id` together, or neither.
The fetch is what makes this safe. Runs are addressed within a
project, so reading the run back under the project being
selected is both the lookup and the proof that the run is in it
— a run from anywhere else simply is not found. With nothing
selected, the run is looked up by ID alone and names its own
project. Either way there is no separate check that the two
agree, and a partial failure must leave nothing selected: a
project with a run that is not in it is a state every log call
would compose into a request for something that does not exist.
`stacklevel` is the number of frames from here to the caller's
line, so the warnings below blame that line: three from a plain
method, four from a generator context manager.
"""
previous = self._project
try:
selected = previous
if project is not None:
selected = get_project_by_name(self.api_config, project)
if selected is None:
raise TrackingUsageError(
f"Project '{project}' does not exist, so run "
f"'{run_id}' cannot be found in it. Adopting a "
f"run never creates a project: a run always has "
f"one already."
)
if selected is None:
run = resolve_run(self.api_config, run_id)
selected = run.project
else:
run = get_run(self.api_config, selected.id, run_id)
except TrackingUsageError:
raise
except Exception as exc: # noqa: BLE001 — re-raised below
# Nothing selected rather than half of it, for the reason in
# the docstring above.
self._project = None
self._active_run = None
if safe:
_warn_tracking_failure(
operation, exc, stacklevel=stacklevel + 1
)
return None
raise
if run.status in _SETTLED_RUN_STATUSES:
raise TrackingUsageError(
f"Run '{run_id}' has already finished with status "
f"{run.status.value}, so it cannot be logged to. Start "
f"a new run, or pass parent_run_id={run_id!r} to record "
f"that it follows this one."
)
if pending and run.status is not RunStatus.PENDING:
raise TrackingUsageError(
f"Run '{run_id}' is {run.status.value}, not pending, so "
f"it cannot be started here: whoever started it owns its "
f"outcome. Use 'use_run' to log to a run something else "
f"is running."
)
# Only a project the caller had already chosen is worth
# mentioning. Filling in an empty selection is what adopting a
# run is for, and warning about it every time -- which is what
# the `exec` path does -- would teach the reader to ignore the
# one case that matters.
if previous is not None and previous.id != selected.id:
warnings.warn(
f"The active project is now '{selected.name}', where "
f"run '{run_id}' lives, replacing '{previous.name}'. "
f"Anything logged from here goes to the new project.",
TrackingProjectChangedWarning,
stacklevel=stacklevel,
)
self._project = selected
self._active_run = run
return run
[docs]
def create_run(
self,
type: RunType | str,
name: str | None = None,
parent_run_id: str | None = None,
custom_type: str | None = None,
description: str | None = None,
*,
pending: bool = False,
safe: bool = False,
) -> Run | None:
"""Create a new run for the current project.
:param type: Built-in run type or a custom type label.
:param name: Optional run display name.
:param parent_run_id: Optional parent run ID for lineage.
:param custom_type: Label used when ``type`` is
:attr:`RunType.CUSTOM`.
:param description: Optional run description. The Hub trims it,
stores blank input as no description, and accepts at most 1,000
characters.
:param pending: When ``True``, the run is created pending rather
than running: it exists and can be logged to, and the Hub
lists it as :attr:`RunStatus.PENDING` until something starts
it. For work launched elsewhere, so the run can be annotated
before the job that will run in it exists. Its ``started_at``
is provisional until then.
:param safe: When ``True``, a failure to reach the Hub emits a
:class:`TrackingFailureWarning` and returns ``None`` instead
of raising. An invalid run type is a caller mistake, not a
Hub failure, so it still raises.
:return: The created run, or ``None`` if it could not be created
and `safe` is ``True``.
:raises ValueError: If a custom run type has no label, whatever
`safe` is set to.
:raises ApiError: If the Hub rejects the request and `safe` is
``False``.
"""
# Outside the guard on purpose: a bad run type is the caller's
# bug to fix, and `ehub run start` turns this into its --type
# error message.
resolved_type, resolved_custom_type = _resolve_run_type(
type, custom_type
)
try:
project = self.project
try:
run = create_run(
self.api_config,
type=resolved_type,
name=name,
project_id=project.id,
parent_run_id=parent_run_id,
custom_type=resolved_custom_type,
description=description,
pending=pending,
)
except ApiError as exc:
raise_if_run_error(exc)
raise
except Exception as exc: # noqa: BLE001 — re-raised below
if not safe:
raise
_warn_tracking_failure("create a run", exc)
return None
return run
[docs]
@_degrades_when_safe("update the run")
def update_active_run(
self,
status: CompletedRunStatus | None = None,
ended_at: datetime | None = None,
metrics: list[Metric] | None = None,
params: list[Parameter] | None = None,
external_links: list[ExternalLink] | None = None,
parent_run_id: str | _RemoveParent | None = None,
description: str | None = None,
) -> None:
"""Update the active run.
Writes metrics, params, links and the description, so it counts
as logging: inside a :meth:`safe_log` scope a failure warns
instead of raising.
:param status: Optional terminal run status.
:param ended_at: Optional completion time.
:param metrics: Metrics to append.
:param params: Parameters to append.
:param external_links: External links to append.
:param parent_run_id: Parent run ID, or :data:`REMOVE_PARENT` to
clear the parent.
:param description: New run description. Pass a blank or
whitespace-only string to clear it. The Hub accepts at most
1,000 characters.
:return: ``None``.
:raises ApiError: If the Hub rejects the request and the call is
not inside a :meth:`safe_log` scope.
"""
self._update_active_run(
status=status,
ended_at=ended_at,
metrics=metrics,
params=params,
external_links=external_links,
parent_run_id=parent_run_id,
description=description,
)
def _update_active_run(
self,
status: CompletedRunStatus | None = None,
ended_at: datetime | None = None,
metrics: list[Metric] | None = None,
params: list[Parameter] | None = None,
external_links: list[ExternalLink] | None = None,
parent_run_id: str | _RemoveParent | None = None,
description: str | None = None,
) -> None:
"""Update the active run, ignoring any safe scope.
The guard lives on the public :meth:`update_active_run`. Callers
inside this class use this instead, so that a `safe_log` scope
cannot quietly absorb a run-lifecycle failure the caller chose to
leave unguarded (`start_run` without ``safe=True``), and so that
`log_batch` reports its own operation name rather than this one.
"""
project = self.project
run = self.active_run
try:
update_run(
self.api_config,
status=status,
ended_at=ended_at,
project_id=project.id,
run_id=run.id,
metrics=metrics,
params=params,
external_links=external_links,
parent_run_id=parent_run_id,
description=description,
)
except ApiError as exc:
raise_if_run_error(exc)
raise
[docs]
@contextmanager
def start_run(
self,
type: RunType | str,
name: str | None = None,
parent_run_id: str | None = None,
custom_type: str | None = None,
description: str | None = None,
*,
safe: bool = False,
) -> Iterator[Run | None]:
"""Start a run and finish it when the context exits.
:param type: Built-in run type or a custom type label.
:param name: Optional run display name.
:param parent_run_id: Optional parent run ID for lineage.
:param custom_type: Label used when ``type`` is
:attr:`RunType.CUSTOM`.
:param description: Optional run description. The Hub trims it,
stores blank input as no description, and accepts at most 1,000
characters.
:param safe: When ``True``, a failure to create or complete the
run emits a :class:`TrackingFailureWarning` instead of
raising. If the run cannot be created the body still runs
as though tracking had never been added, with nothing
active, and the context manager yields ``None``.
Guard the body with :meth:`safe_log` as well if it logs —
``safe`` covers only the run lifecycle. Exceptions from the
body are never suppressed.
:return: An iterator yielding the created run, or ``None`` when
``safe`` is ``True`` and the run could not be created.
:raises RuntimeError: If a run is already active, whatever
``safe`` is set to. Runs do not nest; use `parent_run_id`
to record that one run followed another.
:raises ValueError: If a custom run type has no label, whatever
``safe`` is set to — that is a mistake in the call, not a
Hub failure.
:raises ApiError: If the Hub rejects a create or completion
request and ``safe`` is ``False``.
"""
# Both checks below are caller mistakes rather than Hub
# failures, so they raise whatever `safe` is set to. `safe`
# exists to absorb an unreliable Hub, not to paper over misuse.
if self._active_run is not None:
raise TrackingUsageError(
f"Run '{self._active_run.id}' is still active. Runs do "
f"not nest: leave the current 'start_run' block before "
f"starting another. To record that one run followed "
f"another, pass parent_run_id=<previous run id>."
)
# A bad run type is the caller's bug too, and `ehub exec` turns
# this ValueError into its --type error message. `create_run`
# resolves it again; doing so twice is free and keeps the
# validation visible here.
_resolve_run_type(type, custom_type)
try:
run = self.create_run(
type, name, parent_run_id, custom_type, description
)
# Create a new run log for this run
run_log = RunLog(
run_id=run.id,
run_name=run.name,
started_at=run.started_at,
project_id=self.project.id,
hub_url=self.api_config.base_url,
)
except Exception as exc: # noqa: BLE001 — re-raised below
if not safe:
raise
_warn_tracking_failure(
"start a run", exc, stacklevel=_RUN_STACKLEVEL
)
# Behave as though tracking had never been added: nothing
# is active, so a log call in the body cannot be
# misattributed. Nothing needs restoring afterwards —
# the check above guarantees no run was active here.
yield None
return
self._active_run = run
self._current_run_log = run_log
with self._governing(safe=safe):
yield run
@contextmanager
def _governing(self, *, safe: bool) -> Iterator[None]:
"""Complete the active run from how the body ends.
Shared by :meth:`start_run` and :meth:`use_pending_run`, which
differ only in how the run became active.
"""
status: CompletedRunStatus = RunStatus.FINISHED
body_error: BaseException | None = None
try:
yield
except KeyboardInterrupt as exc:
status = RunStatus.KILLED
body_error = exc
raise
except Exception as exc:
status = RunStatus.FAILED
body_error = exc
raise
except BaseException as exc:
# SystemExit, GeneratorExit and friends. The run status is
# left alone — a shutdown is not a run outcome this can
# characterise — but the exception must be remembered, or an
# escalated finish warning would replace it below.
body_error = exc
raise
finally:
ended_at = datetime.now(UTC)
try:
self._update_active_run(status=status)
except Exception as exc: # noqa: BLE001 — re-raised below
if not safe:
raise
try:
_warn_tracking_failure(
"finish a run", exc, stacklevel=_FINISH_STACKLEVEL
)
except TrackingFailureWarning:
# A filter escalated the warning to an error. We are
# in a `finally`, so raising here would replace the
# exception already leaving the body — the one the
# caller actually needs. Honour the escalation only
# when nothing is in flight.
if body_error is None:
raise
finally:
# Finalize local history and clear the active run even
# when the remote update failed. This is what keeps a
# Hub failure from being reported as misuse later: leave
# the run active and the next `start_run` would raise
# "already active" for something the caller did right.
if self._current_run_log is not None:
self._current_run_log.ended_at = ended_at
self._run_log_history.append(self._current_run_log)
self._current_run_log = None
self._active_run = None
[docs]
@_degrades_when_safe("log a parameter")
def log_param(
self, name: str, value: str, *, save_to_run_log: bool = True
) -> LoggedParam | None:
"""Log a parameter for the current run.
:param name: The parameter name.
:param value: The parameter value.
:param save_to_run_log: Whether to also record this entry in the
current :class:`RunLog`. Defaults to ``True``.
:return: The :class:`LoggedParam` that was recorded, or ``None``
if it failed inside a :meth:`safe_log` scope.
"""
project = self.project
active_run = self.active_run
stored = log_param(
self.api_config,
name=name,
value=value,
project_id=project.id,
run_id=active_run.id,
)
logged_param = LoggedParam(
name=name,
# The server may normalize (truncate) the value; record
# what was actually stored so local history matches it.
value=stored.value,
logged_at=datetime.now(UTC),
)
# Log to the current run log
if save_to_run_log and self._current_run_log is not None:
self._current_run_log.add_param(logged_param)
return logged_param
[docs]
@_degrades_when_safe("log a metric")
def log_metric(
self,
name: str,
value: float,
step: int | None = None,
*,
save_to_run_log: bool = True,
) -> LoggedMetric | None:
"""Log a metric for the current run.
:param name: The metric name.
:param value: The metric value.
:param step: Optional step number.
:param save_to_run_log: Whether to also record this entry in the
current :class:`RunLog`. Defaults to ``True``.
:return: The :class:`LoggedMetric` that was recorded, or ``None``
if it failed inside a :meth:`safe_log` scope.
"""
project = self.project
active_run = self.active_run
log_metric(
self.api_config,
name=name,
value=value,
step=step,
project_id=project.id,
run_id=active_run.id,
)
logged_metric = LoggedMetric(
name=name,
value=value,
step=step,
logged_at=datetime.now(UTC),
)
# Log to the current run log
if save_to_run_log and self._current_run_log is not None:
self._current_run_log.add_metric(logged_metric)
return logged_metric
[docs]
@_degrades_when_safe("log an external link")
def log_link(
self, label: str, url: str, *, save_to_run_log: bool = True
) -> LoggedExternalLink | None:
"""Log an external link for the current run.
:param label: The link label.
:param url: The link target.
:param save_to_run_log: Whether to also record this entry in the
current :class:`RunLog`. Defaults to ``True``.
:return: The :class:`LoggedExternalLink` that was recorded, or
``None`` if it failed inside a :meth:`safe_log` scope.
"""
project = self.project
active_run = self.active_run
log_external_link(
self.api_config,
label=label,
url=url,
project_id=project.id,
run_id=active_run.id,
)
logged_link = LoggedExternalLink(
label=label,
url=url,
logged_at=datetime.now(UTC),
)
if save_to_run_log and self._current_run_log is not None:
self._current_run_log.add_external_link(logged_link)
return logged_link
[docs]
@_degrades_when_safe("log a batch of metrics and parameters")
def log_batch(
self,
*,
metrics: list[tuple[str, float, int | None]] | None = None,
params: list[tuple[str, str]] | None = None,
external_links: list[tuple[str, str]] | None = None,
save_to_run_log: bool = True,
) -> None:
"""Log multiple metrics and params in a single API call.
This is significantly more efficient than calling
:meth:`log_metric` and :meth:`log_param` in a loop, as it
sends all data in one PATCH request instead of one POST per
item.
:param metrics: List of ``(name, value, step)`` tuples.
`step` may be ``None`` for metrics without a step.
:param params: List of ``(name, value)`` tuples.
:param external_links: List of ``(label, url)`` tuples.
:param save_to_run_log: Whether to also record entries in the
current :class:`RunLog`. Defaults to ``True``.
"""
_validate_batch_shapes(metrics, params, external_links)
api_metrics = [
Metric(name=name, value=value, step=step)
for name, value, step in (metrics or [])
]
api_params = [
Parameter(name=name, value=value) for name, value in (params or [])
]
api_external_links = [
ExternalLink(label=label, url=url)
for label, url in (external_links or [])
]
if api_metrics or api_params or api_external_links:
self._update_active_run(
metrics=api_metrics or None,
params=api_params or None,
external_links=api_external_links or None,
)
# Mirror into the local run log
if save_to_run_log and self._current_run_log is not None:
now = datetime.now(UTC)
for m_name, m_value, m_step in metrics or []:
self._current_run_log.add_metric(
LoggedMetric(
name=m_name, value=m_value, step=m_step, logged_at=now
)
)
for p_name, p_value in params or []:
self._current_run_log.add_param(
LoggedParam(name=p_name, value=p_value, logged_at=now)
)
for label, url in external_links or []:
self._current_run_log.add_external_link(
LoggedExternalLink(
label=label,
url=url,
logged_at=now,
)
)
[docs]
@_degrades_when_safe("log a tag")
def log_tag(self, name: str, value: str) -> Tag | None:
"""Log a tag for the current run.
:param name: The tag name.
:param value: The tag value.
:return: The :class:`Tag` that was recorded, or ``None`` if it
failed inside a :meth:`safe_log` scope.
"""
project = self.project
run_id = self.active_run.id
tag = log_tag(
self.api_config,
name=name,
value=value,
project_id=project.id,
run_id=run_id,
)
return tag
[docs]
@_degrades_when_safe("log a dashboard")
def log_dashboard(
self,
dashboard: Dashboard | Path | str,
*,
name: str | None = None,
description: str | None = None,
default: bool = False,
opens_run: bool = False,
) -> RunDashboard | None:
"""Attach a dashboard to the current run.
Accepts a :class:`~embedl_hub.tracking.dashboard.Dashboard` or
the path of a file written by
:meth:`~embedl_hub.tracking.dashboard.Dashboard.save`.
:param dashboard: The dashboard, or the path of its JSON file.
:param name: The name to store it under on the run. Overrides
the name the dashboard carries; required for a file that
holds a bare config.
:param description: Overrides the dashboard's description.
:param default: Makes it the run's default dashboard, the one
the run's Dashboard tab opens on. A run has one.
:param opens_run: Opens the run itself on it rather than on the
run's details. Implies ``default``, so there is no need to
pass both.
:return: The :class:`RunDashboard` the Hub stored, or ``None``
if it failed inside a :meth:`safe_log` scope.
:raises DashboardSchemaError: If the file is not a dashboard this
SDK reads, the dashboard ends up without a name or with one
the Hub would not store, or its sections and panels do not
form a valid layout. A broken
input rather than a Hub failure, so it is raised even inside
a :meth:`safe_log` scope.
:raises FileNotFoundError: If there is no file at the given path.
Degrades to a warning inside a :meth:`safe_log` scope.
"""
if isinstance(dashboard, Dashboard):
overrides = {
key: value
for key, value in (
("name", name),
("description", description),
)
if value is not None
}
if overrides:
dashboard = replace(dashboard, **overrides)
else:
dashboard = Dashboard.load(
dashboard, name=name, description=description
)
project = self.project
run_id = self.active_run.id
body = dashboard.to_dict()
if default or opens_run:
body["default"] = True
if opens_run:
body["opensRun"] = True
return create_run_dashboard(
self.api_config,
project_id=project.id,
run_id=run_id,
dashboard=body,
)
[docs]
@_degrades_when_safe("log an artifact")
def log_artifact(
self,
file_path: Path | str | LoggedArtifact,
name: str | None = None,
file_name: str | None = None,
run_id: str | None = None,
*,
ctx: "HubContext | None" = None,
device_name: str | None = None,
save_to_run_log: bool = True,
) -> LoggedArtifact | None:
"""
Log an artifact file for a run and upload it to artifact storage.
Accepts a local path, a remote path, or an existing
:class:`LoggedArtifact`. Remote artifacts are fetched from the
device and uploaded to the Hub when :attr:`log_remote_artifacts`
is ``True`` (the default). Set it to ``False`` to skip remote
uploads and speed up execution.
If no `name` is given, the artifact is logged as unnamed.
If no `file_name` is given, the name from the file path is used.
If no `run_id` is given, the current active run is used.
:param file_path: Local file path, remote path, or a
:class:`LoggedArtifact`.
:param name: Optional artifact name.
:param file_name: Optional file name override.
:param run_id: Optional run ID (defaults to the active run).
:param ctx: Hub context, required when `file_path` is remote or
a :class:`LoggedArtifact`.
:param device_name: Name of the device that owns the remote
artifact. Required for remote paths; ignored for
:class:`LoggedArtifact` (uses its own device info).
:param save_to_run_log: Whether to also record the artifact in
the current :class:`RunLog`.
:return: The :class:`LoggedArtifact` that was recorded, or
``None`` if it failed inside a :meth:`safe_log` scope.
:raises RuntimeError: If a remote artifact is received without
a `ctx`, or a remote path is received without `device_name`.
:raises TrackingUsageError: If `file_name` is given but blank —
a broken caller rather than a Hub failure, so it is raised
even inside a :meth:`safe_log` scope.
:raises FileTooLargeError: If the file exceeds the maximum
allowed upload size.
:raises StorageQuotaExceededError: If the storage quota has been
exceeded.
:raises ArtifactUploadError: If the file upload fails.
"""
if file_name is not None and not file_name.strip():
# A blank override is a broken caller — usually a shell
# capture upstream that produced nothing — not a request for
# the default. Marked as misuse so a safe scope cannot hide
# it; pass None to get the default name.
raise TrackingUsageError(
"file_name is blank. Pass None to use the file's own "
"name, or a non-blank name to override it."
)
# -- Handle LoggedArtifact input --------------------------------
if isinstance(file_path, LoggedArtifact):
return self._log_logged_artifact(
file_path,
name=name,
file_name=file_name,
run_id=run_id,
ctx=ctx,
save_to_run_log=save_to_run_log,
)
# -- Handle remote path -----------------------------------------
# RemotePath is PurePosixPath; on Linux, Path is also a
# PurePosixPath subclass. Check for concrete Path first so
# that local paths are not misidentified as remote.
if isinstance(file_path, RemotePath) and not isinstance(
file_path, Path
):
return self._log_remote_path(
file_path,
name=name,
file_name=file_name,
run_id=run_id,
ctx=ctx,
device_name=device_name,
save_to_run_log=save_to_run_log,
)
# -- Local path (original behaviour) ----------------------------
return self._upload_and_record(
Path(file_path),
name=name,
file_name=file_name,
run_id=run_id,
save_to_run_log=save_to_run_log,
)
# -- log_artifact private helpers -----------------------------------
def _log_logged_artifact(
self,
artifact: LoggedArtifact,
*,
name: str | None,
file_name: str | None,
run_id: str | None,
ctx: "HubContext | None",
save_to_run_log: bool,
) -> LoggedArtifact:
"""Handle ``log_artifact`` when the input is a :class:`LoggedArtifact`."""
if ctx is None:
raise TrackingUsageError(
"A 'ctx' (HubContext) is required when logging a "
"LoggedArtifact. Pass the current context so the "
"artifact's device information can be resolved."
)
effective_name = name if name is not None else artifact.name
effective_file_name = (
file_name if file_name is not None else artifact.file_name
)
if artifact.is_local:
return self._upload_and_record(
artifact.local(),
name=effective_name,
file_name=effective_file_name,
run_id=run_id,
save_to_run_log=save_to_run_log,
)
# Remote LoggedArtifact
if not self._log_remote_artifacts:
logger.debug(
"Skipping remote artifact '%s' (log_remote_artifacts=False).",
effective_name or effective_file_name,
)
return artifact
device_log = artifact.device
prefix = (
f"remote_{device_log.name}__"
if device_log is not None
else "remote__"
)
prefixed_name = (
f"{prefix}{effective_name}" if effective_name is not None else None
)
prefixed_file_name = f"{prefix}{effective_file_name}"
local_artifact = artifact.to_local(ctx)
return self._upload_and_record(
local_artifact.local(),
name=prefixed_name,
file_name=prefixed_file_name,
run_id=run_id,
save_to_run_log=save_to_run_log,
)
def _log_remote_path(
self,
file_path: RemotePath,
*,
name: str | None,
file_name: str | None,
run_id: str | None,
ctx: "HubContext | None",
device_name: str | None,
save_to_run_log: bool,
) -> LoggedArtifact:
"""Handle ``log_artifact`` when the input is a :class:`RemotePath`."""
if ctx is None:
raise TrackingUsageError(
"A 'ctx' (HubContext) is required when logging a "
"remote artifact path. Pass the current context so "
"the file can be fetched from the device."
)
if device_name is None:
raise TrackingUsageError(
"A 'device_name' is required when logging a remote "
"artifact path so the correct device can be resolved."
)
# Validate device exists in context
if device_name not in ctx.devices:
available = ", ".join(sorted(ctx.devices.keys())) or "(none)"
raise TrackingUsageError(
f"Device '{device_name}' not found in context. "
f"Available devices: {available}."
)
effective_file_name = (
file_name if file_name is not None else file_path.name
)
if not self._log_remote_artifacts:
logger.debug(
"Skipping remote artifact '%s' (log_remote_artifacts=False).",
name or effective_file_name or str(file_path),
)
return LoggedArtifact(
id="",
file_name=effective_file_name,
file_size=0,
logged_at=datetime.now(UTC),
file_path=file_path,
name=name,
)
prefix = f"remote_{device_name}__"
prefixed_name = f"{prefix}{name}" if name is not None else None
prefixed_file_name = f"{prefix}{effective_file_name}"
# Build a temporary LoggedArtifact to leverage to_local()
from embedl_hub._internal.tracking.run_log import (
_resolve_device_by_name,
)
device_obj, _ = _resolve_device_by_name(ctx, device_name)
temp_artifact = LoggedArtifact(
id="",
file_name=effective_file_name,
file_size=0,
logged_at=datetime.now(UTC),
file_path=file_path,
device=device_obj.create_device_log(device_name),
)
local_artifact = temp_artifact.to_local(ctx)
return self._upload_and_record(
local_artifact.local(),
name=prefixed_name,
file_name=prefixed_file_name,
run_id=run_id,
save_to_run_log=save_to_run_log,
)
def _upload_and_record(
self,
file_path: Path,
*,
name: str | None,
file_name: str | None,
run_id: str | None,
save_to_run_log: bool,
) -> LoggedArtifact:
"""Upload a local file to the hub and record it in the run log."""
if file_name is None:
file_name = file_path.name
file_size = file_path.stat().st_size
run_id = run_id or self.active_run.id
artifact = upload_artifact_file(
self.api_config, run_id, file_path, file_name
)
logged_artifact = LoggedArtifact(
id=artifact.id,
file_name=file_name,
file_size=file_size,
logged_at=datetime.now(UTC),
file_path=file_path,
name=name,
artifact_dir=(
self._current_run_log.artifact_dir
if self._current_run_log is not None
else None
),
)
# Log to the current run log
if save_to_run_log and self._current_run_log is not None:
self._current_run_log.add_artifact(logged_artifact)
return logged_artifact
[docs]
def get_devices(self) -> list[Device]:
"""Get the list of supported devices in the Embedl device cloud."""
devices = get_devices(self.api_config)
return devices
[docs]
def validate_device(self, device: str) -> Device:
"""Check if the specified device is supported in the Embedl device cloud.
:param device: The device name to check.
:return: The :class:`Device` if supported.
:raises UnsupportedDeviceError: If `device` is not in the
supported device list.
"""
supported_devices = self.get_devices()
found_device = None
for d in supported_devices:
if d.name == device:
found_device = d
break
if found_device is None:
raise UnsupportedDeviceError(device)
return found_device
[docs]
def list_runs(
self,
*,
params: Mapping[str, str] | None = None,
statuses: RunStatus | Sequence[RunStatus] | None = None,
types: RunType | str | Sequence[RunType | str] | None = None,
limit: int | None = None,
offset: int | None = None,
) -> RunListPage:
"""List the current project's runs, newest first.
Find the run a job logged its own ID on, to link the next job
to it::
page = client.list_runs(params={"vertex_ai_job_id": job_id})
parent = page.items[0] if page.items else None
Like :meth:`get_devices`, a failure raises rather than warns.
:param params: Parameters the runs must have logged, as name to
value. Names and values are normalized the way
:meth:`log_param` stores them, so what was logged is found
again.
:param statuses: Only runs with this status, or one of these.
:param types: Only runs of this type, or one of these. A string
is read as :meth:`start_run` reads it: a built-in type name
selects that type, any other label selects the custom runs
filed under it. Built-in types and labels cannot be
combined in one call.
:param limit: Page size; the hub's default and cap apply.
:param offset: Runs to skip, for paging.
:return: One page of run summaries with pagination info.
:raises ValueError: If a run type label is blank or contains a
comma, built-in types are mixed with labels, a parameter
name is blank, a parameter name or value contains a NUL
character, or two parameter names are the same once trimmed.
:raises RuntimeError: If no project is set.
:raises ApiError: If the hub rejects the request.
"""
if isinstance(statuses, RunStatus):
statuses = [statuses]
if isinstance(types, (RunType, str)):
types = [types]
run_types, custom_types = _split_run_types(types)
normalized = _normalize_param_filters(params) if params else None
project = self.project
return list_runs(
self.api_config,
project.id,
limit=limit,
offset=offset,
statuses=list(statuses) if statuses else None,
types=run_types,
custom_types=custom_types,
params=normalized,
)
[docs]
def get_run_project(self, run_id: str) -> Project:
"""Look up the project a run belongs to.
Needs no project selected first, so a parent run ID is enough to
open the next run in the same project. Select the project with
:meth:`use_project` rather than by name: names are unique per
owner only, so a project shared with you may share its name with
one of yours::
project = client.get_run_project(parent_run_id)
client.use_project(project)
with client.start_run("train", parent_run_id=parent_run_id):
...
Like :meth:`get_devices`, a failure raises rather than warns.
:param run_id: The run's ID.
:return: The project the run belongs to.
:raises ValueError: If `run_id` is blank.
:raises ApiError: If the hub does not know the run, or it is in
a project this API key cannot read; either is a 404.
"""
if not run_id.strip():
raise ValueError(
'The run ID is blank. Pass the ID of a run, e.g. run_id="abc123". '
"A blank ID usually means a shell capture or an environment "
"variable that expanded to nothing."
)
return resolve_run(self.api_config, run_id).project
@property
def api_config(self) -> ApiConfig:
"""Get or create the API config from the resolved settings.
:return: The credentials and Hub address this client uses.
:raises RuntimeError: If no API key is configured.
"""
if self._api_config is None:
api_key = resolve_api_key()
if api_key is None:
raise RuntimeError(
"No API key found. Run "
"'embedl-hub auth --api-key <key>' or set the "
f"{API_KEY_ENV_VAR_NAME} environment variable."
)
self._api_key_source = api_key.source
self._api_config = ApiConfig(
api_key=api_key.value, base_url=resolve_base_url().value
)
return self._api_config
@property
def base_url(self) -> str:
"""The address of the Hub this client sends requests to.
Needs no API key: which Hub is configured is not an
authentication question, so this answers before ``auth`` has
been run.
:return: The address, always ending in ``/``.
"""
if self._api_config is not None:
return self._api_config.base_url
return resolve_base_url().value
@property
def api_key(self) -> str:
"""The API key this client authenticates with.
For handing to something else that has to authenticate. Do not
print or log it: a key that reaches a terminal, a notebook
output or a CI log has to be treated as compromised.
:attr:`api_key_source` answers where the key comes from without
exposing it.
:return: The key in use.
:raises RuntimeError: If no API key is configured.
"""
return self.api_config.api_key
@property
def api_key_source(self) -> SettingSource | None:
"""Which layer supplied the API key this client uses.
Names where the key was read from — the environment or the
config file — rather than the key itself, so it is safe to
print. Use :attr:`api_key` for the value.
:return: ``"env"`` or ``"config"`` for a key this client
resolved, or ``None`` when no layer supplied one: either
nothing is configured, or the client was handed an explicit
configuration, which comes from the caller rather than from
anywhere on this machine.
"""
if self._api_config is not None:
return self._api_key_source
resolved = resolve_api_key()
return resolved.source if resolved else None
@property
def project(self) -> Project:
if self._project is None:
raise RuntimeError("Project is not set. Use set_project() first.")
return self._project
@property
def active_run(self) -> Run:
if self._active_run is None:
raise RuntimeError(
"There is no active run. Use start_run() as a context manager first."
)
return self._active_run
@property
def run_history(self) -> RunLogHistory:
"""Get the run history (read-only).
:return: A :class:`RunLogHistory` containing all completed
:class:`RunLog` entries. The history cannot be modified.
"""
return RunLogHistory(self._run_log_history)
@property
def latest_run_log(self) -> RunLog | None:
"""Get the most recent completed :class:`RunLog`, or ``None`` if no
runs have completed.
"""
if not self._run_log_history:
return None
return self._run_log_history[-1]
@property
def current_run_log(self) -> RunLog | None:
"""Get the current in-progress :class:`RunLog`, or ``None`` if no
run is active.
"""
return self._current_run_log
