Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
exit(1)
base_name = 'document' if base_name == 'B' else base_name
from_file = None if from_file == 'F' else from_file
output_dir = None if output_dir == 'O' else output_dir
compare = 'relaxed' if (compare and relaxed) else compare
threads = int(max(1, available_cpus()/2 if threads == 'T' else int(threads)))
# Do the real work --------------------------------------------------------
try:
body = MainBody(base_name, extended, from_file, output_dir, threads)
body.run(services, files, make_grid, compare)
except (KeyboardInterrupt, UserCancelled) as ex:
if __debug__: log('received {}', sys.exc_info()[0].__name__)
inform('Quitting.')
exit(0)
except Exception as ex:
if debugging:
import traceback
alert('{}\n{}', str(ex), traceback.format_exc())
import pdb; pdb.set_trace()
else:
alert(str(ex))
exit(2)
inform('Done.')
def _resized_image(self, file):
(max_width, max_height) = self._max_dimensions
file_ext = filename_extension(file)
name_tail = '.handprint' + file_ext
new_file = file if name_tail in file else filename_basename(file) + name_tail
if path.exists(new_file) and readable(new_file):
(image_width, image_height) = image_dimensions(new_file)
if image_width < max_width and image_height < max_height:
inform('Using reduced image found in {}', relative(new_file))
return new_file
else:
# We found a "-reduced" file, perhaps from a previous run, but
# for the current set of services, dimension are too large.
if __debug__: log('existing resized file larger than {}x{}: {}',
max_width, max_height, new_file)
inform('Dimensions too large; reducing dimensions: {}', relative(file))
(resized, error) = reduced_image_dimensions(file, new_file, max_width, max_height)
if error:
alert('Failed to re-dimension {}: {}', relative(file), error)
return None
return resized
warn('Pausing {} due to rate limits', service_name)
time.sleep(1/service.max_rate() - time_passed)
# FIXME resend after pause
if output.error:
alert('{} failed: {}', service_name, output.error)
warn('No result from {} for {}', service_name, relative(image.file))
return None
inform('Got result from {}.', service_name)
file_name = path.basename(image.file)
base_path = path.join(image.dest_dir, file_name)
annot_path = None
report_path = None
if self._make_grid:
annot_path = self._renamed(base_path, str(service), 'png')
inform('Creating annotated image for {}.', service_name)
with self._lock:
self._save(annotated_image(image.file, output.boxes, service), annot_path)
if self._extended_results:
txt_file = self._renamed(base_path, str(service), 'txt')
json_file = self._renamed(base_path, str(service), 'json')
inform('Saving all data for {}.', service_name)
self._save(json.dumps(output.data), json_file)
inform('Saving extracted text for {}.', service_name)
self._save(output.text, txt_file)
if self._compare:
gt_file = alt_extension(image.item_file, 'gt.txt')
report_path = self._renamed(image.item_file, str(service), 'tsv')
relaxed = (self._compare == 'relaxed')
if readable(gt_file) and nonempty(gt_file):
if __debug__: log('reading ground truth from {}', gt_file)
gt_text = open(gt_file, 'r').read()
return None
inform('Got result from {}.', service_name)
file_name = path.basename(image.file)
base_path = path.join(image.dest_dir, file_name)
annot_path = None
report_path = None
if self._make_grid:
annot_path = self._renamed(base_path, str(service), 'png')
inform('Creating annotated image for {}.', service_name)
with self._lock:
self._save(annotated_image(image.file, output.boxes, service), annot_path)
if self._extended_results:
txt_file = self._renamed(base_path, str(service), 'txt')
json_file = self._renamed(base_path, str(service), 'json')
inform('Saving all data for {}.', service_name)
self._save(json.dumps(output.data), json_file)
inform('Saving extracted text for {}.', service_name)
self._save(output.text, txt_file)
if self._compare:
gt_file = alt_extension(image.item_file, 'gt.txt')
report_path = self._renamed(image.item_file, str(service), 'tsv')
relaxed = (self._compare == 'relaxed')
if readable(gt_file) and nonempty(gt_file):
if __debug__: log('reading ground truth from {}', gt_file)
gt_text = open(gt_file, 'r').read()
inform('Saving {} comparison to ground truth', service_name)
self._save(text_comparison(output.text, gt_text, relaxed), report_path)
elif not nonempty(gt_file):
warn('Skipping {} comparison because {} is empty',
service_name, relative(gt_file))
else:
def print_intro(ui):
if ui.use_color():
cb = ['chartreuse', 'bold']
name = styled('Handprint', cb)
acronym = '{}written {}age {}ecognit{}o{} {}est'.format(
styled('Hand', cb), styled('p', cb), styled('r', cb),
styled('i', cb), styled('n', cb), styled('t', cb))
else:
name = 'Handprint'
acronym = 'HANDwritten Page RecognItioN Test'
inform('┏' + '━'*68 + '┓')
inform('┃ Welcome to {}, the {}! ┃', name, acronym)
inform('┗' + '━'*68 + '┛')
extended = self._extended
from_file = self._from_file
output_dir = self._output_dir
threads = self._threads
# Gather up some things and get prepared.
targets = self.targets_from_arguments(files, from_file)
if not targets:
raise RuntimeError('No images to process; quitting.')
num_targets = len(targets)
inform('Will apply {} service{} ({}) to {} image{}.',
len(services), 's' if len(services) > 1 else '',
', '.join(services), num_targets, 's' if num_targets > 1 else '')
if self._extended:
inform('Will save extended results.')
inform('Will use up to {} process threads.', threads)
# Get to work.
if __debug__: log('initializing manager and starting processes')
manager = Manager(services, threads, output_dir, make_grid, compare, extended)
print_separators = num_targets > 1
for index, item in enumerate(targets, start = 1):
if print_separators:
inform(styled('━'*70, 'dark'))
manager.run_services(item, index, base_name)
if print_separators:
inform(styled('━'*70, 'dark'))
def _smaller_file(self, file):
if not file:
return None
file_ext = filename_extension(file)
name_tail = '.handprint' + file_ext
new_file = file if name_tail in file else filename_basename(file) + name_tail
if path.exists(new_file):
if image_size(new_file) < self._max_size:
inform('Reusing resized image found in {}', relative(new_file))
return new_file
else:
# We found a ".handprint.ext" file, perhaps from a previous run,
# but for the current set of services, it's larger than allowed.
if __debug__: log('existing resized file larger than {}b: {}',
humanize.intcomma(self._max_size), new_file)
inform('Size too large; reducing size: {}', relative(file))
(resized, error) = reduced_image_size(file, new_file, self._max_size)
if error:
alert('Failed to resize {}: {}', relative(file), error)
return None
return resized