treewide: use pathlib and force path arguments

src/wireviz/tools/build_examples.py

@@ -2,17 +2,17 @@
 # -*- coding: utf-8 -*-
 
 import argparse
-import click
 import os
 import sys
 from pathlib import Path
 
+import click
+
 script_path = Path(__file__).absolute()
 sys.path.insert(0, str(script_path.parent.parent.parent))  # to find wireviz module
 
 from wireviz import APP_NAME, __version__
 from wireviz.wv_cli import cli
-from wireviz.wv_utils import open_file_append, open_file_read, open_file_write
 
 base_dir = script_path.parent.parent.parent.parent
 readme = "readme.md"
@@ -60,22 +60,24 @@ def build_generated(groupkeys):
         if build_readme:
             include_readme = "md" in groups[key][readme]
             include_source = "yml" in groups[key][readme]
-            with open_file_write(path / readme) as out:
+            with (path / readme).open("w") as out:
                 out.write(f'# {groups[key]["title"]}\n\n')
         # collect and iterate input YAML files
         for yaml_file in collect_filenames("Building", key, input_extensions):
             try:
-                res = cli(['--format', 'ghpst', str(yaml_file)])
+                res = cli(["--formats", "ghpst", str(yaml_file)])
             except BaseException as e:
-                if str(e) != '0' and not isinstance(e, (click.ClickException, SystemExit)):
+                if str(e) != "0" and not isinstance(
+                    e, (click.ClickException, SystemExit)
+                ):
                     raise
 
             if build_readme:
                 i = "".join(filter(str.isdigit, yaml_file.stem))
 
-                with open_file_append(path / readme) as out:
+                with (path / readme).open("a") as out:
                     if include_readme:
-                        with open_file_read(yaml_file.with_suffix(".md")) as info:
+                        with yaml_file.with_suffix(".md").open("r") as info:
                             for line in info:
                                 out.write(line.replace("## ", f"## {i} - "))
                             out.write("\n\n")
@@ -83,7 +85,7 @@ def build_generated(groupkeys):
                         out.write(f"## Example {i}\n")
 
                     if include_source:
-                        with open_file_read(yaml_file) as src:
+                        with yaml_file.open("r") as src:
                             out.write("```yaml\n")
                             for line in src:
                                 out.write(line)

src/wireviz/wireviz.py

@@ -2,33 +2,27 @@
 # -*- coding: utf-8 -*-
 
 import logging
-import sys
 from pathlib import Path
 from typing import Any, Dict, List, Tuple, Union
 
 import yaml
 
-if __name__ == "__main__":
-    sys.path.insert(0, str(Path(__file__).parent.parent))  # add src/wireviz to PATH
-
 from wireviz.wv_dataclasses import AUTOGENERATED_PREFIX, Metadata, Options, Tweak
 from wireviz.wv_harness import Harness
 from wireviz.wv_utils import (
     expand,
     get_single_key_and_value,
     is_arrow,
-    open_file_read,
     smart_file_resolve,
 )
 
 
 def parse(
-    inp: Union[Path, str, Dict],
+    inp: List[Path],
     return_types: Union[None, str, Tuple[str]] = None,
     output_formats: Union[None, str, Tuple[str]] = None,
-    output_dir: Union[str, Path] = None,
+    output_dir: Path = None,
     output_name: Union[None, str] = None,
-    image_paths: Union[Path, str, List] = [],
     extra_metadata: Dict = {},
     shared_bom: Dict = {},
 ) -> Any:
@@ -37,9 +31,7 @@ def parse(
     and outputs the result as one or more files and/or as a function return value
 
     Accepted inputs:
-        * A Path object or a path-like string pointing to a YAML source file to parse
+        * A List of Path object pointing to a YAML source file to parse
-        * A string containing the YAML data to parse
-        * A Python Dict containing the pre-parsed YAML data
 
     Supported return types:
         * "png":     the diagram as raw PNG data
@@ -56,7 +48,7 @@ def parse(
         * "tsv":  the BOM, as a tab-separated text file
 
     Args:
-        inp (Path | str | Dict):
+        inp:
             The input to be parsed (see above for accepted inputs).
         return_types (optional):
             One of the supported return types (see above), or a tuple of multiple return types.
@@ -71,10 +63,6 @@ def parse(
             The name to use for the generated output files (without extension).
             Defaults to inp's file name (without extension).
             Required parameter if inp is not a path.
-        image_paths (Path | str | List, optional):
-            Paths to use when resolving any image paths included in the data.
-            Note: If inp is a path to a YAML file,
-            its parent directory will automatically be included in the list.
         extra_metadata (Dict, optional):
             Any metadata to add to the template.
             Normally, this should contain programmatic metadata
@@ -82,28 +70,19 @@ def parse(
     Returns:
         Depending on the return_types parameter, may return:
         * None
-        * one of the following, or a tuple containing two or more of the following:
+        * A dict of {return_type: data}
-            * PNG data
-            * SVG data
-            * a Harness object
     """
 
     if not output_formats and not return_types:
         raise Exception("No output formats or return types specified")
 
-    yaml_data, yaml_file = _get_yaml_data_and_path(inp)
+    yaml_file = inp[-1]
-    if output_formats:
+    yaml_data_str = "\n".join(f.open("r").read() for f in inp)
-        # need to write data to file, determine output directory and filename
+    yaml_data = yaml.safe_load(yaml_data_str)
-        output_dir = _get_output_dir(yaml_file, output_dir)
+    image_paths = {f.parent for f in inp if f.parent.is_dir()}
-        output_name = _get_output_name(yaml_file, output_name)
-        output_file = output_dir / output_name
 
-    if yaml_file:
+    output_dir = yaml_file.parent if not output_dir else output_dir
-        # if reading from file, ensure that input file's parent directory
+    output_name = yaml_file.stem if not output_name else output_name
-        # is included in image_paths
-        default_image_path = yaml_file.parent.resolve()
-        if not default_image_path in [Path(x).resolve() for x in image_paths]:
-            image_paths.append(default_image_path)
 
     # define variables =========================================================
     # containers for parsed component data and connection sets
@@ -124,7 +103,7 @@ def parse(
     autogenerated_designators = {}
 
     if "title" not in harness.metadata:
-        harness.metadata["title"] = Path(yaml_file).stem if yaml_file else ""
+        harness.metadata["title"] = yaml_file.stem
 
     # add items
     # parse YAML input file ====================================================
@@ -140,8 +119,8 @@ def parse(
                         # an image file with a relative path.
                         image = attribs.get("image")
                         if isinstance(image, dict):
-                            image_path = image["src"]
+                            image_path = Path(image["src"])
-                            if image_path and not Path(image_path).is_absolute():
+                            if image_path and not image_path.is_absolute():
                                 # resolve relative image path
                                 image["src"] = smart_file_resolve(
                                     image_path, image_paths
@@ -384,68 +363,25 @@ def parse(
     harness.populate_bom()
 
     if output_formats:
-        harness.output(filename=output_file, fmt=output_formats, view=False)
+        harness.output(
+            filename=output_dir / output_name, fmt=output_formats, view=False
+        )
 
     if return_types:
-        returns = []
         if isinstance(return_types, str):  # only one return type speficied
             return_types = [return_types]
 
         return_types = [t.lower() for t in return_types]
 
+        returns = {}
         for rt in return_types:
             if rt == "png":
-                returns.append(harness.png)
+                returns["png"] = harness.png
             if rt == "svg":
-                returns.append(harness.svg)
+                returns["svg"] = harness.svg
             if rt == "harness":
-                returns.append(harness)
+                returns["harness"] = harness
+            if rt == "shared_bom":
+                returns["shared_bom"] = harness.shared_bom
 
-        return tuple(returns) if len(returns) != 1 else returns[0]
+        return returns
-
-
-def _get_yaml_data_and_path(inp: Union[str, Path, Dict]) -> (Dict, Path):
-    # determine whether inp is a file path, a YAML string, or a Dict
-    if not isinstance(inp, Dict):  # received a str or a Path
-        if isinstance(inp, Path) or (isinstance(inp, str) and not "\n" in inp):
-            yaml_path = Path(inp).expanduser().resolve(strict=True)
-            yaml_str = open_file_read(yaml_path).read()
-        else:
-            yaml_path = None
-            yaml_str = inp
-        yaml_data = yaml.safe_load(yaml_str)
-    else:
-        # received a Dict, use as-is
-        yaml_data = inp
-        yaml_path = None
-    return yaml_data, yaml_path
-
-
-def _get_output_dir(input_file: Path, default_output_dir: Path) -> Path:
-    if default_output_dir:  # user-specified output directory
-        output_dir = Path(default_output_dir)
-    else:  # auto-determine appropriate output directory
-        if input_file:  # input comes from a file; place output in same directory
-            output_dir = input_file.parent
-        else:  # input comes from str or Dict; fall back to cwd
-            output_dir = Path.cwd()
-    return output_dir.resolve()
-
-
-def _get_output_name(input_file: Path, default_output_name: Path) -> str:
-    if default_output_name:  # user-specified output name
-        output_name = default_output_name
-    else:  # auto-determine appropriate output name
-        if input_file:  # input comes from a file; use same file stem
-            output_name = input_file.stem
-        else:  # input comes from str or Dict; no fallback available
-            raise Exception("No output file name provided")
-    return output_name
-
-
-def main():
-    print("When running from the command line, please use wv_cli.py instead.")
-
-
-if __name__ == "__main__":
-    main()

src/wireviz/wv_cli.py

@@ -11,7 +11,8 @@ if __name__ == "__main__":
 
 import wireviz.wireviz as wv
 from wireviz import APP_NAME, __version__
-from wireviz.wv_utils import open_file_read
+from wireviz.wv_bom import bom_list
+from wireviz.wv_utils import bom2tsv
 
 format_codes = {
     "c": "csv",
@@ -24,19 +25,28 @@ format_codes = {
     "b": "shared_bom",
 }
 
-
 epilog = (
-    "The -f or --format option accepts a string containing one or more of the "
+    "The -f or --formats option accepts a string containing one or more of the "
     "following characters to specify which file types to output:\n"
     + f", ".join([f"{key} ({value.upper()})" for key, value in format_codes.items()])
 )
 
 
 @click.command(epilog=epilog, no_args_is_help=True)
-@click.argument("file", nargs=-1)
+@click.argument(
+    "files",
+    type=click.Path(
+        exists=True,
+        readable=True,
+        dir_okay=False,
+        path_type=Path,
+    ),
+    nargs=-1,
+    required=True,
+)
 @click.option(
     "-f",
-    "--format",
+    "--formats",
     default="hpst",
     type=str,
     show_default=True,
@@ -47,14 +57,25 @@ epilog = (
     "--prepend",
     default=[],
     multiple=True,
-    type=Path,
+    type=click.Path(
+        exists=True,
+        readable=True,
+        file_okay=True,
+        path_type=Path,
+    ),
     help="YAML file to prepend to the input file (optional).",
 )
 @click.option(
     "-o",
     "--output-dir",
     default=None,
-    type=Path,
+    type=click.Path(
+        exists=True,
+        readable=True,
+        file_okay=False,
+        dir_okay=True,
+        path_type=Path,
+    ),
     help="Directory to use for output files, if different from input file directory.",
 )
 @click.option(
@@ -74,100 +95,55 @@ epilog = (
     default=False,
     help=f"Output {APP_NAME} version and exit.",
 )
-def cli(file, format, prepend, output_dir, output_name, version):
+def cli(files, formats, prepend, output_dir, output_name, version):
     """
     Parses the provided FILE and generates the specified outputs.
     """
-    print()  # blank line before execution
-    print(f"{APP_NAME} {__version__}")
     if version:
+        print(f"{APP_NAME} {__version__}")
         return  # print version number only and exit
 
-    # get list of files
+    _output_dir = files[0].parent if not output_dir else output_dir
-    try:
-        _ = iter(file)
-    except TypeError:
-        filepaths = [file]
-    else:
-        filepaths = list(file)
 
     # determine output formats
-    output_formats = []
+    output_formats = {format_codes[f] for f in formats if f in format_codes}
-    for code in format:
-        if code in format_codes:
-            output_formats.append(format_codes[code])
-        else:
-            raise Exception(f"Unknown output format: {code}")
-    output_formats = tuple(sorted(set(output_formats)))
-    output_formats_str = (
-        f'[{"|".join(output_formats)}]'
-        if len(output_formats) > 1
-        else output_formats[0]
-    )
-
-    # check prepend file
-    if len(prepend) > 0:
-        prepend_input = ""
-        for prepend_file in prepend:
-            prepend_file = Path(prepend_file)
-            if not prepend_file.exists():
-                raise Exception(f"File does not exist:\n{prepend_file}")
-            if not prepend_file.is_file():
-                raise Exception(f"Path is not a file:\n{prepend_file}")
-            print("Prepend file:", prepend_file)
-
-            prepend_input += open_file_read(prepend_file).read() + "\n"
-    else:
-        prepend_input = ""
 
     harness = None
     shared_bom = {}
     sheet_current = 1
     # run WireVIz on each input file
-    for file in filepaths:
+    for _file in files:
-        file = Path(file)
+        _output_name = _file.stem if not output_name else output_name
-        if not file.exists():
+
-            raise Exception(f"File does not exist:\n{file}")
+        print("Input file:  ", _file)
-        if not file.is_file():
+        print(
-            raise Exception(f"Path is not a file:\n{file}")
+            "Output file: ",
+            f"{_output_dir / _output_name}.[{'|'.join(output_formats)}]",
+        )
 
         extra_metadata = {}
-        extra_metadata["sheet_name"] = file.stem
+        extra_metadata["sheet_name"] = _output_name.upper()
-        extra_metadata["sheet_total"] = len(filepaths)
+        extra_metadata["sheet_total"] = len(files)
         extra_metadata["sheet_current"] = sheet_current
         sheet_current += 1
 
-        # file_out = file.with_suffix("") if not output_file else output_file
+        file_dir = _file.parent
-        _output_dir = file.parent if not output_dir else output_dir
-        _output_name = file.stem if not output_name else output_name
 
-        print("Input file:  ", file)
+        ret = wv.parse(
-        print(
+            prepend + (_file,),
-            "Output file: ", f"{Path(_output_dir / _output_name)}.{output_formats_str}"
+            return_types=("shared_bom"),
-        )
+            output_formats=output_formats,
-
-        yaml_input = open_file_read(file).read()
-        file_dir = file.parent
-
-        yaml_input = prepend_input + yaml_input
-        image_paths = {file_dir}
-        for p in prepend:
-            image_paths.add(Path(p).parent)
-
-        harness = wv.parse(
-            yaml_input,
-            return_types=("harness"),
-            output_formats=[f for f in output_formats if f != "shared_bom"],
             output_dir=_output_dir,
             output_name=_output_name,
-            image_paths=list(image_paths),
             extra_metadata=extra_metadata,
             shared_bom=shared_bom,
         )
+        shared_bom = ret["shared_bom"]
+
     if "shared_bom" in output_formats:
-        _output_dir = file.parent if not output_dir else output_dir
+        shared_bomlist = bom_list(shared_bom)
-        harness.output(str(Path(_output_dir) / "shared_bom"), fmt="shared_bom")
+        shared_bom_tsv = bom2tsv(shared_bomlist)
-        shared_bom = harness.shared_bom
+        (_output_dir / "shared_bom").with_suffix(".tsv").open("w").write(shared_bom_tsv)
 
     print()  # blank line after execution
 

src/wireviz/wv_harness.py

@@ -36,7 +36,7 @@ from wireviz.wv_graphviz import (
     set_dot_basics,
 )
 from wireviz.wv_output import embed_svg_images_file, generate_html_output
-from wireviz.wv_utils import bom2tsv, open_file_write
+from wireviz.wv_utils import bom2tsv
 
 
 @dataclass
@@ -407,35 +407,32 @@ class Harness:
     ) -> None:
         # graphical output
         graph = self.graph
+
+        rendered = set()
         for f in fmt:
             if f in ("png", "svg", "html"):
                 if f == "html":  # if HTML format is specified,
                     f = "svg"  # generate SVG for embedding into HTML
                 # SVG file will be renamed/deleted later
-                _filename = f"{filename}.tmp" if f == "svg" else filename
+                if f in rendered:
-                # TODO: prevent rendering SVG twice when both SVG and HTML are specified
+                    continue
                 graph.format = f
-                graph.render(filename=_filename, view=view, cleanup=cleanup)
+                graph.render(filename=filename, view=view, cleanup=cleanup)
+                rendered.add(f)
         # embed images into SVG output
         if "svg" in fmt or "html" in fmt:
-            embed_svg_images_file(f"{filename}.tmp.svg")
+            embed_svg_images_file(filename.with_suffix(".svg"))
         # GraphViz output
         if "gv" in fmt:
-            graph.save(filename=f"{filename}.gv")
+            graph.save(filename=filename.with_suffix(".gv"))
         # BOM output
         bomlist = bom_list(self.bom)
-        # bomlist = [[]]
         if "tsv" in fmt:
-            tsv = bom2tsv(bomlist)
+            bom_tsv = bom2tsv(bomlist)
-            open_file_write(f"{filename}.tsv").write(tsv)
+            filename.with_suffix(".tsv").open("w").write(bom_tsv)
         if "csv" in fmt:
             # TODO: implement CSV output (preferrably using CSV library)
             print("CSV output is not yet supported")
-        if "shared_bom" in fmt:
-            shared_bomlist = bom_list(self.shared_bom)
-            shared_bom_tsv = bom2tsv(shared_bomlist)
-            open_file_write(f"{filename}.tsv").write(shared_bom_tsv)
-
         # HTML output
         if "html" in fmt:
             generate_html_output(filename, bomlist, self.metadata, self.options)
@@ -446,6 +443,4 @@ class Harness:
         # delete SVG if not needed
         if "html" in fmt and not "svg" in fmt:
             # SVG file was just needed to generate HTML
-            Path(f"{filename}.tmp.svg").unlink()
+            filename.with_suffix(".svg").unlink()
-        elif "svg" in fmt:
-            Path(f"{filename}.tmp.svg").replace(f"{filename}.svg")

src/wireviz/wv_output.py

@@ -10,7 +10,6 @@ import jinja2
 import wireviz  # for doing wireviz.__file__
 from wireviz import APP_NAME, APP_URL, __version__
 from wireviz.wv_dataclasses import Metadata, Options
-from wireviz.wv_utils import open_file_read, open_file_write
 
 mime_subtype_replacements = {"jpg": "jpeg", "tif": "tiff"}
 
@@ -69,21 +68,20 @@ def get_template_html(template_name):
 
 
 def generate_html_output(
-    filename: Union[str, Path],
+    filename: Path,
     bom: List[List[str]],
     metadata: Metadata,
     options: Options,
 ):
     print("Generating html output")
     template_name = metadata.get("template", {}).get("name", "simple")
-    page_template = get_template_html(template_name)
 
     # embed SVG diagram
-    with open_file_read(f"{filename}.tmp.svg") as file:
+    with filename.with_suffix(".svg").open("r") as f:
         svgdata = re.sub(
             "^<[?]xml [^?>]*[?]>[^<]*<!DOCTYPE [^>]*>",
             "<!-- XML and DOCTYPE declarations from SVG file removed -->",
-            file.read(),
+            f.read(),
             1,
         )
 
@@ -175,5 +173,9 @@ def generate_html_output(
     titleblock_template = get_template_html("titleblock")
     replacements["titleblock"] = titleblock_template.render(replacements)
 
+    # generate page template
+    page_template = get_template_html(template_name)
     page_rendered = page_template.render(replacements)
-    open_file_write(f"{filename}.html").write(page_rendered)
+
+    # save generated file
+    filename.with_suffix(".html").open("w").write(page_rendered)

src/wireviz/wv_utils.py

@@ -117,19 +117,6 @@ def clean_whitespace(inp):
     return " ".join(inp.split()).replace(" ,", ",") if isinstance(inp, str) else inp
 
 
-def open_file_read(filename):
-    # TODO: Intelligently determine encoding
-    return open(filename, "r", encoding="UTF-8")
-
-
-def open_file_write(filename):
-    return open(filename, "w", encoding="UTF-8")
-
-
-def open_file_append(filename):
-    return open(filename, "a", encoding="UTF-8")
-
-
 def is_arrow(inp):
     """
     Matches strings of one or multiple `-` or `=` (but not mixed)
@@ -159,25 +146,20 @@ def aspect_ratio(image_src):
     return 1  # Assume 1:1 when unable to read actual image size
 
 
-def smart_file_resolve(filename: str, possible_paths: (str, List[str])) -> Path:
+def smart_file_resolve(filename: Path, possible_paths: (Path, List[Path])) -> Path:
-    if not isinstance(possible_paths, List):
+    if isinstance(possible_paths, Path) or isinstance(possible_paths, str):
         possible_paths = [possible_paths]
-    filename = Path(filename)
     if filename.is_absolute():
         if filename.exists():
             return filename
         else:
             raise Exception(f"{filename} does not exist.")
     else:  # search all possible paths in decreasing order of precedence
-        possible_paths = [
+        for path in possible_paths:
-            Path(path).resolve() for path in possible_paths if path is not None
+            combined_path = (path / filename).resolve()
-        ]
+            if combined_path.exists():
-        for possible_path in possible_paths:
+                return combined_path
-            resolved_path = (possible_path / filename).resolve()
+        raise Exception(
-            if resolved_path.exists():
+            f"{filename} was not found in any of the following locations: \n"
-                return resolved_path
+            + "\n".join(str(p) for p in possible_paths)
-        else:
+        )
-            raise Exception(
-                f"{filename} was not found in any of the following locations: \n"
-                + "\n".join([str(x) for x in possible_paths])
-            )