tf.keras.utils.TimedThread

Time-based interval Threads.

tf.keras.utils.TimedThread(
    interval, **kwargs
)

Runs a timed thread every x seconds. It can be used to run a threaded function alongside model training or any other snippet of code.

interval The interval, in seconds, to wait between calls to the on_interval function.
**kwargs additional args that are passed to threading.Thread. By default, Thread is started as a daemon thread unless overridden by the user in kwargs.

Examples:

class TimedLogIterations(keras.utils.TimedThread):
    def __init__(self, model, interval):
        self.model = model
        super().__init__(interval)

    def on_interval(self):
        # Logs Optimizer iterations every x seconds
        try:
            opt_iterations = self.model.optimizer.iterations.numpy()
            print(f"Epoch: {epoch}, Optimizer Iterations: {opt_iterations}")
        except Exception as e:
            print(str(e))  # To prevent thread from getting killed

# `start` and `stop` the `TimerThread` manually. If the `on_interval` call
# requires access to `model` or other objects, override `__init__` method.
# Wrap it in a `try-except` to handle exceptions and `stop` the thread run.
timed_logs = TimedLogIterations(model=model, interval=5)
timed_logs.start()
try:
    model.fit(...)
finally:
    timed_logs.stop()

# Alternatively, run the `TimedThread` in a context manager
with TimedLogIterations(model=model, interval=5):
    model.fit(...)

# If the timed thread instance needs access to callback events,
# subclass both `TimedThread` and `Callback`.  Note that when calling
# `super`, they will have to called for each parent class if both of them
# have the method that needs to be run. Also, note that `Callback` has
# access to `model` as an attribute and need not be explictly provided.
class LogThreadCallback(
    keras.utils.TimedThread, keras.callbacks.Callback
):
    def __init__(self, interval):
        self._epoch = 0
        keras.utils.TimedThread.__init__(self, interval)
        keras.callbacks.Callback.__init__(self)

    def on_interval(self):
        if self.epoch:
            opt_iter = self.model.optimizer.iterations.numpy()
            logging.info(f"Epoch: {self._epoch}, Opt Iteration: {opt_iter}")

    def on_epoch_begin(self, epoch, logs=None):
        self._epoch = epoch

with LogThreadCallback(interval=5) as thread_callback:
    # It's required to pass `thread_callback` to also `callbacks` arg of
    # `model.fit` to be triggered on callback events.
    model.fit(..., callbacks=[thread_callback])

Methods

is_alive

View source

is_alive()

Returns True if thread is running. Otherwise returns False.

on_interval

View source

@abc.abstractmethod
on_interval()

User-defined behavior that is called in the thread.

start

View source

start()

Creates and starts the thread run.

stop

View source

stop()

Stops the thread run.

__enter__

View source

__enter__()

__exit__

View source

__exit__(
    *args, **kwargs
)