Merge pull request #259 from whitemech/tests

Add more tests

.bandit.yml

@@ -135,7 +135,7 @@ any_other_function_with_shell_equals_true:
   - subprocess.check_output
   - subprocess.run
 assert_used:
-  skips: ["**/test_*.py"]
+  skips: ["**/test_*.py", "tests/helpers/**"]
 hardcoded_tmp_directory:
   tmp_dirs:
   - /tmp

.github/workflows/linting.yml

@@ -14,7 +14,7 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest]
-        python-version: ["3.8", "3.10", "3.11"]
+        python-version: ["3.8", "3.9", "3.10", "3.11"]
 
     timeout-minutes: 30
 
@@ -35,7 +35,6 @@ jobs:
         tox -e black-check
         tox -e isort-check
         tox -e flake8
-        tox -e darglint
     - name: Unused code check
       run: tox -e vulture
     - name: Static type check

examples/pddl/domain.pddl

@@ -3,7 +3,7 @@
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
 (define (domain blocks)
-  (:requirements :strips :typing)
+  (:requirements :strips :typing :disjunctive-preconditions)
   (:types block)
   (:predicates (on ?x - block ?y - block)
 	       (ontable ?x - block)

plan4past/helpers/utils.py

@@ -21,11 +21,7 @@
 #
 
 """Miscellanea utilities."""
-import contextlib
-import os
-from os import PathLike
-from pathlib import Path
-from typing import Dict, Generator
+from typing import Dict
 
 from pddl.logic import Predicate, constants
 from pylogics.syntax.base import Formula
@@ -34,30 +30,25 @@
 from plan4past.utils.atoms_visitor import find_atoms
 
 
-@contextlib.contextmanager
-def cd(path: PathLike) -> Generator:  # pylint: disable=invalid-name
-    """Change working directory temporarily."""
-    old_path = Path.cwd()
-    os.chdir(path)
-    try:
-        yield
-    finally:
-        os.chdir(str(old_path))
-
-
 def add_val_prefix(name: str):
     """Add the 'prime' prefix."""
     return "val-" + name.replace('"', "")
 
 
 def remove_before_prefix(name: str):
     """Remove the 'Y' prefix."""
-    return name.replace("Y-", "") if name[1] == "-" else name.replace("Y", "", 1)
+    return (
+        name.replace("Y-", "")
+        if name.startswith("Y-")
+        else name.replace("Y", "", 1)
+        if name.startswith("Y")
+        else name
+    )
 
 
 def remove_val_prefix(name: str):
     """Remove the 'prime' prefix."""
-    return name.replace("val-", "")
+    return name.replace("val-", "") if name.startswith("val-") else name
 
 
 def replace_symbols(name: str):

plan4past/utils/atoms_visitor.py

@@ -26,7 +26,6 @@
 from typing import Set
 
 from pylogics.syntax.base import And as PLTLAnd
-from pylogics.syntax.base import Formula
 from pylogics.syntax.base import Not as PLTLNot
 from pylogics.syntax.base import Or as PLTLOr
 from pylogics.syntax.base import _BinaryOp, _UnaryOp
@@ -48,9 +47,9 @@ def find_atoms_unaryop(formula: _UnaryOp):
 
 
 @singledispatch
-def find_atoms(_formula: Formula) -> Set[PLTLAtomic]:
+def find_atoms(obj: object) -> Set[PLTLAtomic]:
     """Find atoms for a formula."""
-    return set()
+    raise ValueError(f"object of type {type(obj)} is not supported by this function")
 
 
 @find_atoms.register

plan4past/utils/derived_visitor.py

@@ -27,7 +27,6 @@
 from pddl.logic.base import And, Not, Or
 from pddl.logic.predicates import DerivedPredicate, Predicate
 from pylogics.syntax.base import And as PLTLAnd
-from pylogics.syntax.base import Formula
 from pylogics.syntax.base import Not as PLTLNot
 from pylogics.syntax.base import Or as PLTLOr
 from pylogics.syntax.pltl import Atomic as PLTLAtomic
@@ -45,10 +44,10 @@
 
 @singledispatch
 def derived_predicates(
-    formula: Formula, atoms_to_fluents: Dict[PLTLAtomic, Predicate]
+    formula: object, atoms_to_fluents: Dict[PLTLAtomic, Predicate]
 ) -> Set[DerivedPredicate]:
     """Compute the derived predicate for a formula."""
-    raise NotImplementedError(f"handler not implemented for formula {type(formula)}")
+    raise NotImplementedError(f"handler not implemented for object {type(formula)}")
 
 
 @derived_predicates.register
@@ -132,7 +131,7 @@ def derived_predicates_before(
     """Compute the derived predicate for a Before formula."""
     formula_name = to_string(formula)
     val = Predicate(add_val_prefix(replace_symbols(formula_name)))
-    condition = And(Predicate(replace_symbols(to_string(formula))))
+    condition = Predicate(replace_symbols(to_string(formula)))
     der_pred_arg = derived_predicates(formula.argument, atoms_to_fluents)
     return {DerivedPredicate(val, condition)}.union(der_pred_arg)
 
@@ -167,7 +166,7 @@ def derived_predicates_once(
     val = Predicate(add_val_prefix(replace_symbols(formula_name)))
     condition = Or(
         Predicate(add_val_prefix(replace_symbols(to_string(formula.argument)))),
-        And(Predicate(f"Y-{replace_symbols(to_string(formula))}")),
+        Predicate(f"Y-{replace_symbols(to_string(formula))}"),
     )
     der_pred_arg = derived_predicates(formula.argument, atoms_to_fluents)
     return {DerivedPredicate(val, condition)}.union(der_pred_arg)

plan4past/utils/mapping_parser.py

@@ -21,30 +21,166 @@
 #
 
 """Mapping parser."""
+import re
+from typing import Dict, Optional, Sequence, Set, Tuple
 
-from typing import Dict
-
-from pddl.logic import Predicate, constants
+from pddl.custom_types import name
+from pddl.logic import Constant, Predicate, constants
 from pylogics.syntax.base import Formula
 from pylogics.syntax.pltl import Atomic as PLTLAtomic
 
 from plan4past.utils.atoms_visitor import find_atoms
 
 
+class MappingParserError(Exception):
+    """Mapping parser error."""
+
+    def __init__(self, message: str, *args, row_id: Optional[int] = None) -> None:
+        """
+        Initialize a mapping parser error.
+
+        :param row_id: the row id
+        """
+        self.row_id = row_id
+
+        super().__init__(self._make_message_prefix() + message, *args)
+
+    def _make_message_prefix(self) -> str:
+        """Make the message prefix."""
+        if self.row_id is None:
+            return "invalid mapping: "
+        return f"invalid mapping at row {self.row_id}: "
+
+
+SYMBOL_REGEX = re.compile("[a-z_]([a-zA-Z0-9_-]+[a-zA-Z0-9_])|[a-z_][a-zA-Z0-9_]*")
+"""The following is a sub-regex of AtomName.REGEX in pylogics.syntax.base."""
+
+
+def _check_pddl_name(row_id: int, pddl_name: str, what: str) -> None:
+    """Check that a PDDL name is valid."""
+    try:
+        name(pddl_name)
+    except ValueError as e:
+        raise MappingParserError(
+            f"got an invalid name for {what}: '{pddl_name}'. It must match the regex {name.REGEX}",
+            row_id=row_id,
+        ) from e
+
+
+def _parse_constants(row_id: int, constants_str: str) -> Sequence[Constant]:
+    """
+    Parse a string of constants.
+
+    :param row_id: the row id
+    :param constants_str: the string of constants
+    :return: the sequence of constants
+    """
+    try:
+        return constants(constants_str)
+    except ValueError as e:
+        raise MappingParserError(f"got an invalid constant: {e}", row_id=row_id) from e
+
+
+def _parse_predicate(row_id: int, predicate_str: str) -> Predicate:
+    """
+    Parse a predicate.
+
+    :param predicate_str: the predicate string
+    :return: the predicate
+    """
+    predicate_name, cons = predicate_str.split(" ", maxsplit=1)
+
+    _check_pddl_name(row_id, predicate_name, "a predicate")
+    parsed_constants = _parse_constants(row_id, cons)
+    return Predicate(predicate_name, *parsed_constants)
+
+
+def _process_row(row_id: int, row_str: str) -> Tuple[str, Predicate]:
+    """
+    Process a row of the mapping file.
+
+    :param row_id: the row id
+    :param row_str: the row string
+    :return: the pair (symbol_name, predicate_str)
+    """
+    # check the row is a valid mapping
+    comma_separated_row_parts = row_str.split(",")
+    if len(comma_separated_row_parts) != 2:
+        raise MappingParserError(
+            "expected a mapping of the form '<symbol>,<predicate>'", row_id=row_id
+        )
+
+    symbol_name, predicate = comma_separated_row_parts
+
+    # strip leading and trailing whitespaces
+    symbol_name = symbol_name.strip()
+    predicate_str = predicate.strip()
+
+    if symbol_name == "" or SYMBOL_REGEX.fullmatch(symbol_name) is None:
+        raise MappingParserError(
+            "symbol cannot be empty string and must match the regex {SYMBOL_REGEX}",
+            row_id=row_id,
+        )
+    if predicate_str == "":
+        raise MappingParserError("predicate cannot be empty string", row_id=row_id)
+
+    predicate = _parse_predicate(row_id, predicate_str)
+    return symbol_name, predicate
+
+
+def _check_unmapped_symbols(
+    all_symbol_names: Set[str], mapped_symbol_names: Set[str]
+) -> None:
+    """Check that all symbols are mapped."""
+    # check if there are unmapped symbols (exclude 'true' and 'false')
+    unmapped_symbols = all_symbol_names - {"true", "false"} - mapped_symbol_names
+    # if some symbols are not mapped, raise an error
+    if unmapped_symbols:
+        raise MappingParserError(
+            f"the following symbols of the formula are not mapped: {unmapped_symbols}"
+        )
+
+
+def should_skip_row(row_str: str) -> bool:
+    """Check if a row should be skipped."""
+    # skip empty lines
+    if row_str.strip() == "":
+        return True
+
+    # skip comments
+    if row_str.strip().startswith(";"):
+        return True
+
+    return False
+
+
 def mapping_parser(text: str, formula: Formula) -> Dict[PLTLAtomic, Predicate]:
     """Parse symbols to ground predicates mapping."""
     symbols = find_atoms(formula)
-    maps = text.split("\n")
+    symbols_by_name = {symbol.name: symbol for symbol in symbols}
+    maps = text.splitlines(keepends=False)
     from_atoms_to_fluents = {}
-    for symbol in symbols:
-        for vmap in maps:
-            s, p = vmap.split(",")
-            if symbol.name == s:
-                if " " in p:
-                    name, cons = p.split(" ", maxsplit=1)
-                    from_atoms_to_fluents[symbol] = Predicate(name, *constants(cons))
-                else:
-                    from_atoms_to_fluents[symbol] = Predicate(p)
-            else:
-                continue
+    mapped_symbol_names = set()
+
+    for row_id, vmap in enumerate(maps):
+        if should_skip_row(vmap):
+            continue
+
+        symbol_name, predicate = _process_row(row_id, vmap)
+
+        if symbol_name not in symbols_by_name:
+            # don't need to process this row, since the symbol does not occur in the formula
+            continue
+
+        if symbol_name in mapped_symbol_names:
+            raise MappingParserError(
+                f"symbol '{symbol_name}' is mapped multiple times", row_id=row_id
+            )
+
+        symbol = symbols_by_name[symbol_name]
+        mapped_symbol_names.add(symbol_name)
+
+        from_atoms_to_fluents[symbol] = predicate
+
+    _check_unmapped_symbols(set(symbols_by_name.keys()), mapped_symbol_names)
     return from_atoms_to_fluents

plan4past/utils/predicates_visitor.py

@@ -27,7 +27,6 @@
 
 from pddl.logic.predicates import Predicate
 from pylogics.syntax.base import And as PLTLAnd
-from pylogics.syntax.base import Formula
 from pylogics.syntax.base import Not as PLTLNot
 from pylogics.syntax.base import Or as PLTLOr
 from pylogics.syntax.base import _BinaryOp, _UnaryOp
@@ -55,9 +54,11 @@ def predicates_unaryop(formula: _UnaryOp):
 
 
 @singledispatch
-def predicates(formula: Formula) -> Set[Predicate]:
+def predicates(formula: object) -> Set[Predicate]:
     """Compute predicate for a formula."""
-    raise NotImplementedError(f"handler not implemented for formula {type(formula)}")
+    raise NotImplementedError(
+        f"handler not implemented for object of type {type(formula)}"
+    )
 
 
 @predicates.register

plan4past/utils/rewrite_formula_visitor.py

@@ -24,6 +24,7 @@
 from functools import singledispatch
 
 from pylogics.syntax.base import And as PLTLAnd
+from pylogics.syntax.base import Equivalence as PLTLEquivalence
 from pylogics.syntax.base import Formula
 from pylogics.syntax.base import Implies as PLTLImplies
 from pylogics.syntax.base import Not as PLTLNot
@@ -46,9 +47,11 @@ def rewrite_unaryop(formula: _UnaryOp):
 
 
 @singledispatch
-def rewrite(formula: Formula) -> Formula:
+def rewrite(formula: object) -> Formula:
     """Rewrite a formula."""
-    raise NotImplementedError(f"handler not implemented for formula {type(formula)}")
+    raise NotImplementedError(
+        f"handler not implemented for object of type {type(formula)}"
+    )
 
 
 @rewrite.register
@@ -98,6 +101,14 @@ def rewrite_implies(formula: PLTLImplies) -> Formula:
     return PLTLOr(*head, tail)
 
 
+@rewrite.register
+def rewrite_equivalence(formula: PLTLEquivalence) -> Formula:
+    """Compute the basic formula for an Equivalence formula."""
+    positive = PLTLAnd(*[rewrite(f) for f in formula.operands])
+    negative = PLTLAnd(*[PLTLNot(rewrite(f)) for f in formula.operands])
+    return PLTLOr(positive, negative)
+
+
 @rewrite.register
 def rewrite_before(formula: Before) -> Formula:
     """Compute the basic formula for a Before (Yesterday) formula."""