o i e@sJdZddlZddlZddlmZddlmZedgdGdddZdS) zThread utilities.N)logging) keras_exportzkeras.utils.TimedThread)v1c@sVeZdZdZddZddZddZdd Zd d Zd d Z ddZ e j ddZ dS) TimedThreada Time-based interval Threads. 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. Args: 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: ```python 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]) ``` cKs*||_|dd|_||_d|_d|_dS)NdaemonT)intervalpopr thread_kwargsthreadthread_stop_event)selfrkwargsrL/var/www/myenv/lib/python3.10/site-packages/keras/src/utils/timed_threads.py__init__`s  zTimedThread.__init__cCs2|js||j|j|jrdSdSN)r is_set on_intervalwaitrr rrr_call_on_intervalgs zTimedThread._call_on_intervalcCsT|jr|jrtddStjd|j|jd|j|_t |_ |j dS)z"Creates and starts the thread run.zThread is already running.N)targetrr) r is_aliverwarning threadingThreadrrr Eventr startrrrrrms  zTimedThread.startcCs|jr |jdSdS)zStops the thread run.N)r setrrrrstopzszTimedThread.stopcCs|jr|jSdS)z;Returns True if thread is running. Otherwise returns False.F)r rrrrrrs zTimedThread.is_alivecCs ||Sr)rrrrr __enter__szTimedThread.__enter__cOs |dSr)r)r argsr rrr__exit__s zTimedThread.__exit__cCstd)z3User-defined behavior that is called in the thread.zURuns every x interval seconds. Needs to be implemented in subclasses of `TimedThread`)NotImplementedErrorrrrrrszTimedThread.on_intervalN)__name__ __module__ __qualname____doc__rrrrrr r"abcabstractmethodrrrrrrsF r)r'r(rabslr tensorflow.python.util.tf_exportrrrrrrs