init.py

Overview

The init.py file provides a collection of utility functions primarily focused on file handling, data encoding, comparison, timing control, and data validation. This file appears to serve as a helper module in the larger InfiniFlow project, offering reusable tools that can be imported into other parts of the system.

Key functionalities include:

• Encoding image files to base64 strings for avatar processing.

• Comparing two files by computing and matching their cryptographic hashes.

• Providing a decorator to repeatedly execute a function until success or timeout.

• Checking if a list of dictionaries is sorted by a specific field.

Detailed Descriptions

Function: encode_avatar

def encode_avatar(image_path)

Purpose

Converts an image file into a base64-encoded string, commonly used for embedding images in text-based formats such as JSON or HTML.

Parameters

• image_path (str or Path): The file path of the image to encode.

Returns

• str: The base64-encoded string representation of the image content.

Usage Example

encoded_image = encode_avatar("/path/to/avatar.png")
print(encoded_image)  # Outputs a base64 string

Implementation Details

• Opens the file in binary mode.

• Reads entire contents.

• Uses the base64 standard library to encode the binary data.

• Decodes the result to UTF-8 text.

Function: compare_by_hash

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

Purpose

Determines whether two files are identical by comparing their cryptographic hash values.

Parameters

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

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

• algorithm (str, optional): Hashing algorithm to use (default "sha256"). Must be supported by Python's hashlib.

Returns

• bool: True if the two files have the same hash, False otherwise.

Usage Example

if compare_by_hash("fileA.txt", "fileB.txt"):
    print("Files are identical.")
else:
    print("Files differ.")

Implementation Details

• Defines an inner helper _calc_hash that:

  • Creates a new hashlib object with the specified algorithm.

  • Reads the file in 8KB chunks to avoid loading the entire file into memory.

  • Updates the hash with each chunk.

  • Returns the hexadecimal digest.

• Compares the hex digests of both files.

Function: wait_for

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

Purpose

A decorator factory that retries the decorated function repeatedly until it returns True or a timeout is reached.

Parameters

• timeout (int or float, optional): Maximum time in seconds to keep retrying (default 10).

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

• error_msg (str, optional): Message for assertion error upon timeout (default "Timeout").

Returns

• function: A decorator that wraps the target function with retry logic.

Usage Example

@wait_for(timeout=5, interval=0.5, error_msg="Operation failed after 5 seconds")
def check_condition():
    # Return True when condition met, False otherwise
    return some_condition_met()

check_condition()

Implementation Details

• Uses functools.wraps to preserve metadata of the decorated function.

• Records start time.

• Calls the decorated function repeatedly:

  • If the function returns True, returns immediately.

  • If elapsed time exceeds timeout, raises AssertionError with error_msg.

  • Sleeps for interval seconds between retries.

Function: is_sorted

def is_sorted(data, field, descending=True)

Purpose

Checks if a list of dictionaries is sorted by a specified field in either descending or ascending order.

Parameters

• data (List[dict]): A list where each element is a dictionary.

• field (str): The key in each dictionary to sort by.

• descending (bool, optional): If True, checks for descending order; if False, ascending (default True).

Returns

• bool: True if the list is sorted by the specified field as requested, False otherwise.

Usage Example

logs = [
    {"timestamp": 1620000000},
    {"timestamp": 1619999990},
    {"timestamp": 1619999980},
]

print(is_sorted(logs, "timestamp"))  # True (descending)
print(is_sorted(logs, "timestamp", descending=False))  # False (not ascending)

Implementation Details

• Extracts the values from the specified field in each dictionary.

• Uses zip to compare each pair of adjacent values.

• For descending order, verifies each value is greater than or equal to the next.

• For ascending order, verifies each value is less than or equal to the next.

Important Implementation Details and Algorithms

• File Hashing: The compare_by_hash function reads files in chunks (8192 bytes) to handle large files efficiently without excessive memory usage.

• Retry Decorator: wait_for provides a generic polling mechanism, useful for waiting on transient conditions such as network availability or process completion.

• Sorting Check: is_sorted uses Python's built-in all() and zip() to elegantly compare adjacent items, avoiding explicit loops.

Interactions with Other System Components

• This module is designed as a utility library — it likely supports higher-level modules in InfiniFlow that handle:

  • User avatars or profile images (encode_avatar).

  • File synchronization or integrity verification (compare_by_hash).

  • Polling for asynchronous conditions or resource readiness (wait_for).

  • Data validation or processing steps requiring sorted inputs (is_sorted).

• Other parts of the system can import these functions to simplify file handling, encoding, and control flow logic without duplicating code.

Diagram: Function Flowchart

flowchart TD
    A[encode_avatar(image_path)]
    B[compare_by_hash(file1, file2, algorithm)]
    B_inner[_calc_hash(file_path)]
    C[wait_for(timeout, interval, error_msg)]
    C_decorator[decorator(func)]
    C_wrapper[wrapper(*args, **kwargs)]
    D[is_sorted(data, field, descending)]

    B --> B_inner
    C --> C_decorator --> C_wrapper

Diagram Explanation:

• compare_by_hash contains an inner helper _calc_hash.

• wait_for returns a decorator, which wraps the target function.

• Each function operates independently, serving distinct utilities.

Summary

This init.py file is a utility module providing core reusable functions for encoding images, verifying file identity, controlled waiting/retry logic, and data sorting verification. Its design emphasizes memory efficiency, modularity, and simplicity. It is intended to be imported and leveraged by other components within the InfiniFlow system that require these foundational capabilities.