-
Notifications
You must be signed in to change notification settings - Fork 0
/
run.py
132 lines (106 loc) · 4.23 KB
/
run.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
import json
import logging
import os
import re
import sys
import flywheel
import openslide
# Gear basics
input_folder = 'input/file/'
output_folder = 'output/'
# declare config file path
config_file = 'config.json'
svsfilename = ""
def load_manifest_json():
"""
load the /flywheel/v0/manifest.json file as a dictionary
:return: manifest_json
:rtype: dict
"""
config_file_path = '/flywheel/v0/manifest.json'
with open(config_file_path) as manifest_data:
manifest_json = json.load(manifest_data)
return manifest_json
def log_initializer(gear_context):
"""
Generate a log with level and name configured for the suite, gear name and gear-log-level
:param gear_context: an instance of the flywheel gear context that includes the manifest_json attribute
:type gear_context: flywheel.gear_context.GearContext
:return: log
:rtype: logging.Logger
"""
log_level = gear_context.config.get('gear-log-level', 'INFO')
# Set suite (default to flywheel)
try:
suite = gear_context.manifest_json['custom']['flywheel']['suite']
except KeyError:
suite = 'flywheel'
# Set gear_name
gear_name = gear_context.manifest_json['name']
# Setup logging as per SSE best practices (Thanks Andy!)
log_name = '/'.join([suite, gear_name])
# Tweak the formatting
fmt = '%(asctime)s.%(msecs)03d [%(name)s] %(levelname)-8s: %(message)s'
logging.basicConfig(level=log_level, format=fmt, datefmt='%H:%M:%S')
log = logging.getLogger(log_name)
log.debug('{} log level is {}'.format(log_name, log_level))
return log
def get_output_path(file_path, output_directory, extension_str='png'):
"""
Generates a safe path given an input filepath, output directory and optional extension string
:param file_path: path of the original file
:type file_path: str
:param output_directory: directory for output_path
:type output_directory: str
:param extension_str: file extension to use (default = 'png')
:type extension_str: str
:return:
"""
# Strip leading .
extension_str = extension_str.lstrip('.')
basename = os.path.basename(file_path)
filename, extension = basename.split('.', maxsplit=1)
# Make filename safe
filename = re.sub(r'[^A-Za-z0-9_\-]+', '', filename)
output_basename = '.'.join(filter(None, [filename, extension_str]))
output_path = os.path.join(output_directory, output_basename)
return output_path
def convert_to_png(image_path, output_path, gear_context):
try:
gear_context.log.info(f'Opening file: {image_path}')
s = openslide.OpenSlide(image_path)
except Exception as e:
gear_context.log.error(f'ERROR: Exception encountered while opening file {image_path}: {e}')
os.sys.exit(1)
try:
gear_context.log.info(f'Reading region {s.level_count - 1}')
# t = s.get_thumbnail((512, 512))
t = s.read_region((0, 0), s.level_count - 1, s.level_dimensions[s.level_count - 1])
except Exception as e:
gear_context.log.error(f'ERROR: Exception encountered while reading file region: {e}')
os.sys.exit(2)
try:
gear_context.log.info(f'Saving region to {output_path}')
t.save(output_path)
except Exception as e:
gear_context.log.error(f'ERROR: Exception encountered while saving file to {output_path}: {e}')
os.sys.exit(3)
return output_path
def main():
with flywheel.GearContext() as gear_context:
# Add manifest.json as the manifest_json attribute
setattr(gear_context, 'manifest_json', load_manifest_json())
# Initialize logging
gear_context.log = log_initializer(gear_context)
# Log the gear configuration
gear_context.log.critical('Starting OpenSlide to PNG Converter...')
gear_context.log.info('Gear configuration:')
gear_context.log_config()
input_filepath = gear_context.get_input_path('image')
output_filepath = get_output_path(input_filepath, gear_context.output_dir)
convert_to_png(input_filepath, output_filepath, gear_context)
if os.path.isfile(output_filepath):
gear_context.log.info('Job completed successfully!!')
os.sys.exit(0)
if __name__ == '__main__':
main()