-
Notifications
You must be signed in to change notification settings - Fork 13
/
labeler.py
116 lines (104 loc) · 4.04 KB
/
labeler.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
from rodan.jobs.base import RodanTask
from rodan.models import Input, ResourceLabel
from django.conf import settings as rodan_settings
import re
class Labeler(RodanTask):
name = "Labeler"
author = "Juliette Regimbal"
description = "Add a label to resources"
settings = {
"title": "Parameters",
"type": "object",
"job_queue": "celery",
"properties": {
"Label": {
"type": "string",
"default": "",
"maxLength": 36 # This is specified in the ResourceLabel model
}
}
}
enabled = True
category = "Utility"
interactive = False
input_port_types = [
{
"name": "Resource",
"minimum": 1,
"maximum": 100,
"resource_types": lambda mime: re.match(r"^[-\w]+/[-\w+]+$", mime)
}
]
output_port_types = []
def _inputs(self, runjob, with_urls=False):
"""
Return a dictionary of list of input file path and input resource type.
If with_urls=True, it also includes the resource url and thumbnail urls.
"""
self.runjob = runjob
def _extract_resource(resource, resource_type_mimetype=None):
r = {
# convert 'unicode' object to 'str' object for consistency
"resource_path": str(resource.resource_file.path),
"resource_type": str(
resource_type_mimetype or resource.resource_type.mimetype
),
"resource": resource,
}
if with_urls:
r["resource_url"] = str(resource.resource_url)
r["diva_object_data"] = str(resource.diva_json_url)
r["diva_iip_server"] = getattr(rodan_settings, "IIPSRV_URL")
r["diva_image_dir"] = str(resource.diva_image_dir)
return r
input_objs = (
Input.objects.filter(run_job=runjob)
.select_related("resource", "resource__resource_type", "resource_list")
.prefetch_related("resource_list__resources")
)
inputs = {}
for input in input_objs:
ipt_name = str(input.input_port_type_name)
if ipt_name not in inputs:
inputs[ipt_name] = []
if input.resource is not None: # If resource
inputs[ipt_name].append(_extract_resource(input.resource))
elif input.resource_list is not None: # If resource_list
inputs[ipt_name].append(
map(
lambda x: _extract_resource(
x, input.resource_list.get_resource_type().mimetype
),
input.resource_list.resources.all(),
)
)
else:
raise RuntimeError(
(
"Cannot find any resource or resource list on Input" " {0}"
).format(input.uuid)
)
return inputs
def run_my_task(self, inputs, settings, outputs):
label_text = settings["Label"] if settings["Label"] != "" \
else str(self.runjob.workflow_run.uuid)
label, _ = ResourceLabel.objects.get_or_create(name=label_text)
for input in inputs["Resource"]:
input["resource"].labels.add(label)
def test_my_task(self, testcase):
from model_mommy import mommy
from rodan.models import Resource
rc = mommy.make("Resource")
inputs = {
"Resource": [
{"resource":rc}
]
}
outputs = {}
settings = {"Label":"testcase1"}
self.run_my_task(inputs=inputs, outputs=outputs, settings=settings)
settings = {"Label":"testcase2"}
self.run_my_task(inputs=inputs, outputs=outputs, settings=settings)
testcase.assertEqual(2, len(rc.labels.all()))
testcase.assertEqual("testcase1", rc.labels.all()[0].name)
testcase.assertEqual("testcase2", rc.labels.all()[1].name)