lama/registration_pipeline/run_lama.py
def run_registration_schedule(config: LamaConfig, first_stage_only=False) -> Path:
    """
    Run the registrations specified in the config file

    Parameters
    ----------
    config: Parsed and validated lama config
    first_stage_only: If True, just do the initial rigid stage

    Returns
    -------
    The path to the final registrered images
    """
    st = config['stage_targets']
    if st:
        with open(st, 'r') as stfh:
            stage_targets = cfg_load(stfh)['targets']
        if len(config['registration_stage_params']) != len(stage_targets):
            logging.error(f'Len stage targets: {len(stage_targets)}')
            logging.error(f'Len reg stages: {len(config["registration_stage_params"])}')
            raise LamaConfigError("restage len != number of registration stages")

    # Create a folder to store mid section coronal images to keep an eye on registration process
    if not config['no_qc']:
        qc_metric_dir = config['metric_charts_dir']

    elastix_stage_parameters = generate_elx_parameters(config, do_pairwise=config['pairwise_registration'])
    regenerate_target = config['generate_new_target_each_stage']

    if regenerate_target and st:
        raise LamaConfigError('cannot have regenerate_target and stage_targets')

    if regenerate_target:
        logging.info('Creating new target each stage for population average creation')
    else:
        logging.info('Using same target for each stage')

    # Set the moving volume dir and the fixed image for the first stage
    moving_vols_dir = config['inputs']

    # Set the fixed volume up for the first stage. This will checnge each stage if doing population average
    if st:
        fixed_vol = stage_targets[0]
    else:
        fixed_vol = config['fixed_volume']

    for i, reg_stage in enumerate(config['registration_stage_params']):

        tform_type = reg_stage['elastix_parameters']['Transform']
        euler_stage = True if tform_type == 'EulerTransform' else False
        # affine_similarity_stage = True if tform_type in ['AffineTransform', 'SimilarityTransform'] else False

        if config['pairwise_registration']:
            if not euler_stage:
                logging.info('doing pairwise registration')
                reg_method = PairwiseBasedRegistration
            else:
                reg_method = TargetBasedRegistration
                logging.info(
                    'using target-based registration for initial rigid stage of pairwise registrations')

        else:
            logging.info('using target-based registration')
            reg_method = TargetBasedRegistration

        #  Make the stage output dir
        stage_id = reg_stage['stage_id']
        stage_dir = config.stage_dirs[stage_id]

        common.mkdir_force(stage_dir)

        logging.info("### Current registration step: {} ###".format(stage_id))

        # Make the elastix parameter file for this stage
        elxparam = elastix_stage_parameters[stage_id]
        elxparam_path = join(stage_dir, ELX_PARAM_PREFIX + stage_id + '.txt')

        with open(elxparam_path, 'w') as fh:
            if elxparam:  # Not sure why I put this here
                fh.write(elxparam)

        # if i < 2:  # TODO: shall we keep the fixed mask throughout? I think we should in next release
        #     fixed_mask = config['fixed_mask']

        # # If we are doing target-based phenotype detection, we can used the fixed mask for every stage
        if not config['generate_new_target_each_stage']:
            fixed_mask = config['fixed_mask']
        else:
            fixed_mask = None

        # Do the registrations
        registrator = reg_method(elxparam_path,
                                 moving_vols_dir,
                                 stage_dir,
                                 config['filetype'],
                                 config['threads'],
                                 fixed_mask
                                 )

        if (not config['pairwise_registration']) or (config['pairwise_registration'] and euler_stage):
            registrator.set_target(fixed_vol)

        if reg_stage['elastix_parameters']['Transform'] == 'BSplineTransform':
            if config['fix_folding']:
                logging.info(f'Folding correction for stage {stage_id} set')
            registrator.fix_folding = config['fix_folding']  # Curently only works for TargetBasedRegistration

        registrator.run()  # Do the registrations for a single stage

        # Make average from the stage outputs
        if regenerate_target:
            average_path = join(config['average_folder'], '{0}.{1}'.format(stage_id, config['filetype']))
            registrator.make_average(average_path)

        if not config['no_qc']:

            stage_metrics_dir = qc_metric_dir / stage_id
            common.mkdir_force(stage_metrics_dir)
            make_charts(stage_dir, stage_metrics_dir)

        # Setup the fixed and moving for the next stage, if there is one
        if i + 1 < len(config['registration_stage_params']):
            if regenerate_target:
                fixed_vol = average_path  # The avergae from the previous step
            elif st:
                fixed_vol = stage_targets[i+1]

            moving_vols_dir = stage_dir  # Set the output of the current stage top be the input of the next

        if first_stage_only:
            return stage_dir

    logging.info("### Registration finished ###")

    return stage_dir

    Run the registrations specified in the config file

    Parameters
    ----------
    config: Parsed and validated lama config
    first_stage_only: If True, just do the initial rigid stage

    Returns
    -------
    The path to the final registrered images
    
Run the registrations specified in the config file
Parameters

Parsed and validated lama config
first_stage_only: If True, just do the initial rigid stage

Returns

The path to the final registrered images
Run the registrations specified in the config file Parameters Parsed and validated lama config first_stage_only : If True just do the initial rigid stage Returns The path to the final registrered images
def run_registration_schedule(config: LamaConfig, first_stage_only=False) -> Path:
    st = config['stage_targets']
    if st:
        with open(st, 'r') as stfh:
            stage_targets = cfg_load(stfh)['targets']
        if len(config['registration_stage_params']) != len(stage_targets):
            logging.error(f'Len stage targets: {len(stage_targets)}')
            logging.error(f'Len reg stages: {len(config["registration_stage_params"])}')
            raise LamaConfigError("restage len != number of registration stages")
    if not config['no_qc']:
        qc_metric_dir = config['metric_charts_dir']
    elastix_stage_parameters = generate_elx_parameters(config, do_pairwise=config['pairwise_registration'])
    regenerate_target = config['generate_new_target_each_stage']
    if regenerate_target and st:
        raise LamaConfigError('cannot have regenerate_target and stage_targets')
    if regenerate_target:
        logging.info('Creating new target each stage for population average creation')
    else:
        logging.info('Using same target for each stage')
    moving_vols_dir = config['inputs']
    if st:
        fixed_vol = stage_targets[0]
    else:
        fixed_vol = config['fixed_volume']
    for i, reg_stage in enumerate(config['registration_stage_params']):
        tform_type = reg_stage['elastix_parameters']['Transform']
        euler_stage = True if tform_type == 'EulerTransform' else False
        if config['pairwise_registration']:
            if not euler_stage:
                logging.info('doing pairwise registration')
                reg_method = PairwiseBasedRegistration
            else:
                reg_method = TargetBasedRegistration
                logging.info(
                    'using target-based registration for initial rigid stage of pairwise registrations')
        else:
            logging.info('using target-based registration')
            reg_method = TargetBasedRegistration
        stage_id = reg_stage['stage_id']
        stage_dir = config.stage_dirs[stage_id]
        common.mkdir_force(stage_dir)
        logging.info("### Current registration step: {} ###".format(stage_id))
        elxparam = elastix_stage_parameters[stage_id]
        elxparam_path = join(stage_dir, ELX_PARAM_PREFIX + stage_id + '.txt')
        with open(elxparam_path, 'w') as fh:
            if elxparam:  
                fh.write(elxparam)
        if not config['generate_new_target_each_stage']:
            fixed_mask = config['fixed_mask']
        else:
            fixed_mask = None
        registrator = reg_method(elxparam_path,
                                 moving_vols_dir,
                                 stage_dir,
                                 config['filetype'],
                                 config['threads'],
                                 fixed_mask
                                 )
        if (not config['pairwise_registration']) or (config['pairwise_registration'] and euler_stage):
            registrator.set_target(fixed_vol)
        if reg_stage['elastix_parameters']['Transform'] == 'BSplineTransform':
            if config['fix_folding']:
                logging.info(f'Folding correction for stage {stage_id} set')
            registrator.fix_folding = config['fix_folding']  
        registrator.run()   for a single stage
        if regenerate_target:
            average_path = join(config['average_folder'], '{0}.{1}'.format(stage_id, config['filetype']))
            registrator.make_average(average_path)
        if not config['no_qc']:
            stage_metrics_dir = qc_metric_dir / stage_id
            common.mkdir_force(stage_metrics_dir)
            make_charts(stage_dir, stage_metrics_dir)
        if i + 1 < len(config['registration_stage_params']):
            if regenerate_target:
                fixed_vol = average_path  
            elif st:
                fixed_vol = stage_targets[i+1]
            moving_vols_dir = stage_dir  
        if first_stage_only:
            return stage_dir
    logging.info("### Registration finished ###")
    return stage_dir
def run_registration_schedule ( config : LamaConfig , first_stage_only = False ) -> Path : st = config [ 'stage_targets' ] if st : with open ( st , 'r' ) as stfh : stage_targets = cfg_load ( stfh ) [ 'targets' ] if len ( config [ 'registration_stage_params' ] ) != len ( stage_targets ) : logging . error ( f'Len stage targets: {len(stage_targets)}' ) logging . error ( f'Len reg stages: {len(config["registration_stage_params"])}' ) raise LamaConfigError ( "restage len != number of registration stages" ) if not config [ 'no_qc' ] : qc_metric_dir = config [ 'metric_charts_dir' ] elastix_stage_parameters = generate_elx_parameters ( config , do_pairwise = config [ 'pairwise_registration' ] ) regenerate_target = config [ 'generate_new_target_each_stage' ] if regenerate_target and st : raise LamaConfigError ( 'cannot have regenerate_target and stage_targets' ) if regenerate_target : logging . info ( 'Creating new target each stage for population average creation' ) else : logging . info ( 'Using same target for each stage' ) moving_vols_dir = config [ 'inputs' ] if st : fixed_vol = stage_targets [ 0 ] else : fixed_vol = config [ 'fixed_volume' ] for i , reg_stage in enumerate ( config [ 'registration_stage_params' ] ) : tform_type = reg_stage [ 'elastix_parameters' ] [ 'Transform' ] euler_stage = True if tform_type == 'EulerTransform' else False if config [ 'pairwise_registration' ] : if not euler_stage : logging . info ( 'doing pairwise registration' ) reg_method = PairwiseBasedRegistration else : reg_method = TargetBasedRegistration logging . info ( 'using target-based registration for initial rigid stage of pairwise registrations' ) else : logging . info ( 'using target-based registration' ) reg_method = TargetBasedRegistration stage_id = reg_stage [ 'stage_id' ] stage_dir = config . stage_dirs [ stage_id ] common . mkdir_force ( stage_dir ) logging . info ( "### Current registration step: {} ###" . format ( stage_id ) ) elxparam = elastix_stage_parameters [ stage_id ] elxparam_path = join ( stage_dir , ELX_PARAM_PREFIX + stage_id + '.txt' ) with open ( elxparam_path , 'w' ) as fh : if elxparam : fh . write ( elxparam ) if not config [ 'generate_new_target_each_stage' ] : fixed_mask = config [ 'fixed_mask' ] else : fixed_mask = None registrator = reg_method ( elxparam_path , moving_vols_dir , stage_dir , config [ 'filetype' ] , config [ 'threads' ] , fixed_mask ) if ( not config [ 'pairwise_registration' ] ) or ( config [ 'pairwise_registration' ] and euler_stage ) : registrator . set_target ( fixed_vol ) if reg_stage [ 'elastix_parameters' ] [ 'Transform' ] == 'BSplineTransform' : if config [ 'fix_folding' ] : logging . info ( f'Folding correction for stage {stage_id} set' ) registrator . fix_folding = config [ 'fix_folding' ] registrator . run ( ) if regenerate_target : average_path = join ( config [ 'average_folder' ] , '{0}.{1}' . format ( stage_id , config [ 'filetype' ] ) ) registrator . make_average ( average_path ) if not config [ 'no_qc' ] : stage_metrics_dir = qc_metric_dir / stage_id common . mkdir_force ( stage_metrics_dir ) make_charts ( stage_dir , stage_metrics_dir ) if i + 1 < len ( config [ 'registration_stage_params' ] ) : if regenerate_target : fixed_vol = average_path elif st : fixed_vol = stage_targets [ i + 1 ] moving_vols_dir = stage_dir if first_stage_only : return stage_dir logging . info ( "### Registration finished ###" ) return stage_dir
Run the registrations specified in the config file
Parameters
Run the registrations specified in the config file Parameters
"""
    Run the registrations specified in the config file

    Parameters
    ----------
    config: Parsed and validated lama config
    first_stage_only: If True, just do the initial rigid stage

    Returns
    -------
    The path to the final registrered images
    """ # Create a folder to store mid section coronal images to keep an eye on registration process # Set the moving volume dir and the fixed image for the first stage # Set the fixed volume up for the first stage. This will checnge each stage if doing population average # affine_similarity_stage = True if tform_type in ['AffineTransform', 'SimilarityTransform'] else False #  Make the stage output dir # Make the elastix parameter file for this stage # Not sure why I put this here # if i < 2:  # TODO: shall we keep the fixed mask throughout? I think we should in next release #     fixed_mask = config['fixed_mask'] # # If we are doing target-based phenotype detection, we can used the fixed mask for every stage # Do the registrations # Curently only works for TargetBasedRegistration # Do the registrations for a single stage # Make average from the stage outputs # Setup the fixed and moving for the next stage, if there is one # The avergae from the previous step # Set the output of the current stage top be the input of the next
config
LamaConfig
first_stage_only