You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

142 lines
5.9 KiB

"""Image Transform Processing."""
import os
import sys
import hashlib
from config import Config as Conf
from processing import Processing
from processing.utils import select_phases
from processing.worker import run_worker
from utils import camel_case_to_str, write_image
from loader import Loader
class ImageProcessing(Processing):
"""Image Processing Class."""
def _setup(self, *args):
"""
Process Image Constructor.
:param args: <dict> args parameter to run the image transformation (default use Conf.args)
"""
self.__phases = select_phases(self._args)
self.__input_path = self._args['input']
self.__output_path = self._args['output']
self.__altered_path = self._args.get('altered')
self.__starting_step = self._args['steps'][0] if self._args.get('steps') else 0
self.__ending_step = self._args['steps'][1] if self._args.get('steps') else None
Conf.log.debug("All Phases : {}".format(self.__phases))
Conf.log.debug("To Be Executed Phases : {}".format(self.__phases[self.__starting_step:self.__ending_step]))
imagename_no_ext = os.path.splitext(os.path.basename(self.__input_path))[0]
if (self._args.get('folder_altered')):
folder_name = imagename_no_ext + '_' + str(hashlib.md5(open(self.__input_path, 'rb').read()).hexdigest())
folder_path = os.path.join(self._args['folder_altered'], folder_name)
if (not os.path.isdir(folder_path)):
os.makedirs(folder_path, exist_ok=True)
self._args['folder_altered'] = folder_path
path = self._args['folder_altered']
self.__image_steps = [self.__input_path] + [
os.path.join(path, "{}.png".format(p().__class__.__name__))
for p in self.__phases[:self.__starting_step]
]
elif (self.__altered_path):
folder_name = imagename_no_ext + '_' + str(hashlib.md5(open(self.__input_path, 'rb').read()).hexdigest())
folder_path = os.path.join(self.__altered_path, folder_name)
if (not os.path.isdir(folder_path)):
os.makedirs(folder_path, exist_ok=True)
self.__altered_path = folder_path
path = self.__altered_path
self.__image_steps = [self.__input_path] + [
os.path.join(path, "{}.png".format(p().__class__.__name__))
for p in self.__phases[:self.__starting_step]
]
else:
# TODO: refactor me, please!
self.__image_steps = [self.__input_path] + [
self.__input_path
for p in self.__phases[:self.__starting_step]
]
Conf.log.info("Processing on {}".format(str(self.__image_steps)))
try:
self.__image_steps = [
(Loader.get_loader(x)).load(x) if isinstance(x, str) else x for x in self.__image_steps
]
except (FileNotFoundError, AttributeError) as e:
Conf.log.error(e)
Conf.log.error("{} is not able to resume because it not able to load required images. "
.format(camel_case_to_str(self.__class__.__name__)))
Conf.log.error("Possible source of this error is that --altered argument is not a correct "
"directory path that contains valid images.")
sys.exit(1)
def _execute(self, *args):
"""
Execute all phases on the image.
:return: None
"""
# todo: refactor me, please!
# with this we force the auto-resize for dreamtime, but it is far from ideal
#if self.__starting_step == 5:
# r = run_worker(self.__phases[0], self.__image_steps, config=self._args)
# self.__image_steps.append(r)
for step,p in enumerate(x for x in self.__phases[self.__starting_step:self.__ending_step]):
r = run_worker(p, self.__image_steps, config=self._args)
self.__image_steps.append(r)
# todo: refactor me, please!
if self._args.get('export_step'):
export_step = self._args.get('export_step')
if self._args.get('overlay'):
export_step += 2
#Conf.log.debug("Fixing overlay export_step = {}".format(export_step))
if self._args.get('auto_rescale') or self._args.get('auto_resize') or self._args.get('auto_resize_crop'):
export_step += 1
#Conf.log.debug("Fixing scale export_step = {}".format(export_step))
if export_step == (step-1):
step_path = self._args.get('export_step_path') or os.path.abspath(os.path.join(self.__output_path, '..', 'export.png'))
if self._args.get('overlay'):
r = run_worker(self.__phases[-1], self.__image_steps, config=self._args)
write_image(r, step_path)
Conf.log.debug("Export Step Image Of {} Execution: {}".format(
camel_case_to_str(p.__name__),
step_path
))
if self.__altered_path:
if (self._args.get('folder_altered')):
path = self._args['folder_altered']
else:
path = self.__altered_path
write_image(r, os.path.join(path, "{}.png".format(p.__name__)))
Conf.log.debug("{} Step Image Of {} Execution".format(
os.path.join(path, "{}.png".format(p.__name__)),
camel_case_to_str(p.__name__),
))
write_image(self.__image_steps[-1], self.__output_path)
Conf.log.info("{} Created".format(self.__output_path))
Conf.log.debug("{} Result Image Of {} Execution"
.format(self.__output_path, camel_case_to_str(self.__class__.__name__)))
return self.__image_steps[-1]