Merge pull request #70 from EMMC-ASBL/optional-values

Made the value optional + added some cleanup

tripper/mappings/mappings.py

@@ -14,15 +14,15 @@
 
 from collections import defaultdict
 from enum import Enum
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Sequence
 
 import numpy as np
 from pint import Quantity  # remove
 
 from tripper import DM, EMMO, FNO, MAP, RDF, RDFS
 
 if TYPE_CHECKING:  # pragma: no cover
-    from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
+    from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union
 
     from tripper import Triplestore
 
@@ -83,11 +83,11 @@ class Value:
 
     def __init__(
         self,
-        value: "Any",
+        value: "Any" = None,
         unit: "Optional[str]" = None,
         iri: "Optional[str]" = None,
         property_iri: "Optional[str]" = None,
-        cost: float = 0.0,
+        cost: "Union[float, Callable]" = 0.0,
     ):
         self.value = value
         self.unit = unit
@@ -100,14 +100,19 @@ def show(
         routeno: "Optional[int]" = None,
         name: "Optional[str]" = None,
         indent: int = 0,
-    ):  # pylint: disable=unused-argument
+    ) -> str:
+        # pylint: disable=unused-argument
         """Returns a string representation of the Value.
 
         Arguments:
             routeno: Unused.  The argument exists for consistency with
                 the corresponding method in Step.
             name: Name of value.
             indent: Indentation level.
+
+        Returns:
+            String representation of the value.
+
         """
         strings = []
         ind = " " * indent
@@ -149,7 +154,7 @@ def __init__(
         function: "Optional[Callable]" = None,
         cost: "Union[float, Callable]" = 1.0,
         output_unit: "Optional[str]" = None,
-    ):
+    ) -> None:
         self.output_iri = output_iri
         self.steptype = steptype
         self.function = function
@@ -159,12 +164,12 @@ def __init__(
         self.join_mode = False  # whether to join upcoming input
         self.joined_input: "Inputs" = {}
 
-    def add_inputs(self, inputs: "Inputs"):
+    def add_inputs(self, inputs: "Inputs") -> None:
         """Add input dict for an input route."""
         assert isinstance(inputs, dict)  # nosec B101
         self.input_routes.append(inputs)
 
-    def add_input(self, input: "Input", name: "Optional[str]" = None):
+    def add_input(self, input: "Input", name: "Optional[str]" = None) -> None:
         """Add an input (MappingStep or Value), where `name` is the name
         assigned to the argument.
 
@@ -181,7 +186,7 @@ def add_input(self, input: "Input", name: "Optional[str]" = None):
         else:
             self.add_inputs({argname: input})
 
-    def join_input(self):
+    def join_input(self) -> None:
         """Join all input added with add_input() since `join_mode` was set true.
         Resets `join_mode` to false."""
         if not self.join_mode:
@@ -195,16 +200,16 @@ def eval(
         routeno: "Optional[int]" = None,
         unit: "Optional[str]" = None,
         magnitude: bool = False,
-        quantity: "Any" = Quantity,
+        quantity: "Type[Quantity]" = Quantity,
     ) -> "Any":
         """Returns the evaluated value of given input route number.
 
-        Args:
+        Arguments:
             routeno: The route number to evaluate.  If None (default)
                 the route with the lowest cost is evalueated.
             unit: return the result in the given unit.
                 Implies `magnitude=True`.
-            magnitude: Whether to only return the magitude of the evaluated
+            magnitude: Whether to only return the magnitude of the evaluated
                 value (with no unit).
             quantity: Quantity class to use for evaluation.  Defaults to pint.
 
@@ -230,7 +235,15 @@ def eval(
 
     def get_inputs(self, routeno: int) -> "Tuple[Inputs, int]":
         """Returns input and input index `(inputs, idx)` for route number
-        `routeno`."""
+        `routeno`.
+
+        Arguments:
+            routeno: The route number to return inputs for.
+
+        Returns:
+            Inputs and difference between route number and number of routes for an
+            input dictioary.
+        """
         n = 0
         for inputs in self.input_routes:
             n0 = n
@@ -241,23 +254,42 @@ def get_inputs(self, routeno: int) -> "Tuple[Inputs, int]":
 
     def get_input_iris(self, routeno: int) -> "Dict[str, Optional[str]]":
         """Returns a dict mapping input names to iris for the given route
-        number."""
+        number.
+
+        Arguments:
+            routeno: The route number to return a mapping for.
+
+        Returns:
+            Mapping of input names to IRIs.
+
+        """
         inputs, _ = self.get_inputs(routeno)
         return {
             k: v.output_iri if isinstance(v, MappingStep) else v.iri
             for k, v in inputs.items()
         }
 
     def number_of_routes(self) -> int:
-        """Returns total number of routes to this mapping step."""
+        """Total number of routes to this mapping step.
+
+        Returns:
+            Total number of routes to this mapping step.
+        """
         n = 0
         for inputs in self.input_routes:
             n += get_nroutes(inputs)
         return n
 
     def lowest_costs(self, nresults: int = 5) -> "List[Tuple[float, int]]":
         """Returns a list of `(cost, routeno)` tuples with up to the `nresult`
-        lowest costs and their corresponding route numbers."""
+        lowest costs and their corresponding route numbers.
+
+        Arguments:
+            nresults: Number of results to return.
+
+        Returns:
+            A list of `(cost, routeno)` tuples.
+        """
         result = []
         n = 0  # total number of routes
 
@@ -336,14 +368,17 @@ def show(
         routeno: "Optional[int]" = None,
         name: "Optional[str]" = None,
         indent: int = 0,
-    ):
+    ) -> str:
         """Returns a string representation of the mapping routes to this step.
 
         Arguments:
             routeno: show given route.  The default is to show all routes.
             name: Name of the last mapping step (mainly for internal use).
             indent: How of blanks to prepend each line with (mainly for
                 internal use).
+
+        Returns:
+            String representation of the mapping routes.
         """
         strings = []
         ind = " " * indent
@@ -378,23 +413,43 @@ def show(
 
 
 def get_nroutes(inputs: "Inputs") -> int:
-    """Help function returning the number of routes for an input dict."""
-    m = 1
+    """Help function returning the number of routes for an input dict.
+
+    Arguments:
+        inputs: Input dictionary.
+
+    Returns:
+        Number of routes in the `inputs` input dictionary.
+    """
+    nroutes = 1
     for input in inputs.values():
         if isinstance(input, MappingStep):
-            m *= input.number_of_routes()
-    return m
+            nroutes *= input.number_of_routes()
+    return nroutes
 
 
 def get_values(
-    inputs: "Inputs", routeno: int, quantity: "Any" = Quantity, magnitudes: bool = False
-):
+    inputs: "dict[str, Any]",
+    routeno: int,
+    quantity: "Type[Quantity]" = Quantity,
+    magnitudes: bool = False,
+) -> "dict[str, Any]":
     """Help function returning a dict mapping the input names to actual value
     of expected input unit.
 
     There exists `get_nroutes(inputs)` routes to populate `inputs`.
     `routeno` is the index of the specific route we will use to obtain the
     values.
+
+    Arguments:
+        inputs: Input dictionary.
+        routeno: Route number index.
+        quantity: A unit quantity class.
+        magnitudes: Whether to only return the magnitude of the evaluated
+            value (with no unit).
+
+    Returns:
+        A mapping between input names and values of expected input unit.
     """
     values = {}
     for k, v in inputs.items():
@@ -405,8 +460,13 @@ def get_values(
                 if v.output_unit and isinstance(v, quantity)
                 else value
             )
-        else:
+        elif isinstance(v, Value):
             values[k] = quantity(v.value, v.unit)
+        else:
+            raise TypeError(
+                "Expected values in inputs to be either `MappingStep` or "
+                "`Value` objects."
+            )
 
         if magnitudes:
             values = {
@@ -442,13 +502,13 @@ def fno_mapper(triplestore: "Triplestore") -> "Dict[str, list]":
     """Finds all function definitions in `triplestore` based on the function
     ontololy (FNO).
 
-    Return a dict mapping output IRIs to a list of
-
-        (function_iri, [input_iris, ...])
+    Arguments:
+        triplestore: The triplestore to investigate.
 
-    tuples.
+    Returns:
+        A mapping of output IRIs to a list of `(function_iri, [input_iris, ...])`
+        tuples.
     """
-
     # pylint: disable=too-many-branches
 
     # Temporary dicts for fast lookup
@@ -495,7 +555,7 @@ def mapping_routes(
     target: str,
     sources: "Dict[str, Value]",
     triplestore: "Triplestore",
-    function_repo: "Any" = None,
+    function_repo: "Optional[dict]" = None,
     function_mappers: "Sequence[Callable]" = (fno_mapper,),
     default_costs: "Tuple" = (
         ("function", 10.0),
@@ -511,7 +571,7 @@ def mapping_routes(
     label: str = RDFS.label,
     hasUnit: str = DM.hasUnit,
     hasCost: str = DM.hasCost,  # TODO - add hasCost to the DM ontology
-) -> MappingStep:
+) -> Input:
     """Find routes of mappings from any source in `sources` to `target`.
 
     This implementation supports functions (using FnO) and subclass
@@ -520,7 +580,8 @@ def mapping_routes(
 
     Arguments:
         target: IRI of the target in `triplestore`.
-        sources: Dict mapping source IRIs to source values.
+        sources: Dict mapping source IRIs to source values or a sequence
+            of source IRIs (with no explicit values).
         triplestore: Triplestore instance.
             It is safe to pass a generator expression too.
 
@@ -551,9 +612,14 @@ def mapping_routes(
         MappingStep instances providing an (efficient) internal description
         of all possible mapping routes from `sources` to `target`.
     """
-
     # pylint: disable=too-many-arguments,too-many-locals,too-many-statements
 
+    if target in sources:
+        return Value(iri=target)
+
+    if isinstance(sources, Sequence):
+        sources = {iri: None for iri in sources}
+
     if function_repo is None:
         function_repo = triplestore.function_repo