Reduce Function

      from celline import Project
from celline.functions.reduce import Reduce

# Create project
project = Project("./my-project")

# Create Reduce function instance
reduce_function = Reduce()

# Execute function
result = project.call(reduce_function)

    

      from celline import Project
from celline.functions.count import Count
from celline.functions.reduce import Reduce
from celline.functions.preprocess import Preprocess

# Create project
project = Project("./my-project")

# Complete counting
project.call(Count(nthread=8))

# Reduce storage usage
project.call(Reduce())

# Continue with preprocessing
project.call(Preprocess())

    

      # Process files bottom-up to handle directories properly
for foldername, subfolders, filenames in os.walk(
    target_path, topdown=False, followlinks=True
):
    # Remove files not in keep list
    for filename in filenames:
        rel_path = os.path.relpath(
            os.path.join(foldername, filename), target_path
        )
        if rel_path not in FILES_TO_KEEP:
            os.remove(os.path.join(foldername, filename))
    
    # Remove empty directories and symbolic links
    for subfolder in subfolders:
        full_path = os.path.join(foldername, subfolder)
        if os.path.islink(full_path):
            os.remove(full_path)
        elif os.path.isdir(full_path) and not os.listdir(full_path):
            os.rmdir(full_path)

    

      # Optimal pipeline timing
project.call(Count(nthread=8))           # Count reads
# ... extract any needed secondary analysis
project.call(Reduce())                   # Reduce storage
project.call(Preprocess())               # Continue pipeline

    

      # Check available space before reduction
import shutil

total, used, free = shutil.disk_usage("/shared/project/")
print(f"Free space: {free // (2**30)} GB")

# Run reduction to free space
project.call(Reduce())

# Check space after reduction
total, used, free = shutil.disk_usage("/shared/project/")
print(f"Free space after reduction: {free // (2**30)} GB")

    

      # Automated storage management
def smart_reduce(project, threshold_gb=100):
    """Reduce storage if free space below threshold."""
    import shutil
    
    total, used, free = shutil.disk_usage(Config.PROJ_ROOT)
    free_gb = free // (2**30)
    
    if free_gb < threshold_gb:
        print(f"Low disk space ({free_gb} GB), running reduction...")
        project.call(Reduce())
    else:
        print(f"Sufficient disk space ({free_gb} GB), skipping reduction")

# Use in pipeline
smart_reduce(project, threshold_gb=100)

    

      import os

def storage_report(project_root):
    """Generate storage usage report."""
    total_size = 0
    sample_sizes = {}
    
    for sample_dir in os.listdir(f"{project_root}/resources"):
        sample_path = f"{project_root}/resources/{sample_dir}"
        if os.path.isdir(sample_path):
            size = sum(
                os.path.getsize(os.path.join(dirpath, filename))
                for dirpath, dirnames, filenames in os.walk(sample_path)
                for filename in filenames
            )
            sample_sizes[sample_dir] = size // (2**30)  # GB
            total_size += size
    
    print(f"Total storage: {total_size // (2**30)} GB")
    for sample, size in sorted(sample_sizes.items()):
        print(f"  {sample}: {size} GB")

    

      # Check which samples were processed
processed_samples = []
for sample in project.samples:
    count_dir = f"resources/{sample}/counted"
    if os.path.exists(f"{count_dir}/outs/raw_feature_bc_matrix.h5"):
        print(f"{sample}: Not reduced")
    else:
        print(f"{sample}: Already reduced")
        processed_samples.append(sample)