lama/elastix/invert_transforms.py
def batch_invert_transform_parameters(config: Union[Path, LamaConfig],
                                      clobber=True, new_log:bool=False):
    """
    Create new elastix TransformParameter files that can then be used by transformix to invert labelmaps, stats etc

    Parameters
    ----------
    config
        path to original reg pipeline config file

    clobber
        if True overwrite inverted parameters present

    new_log:
        Whether to create a new log file. If called from another module, logging may happen there
    """
    common.test_installation('elastix')

    if isinstance(config, (Path, str)):
        config = LamaConfig(config)

    threads = str(config['threads'])

    if new_log:
        common.init_logging(config / 'invert_transforms.log')

    reg_dirs = get_reg_dirs(config)

    # Get the image basenames from the first stage registration folder (usually rigid)
    # ignore images in non-relevent folder that may be present
    volume_names = [x.stem for x in common.get_file_paths(reg_dirs[0], ignore_folders=[RESOLUTION_IMGS_DIR, IMG_PYRAMID_DIR])]

    inv_outdir = config.mkdir('inverted_transforms')

    stages_to_invert = defaultdict(list)

    jobs: List[Dict] = []

    reg_stage_dir: Path

    for i, vol_id in enumerate(volume_names):

        for reg_stage_dir in reg_dirs:

            if not reg_stage_dir.is_dir():
                logging.error('cannot find {}'.format(reg_stage_dir))
                raise FileNotFoundError(f'Cannot find registration dir {reg_stage_dir}')

            inv_stage_dir = inv_outdir / reg_stage_dir.name

            specimen_stage_reg_dir = reg_stage_dir / vol_id
            specimen_stage_inversion_dir = inv_stage_dir / vol_id

            transform_file = common.getfile_startswith(specimen_stage_reg_dir, ELX_TRANSFORM_NAME)
            parameter_file = common.getfile_startswith(reg_stage_dir, ELX_PARAM_PREFIX)

            # Create the folder to put the specimen inversion parameter files in.
            inv_stage_dir.mkdir(exist_ok=True)

            # Add the stage to the inversion order config (in reverse order), if not already.
            if reg_stage_dir.name not in stages_to_invert['label_propagation_order']:
                stages_to_invert['label_propagation_order'].insert(0, reg_stage_dir.name)

            if clobber:
                common.mkdir_force(specimen_stage_inversion_dir)  # Overwrite any inversion file that exist for a single specimen

            # Each registration directory contains a metadata file, which contains the relative path to the fixed volume
            reg_metadata = cfg_load(specimen_stage_reg_dir / common.INDV_REG_METADATA)
            fixed_volume = (specimen_stage_reg_dir / reg_metadata['fixed_vol']).resolve()

            # Invert the Transform parameters with options for normal image inversion

            job = {
                'specimen_stage_inversion_dir': specimen_stage_inversion_dir,
                'parameter_file': abspath(parameter_file),
                'transform_file': transform_file,
                'fixed_volume': fixed_volume,
                'param_file_output_name': 'inversion_parameters.txt',
                'image_replacements': IMAGE_REPLACEMENTS,
                'label_replacements': LABEL_REPLACEMENTS,
                'image_transform_file': PROPAGATE_IMAGE_TRANSFORM,
                'label_transform_file': PROPAGATE_LABEL_TRANFORM,
                'clobber': clobber,
                'threads': threads
            }

            jobs.append(job)

    # By putting each inverison job (a single job per registration stage) we can speed things up a bit
    # If we can get multithreded inversion in elastix we can remove this python multithreading
    pool = Pool(8)
    try:
        pool.map(_invert_transform_parameters, jobs)

    except KeyboardInterrupt:
        print('terminating inversion')
        pool.terminate()
        pool.join()

    # TODO: Should we replace the need for this invert.yaml?
    reg_dir = Path(os.path.relpath(reg_stage_dir, inv_outdir))
    stages_to_invert['registration_directory'] = str(reg_dir)  # Doc why we need this
    # Create a yaml config file so that inversions can be run seperatley
    invert_config = config['inverted_transforms'] / PROPAGATE_CONFIG

    with open(invert_config, 'w') as yf:
        yf.write(yaml.dump(dict(stages_to_invert), default_flow_style=False))

    Create new elastix TransformParameter files that can then be used by transformix to invert labelmaps, stats etc

    Parameters
    ----------
    config
        path to original reg pipeline config file

    clobber
        if True overwrite inverted parameters present

    new_log:
        Whether to create a new log file. If called from another module, logging may happen there
    
Create new elastix TransformParameter files that can then be used by transformix to invert labelmaps, stats etc
Parameters

config
path to original reg pipeline config file

clobber
if True overwrite inverted parameters present

Whether to create a new log file. If called from another module, logging may happen there
Create new elastix TransformParameter files that can then be used by transformix to invert labelmaps stats etc Parameters config path to original reg pipeline config file clobber if True overwrite inverted parameters present Whether to create a new log file . If called from another module logging may happen there
def batch_invert_transform_parameters(config: Union[Path, LamaConfig],
                                      clobber=True, new_log:bool=False):
    common.test_installation('elastix')
    if isinstance(config, (Path, str)):
        config = LamaConfig(config)
    threads = str(config['threads'])
    if new_log:
        common.init_logging(config / 'invert_transforms.log')
    reg_dirs = get_reg_dirs(config)
    volume_names = [x.stem for x in common.get_file_paths(reg_dirs[0], ignore_folders=[RESOLUTION_IMGS_DIR, IMG_PYRAMID_DIR])]
    inv_outdir = config.mkdir('inverted_transforms')
    stages_to_invert = defaultdict(list)
    jobs: List[Dict] = []
    reg_stage_dir: Path
    for i, vol_id in enumerate(volume_names):
        for reg_stage_dir in reg_dirs:
            if not reg_stage_dir.is_dir():
                logging.error('cannot find {}'.format(reg_stage_dir))
                raise FileNotFoundError(f'Cannot find registration dir {reg_stage_dir}')
            inv_stage_dir = inv_outdir / reg_stage_dir.name
            specimen_stage_reg_dir = reg_stage_dir / vol_id
            specimen_stage_inversion_dir = inv_stage_dir / vol_id
            transform_file = common.getfile_startswith(specimen_stage_reg_dir, ELX_TRANSFORM_NAME)
            parameter_file = common.getfile_startswith(reg_stage_dir, ELX_PARAM_PREFIX)
            inv_stage_dir.mkdir(exist_ok=True)
            if reg_stage_dir.name not in stages_to_invert['label_propagation_order']:
                stages_to_invert['label_propagation_order'].insert(0, reg_stage_dir.name)
            if clobber:
                common.mkdir_force(specimen_stage_inversion_dir)  
            reg_metadata = cfg_load(specimen_stage_reg_dir / common.INDV_REG_METADATA)
            fixed_volume = (specimen_stage_reg_dir / reg_metadata['fixed_vol']).resolve()
            job = {
                'specimen_stage_inversion_dir': specimen_stage_inversion_dir,
                'parameter_file': abspath(parameter_file),
                'transform_file': transform_file,
                'fixed_volume': fixed_volume,
                'param_file_output_name': 'inversion_parameters.txt',
                'image_replacements': IMAGE_REPLACEMENTS,
                'label_replacements': LABEL_REPLACEMENTS,
                'image_transform_file': PROPAGATE_IMAGE_TRANSFORM,
                'label_transform_file': PROPAGATE_LABEL_TRANFORM,
                'clobber': clobber,
                'threads': threads
            }
            jobs.append(job)
    pool = Pool(8)
    try:
        pool.map(_invert_transform_parameters, jobs)
    except KeyboardInterrupt:
        print('terminating inversion')
        pool.terminate()
        pool.join()
    reg_dir = Path(os.path.relpath(reg_stage_dir, inv_outdir))
    stages_to_invert['registration_directory'] = str(reg_dir)  
    invert_config = config['inverted_transforms'] / PROPAGATE_CONFIG
    with open(invert_config, 'w') as yf:
        yf.write(yaml.dump(dict(stages_to_invert), default_flow_style=False))
def batch_invert_transform_parameters ( config : Union [ Path , LamaConfig ] , clobber = True , new_log : bool = False ) : common . test_installation ( 'elastix' ) if isinstance ( config , ( Path , str ) ) : config = LamaConfig ( config ) threads = str ( config [ 'threads' ] ) if new_log : common . init_logging ( config / 'invert_transforms.log' ) reg_dirs = get_reg_dirs ( config ) volume_names = [ x . stem for x in common . get_file_paths ( reg_dirs [ 0 ] , ignore_folders = [ RESOLUTION_IMGS_DIR , IMG_PYRAMID_DIR ] ) ] inv_outdir = config . mkdir ( 'inverted_transforms' ) stages_to_invert = defaultdict ( list ) jobs : List [ Dict ] = [ ] reg_stage_dir : Path for i , vol_id in enumerate ( volume_names ) : for reg_stage_dir in reg_dirs : if not reg_stage_dir . is_dir ( ) : logging . error ( 'cannot find {}' . format ( reg_stage_dir ) ) raise FileNotFoundError ( f'Cannot find registration dir {reg_stage_dir}' ) inv_stage_dir = inv_outdir / reg_stage_dir . name specimen_stage_reg_dir = reg_stage_dir / vol_id specimen_stage_inversion_dir = inv_stage_dir / vol_id transform_file = common . getfile_startswith ( specimen_stage_reg_dir , ELX_TRANSFORM_NAME ) parameter_file = common . getfile_startswith ( reg_stage_dir , ELX_PARAM_PREFIX ) inv_stage_dir . mkdir ( exist_ok = True ) if reg_stage_dir . name not in stages_to_invert [ 'label_propagation_order' ] : stages_to_invert [ 'label_propagation_order' ] . insert ( 0 , reg_stage_dir . name ) if clobber : common . mkdir_force ( specimen_stage_inversion_dir ) reg_metadata = cfg_load ( specimen_stage_reg_dir / common . INDV_REG_METADATA ) fixed_volume = ( specimen_stage_reg_dir / reg_metadata [ 'fixed_vol' ] ) . resolve ( ) job = { 'specimen_stage_inversion_dir' : specimen_stage_inversion_dir , 'parameter_file' : abspath ( parameter_file ) , 'transform_file' : transform_file , 'fixed_volume' : fixed_volume , 'param_file_output_name' : 'inversion_parameters.txt' , 'image_replacements' : IMAGE_REPLACEMENTS , 'label_replacements' : LABEL_REPLACEMENTS , 'image_transform_file' : PROPAGATE_IMAGE_TRANSFORM , 'label_transform_file' : PROPAGATE_LABEL_TRANFORM , 'clobber' : clobber , 'threads' : threads } jobs . append ( job ) pool = Pool ( 8 ) try : pool . map ( _invert_transform_parameters , jobs ) except KeyboardInterrupt : print ( 'terminating inversion' ) pool . terminate ( ) pool . join ( ) reg_dir = Path ( os . path . relpath ( reg_stage_dir , inv_outdir ) ) stages_to_invert [ 'registration_directory' ] = str ( reg_dir ) invert_config = config [ 'inverted_transforms' ] / PROPAGATE_CONFIG with open ( invert_config , 'w' ) as yf : yf . write ( yaml . dump ( dict ( stages_to_invert ) , default_flow_style = False ) )
Create new elastix TransformParameter files that can then be used by transformix to invert labelmaps, stats etc
Parameters
Create new elastix TransformParameter files that can then be used by transformix to invert labelmaps stats etc Parameters
"""
    Create new elastix TransformParameter files that can then be used by transformix to invert labelmaps, stats etc

    Parameters
    ----------
    config
        path to original reg pipeline config file

    clobber
        if True overwrite inverted parameters present

    new_log:
        Whether to create a new log file. If called from another module, logging may happen there
    """ # Get the image basenames from the first stage registration folder (usually rigid) # ignore images in non-relevent folder that may be present # Create the folder to put the specimen inversion parameter files in. # Add the stage to the inversion order config (in reverse order), if not already. # Overwrite any inversion file that exist for a single specimen # Each registration directory contains a metadata file, which contains the relative path to the fixed volume # Invert the Transform parameters with options for normal image inversion # By putting each inverison job (a single job per registration stage) we can speed things up a bit # If we can get multithreded inversion in elastix we can remove this python multithreading # TODO: Should we replace the need for this invert.yaml? # Doc why we need this # Create a yaml config file so that inversions can be run seperatley
config
Union[Path, LamaConfig]
clobber
new_log
bool