Module pyminflux.combiner

Dataset combining utilities.

Functions

def combine_datasets_with_bead_alignment(reference_dataset: pyminflux.processor._dataset.MinFluxDataset,
moving_dataset: pyminflux.processor._dataset.MinFluxDataset,
bead_correspondence: Dict[str, str] = None,
transform_type: str = 'euclidean',
n_points: int = 3,
next_fluo_id: int | None = None) ‑> pyminflux.processor._dataset.MinFluxDataset | None

Expand source code

def combine_datasets_with_bead_alignment(
    reference_dataset: MinFluxDataset,
    moving_dataset: MinFluxDataset,
    bead_correspondence: Dict[str, str] = None,
    transform_type: str = 'euclidean',
    n_points: int = 3,
    next_fluo_id: Optional[int] = None,
) -> Optional[MinFluxDataset]:
    """
    Perform bead-based alignment and combine two datasets.
    
    The MBM (bead) data is extracted from the datasets themselves.
    
    Args:
        reference_dataset: Reference dataset (must contain mbm_data)
        moving_dataset: Moving dataset (must contain mbm_data)
        bead_correspondence: Mapping of reference bead names to moving bead names
        transform_type: Type of transformation ('euclidean', 'affine', etc.)
        n_points: Number of earliest points to use for bead position averaging
        next_fluo_id: Fluorophore ID to assign (auto-assigned if None)
        
    Returns:
        Combined dataset, or None if alignment fails
    """
    # Extract MBM data from datasets
    if reference_dataset.mbm_data is None:
        print("Error: Reference dataset does not contain MBM (bead) data")
        return None
    
    if moving_dataset.mbm_data is None:
        print("Error: Moving dataset does not contain MBM (bead) data")
        return None
    
    ref_mbm_dict = reference_dataset.mbm_data.get('mbm', {})
    mov_mbm_dict = moving_dataset.mbm_data.get('mbm', {})
    
    if not ref_mbm_dict:
        print("Error: Reference dataset MBM dictionary is empty")
        return None
    
    if not mov_mbm_dict:
        print("Error: Moving dataset MBM dictionary is empty")
        return None
    
    # Perform alignment
    transform_model = align_datasets_using_beads(
        ref_mbm_dict,
        mov_mbm_dict,
        bead_correspondence=bead_correspondence,
        transform_type=transform_type,
        n_points=n_points,
    )
    
    if transform_model is None:
        return None
    
    # Combine the datasets
    return combine_datasets_with_alignment(
        reference_dataset,
        moving_dataset,
        transform_model,
        next_fluo_id=next_fluo_id,
    )

Perform bead-based alignment and combine two datasets.

The MBM (bead) data is extracted from the datasets themselves.

Args

reference_dataset

Reference dataset (must contain mbm_data)

moving_dataset

Moving dataset (must contain mbm_data)

bead_correspondence

Mapping of reference bead names to moving bead names

transform_type

Type of transformation ('euclidean', 'affine', etc.)

n_points

Number of earliest points to use for bead position averaging

next_fluo_id

Fluorophore ID to assign (auto-assigned if None)

Returns

Combined dataset, or None if alignment fails

def get_bead_positions_from_mbm(mbm_dict: Dict, n_points: int = 3) ‑> Dict[str, numpy.ndarray]

Expand source code

def get_bead_positions_from_mbm(mbm_dict: Dict, n_points: int = 3) -> Dict[str, np.ndarray]:
    """
    Extract average bead positions from MBM dictionary.
    
    Args:
        mbm_dict: MBM dictionary from MinFluxReaderV2
        n_points: Number of earliest points to average
        
    Returns:
        Dictionary mapping bead names to positions [z, y, x] in nm
    """
    df = mbm_dict_to_dataframe(mbm_dict)
    
    if df.empty:
        return {}
    
    # Filter to only include beads marked as "used"
    df = df[df['used'] == True]
    
    if df.empty:
        return {}
    
    bead_positions = {}
    for bead_name in df['bead_name'].unique():
        bead_data = df[df['bead_name'] == bead_name]
        # Get earliest n_points
        earliest = bead_data.nsmallest(n_points, 'tim')
        # Calculate mean position as [z, y, x]
        pos = earliest[['z', 'y', 'x']].mean(axis=0).to_numpy()
        bead_positions[bead_name] = pos
    
    return bead_positions

Extract average bead positions from MBM dictionary.

Args

mbm_dict

MBM dictionary from MinFluxReaderV2

n_points

Number of earliest points to average

Returns

Dictionary mapping bead names to positions [z, y, x] in nm

def get_next_fluorophore_id(df: pandas.core.frame.DataFrame) ‑> int

Expand source code

def get_next_fluorophore_id(df: pd.DataFrame) -> int:
    """
    Get the next available fluorophore ID.
    
    Args:
        df: DataFrame with 'fluo' column
        
    Returns:
        Next available fluorophore ID
    """
    if 'fluo' not in df.columns:
        return 1
    
    current_ids = df['fluo'].unique()
    if len(current_ids) == 0:
        return 1
    
    return int(np.max(current_ids)) + 1

Get the next available fluorophore ID.

Args

df

DataFrame with 'fluo' column

Returns

Next available fluorophore ID

def load_zarr_for_beads(zarr_path: str) ‑> Tuple[pyminflux.reader._reader_v2.MinFluxReaderV2 | None, Dict | None]

Expand source code

def load_zarr_for_beads(zarr_path: str) -> Tuple[Optional[MinFluxReaderV2], Optional[Dict]]:
    """
    Load a Zarr file and extract MBM (bead) data.
    
    Args:
        zarr_path: Path to the Zarr file
        
    Returns:
        Tuple of (reader, mbm_dict) or (None, None) if loading fails
    """
    try:
        zarr_path = Path(zarr_path)
        if not zarr_path.exists():
            print(f"Error: Zarr path does not exist: {zarr_path}")
            return None, None
        
        # Load the reader
        reader = MinFluxReaderV2(zarr_path)
        
        # Check if MBM data is available
        if not hasattr(reader, 'mbm_data') or reader.mbm_data is None:
            print(f"Error: No MBM (bead) data found in {zarr_path}")
            return None, None
        
        mbm_dict = reader.mbm_data.get('mbm', {})
        
        if not mbm_dict:
            print(f"Error: MBM dictionary is empty in {zarr_path}")
            return None, None
        
        return reader, mbm_dict
        
    except Exception as e:
        print(f"Error loading Zarr file {zarr_path}: {e}")
        return None, None

Load a Zarr file and extract MBM (bead) data.

Args

zarr_path

Path to the Zarr file

Returns

Tuple of (reader, mbm_dict) or (None, None) if loading fails