striqt.sensor.util

class striqt.sensor.util.DebugOnException(enable: bool = False, verbose: bool = False)

Bases: object

run(etype, exc, tb)
class striqt.sensor.util.ExceptionStack(group_label: str | None = None, cancel_on_except: bool = False)

Bases: object

Creates a context manager object that accumulates exceptions.

Any exception raised within a defer() context is stashed to be raised later. An exception (possibly an ExceptionGroup) is raised when the ExceptionDeferral context exits, or on a call to handle().

defer()
handle()

raise an exception based on any deferred exceptions.

The raised exception type follows these rules: - No exceptions: no exception is raised - Exactly one exception: raise that exception - More than one exception: raise an ExceptionGroup

A ThreadInterruptRequest will only be raised if it was the only deferred exception. ThreadInterruptRequest is never included in an ExceptionGroup.

exception striqt.sensor.util.ThreadInterruptRequest

Bases: Exception

Raised in a thread to indicate the owning thread requested termination

striqt.sensor.util.await_and_ignore(futures: Iterable[concurrent.futures.Future], except_msg: str | None = None)
striqt.sensor.util.cancel_threads()
striqt.sensor.util.log_capture_context(name_suffix, /, capture_index=0, capture_count=None)
striqt.sensor.util.log_to_file(log_path: str | Path, level_name: str)
striqt.sensor.util.log_verbosity(verbose: int = 0)
striqt.sensor.util.propagate_thread_interrupts()
striqt.sensor.util.retry(excs: type[BaseException] | Iterable[type[BaseException]], tries: int, *, delay: float = 0, backoff: float = 0, exception_func=<function <lambda>>, logger: logging.Logger | logging.LoggerAdapter | None = None) → PassThroughWrapper

calls to the decorated function are repeated, suppressing specified exception(s), until a maximum number of retries has been attempted.

If the function raises the exception the specified number of times, the underlying exception is raised. Otherwise, return the result of the function call.

Example

The following retries the telnet connection 5 times on ConnectionRefusedError:

import telnetlib


# Retry a telnet connection 5 times if the telnet library raises ConnectionRefusedError
@retry(ConnectionRefusedError, tries=5)
def open(host, port):
    t = telnetlib.Telnet()
    t.open(host, port, 5)
    return t
Parameters:

  • exception_or_exceptions – Exception (sub)class (or tuple of exception classes) to watch for

  • tries – number of times to try before giving up

  • delay – initial delay between retries in seconds

  • backoff – backoff to multiply to the delay for each retry

  • exception_func – function to call on exception before the next retry

  • logger – if specified, a log info message is emitted on the first retry

striqt.sensor.util.share_thread_interrupts()
striqt.sensor.util.zip_offsets(seq: Iterable[_T], shifts: tuple[int, ...] | list[int], fill: Any, *, squeeze: bool = True) → Generator[tuple[_T, ...]]

a generator that yields from seq at multiple index shifts.

Shifts that would yield an invalid index (i.e., before or beyond the end of seq) are replaced with fill.

Parameters:

  • seq – iterable of values to shift

  • shifts – a sequence of integers indicating the index shifts to apply

  • fill – the value to yield before or after seq yields values

Yields:

tuples of length shifts composed of delayed/advanced values from seq