discrete1.utils.decorators module

Decorators for stopping for-loops early.

Allow for decorating ml/tune and ml/train and stopping long running jobs while saving the current information with Ctrl+C.

discrete1.utils.decorators.break_loop(func)

Interrupt-friendly decorator.

The wrapped function receives the wrapper itself as its first argument, allowing it to query wrapper.stop_event (a threading.Event) to terminate a long‑running loop after the current iteration when the user presses Ctrl+C once. A second Ctrl+C aborts immediately.

Parameters:

func (callable) –

The function to be wrapped. It must accept the wrapper instance as its first positional argument, e.g.:

@break_loop
def process(wrapper, data):
    for i in grange(len(data), stop_event=wrapper.stop_event):
        # do work ...

Returns:

The wrapped function. The wrapper has an attribute stop_event (a threading.Event) that can be inspected inside func.

Return type:

callable

Notes

• The decorator temporarily replaces the process‑wide SIGINT handler. The original handler is restored after the function finishes (or raises).

• wrapper.stop_event is cleared before each call so that the same decorated function can be invoked repeatedly.

Examples

>>> @break_loop
... def count(wrapper, n):
...     for i in grange(n, stop_event=wrapper.stop_event):
...         print(i)
...         time.sleep(0.5)      # simulate work
...
>>> count(10)         # Press Ctrl+C once to stop after the current
...                   # iteration; press twice to raise KeyboardInterrupt

discrete1.utils.decorators.finish_iteration(func)

Interrupt-friendly decorator.

The first Ctrl+C (SIGINT) request sets a flag; after the wrapped function returns, a KeyboardInterrupt is raised if the flag was set. A second Ctrl+C raises KeyboardInterrupt immediately.

Parameters:

func (callable) – The function to be wrapped. Its signature is unchanged.

Returns:

The wrapped function that respects the graceful‑interrupt semantics.

Return type:

callable

Notes

• The decorator temporarily replaces the process‑wide SIGINT handler. The original handler is restored after func completes.

• If the wrapped function itself raises KeyboardInterrupt, the exception propagates unchanged.

Examples

>>> @finish_iteration
... def heavy_task():
...     for i in range(100):
...         print(i)
...         time.sleep(0.1)
...
>>> heavy_task()              # Press Ctrl+C once to finish the current loop
...                           # iteration then raise KeyboardInterrupt.

discrete1.utils.decorators.grange(*args, stop_event)

Interrupt-friendly generator (range type).

Parameters:

• *args (int) – The same positional arguments accepted by the built‑in range(). Typical signatures are stop, start, stop or start, stop, step.

• stop_event (threading.Event) – An event object whose is_set method is consulted before each iteration. If the event is set, the generator stops early.

Yields:

int – Values produced by the underlying range call.

Raises:

TypeError – If stop_event is not provided as a keyword argument.

Notes

This helper is intended to be used together with a function wrapped by break_loop(), which supplies the stop_event tied to a SIGINT handler.

Examples

>>> ev = threading.Event()
>>> for i in grange(0, 10, 2, stop_event=ev):
...     print(i)
0
2
4
6
8