Source code for graphviz.backend

"""Execute rendering subprocesses and open files in viewer."""

import errno
import logging
import os
import platform
import re
import subprocess
import sys
import typing

from . import tools

__all__ = ['render', 'pipe', 'unflatten', 'version', 'view',
           'ENGINES', 'FORMATS', 'RENDERERS', 'FORMATTERS',
           'ExecutableNotFound', 'RequiredArgumentError']

ENGINES = {  # http://www.graphviz.org/pdf/dot.1.pdf
    'dot', 'neato', 'twopi', 'circo', 'fdp', 'sfdp', 'patchwork', 'osage',
}

FORMATS = {  # http://www.graphviz.org/doc/info/output.html
    'bmp',
    'canon', 'dot', 'gv', 'xdot', 'xdot1.2', 'xdot1.4',
    'cgimage',
    'cmap',
    'eps',
    'exr',
    'fig',
    'gd', 'gd2',
    'gif',
    'gtk',
    'ico',
    'imap', 'cmapx',
    'imap_np', 'cmapx_np',
    'ismap',
    'jp2',
    'jpg', 'jpeg', 'jpe',
    'json', 'json0', 'dot_json', 'xdot_json',  # Graphviz 2.40
    'pct', 'pict',
    'pdf',
    'pic',
    'plain', 'plain-ext',
    'png',
    'pov',
    'ps',
    'ps2',
    'psd',
    'sgi',
    'svg', 'svgz',
    'tga',
    'tif', 'tiff',
    'tk',
    'vml', 'vmlz',
    'vrml',
    'wbmp',
    'webp',
    'xlib',
    'x11',
}

RENDERERS = {  # $ dot -T:
    'cairo',
    'dot',
    'fig',
    'gd',
    'gdiplus',
    'map',
    'pic',
    'pov',
    'ps',
    'svg',
    'tk',
    'vml',
    'vrml',
    'xdot',
}

FORMATTERS = {'cairo', 'core', 'gd', 'gdiplus', 'gdwbmp', 'xlib'}

ENCODING = 'utf-8'

PLATFORM = platform.system().lower()


log = logging.getLogger(__name__)


[docs]class ExecutableNotFound(RuntimeError): """Exception raised if the Graphviz executable is not found.""" _msg = ('failed to execute {!r}, ' 'make sure the Graphviz executables are on your systems\' PATH') def __init__(self, args): super().__init__(self._msg.format(*args))
[docs]class RequiredArgumentError(Exception): """Exception raised if a required argument is missing (i.e. ``None``)."""
class CalledProcessError(subprocess.CalledProcessError): def __str__(self): s = super().__str__() return f'{s} [stderr: {self.stderr!r}]' def command(engine: str, format_: str, filepath=None, renderer: typing.Optional[str] = None, formatter: typing.Optional[str] = None): """Return args list for ``subprocess.Popen`` and name of the rendered file.""" if formatter is not None and renderer is None: raise RequiredArgumentError('formatter given without renderer') if engine not in ENGINES: raise ValueError(f'unknown engine: {engine!r}') if format_ not in FORMATS: raise ValueError(f'unknown format: {format_!r}') if renderer is not None and renderer not in RENDERERS: raise ValueError(f'unknown renderer: {renderer!r}') if formatter is not None and formatter not in FORMATTERS: raise ValueError(f'unknown formatter: {formatter!r}') output_format = [f for f in (format_, renderer, formatter) if f is not None] cmd = ['dot', '-K%s' % engine, '-T%s' % ':'.join(output_format)] if filepath is None: rendered = None else: cmd.extend(['-O', filepath]) suffix = '.'.join(reversed(output_format)) rendered = f'{filepath}.{suffix}' return cmd, rendered if PLATFORM == 'windows': # pragma: no cover def get_startupinfo(): """Return subprocess.STARTUPINFO instance hiding the console window.""" startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW startupinfo.wShowWindow = subprocess.SW_HIDE return startupinfo else: def get_startupinfo(): """Return None for startupinfo argument of ``subprocess.Popen``.""" return None def run(cmd, input=None, capture_output: bool = False, check: bool = False, encoding: typing.Optional[str] = None, quiet: bool = False, **kwargs) -> typing.Tuple: """Run the command described by cmd and return its ``(stdout, stderr)`` tuple.""" log.debug('run %r', cmd) if input is not None: kwargs['stdin'] = subprocess.PIPE if encoding is not None: input = input.encode(encoding) if capture_output: # Python 3.6 compat kwargs['stdout'] = kwargs['stderr'] = subprocess.PIPE try: proc = subprocess.Popen(cmd, startupinfo=get_startupinfo(), **kwargs) except OSError as e: if e.errno == errno.ENOENT: raise ExecutableNotFound(cmd) from e else: raise out, err = proc.communicate(input) if not quiet and err: err_encoding = sys.stderr.encoding or sys.getdefaultencoding() sys.stderr.write(err.decode(err_encoding)) sys.stderr.flush() if encoding is not None: if out is not None: out = out.decode(encoding) if err is not None: err = err.decode(encoding) if check and proc.returncode: raise CalledProcessError(proc.returncode, cmd, output=out, stderr=err) return out, err
[docs]def render(engine: str, format: str, filepath, renderer: typing.Optional[str] = None, formatter: typing.Optional[str] = None, quiet: bool = False) -> str: """Render file with Graphviz ``engine`` into ``format``, return result filename. Args: engine: Layout commmand for rendering (``'dot'``, ``'neato'``, ...). format: Output format for rendering (``'pdf'``, ``'png'``, ...). filepath: Path to the DOT source file to render. renderer: Output renderer (``'cairo'``, ``'gd'``, ...). formatter: Output formatter (``'cairo'``, ``'gd'``, ...). quiet: Suppress ``stderr`` output from the layout subprocess. Returns: The (possibly relative) path of the rendered file. Raises: ValueError: If ``engine``, ``format``, ``renderer``, or ``formatter`` are not known. graphviz.RequiredArgumentError: If ``formatter`` is given but ``renderer`` is None. graphviz.ExecutableNotFound: If the Graphviz executable is not found. subprocess.CalledProcessError: If the exit status is non-zero. Note: The layout command is started from the directory of ``filepath``, so that references to external files (e.g. ``[image=...]``) can be given as paths relative to the DOT source file. """ dirname, filename = os.path.split(filepath) del filepath cmd, rendered = command(engine, format, filename, renderer, formatter) if dirname: cwd = dirname rendered = os.path.join(dirname, rendered) else: cwd = None run(cmd, capture_output=True, cwd=cwd, check=True, quiet=quiet) return rendered
[docs]def pipe(engine: str, format: str, data: bytes, renderer: typing.Optional[str] = None, formatter: typing.Optional[str] = None, quiet: bool = False) -> bytes: """Return ``data`` piped through Graphviz ``engine`` into ``format``. Args: engine: Layout commmand for rendering (``'dot'``, ``'neato'``, ...). format: Output format for rendering (``'pdf'``, ``'png'``, ...). data: Binary (encoded) DOT source string to render. renderer: Output renderer (``'cairo'``, ``'gd'``, ...). formatter: Output formatter (``'cairo'``, ``'gd'``, ...). quiet: Suppress ``stderr`` output from the layout subprocess. Returns: Binary (encoded) stdout of the layout command. Raises: ValueError: If ``engine``, ``format``, ``renderer``, or ``formatter`` are not known. graphviz.RequiredArgumentError: If ``formatter`` is given but no ``renderer``. graphviz.ExecutableNotFound: If the Graphviz executable is not found. subprocess.CalledProcessError: If the exit status is non-zero. Example: >>> import graphviz >>> graphviz.pipe('dot', 'svg', b'graph { hello -- world }') b'<?xml version=...' """ cmd, _ = command(engine, format, None, renderer, formatter) out, _ = run(cmd, input=data, capture_output=True, check=True, quiet=quiet) return out
[docs]def unflatten(source: str, stagger: typing.Optional[int] = None, fanout: bool = False, chain: typing.Optional[int] = None, encoding: str = ENCODING) -> str: """Return DOT ``source`` piped through Graphviz *unflatten* preprocessor. Args: source: DOT source to process (improve layout aspect ratio). stagger: Stagger the minimum length of leaf edges between 1 and this small integer. fanout: Fanout nodes with indegree = outdegree = 1 when staggering (requires ``stagger``). chain: Form disconnected nodes into chains of up to this many nodes. encoding: Encoding to encode unflatten stdin and decode its stdout. Returns: Decoded stdout of the Graphviz unflatten command. Raises: graphviz.RequiredArgumentError: If ``fanout`` is given but no ``stagger``. graphviz.ExecutableNotFound: If the Graphviz unflatten executable is not found. subprocess.CalledProcessError: If the exit status is non-zero. See also: https://www.graphviz.org/pdf/unflatten.1.pdf """ if fanout and stagger is None: raise RequiredArgumentError('fanout given without stagger') cmd = ['unflatten'] if stagger is not None: cmd += ['-l', str(stagger)] if fanout: cmd.append('-f') if chain is not None: cmd += ['-c', str(chain)] out, _ = run(cmd, input=source, capture_output=True, encoding=encoding) return out
[docs]def version() -> typing.Tuple[int, ...]: """Return the version number tuple from the ``stderr`` output of ``dot -V``. Returns: Two, three, or four ``int`` version ``tuple``. Raises: graphviz.ExecutableNotFound: If the Graphviz executable is not found. subprocess.CalledProcessError: If the exit status is non-zero. RuntimeError: If the output cannot be parsed into a version number. Example: >>> import graphviz >>> graphviz.version() (...) Note: Ignores the ``~dev.<YYYYmmdd.HHMM>`` portion of development versions. See also: Graphviz Release version entry format: https://gitlab.com/graphviz/graphviz/-/blob/f94e91ba819cef51a4b9dcb2d76153684d06a913/gen_version.py#L17-20 """ cmd = ['dot', '-V'] out, _ = run(cmd, check=True, encoding='ascii', stdout=subprocess.PIPE, stderr=subprocess.STDOUT) ma = re.search(r'graphviz version' r' ' r'(\d+)\.(\d+)' r'(?:\.(\d+)' r'(?:' r'~dev\.\d{8}\.\d{4}' r'|' r'\.(\d+)' r')?' r')?' r' ', out) if ma is None: raise RuntimeError(f'cannot parse {cmd!r} output: {out!r}') return tuple(int(d) for d in ma.groups() if d is not None)
[docs]def view(filepath, quiet: bool = False) -> None: """Open filepath with its default viewing application (platform-specific). Args: filepath: Path to the file to open in viewer. quiet: Suppress ``stderr`` output from the viewer process (ineffective on Windows). Returns: ``None`` Raises: RuntimeError: If the current platform is not supported. Note: There is no option to wait for the application to close, and no way to retrieve the application's exit status. """ try: view_func = getattr(view, PLATFORM) except AttributeError: raise RuntimeError(f'platform {PLATFORM!r} not supported') view_func(filepath, quiet=quiet)
@tools.attach(view, 'darwin') def view_darwin(filepath, *, quiet: bool) -> None: """Open filepath with its default application (mac).""" cmd = ['open', filepath] log.debug('view: %r', cmd) kwargs = {'stderr': subprocess.DEVNULL} if quiet else {} subprocess.Popen(cmd, **kwargs) @tools.attach(view, 'linux') @tools.attach(view, 'freebsd') def view_unixoid(filepath, *, quiet: bool) -> None: """Open filepath in the user's preferred application (linux, freebsd).""" cmd = ['xdg-open', filepath] log.debug('view: %r', cmd) kwargs = {'stderr': subprocess.DEVNULL} if quiet else {} subprocess.Popen(cmd, **kwargs) @tools.attach(view, 'windows') def view_windows(filepath, *, quiet: bool) -> None: """Start filepath with its associated application (windows).""" # TODO: implement quiet=True filepath = os.path.normpath(filepath) log.debug('view: %r', filepath) os.startfile(filepath)