Use pathlib

driver/aliases.py

@@ -1,9 +1,7 @@
-import os
-
 from .util import DRIVER_DIR
 
 
-PORTFOLIO_DIR = os.path.join(DRIVER_DIR, "portfolios")
+PORTFOLIO_DIR = DRIVER_DIR / "portfolios"
 
 ALIASES = {}
 
@@ -143,12 +141,11 @@ def _get_lama(pref):
 
 
 PORTFOLIOS = {}
-for portfolio in os.listdir(PORTFOLIO_DIR):
-    if portfolio == "__pycache__":
+for portfolio in PORTFOLIO_DIR.iterdir():
+    if portfolio.name == "__pycache__":
         continue
-    name, ext = os.path.splitext(portfolio)
-    assert ext == ".py", portfolio
-    PORTFOLIOS[name.replace("_", "-")] = os.path.join(PORTFOLIO_DIR, portfolio)
+    assert portfolio.suffix == ".py", str(portfolio)
+    PORTFOLIOS[portfolio.stem.replace("_", "-")] = PORTFOLIO_DIR / portfolio
 
 
 def show_aliases():

driver/arguments.py

@@ -1,5 +1,5 @@
 import argparse
-import os.path
+from pathlib import Path
 import re
 import sys
 
@@ -47,8 +47,7 @@
 that exceed their time or memory limit are aborted, and the next
 configuration is run."""
 
-EXAMPLE_PORTFOLIO = os.path.relpath(
-    aliases.PORTFOLIOS["seq-opt-fdss-1"], start=util.REPO_ROOT_DIR)
+EXAMPLE_PORTFOLIO = aliases.PORTFOLIOS["seq-opt-fdss-1"].relative_to(util.REPO_ROOT_DIR)
 
 EXAMPLES = [
     ("Translate and find a plan with A* + LM-Cut:",
@@ -60,7 +59,7 @@
     ("Run predefined configuration (LAMA-2011) on translated task:",
      ["--alias", "seq-sat-lama-2011", "output.sas"]),
     ("Run a portfolio on a translated task:",
-     ["--portfolio", EXAMPLE_PORTFOLIO,
+     ["--portfolio", str(EXAMPLE_PORTFOLIO),
       "--search-time-limit", "30m", "output.sas"]),
     ("Run the search component in debug mode (with assertions enabled) "
      "and validate the resulting plan:",
@@ -81,19 +80,29 @@
       "--evaluator", '"hff=ff()"', "--search", '"eager_greedy([hff], preferred=[hff])"']),
 ]
 
-EPILOG = """component options:
+
+def _format_example(description, parameters):
+    call_string = " ".join([Path(sys.argv[0]).name] + parameters)
+    return f"{description}\n{call_string}"
+
+
+def _format_examples(examples):
+    return "\n\n".join(_format_example(*example) for example in examples)
+
+
+EPILOG = f"""component options:
   --translate-options OPTION1 OPTION2 ...
   --search-options OPTION1 OPTION2 ...
                         pass OPTION1 OPTION2 ... to specified planner component
                         (default: pass component options to search)
 
 Examples:
 
-%s
-""" % "\n\n".join("%s\n%s" % (desc, " ".join([os.path.basename(sys.argv[0])] + parameters)) for desc, parameters in EXAMPLES)
+{_format_examples(EXAMPLES)}
+"""
 
 COMPONENTS_PLUS_OVERALL = ["translate", "search", "validate", "overall"]
-DEFAULT_SAS_FILE = "output.sas"
+DEFAULT_SAS_FILE = Path("output.sas")
 
 
 """
@@ -102,7 +111,7 @@
 """
 def print_usage_and_exit_with_driver_input_error(parser, msg):
     parser.print_usage()
-    returncodes.exit_with_driver_input_error("{}: error: {}".format(os.path.basename(sys.argv[0]), msg))
+    returncodes.exit_with_driver_input_error(f"{Path(sys.argv[0]).name}: error: {msg}")
 
 
 class RawHelpFormatter(argparse.HelpFormatter):
@@ -211,8 +220,8 @@ def _set_components_automatically(parser, args):
 
 def _set_components_and_inputs(parser, args):
     """Set args.components to the planner components to be run and set
-    args.translate_inputs and args.search_input to the correct input
-    filenames.
+    args.translate_inputs, args.search_input, and args.validate_inputs
+    to the correct input filenames.
 
     Rules:
     1. If any --run-xxx option is specified, then the union
@@ -239,34 +248,43 @@ def _set_components_and_inputs(parser, args):
 
     assert args.components
     first = args.components[0]
-    num_files = len(args.filenames)
     # When passing --help to any of the components (or -h to the
     # translator), we don't require input filenames and silently
     # swallow any that are provided. This is undocumented to avoid
     # cluttering the driver's --help output.
     if first == "translate":
         if "--help" in args.translate_options or "-h" in args.translate_options:
             args.translate_inputs = []
-        elif num_files == 1:
-            task_file, = args.filenames
-            domain_file = util.find_domain_filename(task_file)
-            args.translate_inputs = [domain_file, task_file]
-        elif num_files == 2:
-            args.translate_inputs = args.filenames
         else:
-            print_usage_and_exit_with_driver_input_error(
-                parser, "translator needs one or two input files")
+            args.translate_inputs = _get_pddl_input_files(args, parser, "translator")
     elif first == "search":
         if "--help" in args.search_options:
             args.search_input = None
-        elif num_files == 1:
-            args.search_input, = args.filenames
+        elif len(args.filenames) == 1:
+            args.search_input = Path(args.filenames[0])
         else:
             print_usage_and_exit_with_driver_input_error(
                 parser, "search needs exactly one input file")
     else:
         assert False, first
 
+    if "validate" in args.components:
+        args.validate_inputs = _get_pddl_input_files(args, parser, "validate")
+
+
+def _get_pddl_input_files(args, parser, component):
+    num_files = len(args.filenames)
+    if num_files == 1:
+        task = Path(args.filenames[0])
+        domain = util.find_domain_path(task)
+    elif num_files == 2:
+        domain = Path(args.filenames[0])
+        task = Path(args.filenames[1])
+    else:
+        print_usage_and_exit_with_driver_input_error(
+            parser, f"{component} needs one or two PDDL input files")
+    return [domain, task]
+
 
 def _set_translator_output_options(parser, args):
     if any("--sas-file" in opt for opt in args.translate_options):
@@ -397,20 +415,20 @@ def parse_args():
         help="set log level (most verbose: debug; least verbose: warning; default: %(default)s)")
 
     driver_other.add_argument(
-        "--plan-file", metavar="FILE", default="sas_plan",
+        "--plan-file", metavar="FILE", default="sas_plan", type=Path,
         help="write plan(s) to FILE (default: %(default)s; anytime configurations append .1, .2, ...)")
 
     driver_other.add_argument(
-        "--sas-file", metavar="FILE",
+        "--sas-file", metavar="FILE", type=Path,
         help="intermediate file for storing the translator output "
-            "(implies --keep-sas-file, default: {})".format(DEFAULT_SAS_FILE))
+            f"(implies --keep-sas-file, default: {DEFAULT_SAS_FILE})")
     driver_other.add_argument(
         "--keep-sas-file", action="store_true",
         help="keep translator output file (implied by --sas-file, default: "
             "delete file if translator and search component are active)")
 
     driver_other.add_argument(
-        "--portfolio", metavar="FILE",
+        "--portfolio", metavar="FILE", type=Path,
         help="run a portfolio specified in FILE")
     driver_other.add_argument(
         "--portfolio-bound", metavar="VALUE", default=None, type=int,

driver/call.py

@@ -10,9 +10,14 @@
 import sys
 
 
+def _replace_paths_with_strings(cmd):
+    return [str(x) for x in cmd]
+
+
 def print_call_settings(nick, cmd, stdin, time_limit, memory_limit):
+    cmd = _replace_paths_with_strings(cmd)
     if stdin is not None:
-        stdin = shlex.quote(stdin)
+        stdin = shlex.quote(str(stdin))
     logging.info("{} stdin: {}".format(nick, stdin))
     limits.print_limits(nick, time_limit, memory_limit)
 
@@ -47,6 +52,7 @@ def fail(exception, exitcode):
 
 
 def check_call(nick, cmd, stdin=None, time_limit=None, memory_limit=None):
+    cmd = _replace_paths_with_strings(cmd)
     print_call_settings(nick, cmd, stdin, time_limit, memory_limit)
 
     kwargs = {"preexec_fn": _get_preexec_function(time_limit, memory_limit)}
@@ -60,6 +66,7 @@ def check_call(nick, cmd, stdin=None, time_limit=None, memory_limit=None):
 
 
 def get_error_output_and_returncode(nick, cmd, time_limit=None, memory_limit=None):
+    cmd = _replace_paths_with_strings(cmd)
     print_call_settings(nick, cmd, None, time_limit, memory_limit)
 
     preexec_fn = _get_preexec_function(time_limit, memory_limit)

driver/cleanup.py

@@ -1,17 +1,5 @@
-from itertools import count
-import os
-
-def _try_remove(f):
-    try:
-        os.remove(f)
-    except OSError:
-        return False
-    return True
+from .plan_manager import PlanManager
 
 def cleanup_temporary_files(args):
-    _try_remove(args.sas_file)
-    _try_remove(args.plan_file)
-
-    for i in count(1):
-        if not _try_remove("%s.%s" % (args.plan_file, i)):
-            break
+    args.sas_file.unlink(missing_ok=True)
+    PlanManager(args.plan_file).delete_existing_plans()

driver/main.py

@@ -1,5 +1,4 @@
 import logging
-import os
 import sys
 
 from . import aliases
@@ -16,7 +15,7 @@ def main():
     logging.basicConfig(level=getattr(logging, args.log_level.upper()),
                         format="%(levelname)-8s %(message)s",
                         stream=sys.stdout)
-    logging.debug("processed args: %s" % args)
+    logging.debug(f"processed args: {args}")
 
     if args.version:
         print(__version__)
@@ -40,16 +39,16 @@ def main():
         elif component == "search":
             (exitcode, continue_execution) = run_components.run_search(args)
             if not args.keep_sas_file:
-                print("Remove intermediate file {}".format(args.sas_file))
-                os.remove(args.sas_file)
+                print(f"Remove intermediate file {args.sas_file}")
+                args.sas_file.unlink()
         elif component == "validate":
             (exitcode, continue_execution) = run_components.run_validate(args)
         else:
-            assert False, "Error: unhandled component: {}".format(component)
-        print("{component} exit code: {exitcode}".format(**locals()))
+            assert False, f"Error: unhandled component: {component}"
+        print(f"{component} exit code: {exitcode}")
         print()
         if not continue_execution:
-            print("Driver aborting after {}".format(component))
+            print(f"Driver aborting after {component}")
             break
 
     try:

driver/plan_manager.py

@@ -1,28 +1,24 @@
 import itertools
-import os
-import os.path
+from pathlib import Path
 import re
 
 from . import returncodes
 
 
-_PLAN_INFO_REGEX = re.compile(r"; cost = (\d+) \((unit cost|general cost)\)\n")
+_PLAN_INFO_REGEX = re.compile(r"; cost = (\d+) \((unit cost|general cost)\)")
 
 
-def _read_last_line(filename):
-    line = None
-    with open(filename) as input_file:
-        for line in input_file:
-            pass
-    return line
+def _read_last_line(path: Path):
+    lines = path.read_text().splitlines()
+    if lines:
+        return lines[-1]
 
 
-def _parse_plan(plan_filename):
+def _parse_plan(plan_path: Path):
     """Parse a plan file and return a pair (cost, problem_type)
     summarizing the salient information. Return (None, None) for
     incomplete plans."""
-
-    last_line = _read_last_line(plan_filename) or ""
+    last_line = _read_last_line(plan_path) or ""
     match = _PLAN_INFO_REGEX.match(last_line)
     if match:
         return int(match.group(1)), match.group(2)
@@ -31,7 +27,7 @@ def _parse_plan(plan_filename):
 
 
 class PlanManager:
-    def __init__(self, plan_prefix, portfolio_bound=None, single_plan=False):
+    def __init__(self, plan_prefix: Path, portfolio_bound=None, single_plan=False):
         self._plan_prefix = plan_prefix
         self._plan_costs = []
         self._problem_type = None
@@ -73,23 +69,22 @@ def process_new_plans(self):
         Read newly generated plans and store the relevant information.
         If the last plan file is incomplete, delete it.
         """
-
         had_incomplete_plan = False
         for counter in itertools.count(self.get_plan_counter() + 1):
-            plan_filename = self._get_plan_file(counter)
+            plan_path = self._get_plan_path(counter)
             def bogus_plan(msg):
-                returncodes.exit_with_driver_critical_error("%s: %s" % (plan_filename, msg))
-            if not os.path.exists(plan_filename):
+                returncodes.exit_with_driver_critical_error(f"{str(plan_path)}: {msg}")
+            if not plan_path.exists():
                 break
             if had_incomplete_plan:
                 bogus_plan("plan found after incomplete plan")
-            cost, problem_type = _parse_plan(plan_filename)
+            cost, problem_type = _parse_plan(plan_path)
             if cost is None:
                 had_incomplete_plan = True
-                print("%s is incomplete. Deleted the file." % plan_filename)
-                os.remove(plan_filename)
+                print(f"{plan_path} is incomplete. Deleted the file.")
+                plan_path.unlink()
             else:
-                print("plan manager: found new plan with cost %d" % cost)
+                print(f"plan manager: found new plan with cost {cost}")
                 if self._problem_type is None:
                     # This is the first plan we found.
                     self._problem_type = problem_type
@@ -103,20 +98,20 @@ def bogus_plan(msg):
 
     def get_existing_plans(self):
         """Yield all plans that match the given plan prefix."""
-        if os.path.exists(self._plan_prefix):
+        if self._plan_prefix.exists():
             yield self._plan_prefix
 
         for counter in itertools.count(start=1):
-            plan_filename = self._get_plan_file(counter)
-            if os.path.exists(plan_filename):
-                yield plan_filename
+            plan_path = self._get_plan_path(counter)
+            if plan_path.exists():
+                yield plan_path
             else:
                 break
 
     def delete_existing_plans(self):
         """Delete all plans that match the given plan prefix."""
         for plan in self.get_existing_plans():
-            os.remove(plan)
+            plan.unlink()
 
-    def _get_plan_file(self, number):
-        return "%s.%d" % (self._plan_prefix, number)
+    def _get_plan_path(self, number):
+        return Path(f"{(self._plan_prefix)}.{number}")