init.py

Overview

The init.py file provides a set of utility functions designed to assist with common file handling, data verification, and timing operations within the InfiniFlow project. These utilities include:

• Encoding image files into base64 strings (commonly used for embedding image data).

• Comparing two files by their cryptographic hash values to verify content equality.

• A decorator to repeatedly invoke a function until a condition is met or a timeout occurs.

• Checking if a list of dictionaries is sorted by a specified field, either ascending or descending.

This file acts as a foundational utility module that can be imported across the project to streamline common workflows related to file processing, data integrity checks, and synchronization/waiting mechanisms.

Detailed Documentation

Functions

encode_avatar(image_path)

Encodes an image file as a base64 string.

Parameters

• image_path (str or Path): The file path to the image to be encoded.

Returns

• str: Base64-encoded string of the image's binary content.

Description

Reads the image file in binary mode, encodes the contents using base64 encoding, and returns the result as a UTF-8 string. This is useful for embedding images directly into data streams or JSON payloads.

Usage Example

from pathlib import Path

encoded_image = encode_avatar(Path("/path/to/avatar.png"))
print(encoded_image)  # Outputs a base64 string representing the image content

compare_by_hash(file1, file2, algorithm="sha256")

Compares two files by computing their cryptographic hash digests.

Parameters

• file1 (str or Path): Path to the first file.

• file2 (str or Path): Path to the second file.

• algorithm (str, optional): Hash algorithm to use (default "sha256"). Supported algorithms depend on the hashlib module.

Returns

• bool: True if both files have identical hash digests (content is the same), otherwise False.

Description

The function reads both files in chunks of 8192 bytes to efficiently handle large files without loading them entirely into memory. It calculates the hash digest of each file using the specified algorithm and compares the results.

Usage Example

result = compare_by_hash("file1.txt", "file2.txt")
if result:
    print("Files are identical.")
else:
    print("Files differ.")

wait_for(timeout=10, interval=1, error_msg="Timeout")

Decorator factory that creates a decorator to repeatedly call a function until it returns True or a timeout is reached.

Parameters

• timeout (int or float, optional): Total time in seconds before giving up (default 10).

• interval (int or float, optional): Time in seconds between function calls (default 1).

• error_msg (str, optional): Error message used when timeout occurs (default "Timeout").

Returns

• decorator (function): A decorator that wraps the target function.

Description

The decorator repeatedly invokes the decorated function until it returns True. If the function does not return True before the timeout expires, it raises an AssertionError with the specified error message. This utility is useful for polling-style waits, for example waiting for resources or conditions to become available.

Usage Example

@wait_for(timeout=5, interval=0.5, error_msg="Condition not met in time")
def check_service_ready():
    # Return True when the service is ready, False otherwise
    return service.is_ready()

# This will keep checking until the service is ready or timeout occurs
check_service_ready()

is_sorted(data, field, descending=True)

Checks if a list of dictionaries is sorted by a specific field.

Parameters

• data (list of dict): List of dictionaries to check.

• field (str): The key in the dictionaries to use for sorting comparison.

• descending (bool, optional): Whether to check for descending order (default True).

Returns

• bool: True if the data is sorted according to the specified field and order, otherwise False.

Description

Extracts the values corresponding to field from each dictionary in data and verifies whether they are sorted either descending or ascending. It uses pairwise comparison with zip to check ordering efficiently.

Usage Example

records = [
    {"timestamp": 1620000000, "value": 10},
    {"timestamp": 1610000000, "value": 20},
    {"timestamp": 1600000000, "value": 5},
]

print(is_sorted(records, "timestamp"))  # True if descending order by timestamp
print(is_sorted(records, "timestamp", descending=False))  # False

Implementation Details & Algorithms

• File Hashing: Uses incremental hashing with a chunk size of 8192 bytes to handle large files efficiently without high memory consumption.

• Base64 Encoding: Reads entire file content at once, suitable for typical avatar images which are generally small.

• Polling Decorator: Implements a time-based retry loop using time.time() for elapsed time measurement and time.sleep() for intervals.

• Sorting Check: Uses Python's built-in all() and zip() functions to verify sorted order in a memory-efficient and Pythonic way.

Interaction with Other Parts of the System

• The functions here are general-purpose utilities likely imported by different modules throughout the InfiniFlow project.

• encode_avatar might be used in user profile or UI components to embed image data.

• compare_by_hash is useful in modules handling file synchronization, caching, or validation.

• wait_for is designed for synchronization in asynchronous or event-driven components.

• is_sorted likely supports data validation or processing pipelines that require time-ordered data verification.

This file acts as a foundational utility layer supporting higher-level business logic and system workflows.

Mermaid Diagram

classDiagram
    class encode_avatar {
        +image_path: str or Path
        +return: str (base64 encoded)
    }
    class compare_by_hash {
        +file1: str or Path
        +file2: str or Path
        +algorithm: str = "sha256"
        +return: bool
        -_calc_hash(file_path)
    }
    class wait_for {
        +timeout: int = 10
        +interval: int = 1
        +error_msg: str = "Timeout"
        +__call__(func)
        +wrapper(*args, **kwargs)
    }
    class is_sorted {
        +data: list of dict
        +field: str
        +descending: bool = True
        +return: bool
    }

Summary

The init.py file is a concise utility module providing essential file encoding, file comparison, polling, and sorting verification functions. These utilities enable consistent, reusable operations across the InfiniFlow project, helping improve code modularity and maintainability.