Source code for pubsub.utils.topictreeprinter

"""
Output various aspects of topic tree to string or file.

:copyright: Copyright since 2006 by Oliver Schoenborn, all rights reserved.
:license: BSD, see LICENSE_BSD_Simple.txt for details.
"""

from typing import TextIO
from textwrap import TextWrapper

from ..core.topictreetraverser import (ITopicTreeVisitor, TopicTreeTraverser)


class TopicTreePrinter(ITopicTreeVisitor):
    """
    Example topic tree visitor that prints a prettified representation
    of topic tree by doing a depth-first traversal of topic tree and
    print information at each (topic) node of tree. Extra info to be
    printed is specified via the 'extra' kwarg. Its value must be a
    list of characters, the order determines output order:
    - D: print description of topic
    - a: print kwarg names only
    - A: print topic kwargs and their description
    - L: print listeners currently subscribed to topic

    E.g. TopicTreePrinter(extra='LaDA') would print, for each topic,
    the list of subscribed listeners, the topic's list of kwargs, the
    topic description, and the description for each kwarg,

        >>> Topic "delTopic"
           >> Listeners:
              > listener1_2880 (from yourModule)
              > listener2_3450 (from yourModule)
           >> Names of Message arguments:
              > arg1
              > arg2
           >> Description: whenever a topic is deleted
           >> Descriptions of Message arguments:
              > arg1: (required) its description
              > arg2: some other description

    """

    allowedExtras = frozenset('DAaL')  # must NOT change
    ALL_TOPICS_NAME = 'ALL_TOPICS'  # output for name of 'all topics' topic

    def __init__(self, extra=None, width: int = 70, indentStep: int = 4,
                 bulletTopic: str = '\\--', bulletTopicItem: str = '|==', bulletTopicArg: str = '-',
                 fileObj: TextIO = None):
        """
        Topic tree printer will print listeners for each topic only
        if printListeners is True. The width will be used to limit
        the width of text output, while indentStep is the number of
        spaces added each time the text is indented further. The
        three bullet parameters define the strings used for each
        item (topic, topic items, and kwargs).
        """
        self.__contentMeth = dict(
            D=self.__printTopicDescription,
            A=self.__printTopicArgsAll,
            a=self.__printTopicArgNames,
            L=self.__printTopicListeners)
        assert self.allowedExtras == set(self.__contentMeth.keys())
        import sys
        self.__destination = fileObj or sys.stdout
        self.__output = []

        self.__content = extra or ''
        unknownSel = set(self.__content) - self.allowedExtras
        if unknownSel:
            msg = 'These extra chars not known: %s' % ','.join(unknownSel)
            raise ValueError(msg)

        self.__width = width
        self.__wrapper = TextWrapper(width)
        self.__indent = 0
        self.__indentStep = indentStep
        self.__topicsBullet = bulletTopic
        self.__topicItemsBullet = bulletTopicItem
        self.__topicArgsBullet = bulletTopicArg

    def getOutput(self):
        return '\n'.join(self.__output)

    def _doneTraversal(self):
        if self.__destination is not None:
            self.__destination.write(self.getOutput())

    def _onTopic(self, topicObj):
        """This gets called for each topic. Print as per specified content."""

        # topic name
        self.__wrapper.width = self.__width
        indent = self.__indent
        if topicObj.isAll():
            topicName = self.ALL_TOPICS_NAME
        else:
            topicName = topicObj.getNodeName()
        head = '%s Topic "%s"' % (self.__topicsBullet, topicName)
        self.__output.append(self.__formatDefn(indent, head))
        indent += self.__indentStep

        # each extra content (assume constructor verified that chars are valid)
        for item in self.__content:
            function = self.__contentMeth[item]
            function(indent, topicObj)

    def _startChildren(self):
        """Increase the indent"""
        self.__indent += self.__indentStep

    def _endChildren(self):
        """Decrease the indent"""
        self.__indent -= self.__indentStep

    def __formatDefn(self, indent, item, defn='', sep=': '):
        """
        Print a definition: a block of text at a certain indent,
        has item name, and an optional definition separated from
        item by sep.
        """
        if defn:
            prefix = '%s%s%s' % (' ' * indent, item, sep)
            self.__wrapper.initial_indent = prefix
            self.__wrapper.subsequent_indent = ' ' * (indent + self.__indentStep)
            return self.__wrapper.fill(defn)
        else:
            return '%s%s' % (' ' * indent, item)

    def __printTopicDescription(self, indent, topicObj):
        # topic description
        defn = '%s Description' % self.__topicItemsBullet
        self.__output.append(
            self.__formatDefn(indent, defn, topicObj.getDescription()))

    def __printTopicArgsAll(self, indent, topicObj, desc=True):
        # topic kwargs
        args = topicObj.getArgDescriptions()
        if args:
            # required, optional, complete = topicObj.getArgs()
            headName = 'Names of Message arguments:'
            if desc:
                headName = 'Descriptions of message arguments:'
            head = '%s %s' % (self.__topicItemsBullet, headName)
            self.__output.append(self.__formatDefn(indent, head))
            tmpIndent = indent + self.__indentStep
            required = topicObj.getArgs()[0]
            for key, arg in args.items():  # iter in 3, list in 2 ok
                if not desc:
                    arg = ''
                elif key in required:
                    arg = '(required) %s' % arg
                msg = '%s %s' % (self.__topicArgsBullet, key)
                self.__output.append(self.__formatDefn(tmpIndent, msg, arg))

    def __printTopicArgNames(self, indent, topicObj):
        self.__printTopicArgsAll(indent, topicObj, False)

    def __printTopicListeners(self, indent, topicObj):
        if topicObj.hasListeners():
            item = '%s Listeners:' % self.__topicItemsBullet
            self.__output.append(self.__formatDefn(indent, item))
            tmpIndent = indent + self.__indentStep
            for listener in topicObj.getListenersIter():
                item = '%s %s (from %s)' % (self.__topicArgsBullet, listener.name(), listener.module())
                self.__output.append(self.__formatDefn(tmpIndent, item))


[docs] def printTreeDocs(rootTopic=None, topicMgr=None, **kwargs): """ Print out the topic tree to a file (or file-like object like a StringIO), starting at rootTopic. If root topic should be root of whole tree, get it from pub.getDefaultTopicTreeRoot(). The treeVisitor is an instance of pub.TopicTreeTraverser. Printing the tree docs would normally involve this:: from pubsub import pub from pubsub.utils.topictreeprinter import TopicTreePrinter traverser = pub.TopicTreeTraverser( TopicTreePrinter(**kwargs) ) traverser.traverse( pub.getDefaultTopicTreeRoot() ) With printTreeDocs, it looks like this:: from pubsub import pub from pubsub.utils import printTreeDocs printTreeDocs() The kwargs are the same as for TopicTreePrinter constructor: extra(None), width(70), indentStep(4), bulletTopic, bulletTopicItem, bulletTopicArg, fileObj(stdout). If fileObj not given, stdout is used. """ if rootTopic is None: if topicMgr is None: from .. import pub topicMgr = pub.getDefaultTopicMgr() rootTopic = topicMgr.getRootAllTopics() printer = TopicTreePrinter(**kwargs) traverser = TopicTreeTraverser(printer) traverser.traverse(rootTopic)