utils API reference

Reference documentation for the utils module of the Neptune API.

Theutilsmodule contains utility functions for logging metadata with the Neptune API.

from neptune import utils

stop_synchronization_callback()

Default callback function that is called if the synchronization lags too much or has not progressed at all.

The function prints a message and stops the synchronization with the Neptune servers. In practice, thestop()function is called and no more operations will be queued, but the synchronization will continue for 60 seconds.

def stop_synchronization_callback(neptune_object: NeptuneObject) -> None:
 logger.error(
 "Threshold for disrupted synchronization exceeded. Stopping the synchronization using the default callback."
 )
 neptune_object.stop(seconds=60.0)

You can use this as inspiration if you want to create your own custom callback function.

For more instructions, seeHow to use callbacks when sync is interrupted.

Usage

There are two ways to enable this default callback:

• Setting one or both of the callback environment variables ( NEPTUNE_ENABLE_DEFAULT_ASYNC_LAG_CALLBACK or NEPTUNE_ENABLE_DEFAULT_ASYNC_NO_PROGRESS_CALLBACK ) to TRUE .
• (not needed if using the environment variables) Passing this function to either the async_lag_callback or async_no_progress_callback argument when initializing a run or other object.

The callback is then triggered if the threshold for the lag or lack of progress is exceeded. You can override the default thresholds with theasync_lag_thresholdandasync_no_progress_thresholdinitialization parameters, respectively.

Examples

import neptune
from neptune.utils import stop_synchronization_callback

run = neptune.init_run(async_no_progress_callback = stop_synchronization_callback)

import neptune
from neptune.utils import stop_synchronization_callback

run = neptune.init_run(async_lag_callback = stop_synchronization_callback)

import neptune
from neptune.utils import stop_synchronization_callback

run = neptune.init_run(
 async_no_progress_callback = stop_synchronization_callback,
 async_lag_callback = stop_synchronization_callback,
)

Note:You do not need to use the aboveinit_run()arguments if you set the corresponding environment variables.

Related

How to use callbacks when sync is interrupted

stringify_unsupported()

Helper function that converts unsupported values in a collection or dictionary to strings.

The output of this function is lazy-evaluated, as the final type depends on the logging method that is called on the stringified object (for example,assign(obj)versusappend(obj)).

Parameters

Examples

Start a run:

>>> import neptune
>>> run = neptune.init_run()

Log a complex dictionary with partially unsupported types:

>>> complex_dict = {"tuple": ("learning_rate", 0.01), "accuracy": 0.87} # (1)!
>>> from neptune.utils import stringify_unsupported
>>> run["complex_dict"] = stringify_unsupported(complex_dict)

1. No error occurs until you upgrade to 1.0. Until then, the tuple is implicitly cast to a string. With neptune 1.0, you need to explicitly log the unsupported type as a string, or use the stringify_unsupported() function as in this example.

When logging lists, sets, or tuples, you can use theexpand=Trueargument to store the keys and values in their proper folder structure in the Neptune run:

>>> run["complex_dict"] = stringify_unsupported(complex_dict, expand=True)

This makes it more convenient to preview the data in the Neptune app and download it later.

You can also use this function to log math constants (math.informath.nan) as strings:

>>> import math
>>> value = math.inf
>>> run["inf"] = stringify_unsupported(value)

>>> value = float(run["inf"].fetch())

Related

This function is related to the deprecation of implicit casting to string.

For details, see theneptune 1.0 upgrade guide.

This page is originally sourced from the legacy docs.