Automatically propagate/condense slots when parsing/writing. (#629)

* Enable automatic propagation and condensation.

Add a `propagate` parameter to all `parse_*` functions. When that
parameter is true, the parsing function will automatically propagate all
condensed slots in the parsed mapping set, prior to returning a
MappingSetDataFrame.

For the `parse_obographs_json` function, the parameter does not actually
make much sense (a mapping set extracted from a OBOGraph-JSON document
cannot be in a condensed state), but having all parsing functions accept
the same `propagate` parameter allows to keep using the
`get_parsing_function` logic without having to make a special case for
the functions that would not accept a `propagate` parameter. Likewise
for the `parse_alignment_xml` function.

The `propagate` parameter defaults to True -- this is the behaviour
recommended by the SSSOM specification --, except for the aforementioned
`parse_obographs_json` and `parse_alignment_xml` functions.

Propagation is normally performed by calling the
`MappingSetDataFrame#propagate()` method, but in the case of the
`parse_sssom_table()` function this cannot work, because propagation
needs to be performed before we can get a MappingSetDataFrame instance
(otherwise we could fail to construct a MappingSetDataFrame if the
mapping set contains literal mappings and the `subject_type` or
`object_type` slot is condensed). So we use a new
`propagate_condensed_slots()` method instead, which does not require a
MappingSetDataFrame object.

Conversely, we also add a `condense` parameter to all the `write_*`
functions, to perform the opposite operation prior to writing a set. The
parameter defaults to True for all writing functions, except `write_rdf`
since writing condensed slots in RDF is most likely not wanted.

* Update tests for the new behaviour of parsing functions.

Now that (most) parsing functions default to propagate condensed slots,
some tests need to be updated. Parsing-time propagation must be disabled
for:

* some tests that are performed on "half-condensed" test files and do
  not expect the structure of the set to be modified;
* the tests for the actual propagation/condensation feature (can't
  properly test propagation if the parser already did it for us).

In addition, we add a test to check that parsing-time condensation does
allow to successfully read a set containing literal mappings when the
object_type slot is condensed (#606).

* Add --[no-]propagate and --[no-]condense options.

Now that parsing and writing functions default to automatically
propagate and condense (respectively) sets, we add `--no-propagate`
and `--no-condense` options to some commands to alter the default
behaviour.

Specifically, we add such options to:

* `sssom convert`,
* `sssom parse`,
* `sssom validate`,
* and `sssom merge`.

It is currently not deemed useful (by me at least) to add those options
to every single command.

Author: gouttegd

src/sssom/cli.py

@@ -103,13 +103,20 @@
     default=("subject_category", "object_category"),
     help="Fields.",
 )
-
 predicate_filter_option = click.option(
     "-F",
     "--mapping-predicate-filter",
     multiple=True,
     help="A list of predicates or a file path containing the list of predicates to be considered.",
 )
+propagate_option = click.option(
+    "--propagate/--no-propagate",
+    default=True,
+    help="Automatically propagate condensed slots upon parsing.",
+)
+condense_option = click.option(
+    "--condense/--no-condense", default=True, help="Automatically condense slots upon writing."
+)
 
 
 @click.group()
@@ -148,9 +155,19 @@ def help(ctx: click.Context, subcommand: str) -> None:
 @input_argument
 @output_option
 @output_format_option
-def convert(input: str, output: TextIO, output_format: str) -> None:
+@propagate_option
+@condense_option
+def convert(
+    input: str, output: TextIO, output_format: str, propagate: bool, condense: bool
+) -> None:
     """Convert a file."""
-    convert_file(input_path=input, output=output, output_format=output_format)
+    convert_file(
+        input_path=input,
+        output=output,
+        output_format=output_format,
+        propagate=propagate,
+        condense=condense,
+    )
 
 
 # Input and metadata would be files (file paths). Check if exists.
@@ -193,6 +210,8 @@ def convert(input: str, output: TextIO, output_format: str) -> None:
 )
 @predicate_filter_option
 @output_option
+@propagate_option
+@condense_option
 def parse(
     input: str,
     input_format: str,
@@ -203,6 +222,8 @@ def parse(
     output: TextIO,
     embedded_mode: bool,
     mapping_predicate_filter: list[str],
+    propagate: bool,
+    condense: bool,
 ) -> None:
     """Parse a file in one of the supported formats (such as obographs) into an SSSOM TSV file."""
     parse_file(
@@ -215,6 +236,8 @@ def parse(
         strict_clean_prefixes=strict_clean_prefixes,
         embedded_mode=embedded_mode,
         mapping_predicate_filter=mapping_predicate_filter,
+        propagate=propagate,
+        condense=condense,
     )
 
 
@@ -227,10 +250,11 @@ def parse(
     multiple=True,
     default=DEFAULT_VALIDATION_TYPES,
 )
-def validate(input: str, validation_types: List[SchemaValidationType]) -> None:
+@propagate_option
+def validate(input: str, validation_types: List[SchemaValidationType], propagate: bool) -> None:
     """Produce an error report for an SSSOM file."""
     validation_type_list = [t for t in validation_types]
-    validate_file(input_path=input, validation_types=validation_type_list)
+    validate_file(input_path=input, validation_types=validation_type_list, propagate=propagate)
 
 
 @main.command()
@@ -522,11 +546,15 @@ def correlations(input: str, output: TextIO, transpose: bool, fields: Tuple[str,
     help="If true, the deduplicate (i.e., remove redundant lower confidence mappings) and reconcile (if msdf contains a higher confidence _negative_ mapping, then remove lower confidence positive one. If confidence is the same, prefer HumanCurated. If both HumanCurated, prefer negative mapping)",
 )
 @output_option
-def merge(inputs: str, output: TextIO, reconcile: bool = False) -> None:
+@propagate_option
+@condense_option
+def merge(
+    inputs: str, output: TextIO, propagate: bool, condense: bool, reconcile: bool = False
+) -> None:
     """Merge multiple MappingSetDataFrames into one."""  # noqa: DAR101
-    msdfs = [parse_sssom_table(i) for i in inputs]
+    msdfs = [parse_sssom_table(i, propagate=propagate) for i in inputs]
     merged_msdf = merge_msdf(*msdfs, reconcile=reconcile)
-    write_table(merged_msdf, output)
+    write_table(merged_msdf, output, condense=condense)
 
 
 @main.command(

src/sssom/io.py

@@ -43,18 +43,22 @@ def convert_file(
     input_path: str,
     output: TextIO,
     output_format: Optional[str] = None,
+    propagate: bool = True,
+    condense: bool = True,
 ) -> None:
     """Convert a file from one format to another.
 
     :param input_path: The path to the input SSSOM tsv file
     :param output: The path to the output file. If none is given, will default to using stdout.
     :param output_format: The format to which the SSSOM TSV should be converted.
+    :param propagate: Propagate condensed slots in the input file.
+    :param condense: Condense slots in the output file.
     """
     raise_for_bad_path(input_path)
-    doc = parse_sssom_table(input_path)
+    doc = parse_sssom_table(input_path, propagate=propagate)
     write_func, fileformat = get_writer_function(output_format=output_format, output=output)
     # TODO cthoyt figure out how to use protocols for this
-    write_func(doc, output, serialisation=fileformat)  # type:ignore
+    write_func(doc, output, serialisation=fileformat, condense=condense)  # type:ignore
 
 
 def parse_file(
@@ -68,6 +72,8 @@ def parse_file(
     strict_clean_prefixes: bool = True,
     embedded_mode: bool = True,
     mapping_predicate_filter: RecursivePathList | None = None,
+    propagate: bool = True,
+    condense: bool = True,
 ) -> None:
     """Parse an SSSOM metadata file and write to a table.
 
@@ -86,6 +92,8 @@ def parse_file(
         (tsv), else two separate files (tsv and yaml).
     :param mapping_predicate_filter: Optional list of mapping predicates or filepath containing the
         same.
+    :param propagate: If true, propagate all condensed slots in the input set.
+    :param condense: If true, condense slots in the output set.
     """
     raise_for_bad_path(input_path)
     converter, meta = _get_converter_and_metadata(
@@ -102,31 +110,34 @@ def parse_file(
         prefix_map=converter,
         meta=meta,
         mapping_predicates=mapping_predicates,
+        propagate=propagate,
     )
     if clean_prefixes:
         # We do this because we got a lot of prefixes from the default SSSOM prefixes!
         doc.clean_prefix_map(strict=strict_clean_prefixes)
-    write_table(doc, output, embedded_mode)
+    write_table(doc, output, embedded_mode, condense=condense)
 
 
 def validate_file(
     input_path: str,
     validation_types: Optional[List[SchemaValidationType]] = None,
     fail_on_error: bool = True,
+    propagate: bool = True,
 ) -> dict[SchemaValidationType, ValidationReport]:
     """Validate the incoming SSSOM TSV according to the SSSOM specification.
 
     :param input_path: The path to the input file in one of the legal formats, eg obographs,
         aligmentapi-xml
     :param validation_types: A list of validation types to run.
     :param fail_on_error: Should an exception be raised on error of _any_ validator?
+    :param propagate: If true, propagate condensed slots in the input set.
 
     :returns: A dictionary from validation types to validation reports
     """
     # Two things to check:
     # 1. All prefixes in the DataFrame are define in prefix_map
     # 2. All columns in the DataFrame abide by sssom-schema.
-    msdf = parse_sssom_table(file_path=input_path)
+    msdf = parse_sssom_table(file_path=input_path, propagate=propagate)
     return validate(msdf=msdf, validation_types=validation_types, fail_on_error=fail_on_error)
 
 

src/sssom/parsers.py

@@ -87,6 +87,7 @@
     MappingSetDataFrame,
     get_file_extension,
     is_multivalued_slot,
+    propagate_condensed_slots,
     raise_for_bad_path,
     safe_compress,
     to_mapping_set_dataframe,
@@ -304,6 +305,7 @@ def parse_sssom_table(
     *,
     strict: bool = False,
     sep: Optional[str] = None,
+    propagate: bool = True,
     **kwargs: Any,
 ) -> MappingSetDataFrame:
     """Parse a SSSOM CSV or TSV file.
@@ -315,6 +317,7 @@ def parse_sssom_table(
         within the document itself. For example, this may come from a companion SSSOM YAML file.
     :param strict: If true, will fail parsing for undefined prefixes, CURIEs, or IRIs
     :param sep: The seperator. If not given, inferred from file name
+    :param propagate: If true, propagate all condensed slots.
     :param kwargs: Additional keyword arguments (unhandled)
 
     :returns: A parsed dataframe wrapper object
@@ -358,6 +361,9 @@ def parse_sssom_table(
         )
     )
 
+    if propagate:
+        propagate_condensed_slots(df, combine_meta)
+
     msdf = from_sssom_dataframe(df, prefix_map=converter, meta=combine_meta)
     return msdf
 
@@ -379,6 +385,7 @@ def parse_sssom_rdf(
     prefix_map: ConverterHint = None,
     meta: Optional[MetadataType] = None,
     serialisation: str = SSSOM_DEFAULT_RDF_SERIALISATION,
+    propagate: bool = True,
     **kwargs: Any,
     # mapping_predicates: Optional[List[str]] = None,
 ) -> MappingSetDataFrame:
@@ -406,6 +413,8 @@ def parse_sssom_rdf(
         ]
     )
     msdf = from_sssom_rdf(g, prefix_map=converter, meta=meta)
+    if propagate:
+        msdf.propagate()
     # df: pd.DataFrame = msdf.df
     # if mapping_predicates and not df.empty():
     #     msdf.df = df[df["predicate_id"].isin(mapping_predicates)]
@@ -416,6 +425,7 @@ def parse_sssom_json(
     file_path: Union[str, Path],
     prefix_map: ConverterHint = None,
     meta: Optional[MetadataType] = None,
+    propagate: bool = True,
     **kwargs: Any,
 ) -> MappingSetDataFrame:
     """Parse a TSV to a :class:`MappingSetDocument` to a :class:`MappingSetDataFrame`."""
@@ -443,6 +453,8 @@ def parse_sssom_json(
     )
 
     msdf = from_sssom_json(jsondoc=jsondoc, prefix_map=converter, meta=meta)
+    if propagate:
+        msdf.propagate()
     return msdf
 
 
@@ -454,13 +466,15 @@ def parse_obographs_json(
     prefix_map: ConverterHint = None,
     meta: Optional[MetadataType] = None,
     mapping_predicates: Optional[List[str]] = None,
+    propagate: bool = False,
 ) -> MappingSetDataFrame:
     """Parse an obographs file as a JSON object and translates it into a MappingSetDataFrame.
 
     :param file_path: The path to the obographs file
     :param prefix_map: an optional prefix map
     :param meta: an optional dictionary of metadata elements
     :param mapping_predicates: an optional list of mapping predicates that should be extracted
+    :param propagate: If true, propagate all condensed slots.
 
     :returns: A SSSOM MappingSetDataFrame
     """
@@ -471,12 +485,15 @@ def parse_obographs_json(
     with open(file_path) as json_file:
         jsondoc = json.load(json_file)
 
-    return from_obographs(
+    msdf = from_obographs(
         jsondoc,
         prefix_map=converter,
         meta=meta,
         mapping_predicates=mapping_predicates,
     )
+    if propagate:
+        msdf.propagate()
+    return msdf
 
 
 def _get_prefix_map_and_metadata(
@@ -539,6 +556,7 @@ def parse_alignment_xml(
     prefix_map: ConverterHint = None,
     meta: Optional[MetadataType] = None,
     mapping_predicates: Optional[List[str]] = None,
+    propagate: bool = False,
 ) -> MappingSetDataFrame:
     """Parse a TSV -> MappingSetDocument -> MappingSetDataFrame."""
     raise_for_bad_path(file_path)
@@ -552,6 +570,8 @@ def parse_alignment_xml(
         meta=meta,
         mapping_predicates=mapping_predicates,
     )
+    if propagate:
+        msdf.propagate()
     return msdf
 
 

src/sssom/util.py

@@ -342,31 +342,7 @@ def propagate(self, fill_empty: bool = False) -> List[str]:
 
         :returns: The list of slots that were effectively propagated.
         """
-        schema = SSSOMSchemaView()
-        propagated = []
-
-        for slot in schema.propagatable_slots:
-            if slot not in self.metadata:  # Nothing to propagate
-                continue
-            is_present = slot in self.df.columns
-            if is_present and not fill_empty:
-                logging.warning(
-                    f"Not propagating value for '{slot}' because the slot is already set on individual records."
-                )
-                continue
-
-            if schema.view.get_slot(slot).multivalued:
-                value = "|".join(self.metadata.pop(slot))
-            else:
-                value = self.metadata.pop(slot)
-
-            if is_present:
-                self.df.loc[self.df[slot].eq("") | self.df[slot].isna(), slot] = value
-            else:
-                self.df[slot] = value
-            propagated.append(slot)
-
-        return propagated
+        return propagate_condensed_slots(self.df, self.metadata, fill_empty)
 
     def condense(self) -> List[str]:
         """Condense record-level slot values to the set whenever possible.
@@ -1306,6 +1282,51 @@ def deal_with_negation(df: pd.DataFrame) -> pd.DataFrame:
     return return_df
 
 
+def propagate_condensed_slots(
+    df: pd.DataFrame, metadata: MetadataType, fill_empty: bool = False
+) -> List[str]:
+    """Propagate slot values from the set level down to individual records.
+
+    This function performs the same operation as the `MappingSetDataFrame#propagate()`
+    method. It is intended to allow propagating a mapping set before an instance of
+    the `MappingSetDataFrame` class can be obtained.
+
+    :param df: The DataFrame into which values should be propagated.
+    :param metadata: The dictionary of set-level metadata.
+    :param fill_empty: If True, propagation of a slot is allowed even if some individual records
+        already have a value for that slot. The set-level value will be propagated to all the
+        records for which the slot is empty. Note that (1) this is not spec-compliant behaviour,
+        and (2) this makes the operation non-reversible by a subsequent condensation.
+
+    :returns: The list of slots that were effectively propagated.
+    """
+    schema = SSSOMSchemaView()
+    propagated = []
+
+    for slot in schema.propagatable_slots:
+        if slot not in metadata:  # Nothing to propagate
+            continue
+        is_present = slot in df.columns
+        if is_present and not fill_empty:
+            logging.warning(
+                f"Not propagating value for '{slot}' because the slot is already set on individual records."
+            )
+            continue
+
+        if schema.view.get_slot(slot).multivalued:
+            value = "|".join(metadata.pop(slot))
+        else:
+            value = metadata.pop(slot)
+
+        if is_present:
+            df.loc[df[slot].eq("") | df[slot].isna(), slot] = value
+        else:
+            df[slot] = value
+        propagated.append(slot)
+
+    return propagated
+
+
 ExtensionLiteral = Literal["tsv", "csv"]