Commit 2ea2a759 authored by Derek Fukumori's avatar Derek Fukumori
Browse files

imagestack: combine unpacking and validation of imagestack

parent ec999a72
import re
from pathlib import Path from pathlib import Path
from subprocess import check_call
from time import time
from .const import PB_TMP from .logger import logger
def get_imagestack_info(task_info): def get_imagestack_info(task_info):
""" """
...@@ -29,27 +29,47 @@ def get_imagestack_info(task_info): ...@@ -29,27 +29,47 @@ def get_imagestack_info(task_info):
return {'image_type': 'tif', 'archive_type': 'tar'} return {'image_type': 'tif', 'archive_type': 'tar'}
raise Exception('Unhandled imagestack format: %s' % source_format) raise Exception('Unhandled imagestack format: %s' % source_format)
def unpack_and_validate_imagestack(imagestack_path, imagestack_info, dst):
""" Unpack and validate an imagestack
def validate_imagestack(identifier, img_ext): An imagestack is valid if it contains at least one directory that contains
""" at least one image of the expected image type.
Validates an image stack by checking if the image stack follows the proper
directory structure, and if the files in the image stack contain the right Args:
file names. imagestack_path (str): The imagestack archive path
imagestack_info (dict): {'archive_type': ..., 'image_type': ...}
dst (str): Destination directory for the unpacked imagestack
Returns:
(str, int): Tuple containing the path to the unpacked image directory
and the image count.
""" """
image_count = 0 logger.info('Unpacking image stack.')
img_dir = Path(PB_TMP)/f'{identifier}_{img_ext}' start_time = time()
if not (img_dir.exists() and img_dir.is_dir()): if imagestack_info['archive_type'] == 'tar':
raise Exception('Imagestack does not follow proper directory structure.') check_call(['tar', '-xf', imagestack_path, '-C', dst])
elif imagestack_info['archive_type'] == 'zip':
check_call(['unzip', '-qq', '-o', imagestack_path, '-d', dst])
else:
raise ValueError('Cannot extract archive_type %s' % imagestack_info['archive_type'])
logger.info('Unpacking image stack took %f seconds', time() - start_time)
for f in Path(dst).iterdir():
if f.name.endswith(f'_{imagestack_info["image_type"]}'):
img_dir = f
break
else:
raise Exception('Unable to locate image directory in imagestack.')
image_count = 0
for img_path in img_dir.iterdir(): for img_path in img_dir.iterdir():
m = re.fullmatch(f'{identifier}_\d\d\d\d\.{img_ext}', img_path.name) if img_path.suffix[1:] == imagestack_info['image_type']:
if m:
image_count += 1 image_count += 1
if image_count == 0: if image_count == 0:
raise Exception('Imagestack contains no valid images.') raise Exception('Imagestack contains no valid images.')
return image_count return str(img_dir), image_count
# TODO: function to unpack an image stack # TODO: function to unpack an image stack
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment