#!/usr/bin/env python
# coding: utf-8
#
# Copyright (c) 2012, Adam Turcotte (adam.turcotte@gmail.com)
# Nicolas Robidoux (nicolas.robidoux@gmail.com)
# License: BSD 2-Clause License
#
# This file is part of the
# EXQUIRES (EXtensible QUantitative Image RESampling) test suite
#
"""A collection of classes used to compute image difference data.
The hierarchy of classes is as follows:
* :class:`Operations` encapsulate a list of :class:`Images`
* :class:`Images` encapsulate a `dict` of images
and a list of :class:`Downsamplers`
* :class:`Downsamplers` encapsulate a `dict` of downsamplers
and a list of :class:`Ratios`
* :class:`Ratios` encapsulate a `dict` of ratios and a list :class:`Images`
* :class:`Images` encapsulate a `dict` of images and a `dict` of metrics
These classes work together to downsample the master images, upsample the
downsampled images, and compare the upsampled images to the master images.
To perform the operations, call :meth:`Operations.compute`.
"""
import os
import shutil
from subprocess import call, check_output
from exquires import database, progress, tools
# pylint: disable-msg=R0903
[docs]class Operations(object):
"""A collection of Image objects to compute data with.
This class is responsible for calling all operations defined in the
specified project file when using :ref:`exquires-run` or
:ref:`exquires-update`.
:param images: images to downsample
:type images: list of :class:`Images`
"""
def __init__(self, images):
"""Create a new :class:`Operations` object."""
self.images = images
self.len = sum(len(image) for image in self.images)
def __len__(self):
"""Return the length of this :class:`Operations` object.
The length of an :class:`Operations` object is the total number of
operations (downsampling, upsampling, and comparing) to be performed.
:return: length of this :class:`Operations` object
:rtype: `integer`
"""
return self.len
[docs] def compute(self, args, old=None):
"""Perform all operations.
:param args: arguments
:param args.prog: name of the calling program
:param args.dbase_file: database file
:param args.proj: name of the current project
:param args.silent: `True` if using silent mode
:param args.met_same: unchanged metrics
:param args.metrics: current metrics
:param args.config_file: current configuration file
:param args.config_bak: previous configuration file
:param old: old configuration entries to be removed
:type args: :class:`argparse.Namespace`
:type args.prog: `string`
:type args.dbase_file: `path`
:type args.proj: `string`
:type args.silent: `boolean`
:type args.met_same: `dict`
:type args.metrics: `dict`
:type args.config_file: `path`
:type args.config_bak: `path`
:type old: :class:`argparse.Namespace`
"""
# Setup verbose mode.
if not args.silent:
prg = progress.Progress(args.prog, args.proj, len(self))
args.do_op = prg.do_op
cleanup = prg.cleanup
complete = prg.complete
else:
prg = []
args.do_op = lambda *a, **k: None
cleanup = lambda: None
complete = lambda: None
# Backup any existing database file.
dbase_bak = '.'.join([args.dbase_file, 'bak'])
if os.path.isfile(args.dbase_file):
shutil.copyfile(args.dbase_file, dbase_bak)
# Open the database connection.
args.dbase = database.Database(args.dbase_file)
success = True
try:
# Remove old database tables.
if old:
cleanup()
args.dbase.drop_tables(old.images,
old.downsamplers, old.ratios)
# Create the project folder if it does not exist.
tools.create_dir(args.proj)
# Compute for all images.
for image in self.images:
image.compute(args)
except StandardError as std_err:
success = False
error = std_err
finally:
# Remove the project directory and close the database.
shutil.rmtree(args.proj, True)
args.dbase.close()
if success:
# Backup the project file.
shutil.copyfile(args.config_file, args.config_bak)
# Delete the database backup.
if os.path.isfile(dbase_bak):
os.remove(dbase_bak)
# Indicate completion and restore the console.
complete()
del prg
else:
# Restore the previous database file.
os.remove(args.dbase_file)
if os.path.isfile(dbase_bak):
shutil.move(dbase_bak, args.dbase_file)
# Restore the console
del prg
# Print an error message.
print error
[docs]class Images(object):
"""This class calls operations for a particular set of images.
:param images: images to downsample
:param downsamplers: downsamplers to use
:param same: `True` if using unchanged images
:type images: `dict`
:type downsamplers: list of :class:`Downsamplers`
:type same: `boolean`
"""
def __init__(self, images, downsamplers, same=False):
"""Create a new :class:`Images` object."""
self.images = images
self.downsamplers = downsamplers
self.same = same
self.len = (len(self.images) *
sum(len(down) + down.ops for down in self.downsamplers))
def __len__(self):
"""Return the length of this :class:`Images` object.
The length of an :class:`Images` object is the number of images times
the sum of the lengths and the number of upsampling and comparison
operations of each :class:`Downsamplers` object.
:return: length of this :class:`Images` object
:rtype: `integer`
"""
return self.len
[docs] def compute(self, args):
"""Perform all operations for this set of images.
:param args: arguments
:param args.dbase_file: database file
:param args.dbase: connected database
:param args.proj: name of the current project
:param args.silent: `True` if using silent mode
:param args.met_same: unchanged metrics
:param args.metrics: current metrics
:param args.do_op: updates the displayed progress
:type args: :class:`argparse.Namespace`
:type args.dbase_file: `path`
:type args.dbase: :class:`database.Database`
:type args.proj: `string`
:type args.silent: `boolean`
:type args.met_same: `dict`
:type args.metrics: `dict`
:type args.do_op: `function`
"""
# Compute for all images.
for args.image in self.images:
# Make a copy of the test image.
if len(self):
args.image_dir = tools.create_dir(args.proj, args.image)
args.master = os.path.join(args.image_dir, 'master.tif')
shutil.copyfile(self.images[args.image], args.master)
# Compute for all downsamplers.
for downsampler in self.downsamplers:
downsampler.compute(args, self.same)
# Remove the directory for this image.
if len(self):
shutil.rmtree(args.image_dir, True)
[docs]class Downsamplers(object):
"""This class calls operations for a particular set of downsamplers.
:param downsamplers: downsamplers to use
:param ratios: ratios to downsample by
:param same: `True` if using unchanged downsamplers
:type downsamplers: `dict`
:type ratios: list of :class:`Ratios`
:type same: `boolean`
"""
def __init__(self, downsamplers, ratios, same=False):
"""Create a new :class:`Downsamplers` object."""
self.downsamplers = downsamplers
self.ratios = ratios
self.same = same
self.ops = len(self.downsamplers) * sum(rat.ops for rat in self.ratios)
self.len = (len(self.downsamplers) *
sum(len(rat) for rat in self.ratios)) if self.ops else 0
def __len__(self):
"""Return the length of this :class:`Downsamplers` object.
The length of a :class:`Downsamplers` object is the number of
downsamplers times the sum of the lengths of each :class:`Ratios`
object.
:return: length of this :class:`Downsamplers` object
:rtype: `integer`
"""
return self.len
[docs] def compute(self, args, same):
"""Perform all operations for this set of downsamplers.
:param args: arguments
:param args.dbase_file: database file
:param args.dbase: connected database
:param args.proj: name of the current project
:param args.silent: `True` if using silent mode
:param args.met_same: unchanged metrics
:param args.metrics: current metrics
:param args.do_op: updates the displayed progress
:param args.image: name of the image
:param args.image_dir: directory to store results for this image
:param args.master: master image to downsample
:param same: `True` if possibly accessing an existing table
:type args: :class:`argparse.Namespace`
:type args.dbase_file: `path`
:type args.dbase: :class:`database.Database`
:type args.proj: `string`
:type args.silent: `boolean`
:type args.met_same: `dict`
:type args.metrics: `dict`
:type args.do_op: `function`
:type args.image: `string`
:type args.image_dir: `path`
:type args.master: `path`
:type same: `boolean`
"""
is_same = self.same and same
# Compute for all downsamplers.
for args.downsampler in self.downsamplers:
# Create a directory for this downsampler if necessary.
if len(self):
args.downsampler_dir = tools.create_dir(args.image_dir,
args.downsampler)
# Compute for all ratios.
for ratio in self.ratios:
ratio.compute(args, self.downsamplers, is_same)
# Remove the directory for this downsampler.
if len(self):
shutil.rmtree(args.downsampler_dir, True)
[docs]class Ratios(object):
"""This class calls operations for a particular set of ratios.
:param ratios: ratios to downsample by
:param upsamplers: upsamplers to use
:param same: `True` if using unchanged ratios
:type ratios: `dict`
:type upsamplers: list of :class:`Upsamplers`
:type same: `boolean`
"""
def __init__(self, ratios, upsamplers, same=False):
"""Create a new :class:`Ratios` object."""
self.ratios = ratios
self.upsamplers = upsamplers
self.same = same
self.ops = len(self.ratios) * sum(len(ups) for ups in self.upsamplers)
self.len = len(self.ratios) if self.ops else 0
def __len__(self):
"""Return the length of this :class:`Ratios` object.
The length of a :class:`Ratios` object is the number of ratios.
:return: length of this :class:`Ratios` object
:rtype: `integer`
"""
return self.len
[docs] def compute(self, args, downsamplers, same):
"""Perform all operations for this set of ratios.
:param args: arguments
:param args.dbase_file: database file
:param args.dbase: connected database
:param args.proj: name of the current project
:param args.silent: `True` if using silent mode
:param args.met_same: unchanged metrics
:param args.metrics: current metrics
:param args.do_op: updates the displayed progress
:param args.image: name of the image
:param args.image_dir: directory to store results for this image
:param args.master: master image to downsample
:param args.downsampler: name of the downsampler
:param args.downsampler_dir: directory to store dowsampled images
:param downsamplers: downsamplers to use
:param same: `True` if accessing an existing table
:type args: :class:`argparse.Namespace`
:type args.dbase_file: `path`
:type args.dbase: :class:`database.Database`
:type args.proj: `string`
:type args.silent: `boolean`
:type args.met_same: `dict`
:type args.metrics: `dict`
:type args.do_op: `function`
:type args.image: `string`
:type args.image_dir: `path`
:type args.master: `path`
:type args.downsampler: `string`
:type args.downsampler_dir: `path`
:type downsamplers: `dict`
:type same: `boolean`
"""
is_same = self.same and same
# Compute for all ratios.
for args.ratio in self.ratios:
if len(self):
args.small = os.path.join(args.downsampler_dir,
'.'.join([args.ratio, 'tif']))
# Create a directory for this ratio.
ratio_dir = tools.create_dir(args.downsampler_dir, args.ratio)
# Downsample master.tif by ratio using downsampler.
# {0} input image path (master)
# {1} output image path (small)
# {2} downsampling ratio
# {3} downsampled size (width or height)
args.do_op(args)
call(
downsamplers[args.downsampler].format(
args.master, args.small,
args.ratio, self.ratios[args.ratio]
).split()
)
if is_same:
# Access the existing database table.
args.table = '_'.join([args.image,
args.downsampler, args.ratio])
args.table_bak = args.dbase.backup_table(args.table,
args.metrics)
else:
# Create a new database table.
args.table = args.dbase.add_table(
args.image, args.downsampler, args.ratio, args.metrics)
# Compute for all upsamplers.
for upsampler in self.upsamplers:
upsampler.compute(args, is_same)
# Remove the directory for this ratio.
if len(self):
shutil.rmtree(ratio_dir, True)
# Delete the backup table.
if is_same:
args.dbase.drop_backup(args.table_bak)
[docs]class Upsamplers(object):
"""This class upsamples an image and compares with its master image.
:param upsamplers: upsamplers to use
:param metrics: metrics to compare with
:param same: `True` if using unchanged upsamplers
:type upsamplers: `dict`
:type metrics: `dict`
:type same: `boolean`
"""
def __init__(self, upsamplers, metrics, same=False):
"""Create a new :class:`Upsamplers` object."""
self.upsamplers = upsamplers
self.metrics = metrics
self.same = same
ops = len(self.metrics)
self.len = len(self.upsamplers) * (ops + 1) if ops else 0
def __len__(self):
"""Return the length of this :class:`Upsamplers` object.
The length of an :class:`Upsamplers` object is the number of upsampling
and comparison operations to perform.
:return: length of this :class:`Upsamplers` object
:rtype: `integer`
"""
return self.len
[docs] def compute(self, args, same):
"""Perform all operations for this set of ratios.
:param args: arguments
:param args.dbase_file: database file
:param args.dbase: connected database
:param args.proj: name of the current project
:param args.silent: `True` if using silent mode
:param args.met_same: unchanged metrics
:param args.metrics: current metrics
:param args.do_op: updates the displayed progress
:param args.image: name of the image
:param args.image_dir: directory to store results for this image
:param args.master: master image to downsample
:param args.downsampler: name of the downsampler
:param args.downsampler_dir: directory to store dowsampled images
:param args.ratio: resampling ratio
:param args.small: downsampled image
:param args.table: name of the table to insert the row into
:param args.table_bak: name of the backup table (if it exists)
:param same: `True` if accessing an existing table
:type args: :class:`argparse.Namespace`
:type args.dbase_file: `path`
:type args.dbase: :class:`database.Database`
:type args.proj: `string`
:type args.silent: `boolean`
:type args.met_same: `dict`
:type args.metrics: `dict`
:type args.do_op: `function`
:type args.image: `string`
:type args.image_dir: `path`
:type args.master: `path`
:type args.downsampler: `string`
:type args.downsampler_dir: `path`
:type args.ratio: `string`
:type args.small: `path`
:type args.table: `string`
:type args.table_bak: `string`
:type same: `boolean`
"""
is_same = self.same and same and args.met_same
# Compute for all upsamplers.
for upsampler in self.upsamplers:
row = {}
if is_same:
# Access the existing table row.
row = args.dbase.get_error_data(args.table_bak, upsampler,
','.join(args.met_same))
if len(self):
if not is_same:
# Start creating a new table row.
row['upsampler'] = upsampler
# Construct the path to the upsampled image.
large = os.path.join(
os.path.dirname(args.small), args.ratio,
'.'.join([upsampler, 'tif'])
)
# Upsample ratio.tif back to 840 using upsampler.
# {0} input image path (small)
# {1} output image path (large)
# {2} upsampling ratio
# {3} upsampled size (always 840)
args.do_op(args, upsampler)
call(self.upsamplers[upsampler].format(
args.small, large, args.ratio, 840).split())
# Compute for all metrics.
for metric in self.metrics:
# Compare master.tif to upsampler.tif.
# {0} reference image path (master)
# {1} test image path (large)
args.do_op(args, upsampler, metric)
row[metric] = float(
check_output(
self.metrics[metric][0].format(args.master,
large).split()
)
)
# Remove the upsampled image.
os.remove(large)
# Add the new row to the table.
if row:
args.dbase.insert(args.table, row)