interface.py 1.1 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849
  1. """Internal APIs for integrating with weave.
  2. The public functions here are intended to be called by weave and care should
  3. be taken to maintain backward compatibility.
  4. """
  5. from __future__ import annotations
  6. import dataclasses
  7. from wandb.sdk import wandb_setup
  8. @dataclasses.dataclass(frozen=True)
  9. class RunPath:
  10. entity: str
  11. """The entity to which the run is logging. Never empty."""
  12. project: str
  13. """The project to which the run is logging. Never empty."""
  14. run_id: str
  15. """The run's ID. Never empty."""
  16. def active_run_path() -> RunPath | None:
  17. """Returns the path of an initialized, unfinished run.
  18. Returns None if all initialized runs are finished. If there is
  19. more than one active run, an arbitrary path is returned.
  20. The run may be finished by the time its path is returned.
  21. Thread-safe.
  22. """
  23. singleton = wandb_setup.singleton()
  24. if (
  25. (run := singleton.most_recent_active_run)
  26. and run.entity
  27. and run.project
  28. and run.id
  29. ):
  30. return RunPath(
  31. entity=run.entity,
  32. project=run.project,
  33. run_id=run.id,
  34. )
  35. return None