1"""Pipe bytes, strings, or string iterators through Graphviz ``dot``."""
2
3import typing
4
5from .. import _tools
6
7from . import dot_command
8from . import execute
9
10__all__ = ['pipe', 'pipe_string',
11 'pipe_lines', 'pipe_lines_string']
12
13
14@_tools.deprecate_positional_args(supported_number=3)
15def pipe(engine: str, format: str, data: bytes,
16 renderer: typing.Optional[str] = None,
17 formatter: typing.Optional[str] = None,
18 neato_no_op: typing.Union[bool, int, None] = None,
19 quiet: bool = False) -> bytes:
20 """Return ``data`` (``bytes``) piped through ``engine`` into ``format`` as ``bytes``.
21
22 Args:
23 engine: Layout engine for rendering (``'dot'``, ``'neato'``, ...).
24 format: Output format for rendering (``'pdf'``, ``'png'``, ...).
25 data: Binary (encoded) DOT source bytes to render.
26 renderer: Output renderer (``'cairo'``, ``'gd'``, ...).
27 formatter: Output formatter (``'cairo'``, ``'gd'``, ...).
28 neato_no_op: Neato layout engine no-op flag.
29 quiet: Suppress ``stderr`` output from the layout subprocess.
30
31 Returns:
32 Binary (encoded) stdout of the layout command.
33
34 Raises:
35 ValueError: If ``engine``, ``format``, ``renderer``, or ``formatter``
36 are unknown.
37 graphviz.RequiredArgumentError: If ``formatter`` is given
38 but ``renderer`` is None.
39 graphviz.ExecutableNotFound: If the Graphviz ``dot`` executable
40 is not found.
41 graphviz.CalledProcessError: If the returncode (exit status)
42 of the rendering ``dot`` subprocess is non-zero.
43
44 Example:
45 >>> doctest_mark_exe()
46 >>> import graphviz
47 >>> graphviz.pipe('dot', 'svg', b'graph { hello -- world }')[:14]
48 b'<?xml version='
49
50 Note:
51 The layout command is started from the current directory.
52 """
53 cmd = dot_command.command(engine, format,
54 renderer=renderer,
55 formatter=formatter,
56 neato_no_op=neato_no_op)
57 kwargs = {'input': data}
58
59 proc = execute.run_check(cmd, capture_output=True, quiet=quiet, **kwargs)
60 return proc.stdout
61
62
63def pipe_string(engine: str, format: str, input_string: str, *,
64 encoding: str,
65 renderer: typing.Optional[str] = None,
66 formatter: typing.Optional[str] = None,
67 neato_no_op: typing.Union[bool, int, None] = None,
68 quiet: bool = False) -> str:
69 """Return ``input_string`` piped through ``engine`` into ``format`` as string.
70
71 Args:
72 engine: Layout engine for rendering (``'dot'``, ``'neato'``, ...).
73 format: Output format for rendering (``'pdf'``, ``'png'``, ...).
74 input_string: Binary (encoded) DOT source bytes to render.
75 encoding: Encoding to en/decode subprocess stdin and stdout (required).
76 renderer: Output renderer (``'cairo'``, ``'gd'``, ...).
77 formatter: Output formatter (``'cairo'``, ``'gd'``, ...).
78 neato_no_op: Neato layout engine no-op flag.
79 quiet: Suppress ``stderr`` output from the layout subprocess.
80
81 Returns:
82 Decoded stdout of the layout command.
83
84 Raises:
85 ValueError: If ``engine``, ``format``, ``renderer``, or ``formatter``
86 are unknown.
87 graphviz.RequiredArgumentError: If ``formatter`` is given
88 but ``renderer`` is None.
89 graphviz.ExecutableNotFound: If the Graphviz ``dot`` executable
90 is not found.
91 graphviz.CalledProcessError: If the returncode (exit status)
92 of the rendering ``dot`` subprocess is non-zero.
93
94 Example:
95 >>> doctest_mark_exe()
96 >>> import graphviz
97 >>> graphviz.pipe_string('dot', 'svg', 'graph { spam }',
98 ... encoding='ascii')[:14]
99 '<?xml version='
100
101 Note:
102 The layout command is started from the current directory.
103 """
104 cmd = dot_command.command(engine, format,
105 renderer=renderer,
106 formatter=formatter,
107 neato_no_op=neato_no_op)
108 kwargs = {'input': input_string, 'encoding': encoding}
109
110 proc = execute.run_check(cmd, capture_output=True, quiet=quiet, **kwargs)
111 return proc.stdout
112
113
114def pipe_lines(engine: str, format: str, input_lines: typing.Iterator[str], *,
115 input_encoding: str,
116 renderer: typing.Optional[str] = None,
117 formatter: typing.Optional[str] = None,
118 neato_no_op: typing.Union[bool, int, None] = None,
119 quiet: bool = False) -> bytes:
120 r"""Return ``input_lines`` piped through ``engine`` into ``format`` as ``bytes``.
121
122 Args:
123 engine: Layout engine for rendering (``'dot'``, ``'neato'``, ...).
124 format: Output format for rendering (``'pdf'``, ``'png'``, ...).
125 input_lines: DOT source lines to render (including final newline).
126 input_encoding: Encode input_lines for subprocess stdin (required).
127 renderer: Output renderer (``'cairo'``, ``'gd'``, ...).
128 formatter: Output formatter (``'cairo'``, ``'gd'``, ...).
129 neato_no_op: Neato layout engine no-op flag.
130 quiet: Suppress ``stderr`` output from the layout subprocess.
131
132 Returns:
133 Binary stdout of the layout command.
134
135 Raises:
136 ValueError: If ``engine``, ``format``, ``renderer``, or ``formatter``
137 are unknown.
138 graphviz.RequiredArgumentError: If ``formatter`` is given
139 but ``renderer`` is None.
140 graphviz.ExecutableNotFound: If the Graphviz ``dot`` executable
141 is not found.
142 graphviz.CalledProcessError: If the returncode (exit status)
143 of the rendering ``dot`` subprocess is non-zero.
144
145 Example:
146 >>> doctest_mark_exe()
147 >>> import graphviz
148 >>> graphviz.pipe_lines('dot', 'svg', iter(['graph { spam }\n']),
149 ... input_encoding='ascii')[:14]
150 b'<?xml version='
151
152 Note:
153 The layout command is started from the current directory.
154 """
155 cmd = dot_command.command(engine, format,
156 renderer=renderer,
157 formatter=formatter,
158 neato_no_op=neato_no_op)
159 kwargs = {'input_lines': (line.encode(input_encoding) for line in input_lines)}
160
161 proc = execute.run_check(cmd, capture_output=True, quiet=quiet, **kwargs)
162 return proc.stdout
163
164
165def pipe_lines_string(engine: str, format: str, input_lines: typing.Iterator[str], *,
166 encoding: str,
167 renderer: typing.Optional[str] = None,
168 formatter: typing.Optional[str] = None,
169 neato_no_op: typing.Union[bool, int, None] = None,
170 quiet: bool = False) -> str:
171 r"""Return ``input_lines`` piped through ``engine`` into ``format`` as string.
172
173 Args:
174 engine: Layout engine for rendering (``'dot'``, ``'neato'``, ...).
175 format: Output format for rendering (``'pdf'``, ``'png'``, ...).
176 input_lines: DOT source lines to render (including final newline).
177 encoding: Encoding to en/decode subprocess stdin and stdout (required).
178 renderer: Output renderer (``'cairo'``, ``'gd'``, ...).
179 formatter: Output formatter (``'cairo'``, ``'gd'``, ...).
180 neato_no_op: Neato layout engine no-op flag.
181 quiet: Suppress ``stderr`` output from the layout subprocess.
182
183 Returns:
184 Decoded stdout of the layout command.
185
186 Raises:
187 ValueError: If ``engine``, ``format``, ``renderer``, or ``formatter``
188 are unknown.
189 graphviz.RequiredArgumentError: If ``formatter`` is given
190 but ``renderer`` is None.
191 graphviz.ExecutableNotFound: If the Graphviz ``dot`` executable
192 is not found.
193 graphviz.CalledProcessError: If the returncode (exit status)
194 of the rendering ``dot`` subprocess is non-zero.
195
196 Example:
197 >>> doctest_mark_exe()
198 >>> import graphviz
199 >>> graphviz.pipe_lines_string('dot', 'svg', iter(['graph { spam }\n']),
200 ... encoding='ascii')[:14]
201 '<?xml version='
202
203 Note:
204 The layout command is started from the current directory.
205 """
206 cmd = dot_command.command(engine, format,
207 renderer=renderer,
208 formatter=formatter,
209 neato_no_op=neato_no_op)
210 kwargs = {'input_lines': input_lines, 'encoding': encoding}
211
212 proc = execute.run_check(cmd, capture_output=True, quiet=quiet, **kwargs)
213 return proc.stdout