Source code for pyalfe.tasks.t1_preprocessing
import os
import logging
from pyalfe.data_structure import PipelineDataDir, Modality
from pyalfe.image_processing import ImageProcessor
[docs]
class T1Preprocessing:
"""This task preforms neck trimming and upsampling on the T1 image
if needed.
Attributes
----------
image_processor: ImageProcessor
The image processor object.
pipeline_dir: PipelineDataDir
The pipeline data directory object.
overwrite: bool
Whether to overwrite any existing output image. Default is True.
"""
logger = logging.getLogger('T1Preprocessing')
def __init__(
self,
image_processor: ImageProcessor,
pipeline_dir: PipelineDataDir,
overwrite=True,
):
self.pipeline_dir = pipeline_dir
self.image_processor = image_processor
self.overwrite = overwrite
[docs]
def run(self, accession):
t1ss = self.pipeline_dir.get_output_image(
accession, Modality.T1, image_type='skullstripped'
)
if not os.path.exists(t1ss):
self.logger.info(
'T1 skullstripped image is missing. Skipping T1Preprocessing.'
)
return
trim_output = self.pipeline_dir.get_output_image(
accession, 'T1', image_type='trim'
)
if self.overwrite or not os.path.exists(trim_output):
self.image_processor.trim_largest_comp(t1ss, trim_output, [15, 15, 15])
output = self.pipeline_dir.get_output_image(
accession, Modality.T1, image_type='trim_upsampled'
)
if self.overwrite or not os.path.exists(output):
dims = self.image_processor.get_dims(t1ss)
upfactors = []
for dim in dims:
if dim < 50:
upfactors.append(500)
else:
upfactors.append(100)
self.image_processor.resample_new_dim(trim_output, output, *upfactors)