Metadata-Version: 2.4
Name: ataraxis-data-structures
Version: 7.0.0
Summary: Provides classes and structures for storing, manipulating, and sharing data between Python processes.
Project-URL: Homepage, https://github.com/Sun-Lab-NBB/ataraxis-data-structures
Project-URL: Documentation, https://ataraxis-data-structures-api-docs.netlify.app/
Author: Ivan Kondratyev
Maintainer-email: Ivan Kondratyev <ik278@cornell.edu>
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: ataraxis,checksum,data logging,data manipulation,data structures,data transfer,interpolation,processing,shared memory
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Operating System :: MacOS :: MacOS X
Classifier: Operating System :: Microsoft :: Windows
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Scientific/Engineering
Classifier: Typing :: Typed
Requires-Python: <3.15,>=3.12
Requires-Dist: ataraxis-base-utilities<8,>=7
Requires-Dist: ataraxis-time<8,>=7
Requires-Dist: dacite<2,>=1
Requires-Dist: filelock<4,>=3
Requires-Dist: numba<1,>=0.64
Requires-Dist: numpy<3,>=2
Requires-Dist: pyyaml<7,>=6
Requires-Dist: xxhash<4,>=3
Description-Content-Type: text/markdown

# ataraxis-data-structures

Provides classes and structures for storing, manipulating, and sharing data between Python processes.

![PyPI - Version](https://img.shields.io/pypi/v/ataraxis-data-structures)
![PyPI - Python Version](https://img.shields.io/pypi/pyversions/ataraxis-data-structures)
[![uv](https://tinyurl.com/uvbadge)](https://github.com/astral-sh/uv)
[![Ruff](https://tinyurl.com/ruffbadge)](https://github.com/astral-sh/ruff)
![type-checked: mypy](https://img.shields.io/badge/type--checked-mypy-blue?style=flat-square&logo=python)
![PyPI - License](https://img.shields.io/pypi/l/ataraxis-data-structures)
![PyPI - Status](https://img.shields.io/pypi/status/ataraxis-data-structures)
![PyPI - Wheel](https://img.shields.io/pypi/wheel/ataraxis-data-structures)

___

## Detailed Description

This library aggregates the classes and methods used by other Ataraxis and Sollertia libraries for working with data.
This includes classes to manipulate the data, share (move) the data between different Python processes, and store the
data in non-volatile memory (on disk). Generally, these classes either implement novel functionality or extend
existing functionality to match the specific needs of other Ataraxis and Sollertia libraries. This library is part of
the [Ataraxis](https://github.com/Sun-Lab-NBB/ataraxis) framework for AI-assisted scientific hardware control.

___

## Features

- Supports Windows, Linux, and macOS.
- Provides a process- and thread-safe way of sharing data between multiple processes through a NumPy array structure.
- Extends the standard Python dataclass to support saving and loading its data to and from YAML files.
- Provides a fast and scalable data logger optimized for saving serialized data from multiple parallel processes in
  non-volatile memory.
- Offers efficient batch processing of log archives with support for parallel workflows.
- Includes a file-based processing pipeline tracker for coordinating multi-process and multi-host data processing jobs.
- Provides utilities for data integrity verification, directory transfer, data asset discovery, time-series
  interpolation, and worker thread limiting.
- Apache 2.0 License.

___

## Table of Contents

- [Dependencies](#dependencies)
- [Installation](#installation)
- [Usage](#usage)
  - [YamlConfig](#yamlconfig)
  - [SharedMemoryArray](#sharedmemoryarray)
  - [DataLogger](#datalogger)
  - [LogArchiveReader](#logarchivereader)
  - [ProcessingTracker](#processingtracker)
  - [Processing Utilities](#processing-utilities)
- [API Documentation](#api-documentation)
- [Developers](#developers)
- [Versioning](#versioning)
- [Authors](#authors)
- [License](#license)
- [Acknowledgments](#acknowledgments)

___

## Dependencies

For users, all library dependencies are installed automatically by all supported installation methods. For developers,
see the [Developers](#developers) section for information on installing additional development dependencies.

___

## Installation

### Source

***Note,*** installation from source is ***highly discouraged*** for anyone who is not an active project developer.

1. Download this repository to the local machine using the preferred method, such as git-cloning. Use one of the
   [stable releases](https://github.com/Sun-Lab-NBB/ataraxis-data-structures/tags) that include precompiled binary
   and source code distribution (sdist) wheels.
2. If the downloaded distribution is stored as a compressed archive, unpack it using the appropriate decompression
   tool.
3. `cd` to the root directory of the prepared project distribution.
4. Run `pip install .` to install the project and its dependencies.

### pip

Use the following command to install the library and all of its dependencies via [pip](https://pip.pypa.io/en/stable/):
`pip install ataraxis-data-structures`

___

## Usage

For detailed information about method signatures and parameters, consult the [API documentation](#api-documentation).

### YamlConfig

The YamlConfig class extends the functionality of the standard Python dataclass module by bundling the dataclass
instances with methods to save and load their data to and from .yaml files. Primarily, this functionality is
implemented to support storing runtime configuration data in a non-volatile, human-readable, and editable format.

The YamlConfig class is designed to be subclassed by custom dataclass instances to gain the .yaml saving and loading
functionality realized through the inherited `to_yaml()` and `from_yaml()` methods:

```python
from ataraxis_data_structures import YamlConfig
from dataclasses import dataclass
from pathlib import Path
import tempfile


# All YamlConfig functionality is accessed via subclassing.
@dataclass
class MyConfig(YamlConfig):
    integer: int = 0
    string: str = "random"


# Instantiates the test class using custom values that do not match the default initialization values.
config = MyConfig(integer=123, string="hello")

# Saves the instance data to a YAML file in a temporary directory. The saved data can be modified by directly
# editing the saved .yaml file.
tempdir = tempfile.TemporaryDirectory()  # Creates a temporary directory for illustration purposes.
out_path = Path(tempdir.name).joinpath("my_config.yaml")  # Resolves the path to the output file.
config.to_yaml(file_path=out_path)

# Ensures that the cache file has been created.
assert out_path.exists()

# Creates a new MyConfig instance using the data inside the .yaml file.
loaded_config = MyConfig.from_yaml(file_path=out_path)

# Ensures that the loaded data matches the original MyConfig instance data.
assert loaded_config.integer == config.integer
assert loaded_config.string == config.string
```

#### Excluding Fields from the Document

A dataclass field marked with the `YAML_EXCLUDE_METADATA` metadata is skipped when the instance is written,
which suits a field that records where the instance lives rather than what it holds. The written document carries no
entry for such a field, so a class whose constructor requires it overrides the `restore_excluded_fields()` class method
to supply the value back. The `from_yaml()` method calls that override between reading the document and building the
instance, so it belongs to the deserialization machinery rather than to the API a caller invokes.

```python
from ataraxis_data_structures import YAML_EXCLUDE_METADATA, YamlConfig
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import tempfile
import yaml


@dataclass
class LocatedConfig(YamlConfig):
    value: int = 0
    # The path records where the instance lives, so it is kept out of the document the instance writes.
    source_path: Path = field(default=Path("/unset"), metadata=YAML_EXCLUDE_METADATA)

    @classmethod
    def restore_excluded_fields(cls, data: dict[Any, Any], file_path: Path) -> dict[Any, Any]:
        """Reattaches the reconstructed instance to the file it was read from."""
        return {**data, "source_path": file_path}


tempdir = tempfile.TemporaryDirectory()  # Creates a temporary directory for illustration purposes.
out_path = Path(tempdir.name).joinpath("located.yaml")
LocatedConfig(value=7, source_path=Path("/the/writing/host/path")).to_yaml(file_path=out_path)

# The writing host's path stays out of the document entirely.
with out_path.open() as yaml_file:
    assert yaml.safe_load(yaml_file) == {"value": 7}

# The reader supplies the path from where it found the file, rather than from what the writer recorded.
loaded = LocatedConfig.from_yaml(file_path=out_path)
assert loaded.value == 7
assert loaded.source_path == out_path
```

### SharedMemoryArray

The SharedMemoryArray class supports sharing data between multiple Python processes in a thread- and process-safe way.
To do so, it implements a shared memory buffer accessed via an n-dimensional NumPy array instance, allowing different
processes to read and write any element(s) of the array.

#### SharedMemoryArray Creation

The SharedMemoryArray only needs to be instantiated once by the main runtime process (thread) and provided to all
child processes as an input. The initialization process uses the specified prototype NumPy array and unique buffer
name to generate a (new) NumPy array whose data is stored in a shared memory buffer accessible from any thread or
process. ***Note,*** the array dimensions and datatype cannot be changed after initialization.

```python
from ataraxis_data_structures import SharedMemoryArray
import numpy as np

# The prototype array and buffer name determine the layout of the SharedMemoryArray for its entire lifetime:
prototype = np.array([[1, 2, 3], [4, 5, 6]], dtype=np.float64)
buffer_name = "unique_buffer_name"  # Has to be unique for all concurrently used SharedMemoryArray instances.

# To initialize the array, use the create_array() method. Do not call the class initialization method directly!
# The returned instance is connected to the shared memory buffer, and it destroys that buffer once it is
# garbage-collected.
sma = SharedMemoryArray.create_array(name=buffer_name, prototype=prototype)

# The instantiated SharedMemoryArray object wraps an n-dimensional NumPy array with the same dimensions and data
# type as the prototype and uses the unique shared memory buffer name to identify the shared memory buffer to
# connect to from different processes.
assert sma.name == buffer_name
assert sma.shape == prototype.shape
assert sma.datatype == prototype.dtype

# Demonstrates the current values for the critical SharedMemoryArray parameters evaluated above:
print(sma)
```

#### SharedMemoryArray Connection, Disconnection, and Destruction

The `create_array()` method returns an instance already connected to the shared memory buffer, and every process the
instance is passed to connects as part of the transfer, so no process has to call `connect()` to reach the array data.
Calling `connect()` anyway remains useful, as it establishes the connection at a chosen point rather than leaving it
implicit, and it reconnects an instance that called `disconnect()`.

At the end of its runtime, each process should call the `disconnect()` method to release its handle on the shared
buffer. A process that exits releases its handle regardless, so this call is good practice rather than a requirement.
The process that created the array destroys the shared memory buffer when its instance is garbage-collected, and
calling `destroy()` ends the buffer's lifetime at a chosen point instead.

```python
import numpy as np
from ataraxis_data_structures import SharedMemoryArray

# Initializes a SharedMemoryArray. The creating process is connected to the buffer once this method returns.
prototype = np.zeros(shape=6, dtype=np.uint64)
buffer_name = "unique_buffer"
sma = SharedMemoryArray.create_array(name=buffer_name, prototype=prototype)

# The connection status of the array can be verified at any time by using is_connected property:
assert sma.is_connected

# Each process that connected to the shared memory buffer should disconnect from it at the end of its runtime. On
# Windows platforms, when all processes are disconnected from the buffer, the buffer is automatically
# garbage-collected.
sma.disconnect()  # Good practice in every process, as it releases the handle as soon as the work is done

assert not sma.is_connected

# On Unix platforms, the buffer persists even after being disconnected by all instances, until the creating instance
# is garbage-collected or the buffer is explicitly destroyed.
sma.destroy()  # Ends the buffer's lifetime immediately rather than waiting for garbage collection
```

#### Reading and Writing SharedMemoryArray Data

For routine data writing or reading operations, the SharedMemoryArray supports accessing its data via indexing or
slicing, just like a regular NumPy array. Critically, accessing the data in this way is process-safe, as the instance
first acquires an exclusive multiprocessing Lock before interfacing with the data. For more complex access scenarios,
it is possible to use the `array()` method to directly access and manipulate the underlying NumPy array object used by
the instance.

```python
import numpy as np
from ataraxis_data_structures import SharedMemoryArray

# Initializes a SharedMemoryArray
prototype = np.array([1, 2, 3, 4, 5, 6], dtype=np.uint64)
buffer_name = "unique_buffer"
sma = SharedMemoryArray.create_array(name=buffer_name, prototype=prototype)

# The SharedMemoryArray data can be accessed directly using indexing or slicing, just like any regular NumPy array
# or Python iterable:

# Index
assert sma[2] == np.uint64(3)
assert isinstance(sma[2], np.uint64)
sma[2] = 123  # Written data must be convertible to the datatype of the underlying NumPy array
assert sma[2] == np.uint64(123)

# Slice
assert np.array_equal(sma[:4], np.array([1, 2, 123, 4], dtype=np.uint64))
assert isinstance(sma[:4], np.ndarray)

# It is also possible to directly access the underlying NumPy array, which allows using the full range of NumPy
# operations. The accessor method can be used from within a context manager to enforce exclusive access to the
# array's data via an internal multiprocessing lock mechanism:
with sma.array(with_lock=True) as array:
    print(f"Before clipping: {array}")

    # Clipping replaces the out-of-bounds value '123' with '10'. The slice assignment writes the clipped values
    # back into the shared memory buffer. Rebinding the name instead would discard the result.
    array[:] = np.clip(array, 0, 10)

    print(f"After clipping: {array}")

# Cleans up the array buffer
sma.disconnect()
sma.destroy()
```

#### Using SharedMemoryArray from Multiple Processes

While all methods showcased above run in the same process, the main advantage of the SharedMemoryArray class is that it
behaves the same way when used from different Python processes.

***Note,*** the main process is connected to the array from the moment `create_array()` returns, and each child process
connects as part of receiving the instance. The example below therefore reduces to creating the array, passing it to the
workers, and ending the runtime. The only ordering requirement is that the buffer is not destroyed until all processes
have finished using it, which means the creating instance has to stay referenced for that whole period.

Passing `auto_connect=False` to `create_array()` opts out of connecting the receiving processes, which makes each of
them call `connect()` before accessing the data. That defers the connection to a point the worker controls, so a
buffer destroyed between the spawn and the worker's first access fails at that access rather than during startup.

```python
from multiprocessing import Process
from ataraxis_base_utilities import console
from ataraxis_time import PrecisionTimer, TimerPrecisions
import numpy as np
from ataraxis_data_structures import SharedMemoryArray


def concurrent_worker(shared_memory_object: SharedMemoryArray, index: int) -> None:
    """This worker runs in a remote process.

    It increments the shared memory array variable by 1 if the variable is even. Since each increment shifts it to
    be odd, to work as intended, this process has to work together with a different process that increments odd
    values. The process shuts down once the value reaches 200.

    Args:
        shared_memory_object: The SharedMemoryArray instance to work with.
        index: The index inside the array to increment
    """
    # The array arrives connected, so this worker reads and writes its data without any setup of its own.

    # Runs until the value becomes 200
    while shared_memory_object[index] < 200:
        # Reads data from the input index
        shared_value = shared_memory_object[index]

        # Checks if the value is even and below 200
        if shared_value % 2 == 0 and shared_value < 200:
            # Increments the value by one and writes it back to the array
            shared_memory_object[index] = shared_value + 1

    # Disconnects and terminates the process
    shared_memory_object.disconnect()


if __name__ == "__main__":
    console.enable()  # Enables terminal printouts

    # Initializes a SharedMemoryArray. This process is connected to the buffer and owns its destruction.
    sma = SharedMemoryArray.create_array(name="test_concurrent", prototype=np.zeros(shape=5, dtype=np.int32))

    # Generates multiple processes and uses each to repeatedly write and read data from different indices of the
    # same array.
    processes = [Process(target=concurrent_worker, args=(sma, index)) for index in range(5)]
    for p in processes:
        p.start()

    # Marks the beginning of the test runtime
    console.echo(f"Running the multiprocessing example on {len(processes)} processes...")
    timer = PrecisionTimer(precision=TimerPrecisions.MILLISECOND)
    timer.reset()

    # For each of the array indices, increments the value of the index if it is odd. Child processes increment
    # even values and ignore odd ones, so the only way for this code to finish is if the child and parent processes
    # take turns incrementing shared values until they reach 200
    while np.any(sma[0:5] < 200):  # Runs as long as any value is below 200
        # Note, while it is possible to index the data from the SharedMemoryArray, it is also possible to retrieve
        # and manipulate the underlying NumPy array directly. This allows using the full range of NumPy operations
        # on the shared memory data:
        with sma.array(with_lock=True) as arr:
            mask = (arr % 2 != 0) & (arr < 200)  # Uses a boolean mask to discover odd values below 200
            arr[mask] += 1  # Increments only the values that meet the condition above

    # Waits for the processes to join
    for p in processes:
        p.join()

    # Verifies that all processes ran as expected and incremented their respective variable
    assert np.all(sma[0:5] == 200)

    # Marks the end of the test runtime.
    time_taken = timer.elapsed
    console.echo(f"Example runtime: complete. Time taken: {time_taken / 1000:.2f} seconds.")

    # Cleans up the shared memory array after all processes are terminated. Ending the runtime without these calls
    # reaches the same state, as the instance destroys the buffer once it is garbage-collected.
    sma.disconnect()
    sma.destroy()
```

### DataLogger

The DataLogger class initializes and manages the runtime of a logger process running in an independent Process and
exposes a shared Queue object for buffering and piping data from any other Process to the logger. Currently, the class
is specifically designed for saving serialized byte arrays used by other Ataraxis libraries, most notably the
ataraxis-video-system and the ataraxis-transport-layer.

#### Creating and Starting the DataLogger

DataLogger is intended to only be initialized once in the main runtime thread (Process) and provided to all child
Processes as an input. ***Note,*** while a single DataLogger instance is typically enough for most use cases, it is
possible to use more than a single DataLogger instance at the same time.

```python
from pathlib import Path
import tempfile
from ataraxis_data_structures import DataLogger

# Due to the internal use of the 'Process' class, each DataLogger call has to be protected by the __main__ guard
# at the highest level of the call hierarchy.
if __name__ == "__main__":
    # As a minimum, each DataLogger has to be given the path to the output directory and a unique name to
    # distinguish the instance from any other concurrently active DataLogger instance.
    tempdir = tempfile.TemporaryDirectory()  # Creates a temporary directory for illustration purposes
    logger = DataLogger(output_directory=Path(tempdir.name), instance_name="my_name")

    # The DataLogger initialized above creates a new directory: 'tempdir/my_name_data_log' to store logged entries.

    # Before the DataLogger starts saving data, its saver process needs to be initialized via the start() method.
    # Until the saver is initialized, the instance buffers all incoming data in RAM (via the internal Queue
    # object), which may eventually exhaust the available memory.
    logger.start()

    # Each call to the start() method must be matched with a corresponding call to the stop() method. This method
    # shuts down the logger process and releases any resources held by the instance.
    logger.stop()
```

#### Data Logging

The DataLogger is explicitly designed to log serialized data of arbitrary size. To enforce the correct data
formatting, all data submitted to the logger must be packaged into a LogPackage class instance before it is put into
the DataLogger's input queue.

```python
from pathlib import Path
import tempfile
import numpy as np
from ataraxis_data_structures import DataLogger, LogPackage
from ataraxis_time import get_timestamp, TimestampFormats

if __name__ == "__main__":
    # Initializes and starts the DataLogger.
    tempdir = tempfile.TemporaryDirectory()
    logger = DataLogger(output_directory=Path(tempdir.name), instance_name="my_name")
    logger.start()

    # The DataLogger uses a multiprocessing Queue to buffer and pipe the incoming data to the saver process. The
    # queue is accessible via the 'input_queue' property of each logger instance.
    logger_queue = logger.input_queue

    # The DataLogger is explicitly designed to log serialized data. All data submitted to the logger must be
    # packaged into a LogPackage instance to ensure that it adheres to the proper format expected by the logger
    # instance.
    source_id = np.uint8(1)  # Has to be an uint8 type
    acquisition_time = np.uint64(get_timestamp(output_format=TimestampFormats.INTEGER))  # Has to be an uint64 type
    data = np.array([1, 2, 3, 4, 5], dtype=np.uint8)  # Has to be an uint8 NumPy array
    logger_queue.put(LogPackage(source_id=source_id, acquisition_time=acquisition_time, serialized_data=data))

    # The timer used to timestamp the log entries has to be precise enough to resolve two consecutive data
    # entries. Due to these constraints, it is recommended to use a nanosecond or microsecond timer, such as the
    # one offered by the ataraxis-time library.
    timestamp = np.uint64(get_timestamp(output_format=TimestampFormats.INTEGER))
    data = np.array([6, 7, 8, 9, 10], dtype=np.uint8)
    logger_queue.put(  # Same source id as the package above
        LogPackage(source_id=source_id, acquisition_time=timestamp, serialized_data=data)
    )

    # Stops the data logger.
    logger.stop()

    # The DataLogger saves the input LogPackage instances as serialized NumPy byte array .npy files. The output
    # directory for the saved files can be queried from the DataLogger instance's 'output_directory' property.
    assert len(list(logger.output_directory.glob("**/*.npy"))) == 2
```

#### Log Archive Assembly

Each log entry is saved to disk as a separate NumPy array .npy file, which keeps the write path short and the time the
data spends in volatile memory small. The `assemble_log_archives()` function aggregates the .npy files from one data
source into a single (uncompressed) .npz archive for storage and transfer.

```python
from pathlib import Path
import tempfile
import numpy as np
from ataraxis_data_structures import DataLogger, LogPackage, assemble_log_archives

if __name__ == "__main__":
    # Creates and starts the DataLogger instance.
    tempdir = tempfile.TemporaryDirectory()
    logger = DataLogger(output_directory=Path(tempdir.name), instance_name="my_name")
    logger.start()
    logger_queue = logger.input_queue

    # Generates and logs 255 data messages. This generates 255 unique .npy files under the logger's output
    # directory.
    for i in range(255):
        logger_queue.put(
            LogPackage(
                source_id=np.uint8(1),
                acquisition_time=np.uint64(i),
                serialized_data=np.array([i, i, i], dtype=np.uint8),
            )
        )

    # Stops the data logger.
    logger.stop()

    # Depending on the runtime context, a DataLogger instance can generate a large number of individual .npy files
    # as part of its runtime. While having advantages for real-time data logging, this format of storing the data
    # is not ideal for later data transfer and manipulation. Therefore, it is recommended to always use the
    # assemble_log_archives() function to aggregate the individual .npy files into one or more .npz archives.
    assemble_log_archives(log_directory=logger.output_directory, remove_sources=True, memory_mapping=True, verbose=True)

    # The archive assembly creates a single .npz file named after the source_id (1_log.npz), using all available
    # .npy files. Generally, each unique data source is assembled into a separate .npz archive.
    assert len(list(logger.output_directory.glob("**/*.npy"))) == 0
    assert len(list(logger.output_directory.glob("**/*.npz"))) == 1
```

### LogArchiveReader

The LogArchiveReader class provides efficient access to .npz log archives generated by DataLogger instances. It
supports onset timestamp discovery, message iteration, and batch assignment for multiprocessing workflows.

#### Basic Usage

Each .npz archive contains messages from a single source (producer). The LogArchiveReader automatically discovers the
onset timestamp stored in the archive and converts elapsed timestamps to absolute UTC timestamps.

```python
from pathlib import Path
from ataraxis_data_structures import LogArchiveReader

# Creates a reader for an existing archive
archive_path = Path("/path/to/1_log.npz")
reader = LogArchiveReader(archive_path=archive_path)

# The onset timestamp is automatically discovered (UTC epoch reference in microseconds)
print(f"Onset timestamp: {reader.onset_timestamp_us}")
print(f"Message count: {reader.message_count}")

# Iterates through all messages in the archive
for message in reader.iter_messages():
    print(f"Timestamp: {message.timestamp_us}, Payload size: {len(message.payload)}")
```

#### Batch Processing for Multiprocessing Workflows

The main process can generate batch assignments, and worker processes can create lightweight reader instances by passing
the pre-discovered onset timestamp to skip redundant scanning.

```python
from pathlib import Path
from concurrent.futures import ProcessPoolExecutor
from ataraxis_data_structures import LogArchiveReader
import numpy as np


def process_batch(archive_path: Path, onset_us: np.uint64, keys: list[str]) -> int:
    """Worker function that processes a batch of messages."""
    # Creates a lightweight reader with pre-discovered onset (skips onset discovery)
    reader = LogArchiveReader(archive_path=archive_path, onset_us=onset_us)

    processed = 0
    for message in reader.iter_messages(keys=keys):
        # Process each message...
        processed += 1
    return processed


if __name__ == "__main__":
    archive_path = Path("/path/to/1_log.npz")

    # Main process discovers onset and generates batches
    reader = LogArchiveReader(archive_path=archive_path)
    onset_us = reader.onset_timestamp_us
    batches = reader.get_batches(workers=4, batch_multiplier=4)

    # Distributes batches to worker processes
    with ProcessPoolExecutor(max_workers=4) as executor:
        futures = [
            executor.submit(process_batch, archive_path=archive_path, onset_us=onset_us, keys=batch)
            for batch in batches
        ]
        total_processed = sum(f.result() for f in futures)

    print(f"Processed {total_processed} messages")
```

#### Reading All Messages at Once

For smaller archives, all messages can be read into memory at once:

```python
from pathlib import Path
from ataraxis_data_structures import LogArchiveReader

reader = LogArchiveReader(archive_path=Path("/path/to/1_log.npz"))

# Returns a tuple of (timestamps_array, payloads_list)
timestamps, payloads = reader.read_all_messages()
print(f"Read {len(timestamps)} messages")
```

### ProcessingTracker

The ProcessingTracker class tracks the state of data processing pipelines and provides tools for communicating this
state between multiple processes and host machines. It uses a file-based approach with a .yaml file for state storage
and a .lock file for thread-safe access.

#### Creating and Initializing the Tracker

The ProcessingTracker extends YamlConfig and uses file locking to ensure safe concurrent access from multiple
processes.

```python
from pathlib import Path
from ataraxis_data_structures import ProcessingTracker

# Creates a tracker pointing to a .yaml file
tracker = ProcessingTracker(file_path=Path("/path/to/tracker.yaml"))

# Initializes jobs to be tracked (each job is a tuple of (job_name, specifier))
# Specifiers differentiate instances of the same job (e.g., different data batches)
job_ids = tracker.initialize_jobs(
    jobs=[
        ("process_video", "session_001"),
        ("process_video", "session_002"),
        ("extract_frames", "session_001"),
        ("extract_frames", "session_002"),
        ("generate_report", ""),  # Empty specifier for jobs without batches
    ]
)

print(f"Initialized {len(job_ids)} jobs")
```

For pipelines that run a selectable subset of their jobs across invocations, `align_jobs()` reconciles an existing
tracker with the jobs a given invocation intends to run. It additively registers any missing jobs, leaves
already-tracked jobs untouched, and discards only those entries that fall outside the provided job universe.
Requesting a job that is absent from the universe raises a `ValueError`.

```python
from pathlib import Path
from ataraxis_data_structures import ProcessingTracker

tracker = ProcessingTracker(file_path=Path("/path/to/tracker.yaml"))

# The full set of jobs the pipeline can produce for its current definition
universe = [
    ("process_video", "session_001"),
    ("process_video", "session_002"),
    ("extract_frames", "session_001"),
]

# Aligns the tracker to run one job while preserving the recorded state of the other universe jobs
job_ids = tracker.align_jobs(jobs=[("process_video", "session_001")], universe=universe)
```

#### Managing Job Lifecycle

Jobs transition through states: SCHEDULED → RUNNING → SUCCEEDED or FAILED.

```python
from pathlib import Path
from ataraxis_data_structures import ProcessingTracker, ProcessingStatus

tracker = ProcessingTracker(file_path=Path("/path/to/tracker.yaml"))

# Generates a job ID using the same name and specifier used during initialization
job_id = ProcessingTracker.generate_job_id(job_name="process_video", specifier="session_001")

# Marks the job as started (optionally with an executor ID like a SLURM job ID)
tracker.start_job(job_id=job_id, executor_id="slurm_12345")

# Queries the current status
status = tracker.get_job_status(job_id=job_id)
print(f"Job status: {status.name}")  # Job status: RUNNING

# Marks the job as completed successfully
tracker.complete_job(job_id=job_id)

# Or, if the job failed:
# tracker.fail_job(job_id=job_id, error_message="Out of memory")
```

***Note,*** when `start_job()` is called without an explicit `executor_id`, the identifier is resolved from the
runtime environment. It records a recognized job scheduler's job ID (SLURM, PBS, LSF, SGE, OAR, HPC Pack, Azure
Batch, or AWS Batch) when one is detected, otherwise the process ID, tagged with its scheme (for example
`slurm:12345` or `pid:4242`). The scheme tag lets a downstream consumer select the matching liveness query.

#### Querying Pipeline State

```python
from pathlib import Path
from ataraxis_data_structures import ProcessingTracker, ProcessingStatus

tracker = ProcessingTracker(file_path=Path("/path/to/tracker.yaml"))

# Checks if all jobs have completed successfully
if tracker.complete:
    print("Pipeline completed successfully!")

# Checks if any job has failed
if tracker.encountered_error:
    print("Pipeline encountered errors!")

# Gets a summary of job counts by status
summary = tracker.get_summary()
for status, count in summary.items():
    print(f"{status.name}: {count}")

# Gets all job IDs with a specific status
failed_jobs = tracker.get_jobs_by_status(status=ProcessingStatus.FAILED)
scheduled_jobs = tracker.get_jobs_by_status(status="SCHEDULED")  # String names also work

# Searches for jobs by name or specifier patterns
matches = tracker.find_jobs(job_name="process", specifier="session_001")
for job_id, (name, spec) in matches.items():
    print(f"Found job: {name} ({spec})")

# Called without arguments, find_jobs matches every tracked job
all_matches = tracker.find_jobs()

# Gets detailed job information
job_info = tracker.get_job_info(job_id=job_id)
print(f"Job: {job_info.job_name}, Status: {job_info.status}")
print(f"Started at: {job_info.started_at}, Completed at: {job_info.completed_at}")

# Reads the entire job registry at once as a consistent, detached snapshot
registry = tracker.snapshot()
for job_id, state in registry.items():
    print(f"{state.job_name} ({state.specifier}): {state.status.name}")
```

#### Retrying and Resetting Jobs

Failed jobs can be reset for retry, and a specific subset of jobs can be reset without disturbing the others:

```python
from pathlib import Path
from ataraxis_data_structures import ProcessingTracker

tracker = ProcessingTracker(file_path=Path("/path/to/tracker.yaml"))

# Resets all failed jobs back to SCHEDULED status
retried_ids = tracker.retry_failed_jobs()
print(f"Reset {len(retried_ids)} failed jobs for retry")

# Resets a specific subset of jobs back to SCHEDULED, preserving every other job's recorded state
target_id = ProcessingTracker.generate_job_id(job_name="process_video", specifier="session_001")
reset_ids = tracker.reset_jobs(job_ids=[target_id])

# Or reset the entire tracker
tracker.reset()
```

### Processing Utilities

#### Directory Checksum Calculation

The `calculate_directory_checksum()` function computes an xxHash3-128 checksum for an entire directory, accounting for
both file contents and directory structure.

```python
from pathlib import Path
from ataraxis_data_structures import calculate_directory_checksum

# Calculates checksum with progress tracking
checksum = calculate_directory_checksum(
    directory=Path("/path/to/data"),
    num_processes=None,  # Uses all available CPU cores minus 2 reserved cores
    progress=True,  # Shows progress bar
    save_checksum=True,  # Saves to ax_checksum.txt in the directory
)
print(f"Directory checksum: {checksum}")

# Calculates checksum without saving or progress tracking (for batch processing)
checksum = calculate_directory_checksum(
    directory=Path("/path/to/data"),
    progress=False,
    save_checksum=False,
    excluded_files={"ax_checksum.txt", ".gitignore"},  # Excludes specific files
)
```

#### Directory Transfer

The `transfer_directory()` function copies directories with optional integrity verification and parallel processing.

```python
from pathlib import Path
from ataraxis_data_structures import transfer_directory

# Transfers with integrity verification
transfer_directory(
    source=Path("/path/to/source"),
    destination=Path("/path/to/destination"),
    num_threads=4,  # Uses 4 threads for parallel copy
    verify_integrity=True,  # Verifies checksum after transfer
    remove_source=False,  # Keeps source after transfer
    progress=True,  # Shows progress bar
)

# Moves data (transfers and removes source)
transfer_directory(
    source=Path("/path/to/source"),
    destination=Path("/path/to/destination"),
    num_threads=0,  # Uses all available CPU cores minus a few reserved for the host
    verify_integrity=True,
    remove_source=True,  # Removes source after successful transfer
)
```

#### Directory Deletion

The `delete_directory()` function removes directories using parallel file deletion for improved performance.

```python
from pathlib import Path
from ataraxis_data_structures import delete_directory

# Deletes a directory and all its contents
delete_directory(directory_path=Path("/path/to/directory"))
```

#### Data Interpolation

The `interpolate_data()` function aligns time-series data to target coordinates using linear interpolation (for
continuous data) or last-known-value interpolation (for discrete data).

```python
import numpy as np
from ataraxis_data_structures import interpolate_data

# Source data with timestamps and values
source_timestamps = np.array([0, 100, 200, 300, 400], dtype=np.uint64)
source_values = np.array([10.0, 20.0, 15.0, 25.0, 30.0], dtype=np.float64)

# Target timestamps for interpolation
target_timestamps = np.array([50, 150, 250, 350], dtype=np.uint64)

# Continuous interpolation (linear)
interpolated_continuous = interpolate_data(
    source_coordinates=source_timestamps,
    source_values=source_values,
    target_coordinates=target_timestamps,
    is_discrete=False,
)
print(f"Continuous: {interpolated_continuous}")  # [15.0, 17.5, 20.0, 27.5]

# Discrete interpolation (last known value)
discrete_values = np.array([1, 2, 3, 4, 5], dtype=np.uint8)
interpolated_discrete = interpolate_data(
    source_coordinates=source_timestamps,
    source_values=discrete_values,
    target_coordinates=target_timestamps,
    is_discrete=True,
)
print(f"Discrete: {interpolated_discrete}")  # [1, 2, 3, 4]
```

#### Worker Thread Limiting

The `limit_worker_threads()` context manager constrains the thread pools that the numeric backends bundled with NumPy,
the polars query engine, and the OpenCV and tifffile decoders open inside worker processes. The NumPy-bundled backends
and polars read their threading environment variable once, while they are being imported, so the value a spawned worker
inherits from its parent is the only value that reaches them. Without the limit, a pool running one worker per core
holds the square of the core count in threads while using one of them.

The context has to enclose the pool's entire lifetime rather than its construction alone, because a pool creates each
worker when work is first submitted to it. The previous values are restored on exit, including when the wrapped block
raises, which keeps the limit from leaking into whatever the calling process does next.

```python
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import get_context
from ataraxis_data_structures import limit_worker_threads


def double_value(value: int) -> int:
    """Stands in for the numeric work a real worker process performs."""
    return value * 2


if __name__ == "__main__":
    # The limit reaches each worker through the environment a spawned child inherits, so the context wraps the pool's
    # whole lifetime rather than only its construction.
    with (
        limit_worker_threads(thread_count=1),
        ProcessPoolExecutor(max_workers=2, mp_context=get_context("spawn")) as executor,
    ):
        results = list(executor.map(double_value, range(4)))

    assert results == [0, 2, 4, 6]
```

___

## API Documentation

See the [API documentation](https://ataraxis-data-structures-api-docs.netlify.app/) for the detailed description of the
methods and classes exposed by components of this library.

___

## Developers

This section provides installation, dependency, and build-system instructions for the developers that want to modify
the source code of this library.

### Installing the Project

***Note,*** this installation method requires **mamba version 2.3.2 or above**. Currently, all automation pipelines
require that mamba is installed through the [miniforge3](https://github.com/conda-forge/miniforge) installer.

1. Download this repository to the local machine using the preferred method, such as git-cloning.
2. If the downloaded distribution is stored as a compressed archive, unpack it using the appropriate decompression
   tool.
3. `cd` to the root directory of the prepared project distribution.
4. Install the core development dependencies into the ***base*** mamba environment via the
   `mamba install tox uv tox-uv` command.
5. Use the `tox -e create` command to create the project-specific development environment followed by `tox -e install`
   command to install the project into that environment as a library.

### Additional Dependencies

In addition to installing the project and all user dependencies, install the following dependencies:

1. [Python](https://www.python.org/downloads/) distributions, one for each version supported by the developed project.
   Currently, this library supports the three latest stable versions. It is recommended to use a tool like
   [pyenv](https://github.com/pyenv/pyenv) to install and manage the required versions.

### Development Automation

This project uses `tox` for development automation. The following tox environments are available:

| Environment          | Description                                                  |
|----------------------|--------------------------------------------------------------|
| `lint`               | Runs ruff formatting, ruff linting, and mypy type checking   |
| `stubs`              | Generates py.typed marker and .pyi stub files                |
| `{py312,...}-test`   | Runs the test suite via pytest for each supported Python     |
| `coverage`           | Aggregates test coverage and applies the 100% coverage gate  |
| `docs`               | Builds the API documentation via Sphinx                      |
| `build`              | Builds sdist and wheel distributions                         |
| `upload`             | Uploads distributions to PyPI via twine                      |
| `deploy`             | Uploads the built documentation to the Netlify site          |
| `install`            | Builds and installs the project into its mamba environment   |
| `uninstall`          | Uninstalls the project from its mamba environment            |
| `create`             | Creates the project's mamba development environment          |
| `remove`             | Removes the project's mamba development environment          |
| `provision`          | Recreates the mamba environment from scratch                 |
| `export`             | Exports the mamba environment as a .yml file                 |
| `import`             | Creates or updates the mamba environment from a .yml file    |

Run any environment using `tox -e ENVIRONMENT`. For example, `tox -e lint`.

***Note,*** all pull requests for this project have to successfully complete the `tox` task before being merged. To
expedite the task's runtime, use the `tox --parallel` command to run some tasks in parallel.

### AI-Assisted Development

Claude Code skills and other AI development assets for this project are distributed through the
[ataraxis](https://github.com/Sun-Lab-NBB/ataraxis) marketplace as part of the **automation** plugin. Install the
plugin from the marketplace to make all associated skills and development tools available to compatible AI coding
agents.

### Automation Troubleshooting

Many packages used in `tox` automation pipelines (uv, mypy, ruff) and `tox` itself may experience runtime failures. In
most cases, this is related to their caching behavior. If an unintelligible error is encountered with any of the
automation components, deleting the corresponding cache directories (`.tox`, `.ruff_cache`, `.mypy_cache`, etc.)
manually or via a CLI command typically resolves the issue.

___

## Versioning

This project uses [semantic versioning](https://semver.org/). See the
[tags on this repository](https://github.com/Sun-Lab-NBB/ataraxis-data-structures/tags) for the available project
releases.

___

## Authors

- Ivan Kondratyev ([Inkaros](https://github.com/Inkaros))

___

## License

This project is licensed under the Apache 2.0 License: see the [LICENSE](LICENSE) file for details.

___

## Acknowledgments

- All Sun lab [members](https://neuroai.github.io/sunlab/people) for providing the inspiration and comments during the
  development of this library.
- The creators of all other dependencies and projects listed in the [pyproject.toml](pyproject.toml) file.
