4911681207

Build Type

github

Committed by GitHub

Commit Message

Merge 9186e10d1 into e48989cce

Pull Request Pull Request #231: Liu 355

Run Details

180 of 229 new or added lines in 17 files covered. (78.6%)

26 existing lines in 5 files now uncovered.

15345 of 19059 relevant lines covered (80.51%)

1.65 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.86

/daliuge-common/dlg/common/reproducibility/reproducibility_fields.py

#
#    ICRAR - International Centre for Radio Astronomy Research
#    (c) UWA - The University of Western Australia, 2017
#    Copyright by UWA (in the framework of the ICRAR)
#    All rights reserved
#
#    This library is free software; you can redistribute it and/or
#    modify it under the terms of the GNU Lesser General Public
#    License as published by the Free Software Foundation; either
#    version 2.1 of the License, or (at your option) any later version.
#
#    This library is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
#    Lesser General Public License for more details.
#
#    You should have received a copy of the GNU Lesser General Public
#    License along with this library; if not, write to the Free Software
#    Foundation, Inc., 59 Temple Place, Suite 330, Boston,
#    MA 02111-1307  USA
#
"""
This module defines the fields each drop takes for each reproducibility standard defined.
Consider this module partially documentation, partially code.
Data generated by instanced drops at runtime are defined with that drop's implementation.
"""

from enum import Enum

from dlg.common.reproducibility.constants import ReproducibilityFlags
import logging

logger = logging.getLogger("__name__")


class FieldOps(Enum):
    """
    Defines the operations possible on drop data for provenance collection.
    """

    STORE = 0
    COUNT = 1
    REMOVE_FIRST = 2  # Removes the first char of an assumed string
    DO_NOT_STORE = None


def extract_fields(drop: dict, fields: dict):
    """
    Attempts to extract fields with the names in fields from the drop description.
    If not found, the key will not be present in the returned dictionary.
    """
    data = {}
    for key, operation in fields.items():
        if drop.get(key) is not None:
            if operation == FieldOps.STORE:
                data[key] = drop.get(key)
            elif operation == FieldOps.COUNT:
                data[key] = len(drop.get(key))
            elif operation == FieldOps.REMOVE_FIRST:
                data[key] = drop.get(key)[1:]
    return data


def lgt_block_fields(rmode: ReproducibilityFlags):
    """
    Collects dict of fields and operations for all drop types at the lgt layer for
    the supplied reproducibility standard.
    :param rmode: The reproducibility level in question
    :return: Dictionary of <str, FieldOp> pairs
    """
    if rmode == ReproducibilityFlags.NOTHING:
        return {}
    data = {
        "categoryType": FieldOps.STORE,
        "category": FieldOps.STORE,
        "inputPorts": FieldOps.COUNT,
        "outputPorts": FieldOps.COUNT,
        "inputLocalPorts": FieldOps.COUNT,
        "outputLocalPorts": FieldOps.COUNT,  # MKN Nodes
    }
    if rmode == ReproducibilityFlags.REPRODUCE:
        del data["inputPorts"]
        del data["outputPorts"]
        del data["inputLocalPorts"]
        del data["outputLocalPorts"]
    return data


def lg_block_fields(
    category_type: str, rmode: ReproducibilityFlags, custom_fields=None
):
    """
    Collects dict of fields and operations for all drop types at the lg layer for
    the supplied reproducibility standard.
    :param category: The broad type of drop
    :param rmode: The reproducibility level in question
    :param custom_fields: Additional application args (used in custom components)
    :return: Dictionary of <str, FieldOp> pairs
    """
    data = {}
    if rmode in (
        ReproducibilityFlags.NOTHING,
        ReproducibilityFlags.RERUN,
        ReproducibilityFlags.REPRODUCE,
        ReproducibilityFlags.REPLICATE_SCI,
    ):
        return data
    # Drop category considerations - Just try to get everything we can, will be filtered later
    data["execution_time"] = FieldOps.STORE
    data["num_cpus"] = FieldOps.STORE
    data["inputApplicationName"] = FieldOps.STORE
    data["inputApplicationType"] = FieldOps.STORE
    data["data_volume"] = FieldOps.STORE

    # Drop type considerations
    if category_type == "Start":
        pass
    elif category_type == "End":
        pass
    elif category_type == "Memory":
        pass
    elif category_type == "ShMem":
        pass
    elif category_type == "File":
        data["check_filepath_exists"] = FieldOps.STORE
        if rmode in (
            ReproducibilityFlags.RECOMPUTE,
            ReproducibilityFlags.REPLICATE_COMP,
        ):
            data["filepath"] = FieldOps.STORE
            data["dirname"] = FieldOps.STORE
    elif category_type == "null":
        pass
    elif category_type == "json":
        pass
    elif category_type == "NGAS":
        pass
    elif category_type == "S3":
        pass
    elif category_type == "Plasma":
        data["plasma_path"] = FieldOps.STORE
        data["object_id"] = FieldOps.STORE
    elif category_type == "PlasmaFlight":
        data["plasma_path"] = FieldOps.STORE
        data["object_id"] = FieldOps.STORE
        data["flight_path"] = FieldOps.STORE
    elif category_type == "ParameterSet":
        pass
    elif category_type == "EnvironmentVariables":
        pass
    elif category_type == "MKN":
        data["m"] = FieldOps.STORE
        data["k"] = FieldOps.STORE
        data["n"] = FieldOps.STORE
    elif category_type == "Scatter":
        data["num_of_copies"] = FieldOps.STORE
        data["scatter_axis"] = FieldOps.STORE
    elif category_type == "Gather":
        data["num_of_inputs"] = FieldOps.STORE
        data["gather_axis"] = FieldOps.STORE
    elif category_type == "Loop":
        data["num_of_iter"] = FieldOps.STORE
    elif category_type == "GroupBy":
        data["group_key"] = FieldOps.STORE
        data["group_axis"] = FieldOps.STORE
    elif category_type == "Variables":
        pass
    elif category_type == "Branch":
        data["dropclass"] = FieldOps.STORE
    elif category_type == "PythonApp":
        data["dropclass"] = FieldOps.STORE
    elif category_type == "Component":
        data["dropclass"] = FieldOps.STORE
    elif category_type == "BashShellApp":
        data["Arg01"] = FieldOps.STORE
    elif category_type == "Mpi":
        data["num_of_procs"] = FieldOps.STORE
    elif category_type == "Docker":
        data["image"] = FieldOps.STORE
        data["command"] = FieldOps.STORE
        data["user"] = FieldOps.STORE
        data["ensureUserAndSwitch"] = FieldOps.STORE
        data["removeContainer"] = FieldOps.STORE
        data["additionalBindings"] = FieldOps.STORE
    elif category_type == "DynlibApp":
        data["libpath"] = FieldOps.STORE
    elif category_type == "DynlibProcApp":
        data["libpath"] = FieldOps.STORE
    if custom_fields is not None and rmode in (
        ReproducibilityFlags.RECOMPUTE,
        ReproducibilityFlags.REPLICATE_COMP,
    ):
        for name in custom_fields:
            data[name] = FieldOps.STORE

    return data


def pgt_unroll_block_fields(category_type, rmode: ReproducibilityFlags):
    """
    Collects dict of fields and operations for all drop types at the pgt unroll layer for
    the supplied reproducibility standard.
    :param category_type: The specific type of drop
    :param rmode: The reproducibility level in question
    :return: Dictionary of <str, FieldOp> pairs
    """
    data = {}
    if rmode == ReproducibilityFlags.NOTHING:
        return data
    if rmode != ReproducibilityFlags.NOTHING:
        data["categoryType"] = FieldOps.STORE
    if rmode != ReproducibilityFlags.REPRODUCE:
        if category_type.lower() != "data":
            data["dt"] = FieldOps.STORE
    if category_type.lower() == "data":
        data["storage"] = FieldOps.STORE
    if rmode in (
        ReproducibilityFlags.RECOMPUTE,
        ReproducibilityFlags.REPLICATE_COMP,
    ):
        data["rank"] = FieldOps.STORE

    return data


def pgt_partition_block_fields(rmode: ReproducibilityFlags):
    """
    Collects dict of fields and operations for all drop types at the pgt partition layer for
    the supplied reproducibility standard.
    :param rmode: The reproducibility level in question
    :return: Dictionary of <str, FieldOp> pairs
    """
    data = {}
    if rmode in (
        ReproducibilityFlags.RECOMPUTE,
        ReproducibilityFlags.REPLICATE_COMP,
    ):
        data["node"] = FieldOps.REMOVE_FIRST
        data["island"] = FieldOps.REMOVE_FIRST
    return data


def pg_block_fields(rmode: ReproducibilityFlags):
    """
    Collects dict of fields and operations for all drop types at the pg layer for
    the supplied reproducibility standard.
    :param rmode: The reproducibility level in question
    :return: Dictionary of <str, FieldOp> pairs
    """
    # These two happen to have the same data.
    data = {}
    if rmode in (
        ReproducibilityFlags.RECOMPUTE,
        ReproducibilityFlags.REPLICATE_COMP,
    ):
        data["node"] = FieldOps.STORE
        data["island"] = FieldOps.STORE
    return data

1	#
2	# ICRAR - International Centre for Radio Astronomy Research
3	# (c) UWA - The University of Western Australia, 2017
4	# Copyright by UWA (in the framework of the ICRAR)
5	# All rights reserved
6	#
7	# This library is free software; you can redistribute it and/or
8	# modify it under the terms of the GNU Lesser General Public
9	# License as published by the Free Software Foundation; either
10	# version 2.1 of the License, or (at your option) any later version.
11	#
12	# This library is distributed in the hope that it will be useful,
13	# but WITHOUT ANY WARRANTY; without even the implied warranty of
14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15	# Lesser General Public License for more details.
16	#
17	# You should have received a copy of the GNU Lesser General Public
18	# License along with this library; if not, write to the Free Software
19	# Foundation, Inc., 59 Temple Place, Suite 330, Boston,
20	# MA 02111-1307 USA
21	#
22	"""	3✔
23	This module defines the fields each drop takes for each reproducibility standard defined.
24	Consider this module partially documentation, partially code.
25	Data generated by instanced drops at runtime are defined with that drop's implementation.
26	"""
27
28	from enum import Enum	3✔
29
30	from dlg.common.reproducibility.constants import ReproducibilityFlags	3✔
31	import logging	3✔
32
33	logger = logging.getLogger("__name__")	3✔
34
35
36	class FieldOps(Enum):	3✔
37	"""
38	Defines the operations possible on drop data for provenance collection.
39	"""
40
41	STORE = 0	3✔
42	COUNT = 1	3✔
43	REMOVE_FIRST = 2 # Removes the first char of an assumed string	3✔
44	DO_NOT_STORE = None	3✔
45
46
47	def extract_fields(drop: dict, fields: dict):	3✔
48	"""
49	Attempts to extract fields with the names in fields from the drop description.
50	If not found, the key will not be present in the returned dictionary.
51	"""
52	data = {}	3✔
53	for key, operation in fields.items():	3✔
54	if drop.get(key) is not None:	3✔
55	if operation == FieldOps.STORE:	3✔
56	data[key] = drop.get(key)	3✔
57	elif operation == FieldOps.COUNT:	3✔
58	data[key] = len(drop.get(key))	2✔
59	elif operation == FieldOps.REMOVE_FIRST:	2✔
60	data[key] = drop.get(key)[1:]	2✔
61	return data	3✔
62
63
64	def lgt_block_fields(rmode: ReproducibilityFlags):	3✔
65	"""
66	Collects dict of fields and operations for all drop types at the lgt layer for
67	the supplied reproducibility standard.
68	:param rmode: The reproducibility level in question
69	:return: Dictionary of <str, FieldOp> pairs
70	"""
71	if rmode == ReproducibilityFlags.NOTHING:	3✔
72	return {}	3✔
73	data = {	3✔
74	"categoryType": FieldOps.STORE,
75	"category": FieldOps.STORE,
76	"inputPorts": FieldOps.COUNT,
77	"outputPorts": FieldOps.COUNT,
78	"inputLocalPorts": FieldOps.COUNT,
79	"outputLocalPorts": FieldOps.COUNT, # MKN Nodes
80	}
81	if rmode == ReproducibilityFlags.REPRODUCE:	3✔
82	del data["inputPorts"]	3✔
83	del data["outputPorts"]	3✔
84	del data["inputLocalPorts"]	3✔
85	del data["outputLocalPorts"]	3✔
86	return data	3✔
87
88
89	def lg_block_fields(	3✔
90	category_type: str, rmode: ReproducibilityFlags, custom_fields=None
91	):
92	"""
93	Collects dict of fields and operations for all drop types at the lg layer for
94	the supplied reproducibility standard.
95	:param category: The broad type of drop
96	:param rmode: The reproducibility level in question
97	:param custom_fields: Additional application args (used in custom components)
98	:return: Dictionary of <str, FieldOp> pairs
99	"""
100	data = {}	3✔
101	if rmode in (	3✔
102	ReproducibilityFlags.NOTHING,
103	ReproducibilityFlags.RERUN,
104	ReproducibilityFlags.REPRODUCE,
105	ReproducibilityFlags.REPLICATE_SCI,
106	):
107	return data	3✔
108	# Drop category considerations - Just try to get everything we can, will be filtered later
109	data["execution_time"] = FieldOps.STORE	3✔
110	data["num_cpus"] = FieldOps.STORE	3✔
111	data["inputApplicationName"] = FieldOps.STORE	3✔
112	data["inputApplicationType"] = FieldOps.STORE	3✔
113	data["data_volume"] = FieldOps.STORE	3✔
114
115	# Drop type considerations
116	if category_type == "Start":	3✔
117	pass	×
118	elif category_type == "End":	3✔
119	pass	×
120	elif category_type == "Memory":	3✔
121	pass	2✔
122	elif category_type == "ShMem":	3✔
123	pass	×
124	elif category_type == "File":	3✔
125	data["check_filepath_exists"] = FieldOps.STORE	3✔
126	if rmode in (	3✔
127	ReproducibilityFlags.RECOMPUTE,
128	ReproducibilityFlags.REPLICATE_COMP,
129	):
130	data["filepath"] = FieldOps.STORE	3✔
131	data["dirname"] = FieldOps.STORE	3✔
132	elif category_type == "null":	3✔
133	pass	×
134	elif category_type == "json":	3✔
135	pass	×
136	elif category_type == "NGAS":	3✔
137	pass	2✔
138	elif category_type == "S3":	3✔
139	pass	×
140	elif category_type == "Plasma":	3✔
141	data["plasma_path"] = FieldOps.STORE	×
142	data["object_id"] = FieldOps.STORE	×
143	elif category_type == "PlasmaFlight":	3✔
144	data["plasma_path"] = FieldOps.STORE	×
145	data["object_id"] = FieldOps.STORE	×
146	data["flight_path"] = FieldOps.STORE	×
147	elif category_type == "ParameterSet":	3✔
148	pass	×
149	elif category_type == "EnvironmentVariables":	3✔
150	pass	×
151	elif category_type == "MKN":	3✔
152	data["m"] = FieldOps.STORE	×
153	data["k"] = FieldOps.STORE	×
154	data["n"] = FieldOps.STORE	×
155	elif category_type == "Scatter":	3✔
156	data["num_of_copies"] = FieldOps.STORE	2✔
157	data["scatter_axis"] = FieldOps.STORE	2✔
158	elif category_type == "Gather":	3✔
159	data["num_of_inputs"] = FieldOps.STORE	2✔
160	data["gather_axis"] = FieldOps.STORE	2✔
161	elif category_type == "Loop":	3✔
162	data["num_of_iter"] = FieldOps.STORE	2✔
163	elif category_type == "GroupBy":	3✔
164	data["group_key"] = FieldOps.STORE	×
165	data["group_axis"] = FieldOps.STORE	×
166	elif category_type == "Variables":	3✔
167	pass	×
168	elif category_type == "Branch":	3✔
NEW 169	data["dropclass"] = FieldOps.STORE	×
170	elif category_type == "PythonApp":	3✔
NEW 171	data["dropclass"] = FieldOps.STORE	×
172	elif category_type == "Component":	3✔
173	data["dropclass"] = FieldOps.STORE	2✔
174	elif category_type == "BashShellApp":	3✔
175	data["Arg01"] = FieldOps.STORE	3✔
176	elif category_type == "Mpi":	2✔
177	data["num_of_procs"] = FieldOps.STORE	2✔
178	elif category_type == "Docker":	2✔
179	data["image"] = FieldOps.STORE	2✔
180	data["command"] = FieldOps.STORE	2✔
181	data["user"] = FieldOps.STORE	2✔
182	data["ensureUserAndSwitch"] = FieldOps.STORE	2✔
183	data["removeContainer"] = FieldOps.STORE	2✔
184	data["additionalBindings"] = FieldOps.STORE	2✔
185	elif category_type == "DynlibApp":	2✔
186	data["libpath"] = FieldOps.STORE	2✔
187	elif category_type == "DynlibProcApp":	2✔
188	data["libpath"] = FieldOps.STORE	×
189	if custom_fields is not None and rmode in (	3✔
190	ReproducibilityFlags.RECOMPUTE,
191	ReproducibilityFlags.REPLICATE_COMP,
192	):
193	for name in custom_fields:	3✔
194	data[name] = FieldOps.STORE	×
195
196	return data	3✔
197
198
199	def pgt_unroll_block_fields(category_type, rmode: ReproducibilityFlags):	3✔
200	"""
201	Collects dict of fields and operations for all drop types at the pgt unroll layer for
202	the supplied reproducibility standard.
203	:param category_type: The specific type of drop
204	:param rmode: The reproducibility level in question
205	:return: Dictionary of <str, FieldOp> pairs
206	"""
207	data = {}	2✔
208	if rmode == ReproducibilityFlags.NOTHING:	2✔
209	return data	×
210	if rmode != ReproducibilityFlags.NOTHING:	2✔
211	data["categoryType"] = FieldOps.STORE	2✔
212	if rmode != ReproducibilityFlags.REPRODUCE:	2✔
213	if category_type.lower() != "data":	2✔
214	data["dt"] = FieldOps.STORE	2✔
215	if category_type.lower() == "data":	2✔
216	data["storage"] = FieldOps.STORE	2✔
217	if rmode in (	2✔
218	ReproducibilityFlags.RECOMPUTE,
219	ReproducibilityFlags.REPLICATE_COMP,
220	):
221	data["rank"] = FieldOps.STORE	2✔
222
223	return data	2✔
224
225
226	def pgt_partition_block_fields(rmode: ReproducibilityFlags):	3✔
227	"""
228	Collects dict of fields and operations for all drop types at the pgt partition layer for
229	the supplied reproducibility standard.
230	:param rmode: The reproducibility level in question
231	:return: Dictionary of <str, FieldOp> pairs
232	"""
233	data = {}	2✔
234	if rmode in (	2✔
235	ReproducibilityFlags.RECOMPUTE,
236	ReproducibilityFlags.REPLICATE_COMP,
237	):
238	data["node"] = FieldOps.REMOVE_FIRST	2✔
239	data["island"] = FieldOps.REMOVE_FIRST	2✔
240	return data	2✔
241
242
243	def pg_block_fields(rmode: ReproducibilityFlags):	3✔
244	"""
245	Collects dict of fields and operations for all drop types at the pg layer for
246	the supplied reproducibility standard.
247	:param rmode: The reproducibility level in question
248	:return: Dictionary of <str, FieldOp> pairs
249	"""
250	# These two happen to have the same data.
251	data = {}	3✔
252	if rmode in (	3✔
253	ReproducibilityFlags.RECOMPUTE,
254	ReproducibilityFlags.REPLICATE_COMP,
255	):
256	data["node"] = FieldOps.STORE	2✔
257	data["island"] = FieldOps.STORE	2✔
258	return data	3✔

ICRAR / daliuge / 4911681207

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous