Add new command: annotate

sssom/cli.py

@@ -25,6 +25,7 @@
 
 from sssom.constants import (
     DEFAULT_VALIDATION_TYPES,
+    MAPPING_SET_SLOTS,
     MAPPING_SLOTS,
     PREFIX_MAP_MODES,
     SchemaValidationType,
@@ -34,6 +35,7 @@
 from . import __version__
 from .cliques import split_into_cliques, summarize_cliques
 from .io import (
+    annotate_file,
     convert_file,
     filter_file,
     parse_file,
@@ -644,5 +646,34 @@ def filter(input: str, output: TextIO, **kwargs):
     filter_file(input=input, output=output, **kwargs)
 
 
+@main.command()
+@input_argument
+@output_option
+# TODO Revist the option below.
+# If a multivalued slot needs to be partially preserved,
+# the users will need to type the ones they need and
+# set --replace-multivalued to True.
+@click.option(
+    "--replace-multivalued",
+    default=False,
+    type=bool,
+    help="Multivalued slots should be replaced or not. [default: False]",
+)
+@dynamically_generate_sssom_options(MAPPING_SET_SLOTS)
+def annotate(input: str, output: TextIO, replace_multivalued: bool, **kwargs):
+    """Annotate metadata of a mapping set.
+
+    :param input: Input path of the SSSOM tsv file.
+    :param output: Output location.
+    :param replace_multivalued: Multivalued slots should be
+        replaced or not, defaults to False
+    :param **kwargs: Options provided by user
+        which are added to the metadata (e.g.: --mapping_set_id http://example.org/abcd)
+    """
+    annotate_file(
+        input=input, output=output, replace_multivalued=replace_multivalued, **kwargs
+    )
+
+
 if __name__ == "__main__":
     main()

sssom/io.py

@@ -28,6 +28,8 @@
 from .typehints import Metadata
 from .util import (
     MappingSetDataFrame,
+    are_params_slots,
+    augment_metadata,
     is_curie,
     is_iri,
     raise_for_bad_path,
@@ -361,3 +363,24 @@ def filter_file(input: str, output: TextIO, **kwargs) -> MappingSetDataFrame:
         if multiple_params and idx != len(params):
             query += " AND ("
     return run_sql_query(query=query, inputs=[input], output=output)
+
+
+def annotate_file(
+    input: str, output: TextIO, replace_multivalued: bool = False, **kwargs
+) -> MappingSetDataFrame:
+    """Annotate a file i.e. add custom metadata to the mapping set.
+
+    :param input: SSSOM tsv file to be queried over.
+    :param output: Output location.
+    :param replace_multivalued: Multivalued slots should be
+        replaced or not, defaults to False
+    :param **kwargs: Options provided by user
+        which are added to the metadata (e.g.: --mapping_set_id http://example.org/abcd)
+    :return: Annotated MappingSetDataFrame object.
+    """
+    params = {k: v for k, v in kwargs.items() if v}
+    are_params_slots(params)
+    input_msdf = parse_sssom_table(input)
+    msdf = augment_metadata(input_msdf, params, replace_multivalued)
+    write_table(msdf, output)
+    return msdf

sssom/util.py

@@ -43,6 +43,7 @@
     ENTITY_REFERENCE_SLOTS,
     MAPPING_JUSTIFICATION,
     MAPPING_SET_ID,
+    MAPPING_SET_SLOTS,
     MAPPING_SET_SOURCE,
     MULTIVALUED_SLOTS,
     OBJECT_CATEGORY,
@@ -1250,3 +1251,61 @@ def get_all_prefixes(msdf: MappingSetDataFrame) -> list:
         prefix_list = list(set(prefix_list))
 
     return prefix_list
+
+
+def augment_metadata(
+    msdf: MappingSetDataFrame, meta: dict, replace_multivalued: bool = False
+) -> MappingSetDataFrame:
+    """Augment metadata with parameters passed.
+
+    :param msdf: MappingSetDataFrame (MSDF) object.
+    :param meta: Dictionary that needs to be added/updated to the metadata of the MSDF.
+    :param replace_multivalued: Multivalued slots should be
+        replaced or not, defaults to False.
+    :raises ValueError: If type of slot is neither str nor list.
+    :return: MSDF with updated metadata.
+    """
+    are_params_slots(meta)
+
+    if msdf.metadata:
+        for k, v in meta.items():
+            # If slot is multivalued, add to list.
+            if k in MULTIVALUED_SLOTS and not replace_multivalued:
+                tmp_value: list = []
+                if isinstance(msdf.metadata[k], str):
+                    tmp_value = [msdf.metadata[k]]
+                elif isinstance(msdf.metadata[k], list):
+                    tmp_value = msdf.metadata[k]
+                else:
+                    raise ValueError(
+                        f"{k} is of type {type(msdf.metadata[k])} and \
+                        as of now only slots of type 'str' or 'list' are handled."
+                    )
+                tmp_value.extend(v)
+                msdf.metadata[k] = list(set(tmp_value))
+            elif k in MULTIVALUED_SLOTS and replace_multivalued:
+                msdf.metadata[k] = list(v)
+            else:
+                msdf.metadata[k] = v[0]
+
+    return msdf
+
+
+def are_params_slots(params: dict) -> bool:
+    """Check if parameters conform to the slots in MAPPING_SET_SLOTS.
+
+    :param params: Dictionary of parameters.
+    :raises ValueError: If params are not slots.
+    :return: True/False
+    """
+    empty_params = {k: v for k, v in params.items() if v is None or v == ""}
+    if len(empty_params) > 0:
+        logging.info(f"Parameters: {empty_params.keys()} has(ve) no value.")
+
+    legit_params = all(p in MAPPING_SET_SLOTS for p in params.keys())
+    if not legit_params:
+        invalids = [p for p in params if p not in MAPPING_SET_SLOTS]
+        raise ValueError(
+            f"The params are invalid: {invalids}. Should be any of the following: {MAPPING_SET_SLOTS}"
+        )
+    return True