Shortcuts

Source code for pytorch_lightning.loggers.neptune

# Copyright The PyTorch Lightning team.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Neptune Logger
--------------
"""
import logging
from argparse import Namespace
from typing import Any, Dict, Iterable, Optional, Union

import torch
from torch import is_tensor

from pytorch_lightning.loggers.base import LightningLoggerBase, rank_zero_experiment
from pytorch_lightning.utilities import _module_available, rank_zero_only

log = logging.getLogger(__name__)
_NEPTUNE_AVAILABLE = _module_available("neptune")

if _NEPTUNE_AVAILABLE:
    import neptune
    from neptune.experiments import Experiment
else:
    # needed for test mocks, these tests shall be updated
    neptune, Experiment = None, None


[docs]class NeptuneLogger(LightningLoggerBase): r""" Log using `Neptune <https://neptune.ai>`_. Install it with pip: .. code-block:: bash pip install neptune-client The Neptune logger can be used in the online mode or offline (silent) mode. To log experiment data in online mode, :class:`NeptuneLogger` requires an API key. In offline mode, the logger does not connect to Neptune. **ONLINE MODE** .. testcode:: from pytorch_lightning import Trainer from pytorch_lightning.loggers import NeptuneLogger # arguments made to NeptuneLogger are passed on to the neptune.experiments.Experiment class # We are using an api_key for the anonymous user "neptuner" but you can use your own. neptune_logger = NeptuneLogger( api_key="ANONYMOUS", project_name="shared/pytorch-lightning-integration", experiment_name="default", # Optional, params={"max_epochs": 10}, # Optional, tags=["pytorch-lightning", "mlp"], # Optional, ) trainer = Trainer(max_epochs=10, logger=neptune_logger) **OFFLINE MODE** .. testcode:: from pytorch_lightning.loggers import NeptuneLogger # arguments made to NeptuneLogger are passed on to the neptune.experiments.Experiment class neptune_logger = NeptuneLogger( offline_mode=True, project_name="USER_NAME/PROJECT_NAME", experiment_name="default", # Optional, params={"max_epochs": 10}, # Optional, tags=["pytorch-lightning", "mlp"], # Optional, ) trainer = Trainer(max_epochs=10, logger=neptune_logger) Use the logger anywhere in you :class:`~pytorch_lightning.core.lightning.LightningModule` as follows: .. code-block:: python class LitModel(LightningModule): def training_step(self, batch, batch_idx): # log metrics self.logger.experiment.log_metric("acc_train", ...) # log images self.logger.experiment.log_image("worse_predictions", ...) # log model checkpoint self.logger.experiment.log_artifact("model_checkpoint.pt", ...) self.logger.experiment.whatever_neptune_supports(...) def any_lightning_module_function_or_hook(self): self.logger.experiment.log_metric("acc_train", ...) self.logger.experiment.log_image("worse_predictions", ...) self.logger.experiment.log_artifact("model_checkpoint.pt", ...) self.logger.experiment.whatever_neptune_supports(...) If you want to log objects after the training is finished use ``close_after_fit=False``: .. code-block:: python neptune_logger = NeptuneLogger(..., close_after_fit=False, ...) trainer = Trainer(logger=neptune_logger) trainer.fit() # Log test metrics trainer.test(model) # Log additional metrics from sklearn.metrics import accuracy_score accuracy = accuracy_score(y_true, y_pred) neptune_logger.experiment.log_metric("test_accuracy", accuracy) # Log charts from scikitplot.metrics import plot_confusion_matrix import matplotlib.pyplot as plt fig, ax = plt.subplots(figsize=(16, 12)) plot_confusion_matrix(y_true, y_pred, ax=ax) neptune_logger.experiment.log_image("confusion_matrix", fig) # Save checkpoints folder neptune_logger.experiment.log_artifact("my/checkpoints") # When you are done, stop the experiment neptune_logger.experiment.stop() See Also: - An `Example experiment <https://ui.neptune.ai/o/shared/org/ pytorch-lightning-integration/e/PYTOR-66/charts>`_ showing the UI of Neptune. - `Tutorial <https://docs.neptune.ai/integrations/pytorch_lightning.html>`_ on how to use Pytorch Lightning with Neptune. Args: api_key: Required in online mode. Neptune API token, found on https://neptune.ai. Read how to get your `API key <https://docs.neptune.ai/python-api/tutorials/get-started.html#copy-api-token>`_. It is recommended to keep it in the `NEPTUNE_API_TOKEN` environment variable and then you can leave ``api_key=None``. project_name: Required in online mode. Qualified name of a project in a form of "namespace/project_name" for example "tom/minst-classification". If ``None``, the value of `NEPTUNE_PROJECT` environment variable will be taken. You need to create the project in https://neptune.ai first. offline_mode: Optional default ``False``. If ``True`` no logs will be sent to Neptune. Usually used for debug purposes. close_after_fit: Optional default ``True``. If ``False`` the experiment will not be closed after training and additional metrics, images or artifacts can be logged. Also, remember to close the experiment explicitly by running ``neptune_logger.experiment.stop()``. experiment_name: Optional. Editable name of the experiment. Name is displayed in the experiment’s Details (Metadata section) and in experiments view as a column. experiment_id: Optional. Default is ``None``. The ID of the existing experiment. If specified, connect to experiment with experiment_id in project_name. Input arguments "experiment_name", "params", "properties" and "tags" will be overriden based on fetched experiment data. prefix: A string to put at the beginning of metric keys. \**kwargs: Additional arguments like `params`, `tags`, `properties`, etc. used by :func:`neptune.Session.create_experiment` can be passed as keyword arguments in this logger. Raises: ImportError: If required Neptune package is not installed on the device. """ LOGGER_JOIN_CHAR = "-" def __init__( self, api_key: Optional[str] = None, project_name: Optional[str] = None, close_after_fit: Optional[bool] = True, offline_mode: bool = False, experiment_name: Optional[str] = None, experiment_id: Optional[str] = None, prefix: str = "", **kwargs, ): if neptune is None: raise ImportError( "You want to use `neptune` logger which is not installed yet," " install it with `pip install neptune-client`." ) super().__init__() self.api_key = api_key self.project_name = project_name self.offline_mode = offline_mode self.close_after_fit = close_after_fit self.experiment_name = experiment_name self._prefix = prefix self._kwargs = kwargs self.experiment_id = experiment_id self._experiment = None log.info(f'NeptuneLogger will work in {"offline" if self.offline_mode else "online"} mode') def __getstate__(self): state = self.__dict__.copy() # Experiment cannot be pickled, and additionally its ID cannot be pickled in offline mode state["_experiment"] = None if self.offline_mode: state["experiment_id"] = None return state @property @rank_zero_experiment def experiment(self) -> Experiment: r""" Actual Neptune object. To use neptune features in your :class:`~pytorch_lightning.core.lightning.LightningModule` do the following. Example:: self.logger.experiment.some_neptune_function() """ # Note that even though we initialize self._experiment in __init__, # it may still end up being None after being pickled and un-pickled if self._experiment is None: self._experiment = self._create_or_get_experiment() return self._experiment
[docs] @rank_zero_only def log_hyperparams(self, params: Union[Dict[str, Any], Namespace]) -> None: params = self._convert_params(params) params = self._flatten_dict(params) for key, val in params.items(): self.experiment.set_property(f"param__{key}", val)
[docs] @rank_zero_only def log_metrics(self, metrics: Dict[str, Union[torch.Tensor, float]], step: Optional[int] = None) -> None: """ Log metrics (numeric values) in Neptune experiments. Args: metrics: Dictionary with metric names as keys and measured quantities as values step: Step number at which the metrics should be recorded, currently ignored """ assert rank_zero_only.rank == 0, "experiment tried to log from global_rank != 0" metrics = self._add_prefix(metrics) for key, val in metrics.items(): # `step` is ignored because Neptune expects strictly increasing step values which # Lighting does not always guarantee. self.log_metric(key, val)
[docs] @rank_zero_only def finalize(self, status: str) -> None: super().finalize(status) if self.close_after_fit: self.experiment.stop()
@property def save_dir(self) -> Optional[str]: # Neptune does not save any local files return None @property def name(self) -> str: if self.offline_mode: return "offline-name" return self.experiment.name @property def version(self) -> str: if self.offline_mode: return "offline-id-1234" return self.experiment.id
[docs] @rank_zero_only def log_metric( self, metric_name: str, metric_value: Union[torch.Tensor, float, str], step: Optional[int] = None ) -> None: """ Log metrics (numeric values) in Neptune experiments. Args: metric_name: The name of log, i.e. mse, loss, accuracy. metric_value: The value of the log (data-point). step: Step number at which the metrics should be recorded, must be strictly increasing """ if is_tensor(metric_value): metric_value = metric_value.cpu().detach() if step is None: self.experiment.log_metric(metric_name, metric_value) else: self.experiment.log_metric(metric_name, x=step, y=metric_value)
[docs] @rank_zero_only def log_text(self, log_name: str, text: str, step: Optional[int] = None) -> None: """ Log text data in Neptune experiments. Args: log_name: The name of log, i.e. mse, my_text_data, timing_info. text: The value of the log (data-point). step: Step number at which the metrics should be recorded, must be strictly increasing """ if step is None: self.experiment.log_text(log_name, text) else: self.experiment.log_text(log_name, x=step, y=text)
[docs] @rank_zero_only def log_image(self, log_name: str, image: Union[str, Any], step: Optional[int] = None) -> None: """ Log image data in Neptune experiment Args: log_name: The name of log, i.e. bboxes, visualisations, sample_images. image: The value of the log (data-point). Can be one of the following types: PIL image, `matplotlib.figure.Figure`, path to image file (str) step: Step number at which the metrics should be recorded, must be strictly increasing """ if step is None: self.experiment.log_image(log_name, image) else: self.experiment.log_image(log_name, x=step, y=image)
[docs] @rank_zero_only def log_artifact(self, artifact: str, destination: Optional[str] = None) -> None: """Save an artifact (file) in Neptune experiment storage. Args: artifact: A path to the file in local filesystem. destination: Optional. Default is ``None``. A destination path. If ``None`` is passed, an artifact file name will be used. """ self.experiment.log_artifact(artifact, destination)
[docs] @rank_zero_only def set_property(self, key: str, value: Any) -> None: """ Set key-value pair as Neptune experiment property. Args: key: Property key. value: New value of a property. """ self.experiment.set_property(key, value)
[docs] @rank_zero_only def append_tags(self, tags: Union[str, Iterable[str]]) -> None: """ Appends tags to the neptune experiment. Args: tags: Tags to add to the current experiment. If str is passed, a single tag is added. If multiple - comma separated - str are passed, all of them are added as tags. If list of str is passed, all elements of the list are added as tags. """ if str(tags) == tags: tags = [tags] # make it as an iterable is if it is not yet self.experiment.append_tags(*tags)
def _create_or_get_experiment(self): if self.offline_mode: project = neptune.Session(backend=neptune.OfflineBackend()).get_project("dry-run/project") else: session = neptune.Session.with_default_backend(api_token=self.api_key) project = session.get_project(self.project_name) if self.experiment_id is None: exp = project.create_experiment(name=self.experiment_name, **self._kwargs) self.experiment_id = exp.id else: exp = project.get_experiments(id=self.experiment_id)[0] self.experiment_name = exp.get_system_properties()["name"] self.params = exp.get_parameters() self.properties = exp.get_properties() self.tags = exp.get_tags() return exp

© Copyright Copyright (c) 2018-2023, William Falcon et al...

Built with Sphinx using a theme provided by Read the Docs.