Merge pull request #2 from ClassicMiniDIY/port-321-stdin-stdout

Add stdin/stdout streaming (port of upstream PR wireviz#321)

Author: SomethingNew71

src/wireviz/DataClasses.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 
+import sys
 from dataclasses import InitVar, dataclass, field
 from enum import Enum, auto
 from pathlib import Path
@@ -343,8 +344,8 @@ def __post_init__(self) -> None:
             self.gauge = g
 
             if self.gauge_unit is not None:
-                print(
-                    f"Warning: Cable {self.name} gauge_unit={self.gauge_unit} is ignored because its gauge contains {u}"
+                sys.stderr.write(
+                    f"Warning: Cable {self.name} gauge_unit={self.gauge_unit} is ignored because its gauge contains {u}\n"
                 )
             if u.upper() == "AWG":
                 self.gauge_unit = u.upper()
@@ -367,8 +368,8 @@ def __post_init__(self) -> None:
                 )
             self.length = L
             if self.length_unit is not None:
-                print(
-                    f"Warning: Cable {self.name} length_unit={self.length_unit} is ignored because its length contains {u}"
+                sys.stderr.write(
+                    f"Warning: Cable {self.name} length_unit={self.length_unit} is ignored because its length contains {u}\n"
                 )
             self.length_unit = u
         elif not isinstance(self.length, (int, float)):

src/wireviz/Harness.py

@@ -1,11 +1,13 @@
 # -*- coding: utf-8 -*-
 
+import base64
 import re
+import sys
 from collections import Counter
 from dataclasses import dataclass
 from itertools import zip_longest
 from pathlib import Path
-from typing import Any, List, Union
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 from graphviz import Graph
 from wireviz import APP_NAME, APP_URL, __version__, wv_colors
@@ -19,7 +21,7 @@
     Side,
     Tweak,
 )
-from wireviz.svgembed import embed_svg_images, embed_svg_images_file
+from wireviz.svgembed import embed_svg_images
 from wireviz.wv_bom import (
     HEADER_MPN,
     HEADER_PN,
@@ -603,12 +605,12 @@ def typecheck(name: str, value: Any, expect: type) -> None:
                                     f'( +)?{attr}=("[^"]*"|[^] ]*)(?(1)| *)', "", entry
                                 )
                                 if n_subs < 1:
-                                    print(
-                                        f"Harness.create_graph() warning: {attr} not found in {keyword}!"
+                                    sys.stderr.write(
+                                        f"Harness.create_graph() warning: {attr} not found in {keyword}!\n"
                                     )
                                 elif n_subs > 1:
-                                    print(
-                                        f"Harness.create_graph() warning: {attr} removed {n_subs} times in {keyword}!"
+                                    sys.stderr.write(
+                                        f"Harness.create_graph() warning: {attr} removed {n_subs} times in {keyword}!\n"
                                     )
                                 continue
 
@@ -622,8 +624,8 @@ def typecheck(name: str, value: Any, expect: type) -> None:
                                 # If attr not found, then append it
                                 entry = re.sub(r"\]$", f" {attr}={value}]", entry)
                             elif n_subs > 1:
-                                print(
-                                    f"Harness.create_graph() warning: {attr} overridden {n_subs} times in {keyword}!"
+                                sys.stderr.write(
+                                    f"Harness.create_graph() warning: {attr} overridden {n_subs} times in {keyword}!\n"
                                 )
 
                         dot.body[i] = entry
@@ -670,54 +672,125 @@ def svg(self):  # TODO?: Verify xml encoding="utf-8" in SVG?
 
     def output(
         self,
-        filename: (str, Path),
+        filename: Optional[Union[str, Path]],
+        fmt: Union[str, Tuple[str, ...], List[str]] = ("html", "png", "svg", "tsv"),
         view: bool = False,
         cleanup: bool = True,
-        fmt: tuple = ("html", "png", "svg", "tsv"),
+        output_dir: Optional[Union[str, Path]] = None,
+        output_name: Optional[str] = None,
     ) -> None:
-        # graphical output
-        graph = self.graph
-        svg_already_exists = Path(
-            f"{filename}.svg"
-        ).exists()  # if SVG already exists, do not delete later
-        # graphical output
-        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
-                # TODO: prevent rendering SVG twice when both SVG and HTML are specified
-                graph.format = f
-                graph.render(filename=_filename, view=view, cleanup=cleanup)
-        # embed images into SVG output
-        if "svg" in fmt or "html" in fmt:
-            embed_svg_images_file(f"{filename}.tmp.svg")
-        # GraphViz output
-        if "gv" in fmt:
-            graph.save(filename=f"{filename}.gv")
-        # BOM output
-        bomlist = bom_list(self.bom())
-        if "tsv" in fmt:
-            file_write_text(f"{filename}.bom.tsv", tuplelist2tsv(bomlist))
+        """Render the harness in the requested formats.
+
+        When ``filename`` is a path, each requested format is written to
+        ``{filename}.{ext}`` (with ``.bom.tsv`` for the BOM). When
+        ``filename`` is None, exactly one format must be requested and
+        its bytes/text are written to stdout — supports piping the CLI
+        into other tools.
+        """
+        if isinstance(fmt, str):
+            fmt = (fmt,)
+        outputs: Dict[str, Union[str, bytes]] = self._render(
+            fmt,
+            output_dir=output_dir,
+            output_name=output_name,
+        )
+
         if "csv" in fmt:
-            # TODO: implement CSV output (preferrably using CSV library)
-            print("CSV output is not yet supported")
-        # HTML output
-        if "html" in fmt:
-            generate_html_output(
-                filename, bomlist, self.metadata, self.options, self.source_path
-            )
-        # PDF output
+            # TODO: implement CSV output (preferably using CSV library)
+            sys.stderr.write("CSV output is not yet supported\n")
         if "pdf" in fmt:
             # TODO: implement PDF output
-            print("PDF output is not yet supported")
-        # 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()
-        elif "svg" in fmt:
-            Path(f"{filename}.tmp.svg").replace(f"{filename}.svg")
+            sys.stderr.write("PDF output is not yet supported\n")
+
+        if filename is None:
+            # stdout mode — emit each rendered format in the user-requested order
+            for f in fmt:
+                content = outputs.get(f)
+                if content is None:
+                    continue
+                if isinstance(content, (bytes, bytearray)):
+                    sys.stdout.buffer.write(content)
+                else:
+                    sys.stdout.write(content)
+            return
+
+        suffix_map = {"tsv": "bom.tsv"}
+        for f, content in outputs.items():
+            ext = suffix_map.get(f, f)
+            out_path = f"{filename}.{ext}"
+            if isinstance(content, (bytes, bytearray)):
+                Path(out_path).write_bytes(content)
+            else:
+                file_write_text(out_path, content)
+
+    def _render(
+        self,
+        fmt: Union[str, Tuple[str, ...], List[str]],
+        output_dir: Optional[Union[str, Path]] = None,
+        output_name: Optional[str] = None,
+    ) -> Dict[str, Union[str, bytes]]:
+        """Produce in-memory representations of each requested format.
+
+        Pipes graphviz once per binary output rather than via ``render()``
+        + temporary files so the caller can write files OR pipe to stdout
+        without the SVG-file roundtrip the previous implementation used.
+        """
+        if isinstance(fmt, str):
+            fmt = (fmt,)
+        graph = self.graph
+        outputs: Dict[str, Union[str, bytes]] = {}
+
+        svg_str: Optional[str] = None
+        if "svg" in fmt or "html" in fmt:
+            # Resolve relative <image src=...> references against the YAML
+            # source's directory when known; fall back to cwd. (In practice
+            # wireviz.parse() rewrites relative image paths to absolute
+            # during YAML parse, so this base path only matters for SVG
+            # produced from already-rendered Harness objects or when a
+            # tweak injects a post-parse relative path.)
+            if self.source_path is not None and str(self.source_path) != "-":
+                base_path: Path = Path(self.source_path).parent
+            else:
+                base_path = Path.cwd()
+            svg_str = embed_svg_images(
+                graph.pipe(format="svg").decode("utf-8"), base_path
+            )
+            if "svg" in fmt:
+                outputs["svg"] = svg_str
+
+        png_bytes: Optional[bytes] = None
+        if "png" in fmt:
+            png_bytes = graph.pipe(format="png")
+            outputs["png"] = png_bytes
+
+        if "gv" in fmt:
+            outputs["gv"] = graph.source
+
+        if "tsv" in fmt or "html" in fmt:
+            bomlist = bom_list(self.bom())
+            if "tsv" in fmt:
+                outputs["tsv"] = tuplelist2tsv(bomlist)
+            if "html" in fmt:
+                # Inline PNG as base64 in the HTML only when the PNG was
+                # rendered in this same call; otherwise let the template
+                # fall back to reading {output_dir}/{output_name}.png.
+                png_b64 = (
+                    f"data:image/png;base64, {base64.b64encode(png_bytes).decode('utf-8')}"
+                    if png_bytes is not None
+                    else None
+                )
+                outputs["html"] = generate_html_output(
+                    svg_str,
+                    bomlist,
+                    self.metadata,
+                    self.options,
+                    output_dir=output_dir,
+                    output_name=output_name,
+                    png_b64=png_b64,
+                    source_path=self.source_path,
+                )
+
+        return outputs
 
     def bom(self):
         if not self._bom:

src/wireviz/svgembed.py

@@ -54,13 +54,3 @@ def get_mime_subtype(filename: Union[str, Path]) -> str:
     return mime_subtype
 
 
-def embed_svg_images_file(
-    filename_in: Union[str, Path], overwrite: bool = True
-) -> None:
-    filename_in = Path(filename_in).resolve()
-    filename_out = filename_in.with_suffix(".b64.svg")
-    filename_out.write_text(  # TODO?: Verify xml encoding="utf-8" in SVG?
-        embed_svg_images(filename_in.read_text(), filename_in.parent)
-    )  # TODO: Use encoding="utf-8" in both read_text() and write_text()
-    if overwrite:
-        filename_out.replace(filename_in)

src/wireviz/wireviz.py

@@ -94,11 +94,18 @@ def parse(
         raise TypeError(
             f"Expected a dict as top-level YAML input, but got: {type(yaml_data)}"
         )
-    if output_formats:
+    write_to_stdout = (
+        output_formats and (str(output_dir) == "-" or str(output_name) == "-")
+    )
+    if output_formats and not write_to_stdout:
         # need to write data to file, determine output directory and filename
         output_dir = _get_output_dir(yaml_file, output_dir)
         output_name = _get_output_name(yaml_file, output_name)
         output_file = output_dir / output_name
+    else:
+        output_dir = None
+        output_name = None
+        output_file = None
 
     if yaml_file:
         # if reading from file, ensure that input file's parent directory is included in image_paths
@@ -397,10 +404,10 @@ def alternate_type():  # flip between connector and cable/arrow
     used_components = set(designators_and_templates.values())
     forgotten_components = [c for c in proposed_components if not c in used_components]
     if len(forgotten_components) > 0:
-        print(
-            "Warning: The following components are not referenced in any connection set:"
+        sys.stderr.write(
+            "Warning: The following components are not referenced in any connection set:\n"
         )
-        print(", ".join(forgotten_components))
+        sys.stderr.write(", ".join(forgotten_components) + "\n")
 
     # harness population completed =============================================
 
@@ -409,7 +416,20 @@ def alternate_type():  # flip between connector and cable/arrow
             harness.add_bom_item(line)
 
     if output_formats:
-        harness.output(filename=output_file, fmt=output_formats, view=False)
+        if write_to_stdout:
+            if len(output_formats) != 1:
+                raise ValueError(
+                    "Exactly one output format must be specified when writing to stdout."
+                )
+            harness.output(filename=None, fmt=output_formats, view=False)
+        else:
+            harness.output(
+                filename=output_file,
+                fmt=output_formats,
+                view=False,
+                output_dir=output_dir,
+                output_name=output_name,
+            )
 
     if return_types:
         returns = []
@@ -447,8 +467,8 @@ def _get_yaml_data_and_path(inp: Union[str, Path, Dict]) -> (Dict, Path):
             from errno import EINVAL, ENAMETOOLONG
 
             if type(e) is OSError and e.errno not in (EINVAL, ENAMETOOLONG, None):
-                print(
-                    f"OSError(errno={e.errno}) in Python {sys.version} at {platform.platform()}"
+                sys.stderr.write(
+                    f"OSError(errno={e.errno}) in Python {sys.version} at {platform.platform()}\n"
                 )
                 raise e
             # file does not exist; assume inp is a YAML string
@@ -485,7 +505,7 @@ def _get_output_name(input_file: Path, default_output_name: Path) -> str:
 
 
 def main():
-    print("When running from the command line, please use wv_cli.py instead.")
+    sys.stderr.write("When running from the command line, please use wv_cli.py instead.\n")
 
 
 if __name__ == "__main__":