avdb-py/avdb/analysis.py

import os
import sys
import numpy as np
import csv
import uuid
from typing import List, Tuple

from .db import DBConnection, DBOps

BATCH_SIZE = 250
def run_analysis():
    """
    Iterate over all images in the 'img/' directory and process them using the DBOps class.
    
    This function processes each image in the 'img/' directory, initializes the DBOps,
    and performs the analysis for each image.
    """
    img_folder = "img"
    if not os.path.exists(img_folder):
        print(f"Error: The folder '{img_folder}' does not exist.")
        return

    # Get all subdirectories in the 'img/' folder
    subdirectories = [
        d for d in os.listdir(img_folder)
        if os.path.isdir(os.path.join(img_folder, d))
    ]

    if not subdirectories:
        print(f"No subdirectories found in the folder '{img_folder}'.")
        return

    print(f"Found {len(subdirectories)} subdirectories in the folder '{img_folder}'.")

    # Initialize the database connection and operations
    db_conn = DBConnection('testdb')
    db_ops = DBOps(db_conn)

    for subdir in subdirectories:
    # subdirs = ["TESTIMG", "IMG_9997", "IMG_9998", "IMG_9999", "IMG_2327"]
    # for subdir in subdirs:
        subdir_path = os.path.join(img_folder, subdir)
        # Look for the image file in the subdirectory

        img_file = f"{subdir}.jpeg"
        img_path = os.path.join(subdir_path, img_file)
        print(f"\nProcessing image: {img_path}")

        try:
            # Process the image analysis
            db_ops.process_image_analysis(img_path)
            print(f"Successfully processed image: {img_path}")
        except Exception as e:
            print(f"Error processing image {img_path}: {e}")

        print("\nAnalysis of all images completed successfully.")

def convert_mmap_to_csv(mmap_file_path: str, csv_file_path: str, rows: int = 81) -> None:
    """
    Convert a memory-mapped file to a CSV file.

    Args:
        mmap_file_path (str): Path to the memory-mapped file.
        csv_file_path (str): Path to the output CSV file.
        rows (int): Number of rows in the memory-mapped data.
    """
    # Load the memory-mapped data
    mmap_data = np.memmap(mmap_file_path, mode='r', dtype='float32', shape=(rows, 48))

    # Convert the data to a list of lists for CSV writing
    data_for_csv = mmap_data.tolist()

    # Write the data to a CSV file
    with open(csv_file_path, 'w', newline='') as csv_file:
        writer = csv.writer(csv_file)
        writer.writerows(data_for_csv)

def insert_tile_data(self, rows: List[Tuple]) -> None:
    """
    Insert tile data into the pyramiddata table using batch insertion.

    Args:
        rows (List[Tuple]): List of tuples containing tile data to insert.
    """
    query_str = """
        INSERT INTO pyramiddata (
            img_id, level, tilex, tiley, x, y, x0, y0, w, h, hash, tile_data
        )
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    """
    print(len(rows))

    with self._connection.get_cursor() as cursor:
        # Use execute_batch for efficient batch insertion
        extras.execute_batch(cursor, query_str, rows, page_size=BATCH_SIZE)
        extras.commit()

# def process_and_insert_tile_data_batched(
#     self,
#     img_id: int,
#     img_dims: Tuple[int, int],
#     level: int,
#     tile_file: str
# ) -> None:
#     """
#     Process and insert tile data for a specific level of an image.
#     Args:
#         img_id (int): ID of the image.
#         img_dims (Tuple[int, int]): Dimensions of the image (width, height).
#         level (int): Pyramid level.
#         tile_file (str): The memory-mapped tile file.
#     """
#     # Calculate the downsampled resolution for this level
#     resolution = 2 ** level
#     img_w, img_h = img_dims

#     # Load the memory-mapped tile data
#     tile_data = self.load_memmapped_tile_data(tile_file)

#     print(f"img dims: {img_dims}")
#     print(f"tile data shape: {tile_data.shape}")

#     tilew = img_w // resolution
#     tileh = img_h // resolution
#     width_t = img_w // tilew
#     height_t = img_h // tileh

#     grid_size = 1

#     def gen_tiles():
#         counter = 0
#         # Loop through each tile
#         for tiley in range(tileh):
#             for tilex in range(tilew):
#                 x0 = tilex * width_t
#                 y0 = tiley * height_t
#                 w = grid_size
#                 h = 1

#                 try:
#                     tiledata = list(
#                         (
#                             np.r_[
#                                 tile_data[
#                                     (counter * 3):
#                                     ((counter + 1) * 3)
#                                 ]
#                             ]
#                         ).flatten().tolist()
#                     )
#                     counter += 1
#                     yield (
#                         img_id,
#                         level,
#                         tilex,
#                         tiley,
#                         width_t,
#                         height_t,
#                         x0,
#                         y0,
#                         w,
#                         h,
#                         str(uuid.uuid4()),
#                         list(map(self.to_unsigned, tiledata))
#                     )
#                 except Exception as e:
#                     print(f"ERROR: {e}")
#                 counter += 1

#     all_for_insertion = list(gen_tiles())
#     print(f"Inserting {len(all_for_insertion)} tiles")

#     # # Batch the data for insertion
#     # batched = [
#     #     all_for_insertion[BATCH_SIZE * i: BATCH_SIZE * (i + 1)]
#     #     for i in range(int(len(all_for_insertion) / BATCH_SIZE) + 1)
#     # ]
#     # print(f"Inserting {len(batched)} batches")

#     # for batch in batched:
#     #     if batch:  # Ensure the batch is not empty
#     #         self.insert_tile_data(batch)

#     self.insert_tile_data(all_for_insertion)

if __name__ == "__main__":
    run_analysis()