Table of Contents

Basic Examples

Example 1: Generate a Simple Dataset

python scripts/generate_dataset.py --samples 100 --output ./my_first_dataset

my_first_dataset/
├── npz/
│ ├── sample_00000_circle.npz
│ ├── sample_00001_rectangle.npz
│ └── ...
└── preview/
 ├── object/
 ├── hologram/
 └── reconstruction/

Example 2: Load and Visualize Samples

import numpy as np
import matplotlib.pyplot as plt
from pathlib import Path
# Load a sample
data_path = Path("my_first_dataset/npz/sample_00000_circle.npz")
data = np.load(data_path)
# Extract arrays
object_img = data['object']
hologram = data['hologram']
reconstruction = data['reconstruction']
# Create visualization
fig, axes = plt.subplots(1, 3, figsize=(15, 5))
axes[0].imshow(object_img, cmap='gray')
axes[0].set_title('Object Domain')
axes[0].axis('off')
axes[1].imshow(hologram, cmap='gray')
axes[1].set_title('Hologram')
axes[1].axis('off')
axes[2].imshow(reconstruction, cmap='gray')
axes[2].set_title('Reconstruction')
axes[2].axis('off')
plt.tight_layout()
plt.savefig('sample_visualization.png', dpi=150, bbox_inches='tight')
plt.show()
print(f"Object shape: {object_img.shape}")
print(f"Hologram intensity range: [{hologram.min():.3f}, {hologram.max():.3f}]")
print(f"Reconstruction quality (MSE): {np.mean((object_img - reconstruction)**2):.6f}")

Object shape: (256, 256)
Hologram intensity range: [0.000, 1.000]
Reconstruction quality (MSE): 0.001234

Example 3: Train a Basic Reconstruction Model

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import numpy as np
from pathlib import Path
# Define dataset
class HologramDataset(Dataset):
 def __init__(self, data_dir):
 self.samples = sorted(Path(data_dir).glob("*.npz"))
 
 def __len__(self):
 return len(self.samples)
 
 def __getitem__(self, idx):
 data = np.load(self.samples[idx])
 hologram = torch.from_numpy(data['hologram']).float().unsqueeze(0)
 target = torch.from_numpy(data['object']).float().unsqueeze(0)
 return hologram, target
# Simple U-Net style model
class SimpleReconstructionNet(nn.Module):
 def __init__(self):
 super().__init__()
 self.encoder = nn.Sequential(
 nn.Conv2d(1, 32, 3, padding=1),
 nn.ReLU(),
 nn.Conv2d(32, 64, 3, padding=1),
 nn.ReLU(),
 nn.MaxPool2d(2)
 )
 self.decoder = nn.Sequential(
 nn.Upsample(scale_factor=2, mode='bilinear', align_corners=False),
 nn.Conv2d(64, 32, 3, padding=1),
 nn.ReLU(),
 nn.Conv2d(32, 1, 3, padding=1),
 nn.Sigmoid()
 )
 
 def forward(self, x):
 x = self.encoder(x)
 x = self.decoder(x)
 return x
# Training setup
dataset = HologramDataset('my_first_dataset/npz')
train_size = int(0.8 * len(dataset))
val_size = len(dataset) - train_size
train_dataset, val_dataset = torch.utils.data.random_split(dataset, [train_size, val_size])
train_loader = DataLoader(train_dataset, batch_size=16, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=16)
model = SimpleReconstructionNet()
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
# Training loop
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model.to(device)
for epoch in range(10):
 model.train()
 train_loss = 0.0
 for holograms, targets in train_loader:
 holograms, targets = holograms.to(device), targets.to(device)
 
 optimizer.zero_grad()
 outputs = model(holograms)
 loss = criterion(outputs, targets)
 loss.backward()
 optimizer.step()
 
 train_loss += loss.item()
 
 # Validation
 model.eval()
 val_loss = 0.0
 with torch.no_grad():
 for holograms, targets in val_loader:
 holograms, targets = holograms.to(device), targets.to(device)
 outputs = model(holograms)
 val_loss += criterion(outputs, targets).item()
 
 print(f"Epoch {epoch+1}/10 - Train Loss: {train_loss/len(train_loader):.6f}, "
 f"Val Loss: {val_loss/len(val_loader):.6f}")
# Save model
torch.save(model.state_dict(), 'reconstruction_model.pth')
print("Model saved to reconstruction_model.pth")

Epoch 1/10 - Train Loss: 0.045123, Val Loss: 0.038456
Epoch 2/10 - Train Loss: 0.032145, Val Loss: 0.029876
...
Epoch 10/10 - Train Loss: 0.012345, Val Loss: 0.015678
Model saved to reconstruction_model.pth

Intermediate Examples

Example 4: Custom Shape Generator

from hologen.shapes import BaseShapeGenerator
from hologen.types import GridSpec, ArrayFloat, ArrayComplex
from numpy.random import Generator
import numpy as np
class StarGenerator(BaseShapeGenerator):
 """Generate star-shaped objects with configurable points."""
 
 __slots__ = ("min_radius", "max_radius", "num_points")
 
 def __init__(self, name: str = "star", min_radius: float = 0.1, 
 max_radius: float = 0.2, num_points: int = 5):
 super().__init__(name=name)
 self.min_radius = min_radius
 self.max_radius = max_radius
 self.num_points = num_points
 
 def generate(self, grid: GridSpec, rng: Generator) -> ArrayFloat:
 """Generate a star-shaped binary amplitude pattern."""
 canvas = self._empty_canvas(grid)
 
 # Random parameters
 outer_radius = rng.uniform(self.min_radius, self.max_radius) * min(grid.height, grid.width)
 inner_radius = outer_radius * 0.4
 center_y = rng.uniform(0.3, 0.7) * grid.height
 center_x = rng.uniform(0.3, 0.7) * grid.width
 rotation = rng.uniform(0, 2 * np.pi)
 
 # Create star vertices (alternating outer and inner radii)
 angles = np.linspace(0, 2*np.pi, 2*self.num_points, endpoint=False) + rotation
 radii = np.tile([outer_radius, inner_radius], self.num_points)
 
 vertices_y = center_y + radii * np.sin(angles)
 vertices_x = center_x + radii * np.cos(angles)
 
 # Fill star using polygon rasterization
 yy, xx = np.ogrid[:grid.height, :grid.width]
 
 # Simple point-in-polygon test (for each pixel, check if inside star)
 for i in range(len(vertices_y)):
 j = (i + 1) % len(vertices_y)
 # Create triangle from center to edge
 mask = self._point_in_triangle(
 xx, yy, 
 center_x, center_y,
 vertices_x[i], vertices_y[i],
 vertices_x[j], vertices_y[j]
 )
 canvas[mask] = 1.0
 
 return self._clamp(canvas)
 
 def _point_in_triangle(self, px, py, x1, y1, x2, y2, x3, y3):
 """Check if points (px, py) are inside triangle (x1,y1)-(x2,y2)-(x3,y3)."""
 def sign(px, py, x1, y1, x2, y2):
 return (px - x2) * (y1 - y2) - (x1 - x2) * (py - y2)
 
 d1 = sign(px, py, x1, y1, x2, y2)
 d2 = sign(px, py, x2, y2, x3, y3)
 d3 = sign(px, py, x3, y3, x1, y1)
 
 has_neg = (d1 < 0) | (d2 < 0) | (d3 < 0)
 has_pos = (d1 > 0) | (d2 > 0) | (d3 > 0)
 
 return ~(has_neg & has_pos)
# Use custom generator in pipeline
from hologen.converters import ObjectDomainProducer, ObjectToHologramConverter, HologramDatasetGenerator
from hologen.converters import default_converter
from hologen.types import HolographyConfig, HolographyMethod, OpticalConfig
from hologen.utils.io import NumpyDatasetWriter
from pathlib import Path
# Create custom producer with star generator
star_gen = StarGenerator(name="star", min_radius=0.1, max_radius=0.2, num_points=5)
producer = ObjectDomainProducer(shape_generators=(star_gen,))
# Create pipeline
grid = GridSpec(height=512, width=512, pixel_pitch=6.4e-6)
optics = OpticalConfig(wavelength=532e-9, propagation_distance=0.05)
config = HolographyConfig(grid=grid, optics=optics, method=HolographyMethod.INLINE)
converter = default_converter()
generator = HologramDatasetGenerator(object_producer=producer, converter=converter)
# Generate dataset with star shapes
rng = np.random.default_rng(42)
samples = list(generator.generate(count=50, config=config, rng=rng))
# Save dataset
writer = NumpyDatasetWriter(save_preview=True)
writer.save(samples, output_dir=Path("star_dataset"))
print(f"Generated {len(samples)} star-shaped samples")

Example 5: Custom Noise Model

from hologen.noise.base import BaseNoiseModel
from hologen.types import ArrayFloat, HolographyConfig
from numpy.random import Generator
import numpy as np
from scipy.ndimage import gaussian_filter
class AtmosphericTurbulenceModel(BaseNoiseModel):
 """Simulate atmospheric turbulence using phase screens."""
 
 __slots__ = ("turbulence_strength", "correlation_length")
 
 def __init__(self, name: str = "turbulence", 
 turbulence_strength: float = 0.5,
 correlation_length: float = 10.0):
 super().__init__(name=name)
 self.turbulence_strength = turbulence_strength
 self.correlation_length = correlation_length
 
 def apply(self, hologram: ArrayFloat, config: HolographyConfig, 
 rng: Generator) -> ArrayFloat:
 """Apply atmospheric turbulence to hologram intensity."""
 
 # Generate random phase screen
 phase_screen = rng.normal(0, 1, hologram.shape)
 
 # Apply spatial correlation
 phase_screen = gaussian_filter(phase_screen, sigma=self.correlation_length)
 
 # Scale by turbulence strength
 phase_screen *= self.turbulence_strength
 
 # Convert hologram intensity to complex field
 amplitude = np.sqrt(hologram)
 
 # Apply phase distortion
 distorted_field = amplitude * np.exp(1j * phase_screen)
 
 # Return distorted intensity
 return np.abs(distorted_field) ** 2
# Use custom noise model in pipeline
from hologen.converters import (
 ObjectDomainProducer, ObjectToHologramConverter, 
 HologramDatasetGenerator, default_object_producer
)
from hologen.holography.inline import InlineHolographyStrategy
from hologen.types import HolographyMethod
from hologen.utils.io import NumpyDatasetWriter
from pathlib import Path
# Create noise model
turbulence = AtmosphericTurbulenceModel(
 name="turbulence",
 turbulence_strength=0.3,
 correlation_length=15.0
)
# Create converter with custom noise
strategies = {HolographyMethod.INLINE: InlineHolographyStrategy()}
converter = ObjectToHologramConverter(
 strategy_mapping=strategies,
 noise_model=turbulence
)
# Create pipeline
producer = default_object_producer()
generator = HologramDatasetGenerator(object_producer=producer, converter=converter)
# Generate dataset with turbulence
grid = GridSpec(height=512, width=512, pixel_pitch=6.4e-6)
optics = OpticalConfig(wavelength=532e-9, propagation_distance=0.05)
config = HolographyConfig(grid=grid, optics=optics, method=HolographyMethod.INLINE)
rng = np.random.default_rng(42)
samples = list(generator.generate(count=100, config=config, rng=rng))
# Save dataset
writer = NumpyDatasetWriter(save_preview=True)
writer.save(samples, output_dir=Path("turbulent_dataset"))
print(f"Generated {len(samples)} samples with atmospheric turbulence")

Example 6: Pipeline Customization

from hologen.converters import (
 ObjectDomainProducer, ObjectToHologramConverter, 
 HologramDatasetGenerator, default_object_producer, default_converter
)
from hologen.types import HologramSample, HolographyConfig
from hologen.utils.io import NumpyDatasetWriter
from pathlib import Path
import numpy as np
from scipy.ndimage import gaussian_filter
from collections.abc import Iterable
class CustomHologramGenerator(HologramDatasetGenerator):
 """Extended generator with preprocessing and postprocessing."""
 
 def __init__(self, object_producer, converter, 
 blur_sigma: float = 0.0,
 normalize_output: bool = True):
 super().__init__(object_producer, converter)
 self.blur_sigma = blur_sigma
 self.normalize_output = normalize_output
 
 def generate(self, count: int, config: HolographyConfig, 
 rng, **kwargs) -> Iterable[HologramSample]:
 """Generate samples with custom processing."""
 
 for sample in super().generate(count, config, rng, **kwargs):
 # Preprocessing: blur object slightly
 if self.blur_sigma > 0:
 blurred_object = gaussian_filter(
 sample.object_sample.pixels, 
 sigma=self.blur_sigma
 )
 sample.object_sample.pixels[:] = blurred_object
 
 # Postprocessing: normalize hologram to [0, 1]
 if self.normalize_output:
 hologram_min = sample.hologram.min()
 hologram_max = sample.hologram.max()
 if hologram_max > hologram_min:
 sample.hologram[:] = (sample.hologram - hologram_min) / (hologram_max - hologram_min)
 
 yield sample
# Create custom pipeline
producer = default_object_producer()
converter = default_converter()
custom_generator = CustomHologramGenerator(
 object_producer=producer,
 converter=converter,
 blur_sigma=1.0, # Slight blur on objects
 normalize_output=True # Normalize holograms
)
# Generate dataset
grid = GridSpec(height=512, width=512, pixel_pitch=6.4e-6)
optics = OpticalConfig(wavelength=532e-9, propagation_distance=0.05)
config = HolographyConfig(grid=grid, optics=optics, method=HolographyMethod.INLINE)
rng = np.random.default_rng(42)
samples = list(custom_generator.generate(count=100, config=config, rng=rng))
# Save dataset
writer = NumpyDatasetWriter(save_preview=True)
writer.save(samples, output_dir=Path("custom_pipeline_dataset"))
print(f"Generated {len(samples)} samples with custom pipeline")

Advanced Examples

Example 7: Multi-Scale Datasets

from hologen.converters import (
 HologramDatasetGenerator, default_object_producer, default_converter
)
from hologen.types import GridSpec, OpticalConfig, HolographyConfig, HolographyMethod
from hologen.utils.io import NumpyDatasetWriter
from pathlib import Path
import numpy as np
def generate_multiscale_dataset(scales: list[int], samples_per_scale: int, 
 base_output_dir: Path):
 """Generate datasets at multiple resolutions."""
 
 producer = default_object_producer()
 converter = default_converter()
 generator = HologramDatasetGenerator(object_producer=producer, converter=converter)
 
 optics = OpticalConfig(wavelength=532e-9, propagation_distance=0.05)
 rng = np.random.default_rng(42)
 
 for scale in scales:
 print(f"Generating {samples_per_scale} samples at {scale}x{scale} resolution...")
 
 # Create grid for this scale
 grid = GridSpec(height=scale, width=scale, pixel_pitch=6.4e-6)
 config = HolographyConfig(grid=grid, optics=optics, method=HolographyMethod.INLINE)
 
 # Generate samples
 samples = list(generator.generate(count=samples_per_scale, config=config, rng=rng))
 
 # Save to scale-specific directory
 output_dir = base_output_dir / f"scale_{scale}"
 writer = NumpyDatasetWriter(save_preview=True)
 writer.save(samples, output_dir=output_dir)
 
 print(f" Saved to {output_dir}")
# Generate multi-scale dataset
scales = [128, 256, 512, 1024]
generate_multiscale_dataset(
 scales=scales,
 samples_per_scale=50,
 base_output_dir=Path("multiscale_dataset")
)
print("Multi-scale dataset generation complete!")

multiscale_dataset/
├── scale_128/
│ ├── npz/
│ └── preview/
├── scale_256/
│ ├── npz/
│ └── preview/
├── scale_512/
│ ├── npz/
│ └── preview/
└── scale_1024/
 ├── npz/
 └── preview/

Example 8: Hybrid Object Types

from hologen.converters import (
 ObjectDomainProducer, ObjectToHologramConverter, 
 HologramDatasetGenerator, default_converter
)
from hologen.shapes import CircleGenerator, RectangleGenerator
from hologen.types import (
 GridSpec, OpticalConfig, HolographyConfig, HolographyMethod,
 OutputConfig, FieldRepresentation, ComplexHologramSample
)
from hologen.utils.io import ComplexFieldWriter
from pathlib import Path
import numpy as np
def generate_hybrid_dataset(amplitude_samples: int, phase_samples: int, 
 output_dir: Path):
 """Generate dataset with mixed amplitude and phase objects."""
 
 # Setup
 grid = GridSpec(height=512, width=512, pixel_pitch=6.4e-6)
 optics = OpticalConfig(wavelength=532e-9, propagation_distance=0.05)
 config = HolographyConfig(grid=grid, optics=optics, method=HolographyMethod.INLINE)
 
 # Create generators
 generators = (
 CircleGenerator(name="circle", min_radius=0.08, max_radius=0.18),
 RectangleGenerator(name="rectangle", min_scale=0.1, max_scale=0.35)
 )
 producer = ObjectDomainProducer(shape_generators=generators)
 
 # Output configuration for complex fields
 output_config = OutputConfig(
 object_representation=FieldRepresentation.COMPLEX,
 hologram_representation=FieldRepresentation.COMPLEX,
 reconstruction_representation=FieldRepresentation.COMPLEX
 )
 
 converter = default_converter()
 converter.output_config = output_config
 
 generator = HologramDatasetGenerator(object_producer=producer, converter=converter)
 
 rng = np.random.default_rng(42)
 all_samples = []
 
 # Generate amplitude-only objects
 print(f"Generating {amplitude_samples} amplitude-only samples...")
 amplitude_samples_list = list(generator.generate(
 count=amplitude_samples,
 config=config,
 rng=rng,
 mode="amplitude",
 use_complex=True
 ))
 all_samples.extend(amplitude_samples_list)
 
 # Generate phase-only objects
 print(f"Generating {phase_samples} phase-only samples...")
 phase_samples_list = list(generator.generate(
 count=phase_samples,
 config=config,
 rng=rng,
 mode="phase",
 phase_shift=np.pi/2,
 use_complex=True
 ))
 all_samples.extend(phase_samples_list)
 
 # Shuffle samples
 rng.shuffle(all_samples)
 
 # Save dataset
 writer = ComplexFieldWriter(save_preview=True, phase_colormap="twilight")
 writer.save(all_samples, output_dir=output_dir)
 
 print(f"Generated {len(all_samples)} hybrid samples")
 print(f" - {amplitude_samples} amplitude-only")
 print(f" - {phase_samples} phase-only")
# Generate hybrid dataset
generate_hybrid_dataset(
 amplitude_samples=50,
 phase_samples=50,
 output_dir=Path("hybrid_dataset")
)

Example 9: Custom Reconstruction Algorithm

import numpy as np
from scipy.fft import fft2, ifft2, fftshift, ifftshift
from hologen.types import GridSpec, OpticalConfig
def angular_spectrum_propagation(field: np.ndarray, distance: float, 
 wavelength: float, pixel_pitch: float) -> np.ndarray:
 """Propagate complex field using angular spectrum method.

 Args:
 field: Complex field to propagate.
 distance: Propagation distance in meters.
 wavelength: Wavelength in meters.
 pixel_pitch: Pixel pitch in meters.

 Returns:
 Propagated complex field.
 """
 height, width = field.shape
 
 # Frequency coordinates
 fy = np.fft.fftfreq(height, pixel_pitch)
 fx = np.fft.fftfreq(width, pixel_pitch)
 FX, FY = np.meshgrid(fx, fy)
 
 # Wave number
 k = 2 * np.pi / wavelength
 
 # Transfer function
 kz = np.sqrt(k**2 - (2*np.pi*FX)**2 - (2*np.pi*FY)**2 + 0j)
 H = np.exp(1j * kz * distance)
 
 # Propagate
 field_fft = fft2(field)
 propagated_fft = field_fft * H
 propagated = ifft2(propagated_fft)
 
 return propagated
def custom_reconstruction(hologram: np.ndarray, grid: GridSpec, 
 optics: OpticalConfig) -> np.ndarray:
 """Custom reconstruction with preprocessing and filtering.

 Args:
 hologram: Hologram intensity.
 grid: Grid specification.
 optics: Optical configuration.

 Returns:
 Reconstructed object amplitude.
 """
 # Convert intensity to complex field (assume amplitude from sqrt)
 hologram_field = np.sqrt(hologram).astype(np.complex128)
 
 # Apply preprocessing: background subtraction
 background = np.median(hologram)
 hologram_field = hologram_field - background
 
 # Propagate back to object plane
 reconstructed = angular_spectrum_propagation(
 hologram_field,
 distance=-optics.propagation_distance, # Negative for back-propagation
 wavelength=optics.wavelength,
 pixel_pitch=grid.pixel_pitch
 )
 
 # Apply postprocessing: Wiener filter
 amplitude = np.abs(reconstructed)
 
 # Frequency domain filtering
 amplitude_fft = fft2(amplitude)
 
 # Create low-pass filter
 fy = np.fft.fftfreq(grid.height, grid.pixel_pitch)
 fx = np.fft.fftfreq(grid.width, grid.pixel_pitch)
 FX, FY = np.meshgrid(fx, fy)
 freq_radius = np.sqrt(FX**2 + FY**2)
 cutoff = 1.0 / (10 * grid.pixel_pitch) # Cutoff frequency
 filter_mask = np.exp(-(freq_radius / cutoff)**2)
 
 # Apply filter
 filtered_fft = amplitude_fft * filter_mask
 filtered_amplitude = np.abs(ifft2(filtered_fft))
 
 # Normalize
 filtered_amplitude = (filtered_amplitude - filtered_amplitude.min()) / \
 (filtered_amplitude.max() - filtered_amplitude.min() + 1e-10)
 
 return filtered_amplitude
# Test custom reconstruction
from pathlib import Path
# Load sample
data = np.load("my_first_dataset/npz/sample_00000_circle.npz")
hologram = data['hologram']
ground_truth = data['object']
# Reconstruct using custom algorithm
grid = GridSpec(height=256, width=256, pixel_pitch=4.65e-6)
optics = OpticalConfig(wavelength=532e-9, propagation_distance=0.02)
reconstruction = custom_reconstruction(hologram, grid, optics)
# Evaluate quality
mse = np.mean((ground_truth - reconstruction)**2)
psnr = 10 * np.log10(1.0 / (mse + 1e-10))
print(f"Reconstruction MSE: {mse:.6f}")
print(f"Reconstruction PSNR: {psnr:.2f} dB")
# Visualize
import matplotlib.pyplot as plt
fig, axes = plt.subplots(1, 3, figsize=(15, 5))
axes[0].imshow(ground_truth, cmap='gray')
axes[0].set_title('Ground Truth')
axes[1].imshow(hologram, cmap='gray')
axes[1].set_title('Hologram')
axes[2].imshow(reconstruction, cmap='gray')
axes[2].set_title('Custom Reconstruction')
plt.tight_layout()
plt.savefig('custom_reconstruction.png', dpi=150)
plt.show()

Example 10: ML Framework Integration

import torch
from torch.utils.data import Dataset, DataLoader
import numpy as np
from pathlib import Path
class ComplexHologramDataset(Dataset):
 """PyTorch dataset for complex hologram data with augmentation."""
 
 def __init__(self, data_dir: Path, transform=None):
 self.data_dir = Path(data_dir)
 self.samples = sorted(self.data_dir.glob("*_hologram.npz"))
 self.transform = transform
 
 def __len__(self):
 return len(self.samples)
 
 def __getitem__(self, idx):
 # Load hologram
 hologram_data = np.load(self.samples[idx])
 
 if 'real' in hologram_data and 'imag' in hologram_data:
 # Complex representation: stack real and imaginary as channels
 hologram = torch.stack([
 torch.from_numpy(hologram_data['real']).float(),
 torch.from_numpy(hologram_data['imag']).float()
 ], dim=0)
 else:
 # Intensity representation
 hologram = torch.from_numpy(hologram_data['hologram']).float().unsqueeze(0)
 
 # Load object
 object_path = str(self.samples[idx]).replace('_hologram', '_object')
 object_data = np.load(object_path)
 
 if 'real' in object_data and 'imag' in object_data:
 obj = torch.stack([
 torch.from_numpy(object_data['real']).float(),
 torch.from_numpy(object_data['imag']).float()
 ], dim=0)
 else:
 obj = torch.from_numpy(object_data['object']).float().unsqueeze(0)
 
 # Apply transforms
 if self.transform:
 hologram = self.transform(hologram)
 obj = self.transform(obj)
 
 return hologram, obj
# Data augmentation
import torchvision.transforms as transforms
transform = transforms.Compose([
 transforms.RandomHorizontalFlip(p=0.5),
 transforms.RandomVerticalFlip(p=0.5),
 transforms.RandomRotation(degrees=90),
])
# Create DataLoader
dataset = ComplexHologramDataset('hybrid_dataset/npz', transform=transform)
dataloader = DataLoader(
 dataset, 
 batch_size=32, 
 shuffle=True, 
 num_workers=4,
 pin_memory=True
)
print(f"Dataset size: {len(dataset)}")
for batch_idx, (holograms, objects) in enumerate(dataloader):
 print(f"Batch {batch_idx}: holograms {holograms.shape}, objects {objects.shape}")
 if batch_idx >= 2:
 break

import tensorflow as tf
import numpy as np
from pathlib import Path
def load_complex_sample(hologram_path, object_path):
 """Load complex hologram-object pair."""
 # Load hologram
 h_data = np.load(hologram_path.numpy().decode())
 if b'real' in h_data.files and b'imag' in h_data.files:
 hologram = np.stack([h_data['real'], h_data['imag']], axis=-1)
 else:
 hologram = h_data['hologram'][..., np.newaxis]
 
 # Load object
 o_data = np.load(object_path.numpy().decode())
 if b'real' in o_data.files and b'imag' in o_data.files:
 obj = np.stack([o_data['real'], o_data['imag']], axis=-1)
 else:
 obj = o_data['object'][..., np.newaxis]
 
 return hologram.astype(np.float32), obj.astype(np.float32)
def augment(hologram, obj):
 """Apply data augmentation."""
 # Random flip
 if tf.random.uniform(()) > 0.5:
 hologram = tf.image.flip_left_right(hologram)
 obj = tf.image.flip_left_right(obj)
 
 if tf.random.uniform(()) > 0.5:
 hologram = tf.image.flip_up_down(hologram)
 obj = tf.image.flip_up_down(obj)
 
 # Random rotation (90 degree increments)
 k = tf.random.uniform((), minval=0, maxval=4, dtype=tf.int32)
 hologram = tf.image.rot90(hologram, k=k)
 obj = tf.image.rot90(obj, k=k)
 
 return hologram, obj
# Create dataset
data_dir = Path("hybrid_dataset/npz")
hologram_paths = sorted(data_dir.glob("*_hologram.npz"))
object_paths = [str(p).replace('_hologram', '_object') for p in hologram_paths]
dataset = tf.data.Dataset.from_tensor_slices((
 [str(p) for p in hologram_paths],
 object_paths
))
dataset = dataset.map(
 lambda h, o: tf.py_function(load_complex_sample, [h, o], [tf.float32, tf.float32]),
 num_parallel_calls=tf.data.AUTOTUNE
)
dataset = dataset.map(augment, num_parallel_calls=tf.data.AUTOTUNE)
dataset = dataset.batch(32).prefetch(tf.data.AUTOTUNE)
print(f"Dataset created with {len(hologram_paths)} samples")
for batch_idx, (holograms, objects) in enumerate(dataset.take(3)):
 print(f"Batch {batch_idx}: holograms {holograms.shape}, objects {objects.shape}")

import jax
import jax.numpy as jnp
import numpy as np
from pathlib import Path
from typing import Iterator, Tuple
def load_dataset_jax(data_dir: Path) -> Tuple[jnp.ndarray, jnp.ndarray]:
 """Load entire dataset into JAX arrays."""
 hologram_files = sorted(Path(data_dir).glob("*_hologram.npz"))
 
 holograms = []
 objects = []
 
 for h_path in hologram_files:
 h_data = np.load(h_path)
 o_path = str(h_path).replace('_hologram', '_object')
 o_data = np.load(o_path)
 
 # Load complex or intensity
 if 'real' in h_data and 'imag' in h_data:
 hologram = np.stack([h_data['real'], h_data['imag']], axis=0)
 else:
 hologram = h_data['hologram'][np.newaxis, ...]
 
 if 'real' in o_data and 'imag' in o_data:
 obj = np.stack([o_data['real'], o_data['imag']], axis=0)
 else:
 obj = o_data['object'][np.newaxis, ...]
 
 holograms.append(hologram)
 objects.append(obj)
 
 return jnp.array(holograms), jnp.array(objects)
def batch_iterator(X: jnp.ndarray, y: jnp.ndarray, 
 batch_size: int, key: jax.random.PRNGKey) -> Iterator:
 """Create batched iterator with shuffling."""
 n_samples = X.shape[0]
 indices = jax.random.permutation(key, n_samples)
 
 for i in range(0, n_samples, batch_size):
 batch_indices = indices[i:i+batch_size]
 yield X[batch_indices], y[batch_indices]
# Load dataset
holograms, objects = load_dataset_jax(Path("hybrid_dataset/npz"))
print(f"Loaded dataset: holograms {holograms.shape}, objects {objects.shape}")
# Create batches
key = jax.random.PRNGKey(42)
for batch_idx, (h_batch, o_batch) in enumerate(batch_iterator(holograms, objects, 32, key)):
 print(f"Batch {batch_idx}: holograms {h_batch.shape}, objects {o_batch.shape}")
 if batch_idx >= 2:
 break

Research Examples

Example 11: Reproducing Experimental Conditions

from hologen.converters import (
 HologramDatasetGenerator, default_object_producer, default_converter,
 create_noise_model
)
from hologen.types import (
 GridSpec, OpticalConfig, HolographyConfig, HolographyMethod,
 NoiseConfig, OffAxisCarrier
)
from hologen.utils.io import NumpyDatasetWriter
from pathlib import Path
import numpy as np
def generate_experimental_dataset(output_dir: Path):
 """Generate dataset matching experimental setup.

 Experimental parameters:
 - Camera: Basler acA2040-90um (2048x2048, 5.5 μm pixels)
 - Laser: 632.8 nm HeNe
 - Distance: 150 mm
 - Off-axis angle: 2.5 degrees
 - Noise: Read noise 3.2 e-, 12-bit ADC, shot noise
 """
 
 # Grid matching camera sensor
 grid = GridSpec(
 height=2048,
 width=2048,
 pixel_pitch=5.5e-6 # 5.5 μm pixels
 )
 
 # Optical parameters matching setup
 optics = OpticalConfig(
 wavelength=632.8e-9, # HeNe laser
 propagation_distance=0.150 # 150 mm
 )
 
 # Off-axis carrier (2.5 degree angle)
 # carrier_frequency = sin(angle) / wavelength
 angle_rad = np.deg2rad(2.5)
 carrier_freq = np.sin(angle_rad) / optics.wavelength
 
 carrier = OffAxisCarrier(
 frequency_x=carrier_freq,
 frequency_y=0.0,
 gaussian_width=carrier_freq * 0.3 # 30% of carrier for filtering
 )
 
 config = HolographyConfig(
 grid=grid,
 optics=optics,
 method=HolographyMethod.OFF_AXIS,
 carrier=carrier
 )
 
 # Noise matching camera characteristics
 noise_config = NoiseConfig(
 sensor_read_noise=3.2, # 3.2 electrons RMS
 sensor_shot_noise=True, # Poisson noise
 sensor_bit_depth=12, # 12-bit ADC
 speckle_contrast=0.0, # No speckle in this setup
 aberration_defocus=0.0 # Well-focused system
 )
 
 noise_model = create_noise_model(noise_config)
 
 # Create pipeline
 producer = default_object_producer()
 converter = default_converter()
 converter.noise_model = noise_model
 
 generator = HologramDatasetGenerator(
 object_producer=producer,
 converter=converter
 )
 
 # Generate dataset
 rng = np.random.default_rng(42)
 samples = list(generator.generate(count=500, config=config, rng=rng))
 
 # Save dataset
 writer = NumpyDatasetWriter(save_preview=True)
 writer.save(samples, output_dir=output_dir)
 
 print(f"Generated {len(samples)} samples matching experimental conditions")
 print(f" Camera: 2048x2048, 5.5 μm pixels")
 print(f" Laser: 632.8 nm HeNe")
 print(f" Distance: 150 mm")
 print(f" Off-axis angle: 2.5°")
 print(f" Noise: Read 3.2e-, shot noise, 12-bit ADC")
# Generate experimental dataset
generate_experimental_dataset(Path("experimental_match_dataset"))

Example 12: Parameter Studies

from hologen.converters import (
 HologramDatasetGenerator, default_object_producer, default_converter
)
from hologen.types import GridSpec, OpticalConfig, HolographyConfig, HolographyMethod
from hologen.utils.io import NumpyDatasetWriter
from pathlib import Path
import numpy as np
import matplotlib.pyplot as plt
def parameter_study_distance(distances: list[float], samples_per_distance: int,
 output_dir: Path):
 """Study effect of propagation distance on reconstruction quality."""
 
 grid = GridSpec(height=512, width=512, pixel_pitch=6.4e-6)
 wavelength = 532e-9
 
 producer = default_object_producer()
 converter = default_converter()
 generator = HologramDatasetGenerator(object_producer=producer, converter=converter)
 
 results = []
 
 for distance in distances:
 print(f"Testing distance: {distance*1000:.1f} mm")
 
 optics = OpticalConfig(wavelength=wavelength, propagation_distance=distance)
 config = HolographyConfig(grid=grid, optics=optics, method=HolographyMethod.INLINE)
 
 # Generate samples
 rng = np.random.default_rng(42) # Same seed for fair comparison
 samples = list(generator.generate(count=samples_per_distance, config=config, rng=rng))
 
 # Compute reconstruction quality metrics
 mse_values = []
 for sample in samples:
 mse = np.mean((sample.object_sample.pixels - sample.reconstruction)**2)
 mse_values.append(mse)
 
 avg_mse = np.mean(mse_values)
 std_mse = np.std(mse_values)
 
 results.append({
 'distance': distance,
 'avg_mse': avg_mse,
 'std_mse': std_mse
 })
 
 print(f" Average MSE: {avg_mse:.6f} ± {std_mse:.6f}")
 
 # Save samples for this distance
 distance_dir = output_dir / f"distance_{int(distance*1000)}mm"
 writer = NumpyDatasetWriter(save_preview=True)
 writer.save(samples, output_dir=distance_dir)
 
 # Plot results
 distances_mm = [r['distance'] * 1000 for r in results]
 avg_mses = [r['avg_mse'] for r in results]
 std_mses = [r['std_mse'] for r in results]
 
 plt.figure(figsize=(10, 6))
 plt.errorbar(distances_mm, avg_mses, yerr=std_mses, marker='o', capsize=5)
 plt.xlabel('Propagation Distance (mm)')
 plt.ylabel('Reconstruction MSE')
 plt.title('Effect of Propagation Distance on Reconstruction Quality')
 plt.grid(True, alpha=0.3)
 plt.savefig(output_dir / 'distance_study.png', dpi=150, bbox_inches='tight')
 plt.close()
 
 print(f"\nParameter study complete. Results saved to {output_dir}")
 return results
# Run parameter study
distances = [0.01, 0.02, 0.05, 0.10, 0.15, 0.20] # 10mm to 200mm
results = parameter_study_distance(
 distances=distances,
 samples_per_distance=20,
 output_dir=Path("parameter_study_distance")
)
# Find optimal distance
optimal = min(results, key=lambda x: x['avg_mse'])
print(f"\nOptimal distance: {optimal['distance']*1000:.1f} mm (MSE: {optimal['avg_mse']:.6f})")

# Study wavelength effect
def parameter_study_wavelength(wavelengths: list[float], ...):
 """Study effect of illumination wavelength."""
 # Similar structure, vary wavelength instead of distance
 pass
# Study pixel pitch effect
def parameter_study_resolution(pixel_pitches: list[float], ...):
 """Study effect of sensor resolution."""
 # Similar structure, vary pixel_pitch instead of distance
 pass
# Study noise effect
def parameter_study_noise(noise_levels: list[float], ...):
 """Study effect of noise on reconstruction."""
 # Similar structure, vary noise parameters
 pass

Example 13: Ablation Studies

from hologen.converters import (
 HologramDatasetGenerator, default_object_producer, default_converter,
 create_noise_model
)
from hologen.types import GridSpec, OpticalConfig, HolographyConfig, HolographyMethod, NoiseConfig
from hologen.noise import SensorNoiseModel, SpeckleNoiseModel, AberrationNoiseModel, CompositeNoiseModel
from hologen.utils.io import NumpyDatasetWriter
from pathlib import Path
import numpy as np
import matplotlib.pyplot as plt
def ablation_study_noise(samples_per_condition: int, output_dir: Path):
 """Ablation study: effect of different noise sources."""
 
 grid = GridSpec(height=512, width=512, pixel_pitch=6.4e-6)
 optics = OpticalConfig(wavelength=532e-9, propagation_distance=0.05)
 config = HolographyConfig(grid=grid, optics=optics, method=HolographyMethod.INLINE)
 
 producer = default_object_producer()
 
 # Define noise conditions
 conditions = {
 'no_noise': None,
 'sensor_only': SensorNoiseModel(
 name="sensor",
 read_noise=3.0,
 shot_noise=True,
 dark_current=0.5,
 bit_depth=12
 ),
 'speckle_only': SpeckleNoiseModel(
 name="speckle",
 contrast=0.8,
 correlation_length=1.0
 ),
 'aberration_only': AberrationNoiseModel(
 name="aberration",
 defocus=0.5,
 astigmatism_x=0.2,
 astigmatism_y=0.2,
 coma_x=0.0,
 coma_y=0.0
 ),
 'all_noise': CompositeNoiseModel(
 name="composite",
 models=(
 SensorNoiseModel(name="sensor", read_noise=3.0, shot_noise=True, 
 dark_current=0.5, bit_depth=12),
 SpeckleNoiseModel(name="speckle", contrast=0.8, correlation_length=1.0),
 AberrationNoiseModel(name="aberration", defocus=0.5, 
 astigmatism_x=0.2, astigmatism_y=0.2)
 )
 )
 }
 
 results = {}
 
 for condition_name, noise_model in conditions.items():
 print(f"\nTesting condition: {condition_name}")
 
 # Create converter with this noise model
 converter = default_converter()
 converter.noise_model = noise_model
 
 generator = HologramDatasetGenerator(object_producer=producer, converter=converter)
 
 # Generate samples
 rng = np.random.default_rng(42) # Same seed for fair comparison
 samples = list(generator.generate(count=samples_per_condition, config=config, rng=rng))
 
 # Compute metrics
 mse_values = []
 snr_values = []
 
 for sample in samples:
 # Reconstruction quality
 mse = np.mean((sample.object_sample.pixels - sample.reconstruction)**2)
 mse_values.append(mse)
 
 # Signal-to-noise ratio
 signal_power = np.mean(sample.hologram**2)
 noise_power = np.var(sample.hologram)
 snr = 10 * np.log10(signal_power / (noise_power + 1e-10))
 snr_values.append(snr)
 
 results[condition_name] = {
 'mse_mean': np.mean(mse_values),
 'mse_std': np.std(mse_values),
 'snr_mean': np.mean(snr_values),
 'snr_std': np.std(snr_values)
 }
 
 print(f" MSE: {results[condition_name]['mse_mean']:.6f} ± {results[condition_name]['mse_std']:.6f}")
 print(f" SNR: {results[condition_name]['snr_mean']:.2f} ± {results[condition_name]['snr_std']:.2f} dB")
 
 # Save samples
 condition_dir = output_dir / condition_name
 writer = NumpyDatasetWriter(save_preview=True)
 writer.save(samples, output_dir=condition_dir)
 
 # Plot results
 fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 5))
 
 conditions_list = list(results.keys())
 mse_means = [results[c]['mse_mean'] for c in conditions_list]
 mse_stds = [results[c]['mse_std'] for c in conditions_list]
 snr_means = [results[c]['snr_mean'] for c in conditions_list]
 snr_stds = [results[c]['snr_std'] for c in conditions_list]
 
 x_pos = np.arange(len(conditions_list))
 
 ax1.bar(x_pos, mse_means, yerr=mse_stds, capsize=5, alpha=0.7)
 ax1.set_xticks(x_pos)
 ax1.set_xticklabels(conditions_list, rotation=45, ha='right')
 ax1.set_ylabel('Reconstruction MSE')
 ax1.set_title('Effect of Noise on Reconstruction Quality')
 ax1.grid(True, alpha=0.3, axis='y')
 
 ax2.bar(x_pos, snr_means, yerr=snr_stds, capsize=5, alpha=0.7, color='orange')
 ax2.set_xticks(x_pos)
 ax2.set_xticklabels(conditions_list, rotation=45, ha='right')
 ax2.set_ylabel('SNR (dB)')
 ax2.set_title('Effect of Noise on Signal Quality')
 ax2.grid(True, alpha=0.3, axis='y')
 
 plt.tight_layout()
 plt.savefig(output_dir / 'ablation_study.png', dpi=150, bbox_inches='tight')
 plt.close()
 
 print(f"\nAblation study complete. Results saved to {output_dir}")
 return results
# Run ablation study
results = ablation_study_noise(
 samples_per_condition=50,
 output_dir=Path("ablation_study_noise")
)
# Analyze results
print("\n=== Ablation Study Summary ===")
baseline_mse = results['no_noise']['mse_mean']
for condition, metrics in results.items():
 if condition != 'no_noise':
 mse_increase = (metrics['mse_mean'] - baseline_mse) / baseline_mse * 100
 print(f"{condition}: MSE increased by {mse_increase:.1f}%")

Code Recipes

Recipe 1: Batch Processing

from hologen.converters import (
 HologramDatasetGenerator, default_object_producer, default_converter
)
from hologen.types import GridSpec, OpticalConfig, HolographyConfig, HolographyMethod
from hologen.utils.io import NumpyDatasetWriter
from pathlib import Path
import numpy as np
from tqdm import tqdm
def generate_large_dataset_batched(total_samples: int, batch_size: int, 
 output_dir: Path):
 """Generate large dataset in batches to manage memory."""
 
 grid = GridSpec(height=512, width=512, pixel_pitch=6.4e-6)
 optics = OpticalConfig(wavelength=532e-9, propagation_distance=0.05)
 config = HolographyConfig(grid=grid, optics=optics, method=HolographyMethod.INLINE)
 
 producer = default_object_producer()
 converter = default_converter()
 generator = HologramDatasetGenerator(object_producer=producer, converter=converter)
 
 writer = NumpyDatasetWriter(save_preview=False) # Disable previews for speed
 rng = np.random.default_rng(42)
 
 num_batches = (total_samples + batch_size - 1) // batch_size
 
 print(f"Generating {total_samples} samples in {num_batches} batches of {batch_size}")
 
 for batch_idx in tqdm(range(num_batches), desc="Batches"):
 # Determine batch size (last batch may be smaller)
 current_batch_size = min(batch_size, total_samples - batch_idx * batch_size)
 
 # Generate batch
 samples = list(generator.generate(count=current_batch_size, config=config, rng=rng))
 
 # Save batch to separate directory
 batch_dir = output_dir / f"batch_{batch_idx:04d}"
 writer.save(samples, output_dir=batch_dir)
 
 # Clear memory
 del samples
 
 print(f"Dataset generation complete: {total_samples} samples in {output_dir}")
# Generate 10,000 samples in batches of 100
generate_large_dataset_batched(
 total_samples=10000,
 batch_size=100,
 output_dir=Path("large_dataset_batched")
)

Recipe 2: Parallel Generation

from hologen.converters import (
 HologramDatasetGenerator, default_object_producer, default_converter
)
from hologen.types import GridSpec, OpticalConfig, HolographyConfig, HolographyMethod
from hologen.utils.io import NumpyDatasetWriter
from pathlib import Path
import numpy as np
from multiprocessing import Pool, cpu_count
from functools import partial
def generate_batch_worker(batch_id: int, samples_per_batch: int, 
 config: HolographyConfig, seed: int, 
 output_dir: Path) -> int:
 """Worker function to generate one batch of samples."""
 
 # Create fresh instances for this worker
 producer = default_object_producer()
 converter = default_converter()
 generator = HologramDatasetGenerator(object_producer=producer, converter=converter)
 
 # Use unique seed for this batch
 rng = np.random.default_rng(seed + batch_id)
 
 # Generate samples
 samples = list(generator.generate(count=samples_per_batch, config=config, rng=rng))
 
 # Save batch
 batch_dir = output_dir / f"batch_{batch_id:04d}"
 writer = NumpyDatasetWriter(save_preview=False)
 writer.save(samples, output_dir=batch_dir)
 
 return len(samples)
def generate_dataset_parallel(total_samples: int, num_workers: int, 
 output_dir: Path, seed: int = 42):
 """Generate dataset using parallel workers."""
 
 # Configuration
 grid = GridSpec(height=512, width=512, pixel_pitch=6.4e-6)
 optics = OpticalConfig(wavelength=532e-9, propagation_distance=0.05)
 config = HolographyConfig(grid=grid, optics=optics, method=HolographyMethod.INLINE)
 
 # Determine batch distribution
 samples_per_worker = total_samples // num_workers
 num_batches = num_workers
 
 print(f"Generating {total_samples} samples using {num_workers} parallel workers")
 print(f"Each worker generates {samples_per_worker} samples")
 
 # Create worker function with fixed arguments
 worker_func = partial(
 generate_batch_worker,
 samples_per_batch=samples_per_worker,
 config=config,
 seed=seed,
 output_dir=output_dir
 )
 
 # Run parallel generation
 with Pool(processes=num_workers) as pool:
 results = pool.map(worker_func, range(num_batches))
 
 total_generated = sum(results)
 print(f"Dataset generation complete: {total_generated} samples in {output_dir}")
# Generate 10,000 samples using all available CPU cores
num_workers = cpu_count()
generate_dataset_parallel(
 total_samples=10000,
 num_workers=num_workers,
 output_dir=Path("parallel_dataset"),
 seed=42
)

def combine_batches(batched_dir: Path, output_dir: Path):
 """Combine batched datasets into single directory."""
 import shutil
 
 output_npz = output_dir / "npz"
 output_npz.mkdir(parents=True, exist_ok=True)
 
 batch_dirs = sorted(batched_dir.glob("batch_*"))
 
 sample_idx = 0
 for batch_dir in batch_dirs:
 npz_files = sorted((batch_dir / "npz").glob("*.npz"))
 for npz_file in npz_files:
 # Rename with sequential index
 new_name = f"sample_{sample_idx:05d}_{npz_file.stem.split('_', 2)[-1]}.npz"
 shutil.copy(npz_file, output_npz / new_name)
 sample_idx += 1
 
 print(f"Combined {sample_idx} samples into {output_dir}")
# Combine parallel batches
combine_batches(
 batched_dir=Path("parallel_dataset"),
 output_dir=Path("combined_dataset")
)

Recipe 3: Memory-Efficient Loading

import numpy as np
from pathlib import Path
from typing import Iterator, Tuple
class MemoryEfficientDataLoader:
 """Memory-efficient data loader using generators."""
 
 def __init__(self, data_dir: Path, batch_size: int = 32):
 self.data_dir = Path(data_dir)
 self.batch_size = batch_size
 self.sample_files = sorted(self.data_dir.glob("*.npz"))
 self.num_samples = len(self.sample_files)
 
 def __len__(self) -> int:
 return (self.num_samples + self.batch_size - 1) // self.batch_size
 
 def __iter__(self) -> Iterator[Tuple[np.ndarray, np.ndarray]]:
 """Iterate over batches without loading entire dataset."""
 batch_holograms = []
 batch_objects = []
 
 for sample_file in self.sample_files:
 # Load single sample
 data = np.load(sample_file)
 
 if 'hologram' in data:
 # Legacy format
 hologram = data['hologram']
 obj = data['object']
 else:
 # Complex format
 hologram = data['real'] + 1j * data['imag']
 obj_file = str(sample_file).replace('_hologram', '_object')
 obj_data = np.load(obj_file)
 obj = obj_data['real'] + 1j * obj_data['imag']
 
 batch_holograms.append(hologram)
 batch_objects.append(obj)
 
 # Yield batch when full
 if len(batch_holograms) == self.batch_size:
 yield np.array(batch_holograms), np.array(batch_objects)
 batch_holograms = []
 batch_objects = []
 
 # Yield remaining samples
 if batch_holograms:
 yield np.array(batch_holograms), np.array(batch_objects)
# Usage example
loader = MemoryEfficientDataLoader(
 data_dir=Path("large_dataset_batched/batch_0000/npz"),
 batch_size=32
)
print(f"Dataset has {loader.num_samples} samples, {len(loader)} batches")
for batch_idx, (holograms, objects) in enumerate(loader):
 print(f"Batch {batch_idx}: holograms {holograms.shape}, objects {objects.shape}")
 
 # Process batch (e.g., train model)
 # ...
 
 if batch_idx >= 2:
 break

class MultiDirectoryLoader:
 """Load from multiple batch directories sequentially."""
 
 def __init__(self, base_dir: Path, batch_size: int = 32):
 self.base_dir = Path(base_dir)
 self.batch_size = batch_size
 self.batch_dirs = sorted(self.base_dir.glob("batch_*"))
 
 def __iter__(self):
 for batch_dir in self.batch_dirs:
 npz_dir = batch_dir / "npz"
 if npz_dir.exists():
 loader = MemoryEfficientDataLoader(npz_dir, self.batch_size)
 yield from loader
# Usage
multi_loader = MultiDirectoryLoader(
 base_dir=Path("large_dataset_batched"),
 batch_size=32
)
for batch_idx, (holograms, objects) in enumerate(multi_loader):
 print(f"Batch {batch_idx}: {holograms.shape}")
 if batch_idx >= 5:
 break

Recipe 4: Data Augmentation

import numpy as np
from scipy.ndimage import rotate, gaussian_filter
from typing import Tuple
class HologramAugmenter:
 """Data augmentation for hologram-object pairs."""
 
 def __init__(self, rotation_range: float = 90.0,
 flip_horizontal: bool = True,
 flip_vertical: bool = True,
 noise_std: float = 0.01,
 blur_sigma: float = 0.5):
 self.rotation_range = rotation_range
 self.flip_horizontal = flip_horizontal
 self.flip_vertical = flip_vertical
 self.noise_std = noise_std
 self.blur_sigma = blur_sigma
 
 def augment(self, hologram: np.ndarray, obj: np.ndarray, 
 rng: np.random.Generator) -> Tuple[np.ndarray, np.ndarray]:
 """Apply random augmentation to hologram-object pair."""
 
 # Random rotation (same for both)
 if self.rotation_range > 0:
 angle = rng.uniform(-self.rotation_range, self.rotation_range)
 hologram = rotate(hologram, angle, reshape=False, order=1)
 obj = rotate(obj, angle, reshape=False, order=1)
 
 # Random horizontal flip
 if self.flip_horizontal and rng.random() > 0.5:
 hologram = np.fliplr(hologram)
 obj = np.fliplr(obj)
 
 # Random vertical flip
 if self.flip_vertical and rng.random() > 0.5:
 hologram = np.flipud(hologram)
 obj = np.flipud(obj)
 
 # Add random noise to hologram only
 if self.noise_std > 0:
 noise = rng.normal(0, self.noise_std, hologram.shape)
 hologram = hologram + noise
 hologram = np.clip(hologram, 0, 1)
 
 # Random blur to hologram only
 if self.blur_sigma > 0 and rng.random() > 0.5:
 sigma = rng.uniform(0, self.blur_sigma)
 hologram = gaussian_filter(hologram, sigma=sigma)
 
 return hologram, obj
# Usage with PyTorch
import torch
from torch.utils.data import Dataset, DataLoader
from pathlib import Path
class AugmentedHologramDataset(Dataset):
 """Dataset with on-the-fly augmentation."""
 
 def __init__(self, data_dir: Path, augmenter: HologramAugmenter = None):
 self.samples = sorted(Path(data_dir).glob("*.npz"))
 self.augmenter = augmenter
 self.rng = np.random.default_rng()
 
 def __len__(self):
 return len(self.samples)
 
 def __getitem__(self, idx):
 data = np.load(self.samples[idx])
 hologram = data['hologram']
 obj = data['object']
 
 # Apply augmentation
 if self.augmenter is not None:
 hologram, obj = self.augmenter.augment(hologram, obj, self.rng)
 
 # Convert to tensors
 hologram = torch.from_numpy(hologram).float().unsqueeze(0)
 obj = torch.from_numpy(obj).float().unsqueeze(0)
 
 return hologram, obj
# Create augmented dataset
augmenter = HologramAugmenter(
 rotation_range=45.0,
 flip_horizontal=True,
 flip_vertical=True,
 noise_std=0.01,
 blur_sigma=0.5
)
dataset = AugmentedHologramDataset(
 data_dir=Path("my_first_dataset/npz"),
 augmenter=augmenter
)
dataloader = DataLoader(dataset, batch_size=16, shuffle=True)
# Visualize augmentation effect
import matplotlib.pyplot as plt
sample_idx = 0
original_data = np.load(dataset.samples[sample_idx])
original_hologram = original_data['hologram']
# Get augmented version
augmented_hologram, _ = dataset[sample_idx]
augmented_hologram = augmented_hologram.squeeze().numpy()
fig, axes = plt.subplots(1, 2, figsize=(10, 5))
axes[0].imshow(original_hologram, cmap='gray')
axes[0].set_title('Original')
axes[1].imshow(augmented_hologram, cmap='gray')
axes[1].set_title('Augmented')
plt.tight_layout()
plt.savefig('augmentation_example.png', dpi=150)
plt.show()

Recipe 5: Dataset Validation and Quality Checks

import numpy as np
from pathlib import Path
from typing import Dict, List
import matplotlib.pyplot as plt
class DatasetValidator:
 """Validate hologram dataset quality."""
 
 def __init__(self, data_dir: Path):
 self.data_dir = Path(data_dir)
 self.sample_files = sorted(self.data_dir.glob("*.npz"))
 
 def validate_all(self) -> Dict[str, any]:
 """Run all validation checks."""
 results = {
 'num_samples': len(self.sample_files),
 'file_integrity': self.check_file_integrity(),
 'value_ranges': self.check_value_ranges(),
 'reconstruction_quality': self.check_reconstruction_quality(),
 'shape_consistency': self.check_shape_consistency(),
 'statistics': self.compute_statistics()
 }
 return results
 
 def check_file_integrity(self) -> Dict[str, any]:
 """Check if all files can be loaded."""
 corrupted = []
 missing_keys = []
 
 for sample_file in self.sample_files:
 try:
 data = np.load(sample_file)
 required_keys = ['object', 'hologram', 'reconstruction']
 
 for key in required_keys:
 if key not in data:
 missing_keys.append((sample_file.name, key))
 
 except Exception as e:
 corrupted.append((sample_file.name, str(e)))
 
 return {
 'corrupted_files': corrupted,
 'missing_keys': missing_keys,
 'all_valid': len(corrupted) == 0 and len(missing_keys) == 0
 }
 
 def check_value_ranges(self) -> Dict[str, any]:
 """Check if values are in expected ranges."""
 issues = []
 
 for sample_file in self.sample_files[:10]: # Check first 10
 data = np.load(sample_file)
 
 # Check for NaN or Inf
 for key in ['object', 'hologram', 'reconstruction']:
 if not np.isfinite(data[key]).all():
 issues.append(f"{sample_file.name}: {key} contains NaN or Inf")
 
 # Check value ranges
 if data['object'].min() < 0 or data['object'].max() > 1:
 issues.append(f"{sample_file.name}: object values out of [0,1]")
 
 if data['hologram'].min() < 0:
 issues.append(f"{sample_file.name}: hologram has negative values")
 
 return {
 'issues': issues,
 'all_valid': len(issues) == 0
 }
 
 def check_reconstruction_quality(self) -> Dict[str, any]:
 """Check reconstruction quality metrics."""
 mse_values = []
 psnr_values = []
 
 for sample_file in self.sample_files:
 data = np.load(sample_file)
 obj = data['object']
 recon = data['reconstruction']
 
 mse = np.mean((obj - recon)**2)
 psnr = 10 * np.log10(1.0 / (mse + 1e-10))
 
 mse_values.append(mse)
 psnr_values.append(psnr)
 
 return {
 'mse_mean': np.mean(mse_values),
 'mse_std': np.std(mse_values),
 'mse_min': np.min(mse_values),
 'mse_max': np.max(mse_values),
 'psnr_mean': np.mean(psnr_values),
 'psnr_std': np.std(psnr_values),
 'poor_quality_count': sum(1 for mse in mse_values if mse > 0.1)
 }
 
 def check_shape_consistency(self) -> Dict[str, any]:
 """Check if all samples have consistent shapes."""
 shapes = set()
 
 for sample_file in self.sample_files:
 data = np.load(sample_file)
 shapes.add(data['object'].shape)
 
 return {
 'unique_shapes': list(shapes),
 'consistent': len(shapes) == 1
 }
 
 def compute_statistics(self) -> Dict[str, any]:
 """Compute dataset statistics."""
 object_means = []
 hologram_means = []
 object_stds = []
 hologram_stds = []
 
 for sample_file in self.sample_files:
 data = np.load(sample_file)
 
 object_means.append(data['object'].mean())
 hologram_means.append(data['hologram'].mean())
 object_stds.append(data['object'].std())
 hologram_stds.append(data['hologram'].std())
 
 return {
 'object_mean': np.mean(object_means),
 'object_std': np.mean(object_stds),
 'hologram_mean': np.mean(hologram_means),
 'hologram_std': np.mean(hologram_stds)
 }
 
 def generate_report(self, output_path: Path):
 """Generate validation report with visualizations."""
 results = self.validate_all()
 
 # Print text report
 print("=" * 60)
 print("DATASET VALIDATION REPORT")
 print("=" * 60)
 print(f"\nDataset: {self.data_dir}")
 print(f"Number of samples: {results['num_samples']}")
 
 print("\n--- File Integrity ---")
 print(f"All files valid: {results['file_integrity']['all_valid']}")
 if results['file_integrity']['corrupted_files']:
 print(f"Corrupted files: {results['file_integrity']['corrupted_files']}")
 
 print("\n--- Value Ranges ---")
 print(f"All values valid: {results['value_ranges']['all_valid']}")
 if results['value_ranges']['issues']:
 for issue in results['value_ranges']['issues']:
 print(f" - {issue}")
 
 print("\n--- Reconstruction Quality ---")
 print(f"Mean MSE: {results['reconstruction_quality']['mse_mean']:.6f} "
 f"± {results['reconstruction_quality']['mse_std']:.6f}")
 print(f"Mean PSNR: {results['reconstruction_quality']['psnr_mean']:.2f} "
 f"± {results['reconstruction_quality']['psnr_std']:.2f} dB")
 print(f"Poor quality samples (MSE > 0.1): "
 f"{results['reconstruction_quality']['poor_quality_count']}")
 
 print("\n--- Shape Consistency ---")
 print(f"Consistent shapes: {results['shape_consistency']['consistent']}")
 print(f"Shapes: {results['shape_consistency']['unique_shapes']}")
 
 print("\n--- Statistics ---")
 print(f"Object mean: {results['statistics']['object_mean']:.4f}")
 print(f"Object std: {results['statistics']['object_std']:.4f}")
 print(f"Hologram mean: {results['statistics']['hologram_mean']:.4f}")
 print(f"Hologram std: {results['statistics']['hologram_std']:.4f}")
 
 # Generate visualization
 self._plot_quality_distribution(output_path)
 
 print(f"\nValidation report saved to {output_path}")
 
 def _plot_quality_distribution(self, output_path: Path):
 """Plot reconstruction quality distribution."""
 mse_values = []
 
 for sample_file in self.sample_files:
 data = np.load(sample_file)
 mse = np.mean((data['object'] - data['reconstruction'])**2)
 mse_values.append(mse)
 
 plt.figure(figsize=(10, 6))
 plt.hist(mse_values, bins=50, alpha=0.7, edgecolor='black')
 plt.xlabel('Reconstruction MSE')
 plt.ylabel('Frequency')
 plt.title('Distribution of Reconstruction Quality')
 plt.axvline(np.mean(mse_values), color='red', linestyle='--', 
 label=f'Mean: {np.mean(mse_values):.6f}')
 plt.legend()
 plt.grid(True, alpha=0.3)
 plt.savefig(output_path / 'quality_distribution.png', dpi=150, bbox_inches='tight')
 plt.close()
# Run validation
validator = DatasetValidator(Path("my_first_dataset/npz"))
validator.generate_report(Path("my_first_dataset"))

Summary

Next Steps

Contributing Examples

Getting Help