Assorted minor code improvements

src/nidm/core/Constants.py

@@ -588,17 +588,13 @@ def __init__(self, namespaces=None):
 PROVONE_ATTRIBUTES = (
     PROVONE_ATTRIBUTE_QNAMES | PROV_ATTRIBUTE_QNAMES | PROV_ATTRIBUTE_LITERALS
 )
-PROVONE_RECORD_ATTRIBUTES = list((attr, str(attr)) for attr in PROVONE_ATTRIBUTES)
+PROVONE_RECORD_ATTRIBUTES = [(attr, str(attr)) for attr in PROVONE_ATTRIBUTES]
 
-PROV_RECORD_IDS_MAP = dict(
-    (PROV_N_MAP[rec_type_id], rec_type_id) for rec_type_id in PROV_N_MAP
-)
-PROVONE_ID_ATTRIBUTES_MAP = dict(
-    (prov_id, attribute) for (prov_id, attribute) in PROVONE_RECORD_ATTRIBUTES
-)
-PROVONE_ATTRIBUTES_ID_MAP = dict(
-    (attribute, prov_id) for (prov_id, attribute) in PROVONE_RECORD_ATTRIBUTES
-)
+PROV_RECORD_IDS_MAP = {value: rec_type_id for rec_type_id, value in PROV_N_MAP.items()}
+PROVONE_ID_ATTRIBUTES_MAP = dict(PROVONE_RECORD_ATTRIBUTES)
+PROVONE_ATTRIBUTES_ID_MAP = {
+    attribute: prov_id for (prov_id, attribute) in PROVONE_RECORD_ATTRIBUTES
+}
 
 
 # ADDED BY DBK to make searching NIDM-Experiment Terms easier...temporary, should be done in the OWL file #

src/nidm/core/provone.py

@@ -100,8 +100,6 @@ class ProvPlan(ProvEntity):
     ProvONE Plan element
     """
 
-    pass
-
 
 class Process(ProvEntity):
     """
@@ -905,7 +903,7 @@ def serialize(self, destination=None, format="json", **args):  # noqa: A002
             serializer.serialize(stream, **args)
         else:
             location = destination
-            scheme, netloc, path, params, _query, fragment = urlparse(location)
+            _, netloc, path, _, _, _ = urlparse(location)
             if netloc != "":
                 print(
                     "WARNING: not saving as location " + "is not a local file reference"

src/nidm/core/serializers/__init__.py

@@ -38,8 +38,6 @@ def deserialize(self, stream, **kwargs):
 class DoNotExist(Error):
     """Exception for the case a serializer is not available."""
 
-    pass
-
 
 class Registry:
     """Registry of serializers."""

src/nidm/core/serializers/provonerdf.py

@@ -309,10 +309,10 @@ def encode_container(self, bundle, container=None, identifier=None):
                                 rec_uri = rec_type.uri
                                 for attr_name, val in record.extra_attributes:
                                     if attr_name == PROV["type"]:
-                                        if (
-                                            PROV["Revision"] == val
-                                            or PROV["Quotation"] == val
-                                            or PROV["PrimarySource"] == val
+                                        if val in (
+                                            PROV["Revision"],
+                                            PROV["Quotation"],
+                                            PROV["PrimarySource"],
                                         ):
                                             qualifier = val._localpart
                                             rec_uri = val.uri
@@ -453,8 +453,8 @@ def decode_container(self, graph, bundle):
         PROV_CLS_MAP = {}
         formal_attributes = {}
         unique_sets = {}
-        for key, _ in PROV_BASE_CLS.items():
-            PROV_CLS_MAP[key.uri] = PROV_BASE_CLS[key]
+        for key, value in PROV_BASE_CLS.items():
+            PROV_CLS_MAP[key.uri] = value
         relation_mapper = {
             URIRef(PROV["alternateOf"].uri): "alternate",
             URIRef(PROV["actedOnBehalfOf"].uri): "delegation",
@@ -575,9 +575,7 @@ def decode_container(self, graph, bundle):
                 obj1 = self.decode_rdf_representation(obj, graph)
                 if obj is not None and obj1 is None:
                     raise ValueError(("Error transforming", obj))
-                pred_new = pred
-                if pred in predicate_mapper:
-                    pred_new = predicate_mapper[pred]
+                pred_new = predicate_mapper.get(pred, pred)
                 if ids[id_] == PROV_COMMUNICATION and "activity" in str(pred_new):
                     pred_new = PROV_ATTR_INFORMANT
                 if ids[id_] == PROV_DELEGATION and "agent" in str(pred_new):
@@ -605,12 +603,10 @@ def decode_container(self, graph, bundle):
             if local_key in ids:
                 if "qualified" in pred:
                     formal_attributes[local_key][
-                        list(formal_attributes[local_key].keys())[0]
+                        next(iter(formal_attributes[local_key]))
                     ] = id_
-        for id_ in ids:
-            attrs = None
-            if id_ in other_attributes:
-                attrs = other_attributes[id_]
+        for id_, idvalue in ids.items():
+            attrs = other_attributes.get(id_)
             items_to_walk = []
             for qname, values in unique_sets[id_].items():
                 if values and len(values) > 1:
@@ -619,9 +615,9 @@ def decode_container(self, graph, bundle):
                 for subset in list(walk(items_to_walk)):
                     for key, value in subset.items():
                         formal_attributes[id_][key] = value
-                    bundle.new_record(ids[id_], id_, formal_attributes[id_], attrs)
+                    bundle.new_record(idvalue, id_, formal_attributes[id_], attrs)
             else:
-                bundle.new_record(ids[id_], id_, formal_attributes[id_], attrs)
+                bundle.new_record(idvalue, id_, formal_attributes[id_], attrs)
             ids[id_] = None
             if attrs is not None:
                 other_attributes[id_] = []

src/nidm/experiment/Acquisition.py

@@ -83,10 +83,7 @@ def acquisition_object_exists(self, uuid):
         :param uuid: full uuid of acquisition
         :return: True if exists, False otherwise
         """
-        if uuid in self._acquisition_objects:
-            return True
-        else:
-            return False
+        return bool(uuid in self._acquisition_objects)
 
     def __str__(self):
         return "NIDM-Experiment Acquisition Class"

src/nidm/experiment/Core.py

@@ -16,7 +16,7 @@ def getUUID():
     uid = str(uuid.uuid1())
     # added to address some weird bug in rdflib where if the uuid starts with a number, everything up until the first
     # alapha character becomes a prefix...
-    if not (re.match("^[a-fA-F]+.*", uid)):
+    if not re.match("^[a-fA-F]+.*", uid):
         # if first digit is not a character than replace it with a randomly selected hex character (a-f).
         uid_temp = uid
         randint = random.randint(0, 5)
@@ -116,11 +116,7 @@ def checkNamespacePrefix(self, prefix):
         :return: True if prefix exists, False if not
         """
         # check if prefix already exists
-        if prefix in self.graph._namespaces.keys():
-            # prefix already exists
-            return True
-        else:
-            return False
+        return bool(prefix in self.graph._namespaces)
 
     def safe_string(self, string):
         return (
@@ -538,53 +534,42 @@ def save_DotGraph(self, filename, format=None):  # noqa: A002
             project_uuid = str(row[0])
             # for each Project uuid search dot structure for Project uuid
             project_node = None
-            for key, _ in dot.obj_dict["nodes"].items():
+            for key, value in dot.obj_dict["nodes"].items():
                 # get node number in DOT graph for Project
-                if "URL" in dot.obj_dict["nodes"][key][0]["attributes"]:
-                    if project_uuid in str(
-                        dot.obj_dict["nodes"][key][0]["attributes"]["URL"]
-                    ):
-                        project_node = key
-                        break
+                if project_uuid in str(value[0]["attributes"].get("URL", "")):
+                    project_node = key
+                    break
 
         # for each Session in Project class self.sessions list, find node numbers in DOT graph
 
         for session in self.sessions:
             print(session)
-            for key, _ in dot.obj_dict["nodes"].items():
+            for key, value in dot.obj_dict["nodes"].items():
                 # get node number in DOT graph for Project
-                if "URL" in dot.obj_dict["nodes"][key][0]["attributes"]:
-                    if session.identifier.uri in str(
-                        dot.obj_dict["nodes"][key][0]["attributes"]["URL"]
-                    ):
-                        session_node = key
-                        # print(f"session node = {key}")
-
-                        # add to DOT structure edge between project_node and session_node
-                        dot.add_edge(Edge(session_node, project_node, **style))
-
-                        # for each Acquisition in Session class ._acquisitions list, find node numbers in DOT graph
-                        for acquisition in session.get_acquisitions():
-                            # search through the nodes again to figure out node number for acquisition
-                            for key, _ in dot.obj_dict["nodes"].items():
-                                # get node number in DOT graph for Project
-                                if "URL" in dot.obj_dict["nodes"][key][0]["attributes"]:
-                                    if acquisition.identifier.uri in str(
-                                        dot.obj_dict["nodes"][key][0]["attributes"][
-                                            "URL"
-                                        ]
-                                    ):
-                                        acquisition_node = key
-                                        # print(f"acquisition node = {key}")
-
-                                        dot.add_edge(
-                                            Edge(
-                                                acquisition_node, session_node, **style
-                                            )
-                                        )
+                if session.identifier.uri in str(value[0]["attributes"].get("URL", "")):
+                    session_node = key
+                    # print(f"session node = {key}")
+
+                    # add to DOT structure edge between project_node and session_node
+                    dot.add_edge(Edge(session_node, project_node, **style))
+
+                    # for each Acquisition in Session class ._acquisitions list, find node numbers in DOT graph
+                    for acquisition in session.get_acquisitions():
+                        # search through the nodes again to figure out node number for acquisition
+                        for key, value in dot.obj_dict["nodes"].items():
+                            # get node number in DOT graph for Project
+                            if acquisition.identifier.uri in str(
+                                value[0]["attributes"].get("URL", "")
+                            ):
+                                acquisition_node = key
+                                # print(f"acquisition node = {key}")
+
+                                dot.add_edge(
+                                    Edge(acquisition_node, session_node, **style)
+                                )
 
         # add some logic to find nodes with dct:hasPart relation and add those edges to graph...prov_to_dot ignores these
-        if not (format == "None"):
+        if format != "None":
             dot.write(filename, format=format)
         else:
             dot.write(filename, format="pdf")

src/nidm/experiment/Derivative.py

@@ -72,10 +72,7 @@ def derivative_object_exists(self, uuid):
         :param uuid: full uuid of derivative object
         :return: True if exists, False otherwise
         """
-        if uuid in self._derivative_objects:
-            return True
-        else:
-            return False
+        return bool(uuid in self._derivative_objects)
 
     def __str__(self):
         return "NIDM-Experiment Derivative Activity Class"

src/nidm/experiment/Navigate.py

@@ -139,7 +139,7 @@ def getNamespaceLookup(nidm_file_tuples):
         for f in nidm_file_tuples:
             rdf_graph = OpenGraph(f)
             for prefix, uri in rdf_graph.namespace_manager.namespaces():
-                if not str(uri) in names:
+                if str(uri) not in names:
                     names[str(uri)] = prefix
         return names
 
@@ -504,18 +504,14 @@ def GetDataelementDetails(nidm_files_tuple, dataelement):
             dti = getDataTypeInfo(rdf_graph, de_uri)
 
             # check if this is the correct one
-            if not (
-                dataelement
-                in [
-                    str(dti["label"]),
-                    str(dti["dataElement"]),
-                    str(dti["dataElementURI"]),
-                ]
-            ):
+            if dataelement not in [
+                str(dti["label"]),
+                str(dti["dataElement"]),
+                str(dti["dataElementURI"]),
+            ]:
                 continue
 
-            for key in dti.keys():
-                result[key] = dti[key]
+            result.update(dti)
             result["inProjects"] = set()
 
             # figure out what project the dataelement was used in
@@ -542,24 +538,20 @@ def GetDataelementDetails(nidm_files_tuple, dataelement):
 
             return result  # found it, we are done
 
-    if result == {}:  # didn't find it yet, check the CDEs
+    if not result:  # didn't find it yet, check the CDEs
         cde_graph = nidm.experiment.CDE.getCDEs()
         for de_uri in cde_graph.subjects(predicate=isa):
             dti = getDataTypeInfo(cde_graph, de_uri)
 
             # check if this is the correct one
-            if not (
-                dataelement
-                in [
-                    str(dti["label"]),
-                    str(dti["dataElement"]),
-                    str(dti["dataElementURI"]),
-                ]
-            ):
+            if dataelement not in [
+                str(dti["label"]),
+                str(dti["dataElement"]),
+                str(dti["dataElementURI"]),
+            ]:
                 continue
 
-            for key in dti.keys():
-                result[key] = dti[key]
+            result.update(dti)
             result["inProjects"] = set()
             result["inProjects"].add("Common Data Element")
 

src/nidm/experiment/Query.py

@@ -36,7 +36,7 @@ def sparql_query_nidm(nidm_file_list, query, output_file=None, return_graph=Fals
     :return: dataframe | graph depending on return_graph parameter
     """
 
-    if "BLAZEGRAPH_URL" in environ.keys():
+    if "BLAZEGRAPH_URL" in environ:
         try:
             # first make sure all files are loaded into blazegraph
             for nidm_file in nidm_file_list:
@@ -902,14 +902,10 @@ def CheckSubjectMatchesFilter(
             instrument_details = GetParticipantInstrumentData(
                 nidm_file_list, project_uuid, subject_uuid
             )
-            for instrument_uuid in instrument_details:
-                for instrument_term in instrument_details[instrument_uuid]:
+            for terms in instrument_details.values():
+                for instrument_term, v in terms.items():
                     if instrument_term in synonyms:
-                        found_match = filterCompare(
-                            instrument_details[instrument_uuid][instrument_term],
-                            op,
-                            value,
-                        )
+                        found_match = filterCompare(v, op, value)
                     if found_match:
                         break
 
@@ -921,8 +917,8 @@ def CheckSubjectMatchesFilter(
             derivatives_details = GetDerivativesDataForSubject(
                 nidm_file_list, project_uuid, subject_uuid
             )
-            for key in derivatives_details:
-                derivatives = derivatives_details[key]["values"]
+            for details in derivatives_details.values():
+                derivatives = details["values"]
                 for (
                     vkey
                 ) in (
@@ -1345,7 +1341,7 @@ def OpenGraph(file):
         return file
 
     # If we have a Blazegraph instance, load the data then do the rest
-    if "BLAZEGRAPH_URL" in environ.keys():
+    if "BLAZEGRAPH_URL" in environ:
         try:
             with open(file, encoding="utf-8") as f:
                 data = f.read()