repype.status

class repype.status.Cursor(data: list | None = None, other: Self | None = None)

Bases: object

A cursor to navigate nested list structures.

data: list: The data structure that the cursor navigates.

find_next_child_or_sibling() → Self | None

Return a cursor to the next child or sibling.

This cursor is not changed, but a new cursor is returned.

Returns:: The cursor to the next child or sibling, if such exists, or None otherwise.

find_next_element() → Self | None

Return a cursor to the next element.

Precedentially, the next element is the next child or subling. If no next child or subling exists, then the next element is the next sibling of the parent. If no next sibling of the parent exists, then the next element is the next sibling of the grandparent, and so on.

This cursor is not changed, but a new cursor is returned.

Returns:: The cursor to the next element, if such exists, or None otherwise.

get_elements() → List[list] | None

Get the sequence of elements which represent the path to the element, that this cursor points to.

Returns:: The sequence of elements along the path to where this cursor points, if the cursor points to a valid element, or None otherwise.

has_subsequent_non_intermediate() → bool

Check if there is a subsequent non-intermediate element.

Returns:: True if calling find_next_element() once or repeatedly will yield a non-intermediate element, and False otherwise.

increment() → Self | None

Move the cursor to the next sibling.

Returns:: The cursor, if it points to a valid element, or None otherwise.

property intermediate: bool | None

Check if the cursor points to an intermediate element.

The value is None if the cursor is invalid, True if the cursor points to an intermediate element, and False otherwise.

property parent: Self: Get the cursor to the parent element.

property parents: Iterator[Self]: List of cursors to the parent elements.

path: List[int]: Sequence of elements along the path to where this cursor points, represented by the positions of the elements within the parent lists.

property valid: bool: Check if the cursor points to an existing element.

class repype.status.Status(parent: Self | None = None, path: PathLike | None = None)

Bases: object

A status object that can be used to report the progress of a computation.

Status updates should be made via the repype.status.update(), repype.status.progress(), and repype.status.derive() shortcuts. The updates can be monitored by a StatusReader object.

Status objects can be nested, so that the progress of a sub-computation can be reported within the progress of a parent computation. In addition, since each status object fosters its own status file, the amount of I/O operations required to write and read status updates is reduced.

data: list: The data structure that represents the progress of the computation.

derive() → Self: Create a child status object that is nested within this status object.

property filepath: Path: The path to the status file written by this status object.

id: UUID: The unique identifier of the status object.

intermediate(status: str | dict | None = None) → None

Write an intermediate status update to the status object.

Intermediate status updates are overwritten by subsequent status updates (intermediate or permanent). If status is None, any previous intermediate status is cleared without writing a new one.

parent: Self | None: The parent status object, if this status object is nested within another status object.

path: Path | None: The path to the directory where the status file is written by this status object, or None if the path of the parent status object is adopted.

progress(iterable: Iterable, iterations: int | None = None, details: str | dict | None = None) → Iterator[dict]

Write an intermediate progress update for each item in the iterable.

The intermediate status is cleared after yielding the last item from the iterable, after exiting the generator (e.g., breaking the loop), or if an error is raised.

Parameters:

iterable – The iterable to be processed.
iterations – The number of iterations to make (e.g., if this cannot be determined by calling len on the iterable). Defaults to the len of the iterable.
details – Additional status details.

Yields:

The items from the iterable, while making intermediate progress updates to the status object.

Raises:

AssertionError – If the iterable has more items than the number of iterations.

property root: Path | None: The path to the directory where the status file is written by this status object.

update() → None: Write the status data to the status file.

write(status: str | dict | list) → None: Write a permanent status update to the status object.

class repype.status.StatusReader(filepath: PathLike, loop: AbstractEventLoop = None, blocking=False)

Bases: FileSystemEventHandler

A status reader that can be used to monitor the progress of a computation by tracking the updates of a Status object, including its nested status objects.

The monitored status object can reside in a different process and is accessed by reading the corresponding status file. The progress of the computation is represented by a nested list structure, where each list directly corresponds to the state of a nested status object.

Parameters:

filepath – The status file written by the status object to be monitored.
loop – The event loop to be used for processing the status updates (usually the loop of the main thread). Defaults to the event loop of the thread used to create the status reader object.
blocking – If True, the status updates are processed on a separate thread (the main thread is assumed to be used for long-running, blocking operations). If False, the status updates are posted to the main thread.

See also

The implementation repype.cli.StatusReaderConsoleAdapter writes status updates to the standard output.

blocking: bool: If True, the status updates are processed on a separate thread (the main thread is assumed to be used for long-running, blocking operations). If False, the status updates are posted to the main thread.

check_new_status() → None

Check the data for new status updates, based on the current position of the cursor.

The cursor is advanced to the next element (if any), and the new status update is processed by the handle_new_status() method. This procedure is repated until the cursor points to the end of data. If the cursor then points to an intermediate status, it is rewinded to the last non-intermediate position (i.e. permanent). This assures that future intermediate status updates will be again proclaimed to the handle_new_status() method.

cursor: Cursor: Points to the latest permanent (i.e. non-intermediate) status update within data.

data: list: The data structure that represents the progress of the computation.

data_frames: Dict[Path, list]: The data structures that represent the progress of the nested status objects, indexed by the paths to the corresponding status files. This also contains the progress of the status object that corresponds to the filepath attribute.

file_hashes: Dict[Path, str]: The hashes of the status files when they were last read.

filepath: PathLike: The status file written by the monitored status object.

handle_new_status(positions: List[int], status: str | dict | None, intermediate: bool) → None

Process a new status update.

Parameters:

positions – The sequence of elements along the path, represented by the positions of the elements within the parent lists.
status – The new status update. Can only be None if intermediate is True, indicating that the intermediate status is cleared.
intermediate – True if the status update is intermediate, and False otherwise.

loop: AbstractEventLoop: The event loop to be used for processing the status updates.

on_modified(event: DirModifiedEvent | FileModifiedEvent) → None

Handle status file updates within the directory of the monitored status file.

If the monitored or a nested status file is updated, the status data is reloaded by the update() method. Only if the status data has changed according to the file_hashes, the new status data is processed by the check_new_status() method.

Parameters:: event – The file modification event.

update(filepath: Path) → bool

Update the nested list structure that represents the progress of the computation.

Only the list in data_frames is updated that corresponds to the status object that writes the status file at filepath. The status file is only read if its content has changed according to the file_hashes.

Parameters:: filepath – The path to the status file to be read.
Returns:: True if the status data has changed, and False otherwise.

repype.status.create() → ContextManager[Status]

Create a status object associated with a temporary directory.

>>> import repype.status
>>> with repype.status.create() as status:
...    status.write('Hello, World!')
...    print(status.filepath.read_text())
... 
["Hello, World!"]

repype.status.derive(status: Status | None) → Status | None

Shortcut for Status.derive().

Does nothing if status is None.

repype.status.progress(status: Status | None, iterable: Iterable, iterations: int | None = None, details: str | dict | None = None) → Iterator[dict]

Shortcut for Status.progress().

Yields the items from the iterable directly if status is None.

repype.status.update(status: Status | None, plain_text=None, intermediate: bool = False, **kwargs) → None

Shortcut for Status.write() and Status.intermediate().

Does nothing if status is None.

Raises:: AssertionError – If both plain_text and kwargs are provided (or neither, and intermediate is True).