"""
:copyright: Copyright since 2006 by Oliver Schoenborn, all rights reserved.
:license: BSD, see LICENSE_BSD_Simple.txt for details.
"""
import os, re, inspect, io
from textwrap import TextWrapper, dedent
import sys
from typing import Tuple, List, Sequence, Mapping, Dict, Callable, Any, Optional, Union, TextIO
try:
from importlib.util import cache_from_source
except ImportError:
from imp import cache_from_source
try:
# raise ImportError
from textwrap import indent
except ImportError:
def indent(text, prefix):
new_text = ''.join(((prefix + line) if line.strip() else '\n') for line in text.splitlines(True))
return new_text
from .topicargspec import (
topicArgsFromCallable,
ArgSpecGiven,
ArgsDocs
)
from .topictreetraverser import TopicTreeTraverser
from .topicexc import UnrecognizedSourceFormatError
from .topicobj import Topic
[docs]class ITopicDefnProvider:
"""
All topic definition providers added via pub.addTopicDefnProvider()
must have this interface. Derived classes must override the getDefn(),
getTreeDoc() and topicNames() methods.
"""
def getDefn(self, topicNameTuple: Sequence[str]) -> Tuple[str, ArgSpecGiven]:
"""
Must return a pair (string, ArgSpecGiven) for given topic.
The first item is a description for topic, the second item
contains the message data specification (MDS). Note topic name
is in tuple format ('a', 'b', 'c') rather than 'a.b.c'.
"""
msg = 'Must return (string, ArgSpecGiven), or (None, None)'
raise NotImplementedError(msg)
def topicNames(self) -> List[str]:
"""
Return an iterator over topic names available from this provider.
Note that the topic names should be in tuple rather than dotted-string
format so as to be compatible with getDefn().
"""
msg = 'Must return a list of topic names available from this provider'
raise NotImplementedError(msg)
def getTreeDoc(self) -> str:
"""Get the docstring for the topic tree."""
msg = 'Must return documentation string for root topic (tree)'
raise NotImplementedError(msg)
def __iter__(self):
"""Same as self.topicNames(), do NOT override."""
return iter(self.topicNames())
# name of method in class name assumed to represent topic's listener signature
# which will get checked against topic's Message Data Specification (MDS)
SPEC_METHOD_NAME = 'msgDataSpec'
[docs]class ITopicDefnDeserializer:
"""
Interface class for all topic definition de-serializers that can be
accepted by TopicDefnProvider. A deserializer
creates a topic tree from something such as file, module, or string.
"""
class TopicDefn:
"""
Encapsulate date for a topic definition. Used by
getNextTopic().
"""
def __init__(self, nameTuple: Sequence[str], description: str, argsDocs: ArgsDocs, required: Sequence[str]):
self.nameTuple = nameTuple
self.description = description
self.argsDocs = argsDocs
self.required = required
def isComplete(self):
return (self.description is not None) and (self.argsDocs is not None)
def getTreeDoc(self) -> str:
"""Get the docstring for the topic tree."""
raise NotImplementedError
def getNextTopic(self) -> TopicDefn:
"""
Get the next topic definition available from the data. The return
must be an instance of TopicDefn. Must return None when no topics
are left.
"""
raise NotImplementedError
def doneIter(self):
"""
Called automatically by TopicDefnProvider once
it considers the iteration completed. Override this only if
deserializer needs to take action, such as closing a file.
"""
pass
def resetIter(self):
"""
Called by the TopicDefnProvider if it needs to
restart the topic iteration. Override this only if special action needed,
such as resetting a file pointer to beginning of file.
"""
pass
class TopicDefnDeserialClass(ITopicDefnDeserializer):
"""
Convert a nested class tree as a topic definition tree. Format: the class
name is the topic name, its doc string is its description. The topic's
message data specification is determined by inspecting a class method called
the same as SPEC_METHOD_NAME. The doc string of that method is parsed to
extract the description for each message data.
"""
def __init__(self, pyClassObj: type = None):
"""
If pyClassObj is given, it is an object that contains nested
classes defining root topics; the root topics contain nested
classes defining subtopics; etc. The class name is the topic
name, and the class docstring is the topic documentation string.
"""
self.__rootTopics = []
self.__iterStarted = False
self.__nextTopic = iter(self.__rootTopics)
self.__rootDoc = None
if pyClassObj is not None:
self.__rootDoc = pyClassObj.__doc__
topicClasses = self.__getTopicClasses(pyClassObj)
for topicName, pyClassObj in topicClasses:
self.__addDefnFromClassObj(pyClassObj)
def getTreeDoc(self) -> str:
return self.__rootDoc
def getNextTopic(self) -> ITopicDefnDeserializer.TopicDefn:
self.__iterStarted = True
try:
topicNameTuple, topicClassObj = next(self.__nextTopic)
except StopIteration:
return None
# ok get the info from class
if hasattr(topicClassObj, SPEC_METHOD_NAME):
protoListener = getattr(topicClassObj, SPEC_METHOD_NAME)
argsDocs, required = topicArgsFromCallable(protoListener)
if protoListener.__doc__:
self.__setArgsDocsFromProtoDocs(argsDocs, protoListener.__doc__)
else:
# assume definition is implicitly that listener has no args
argsDocs = {}
required = ()
desc = None
if topicClassObj.__doc__:
desc = dedent(topicClassObj.__doc__)
return self.TopicDefn(topicNameTuple, desc, argsDocs, required)
def resetIter(self):
self.__iterStarted = False
self.__nextTopic = iter(self.__rootTopics)
def getDefinedTopics(self) -> List[str]:
return [nt for (nt, defn) in self.__rootTopics]
def __addDefnFromClassObj(self, pyClassObj: type):
"""
Extract a topic definition from a Python class: topic name,
docstring, and MDS, and docstring for each message data.
The class name is the topic name, assumed to be a root topic, and
descends recursively into nested classes to define subtopic etc.
"""
if self.__iterStarted:
raise RuntimeError('addDefnFromClassObj must be called before iteration started!')
parentNameTuple = (pyClassObj.__name__,)
if pyClassObj.__doc__ is not None:
self.__rootTopics.append((parentNameTuple, pyClassObj))
if self.__rootDoc is None:
self.__rootDoc = pyClassObj.__doc__
self.__findTopics(pyClassObj, parentNameTuple)
# iterator is now out of sync, so reset it; obviously this would
# screw up getNextTopic which is why we had to test for self.__iterStarted
self.__nextTopic = iter(self.__rootTopics)
def __findTopics(self, pyClassObj: type, parentNameTuple: Sequence[str]):
assert not self.__iterStarted
assert parentNameTuple
assert pyClassObj.__name__ == parentNameTuple[-1]
topicClasses = self.__getTopicClasses(pyClassObj, parentNameTuple)
pyClassObj._topicNameStr = '.'.join(parentNameTuple)
# make sure to update rootTopics BEFORE we recurse, so that toplevel
# topics come first in the list
for parentNameTuple2, topicClassObj in topicClasses:
# we only keep track of topics that are documented, so that
# multiple providers can co-exist without having to duplicate
# information
if topicClassObj.__doc__ is not None:
self.__rootTopics.append((parentNameTuple2, topicClassObj))
# now can find its subtopics
self.__findTopics(topicClassObj, parentNameTuple2)
def __getTopicClasses(self, pyClassObj: type, parentNameTuple: Sequence[str] = ()):
"""Returns a list of pairs, (topicNameTuple, memberClassObj)"""
memberNames = dir(pyClassObj)
topicClasses = []
for memberName in memberNames:
if memberName.startswith('_'):
continue # ignore special and non-public methods
member = getattr(pyClassObj, memberName)
if inspect.isclass(member):
topicNameTuple = parentNameTuple + (memberName,)
topicClasses.append((topicNameTuple, member))
return topicClasses
def __setArgsDocsFromProtoDocs(self, argsDocs: ArgsDocs, protoDocs: str):
PAT_ITEM_STR = r'\A-\s*' # hyphen and any number of blanks
PAT_ARG_NAME = r'(?P<argName>\w*)'
PAT_DOC_STR = r'(?P<doc1>.*)'
PAT_BLANK = r'\s*'
PAT_ITEM_SEP = r':'
argNamePat = re.compile(
PAT_ITEM_STR + PAT_ARG_NAME + PAT_BLANK + PAT_ITEM_SEP
+ PAT_BLANK + PAT_DOC_STR)
protoDocs = dedent(protoDocs)
lines = protoDocs.splitlines()
argName = None
namesFound = []
for line in lines:
match = argNamePat.match(line)
if match:
argName = match.group('argName')
namesFound.append(argName)
argsDocs[argName] = [match.group('doc1')]
elif argName:
argsDocs[argName].append(line)
for name in namesFound:
argsDocs[name] = '\n'.join(argsDocs[name])
class TopicDefnDeserialModule(ITopicDefnDeserializer):
"""
Deserialize a module containing Python source code defining a topic tree.
This loads the module and gives it to an instance of TopicDefnDeserialClass.
"""
def __init__(self, moduleName: str, searchPath: Sequence[str] = None):
"""
Load the given named module, searched for in searchPath or, if not
specified, in sys.path. Give it to a TopicDefnDeserialClass.
"""
if searchPath is not None:
old_path = sys.path
sys.path = searchPath
try:
from importlib import import_module
module = import_module(moduleName)
finally:
if searchPath is not None:
sys.path = old_path
self.__classDeserial = TopicDefnDeserialClass(module)
def getTreeDoc(self) -> str:
return self.__classDeserial.getTreeDoc()
def getNextTopic(self) -> ITopicDefnDeserializer.TopicDefn:
return self.__classDeserial.getNextTopic()
def doneIter(self):
self.__classDeserial.doneIter()
def resetIter(self):
self.__classDeserial.resetIter()
def getDefinedTopics(self) -> List[str]:
return self.__classDeserial.getDefinedTopics()
class TopicDefnDeserialString(ITopicDefnDeserializer):
"""
Deserialize a string containing Python source code defining a topic tree.
The string has the same format as expected by TopicDefnDeserialModule.
"""
def __init__(self, source: str):
"""
This just saves the string into a temporary file created in
os.getcwd(), and the rest is delegated to TopicDefnDeserialModule.
The temporary file (module -- as well as its byte-compiled
version) will be deleted when the doneIter() method is called.
"""
source = "class TopicTree:\n" + indent(dedent(source), ' ' * 4)
namespace = {}
exec(source, namespace)
self.__clsDeserial = TopicDefnDeserialClass(namespace['TopicTree'])
def getTreeDoc(self) -> str:
return self.__clsDeserial.getTreeDoc()
def getNextTopic(self) -> ITopicDefnDeserializer.TopicDefn:
return self.__clsDeserial.getNextTopic()
def doneIter(self):
self.__clsDeserial.doneIter()
def resetIter(self):
self.__clsDeserial.resetIter()
def getDefinedTopics(self) -> List[str]:
return self.__clsDeserial.getDefinedTopics()
TOPIC_TREE_FROM_MODULE = 'module'
TOPIC_TREE_FROM_STRING = 'string'
TOPIC_TREE_FROM_CLASS = 'class'
[docs]class TopicDefnProvider(ITopicDefnProvider):
"""
Default implementation of the ITopicDefnProvider API. This
implementation accepts several formats for the topic tree
source data and delegates to a registered ITopicDefnDeserializer
that converts source data into topic definitions.
This provider is instantiated automatically by
``pub.addTopicDefnProvider(source, format)``
when source is *not* an ITopicDefnProvider.
Additional de-serializers can be registered via registerTypeForImport().
"""
_typeRegistry = {}
def __init__(self, source: Any, format: str, **providerKwargs):
"""
Find the correct de-serializer class from registry for the given
format; instantiate it with given source and providerKwargs; get
all available topic definitions.
"""
if format not in self._typeRegistry:
raise UnrecognizedSourceFormatError()
providerClassObj = self._typeRegistry[format]
provider = providerClassObj(source, **providerKwargs)
self.__topicDefns = {}
self.__treeDocs = provider.getTreeDoc()
try:
topicDefn = provider.getNextTopic()
while topicDefn is not None:
self.__topicDefns[topicDefn.nameTuple] = topicDefn
topicDefn = provider.getNextTopic()
finally:
provider.doneIter()
def getDefn(self, topicNameTuple: Sequence[str]) -> Tuple[str, ArgSpecGiven]:
desc, spec = None, None
defn = self.__topicDefns.get(topicNameTuple, None)
if defn is not None:
assert defn.isComplete()
desc = defn.description
spec = ArgSpecGiven(defn.argsDocs, defn.required)
return desc, spec
def topicNames(self) -> Sequence[str]:
return self.__topicDefns.keys()
def getTreeDoc(self) -> str:
return self.__treeDocs
@classmethod
def registerTypeForImport(cls, typeName: str, providerClassObj: type):
"""
If a new type of importer is defined for topic definitions, it
can be registered with pubsub by providing a name for the new
importer (typeName), and the class to instantiate when
pub.addTopicDefnProvider(obj, typeName) is called. For instance, ::
from pubsub.core.topicdefnprovider import ITopicDefnDeserializer
class SomeNewImporter(ITopicDefnDeserializer):
...
TopicDefnProvider.registerTypeForImport('some name', SomeNewImporter)
# will instantiate SomeNewImporter(source)
pub.addTopicDefnProvider(source, 'some name')
"""
assert issubclass(providerClassObj, ITopicDefnDeserializer)
cls._typeRegistry[typeName] = providerClassObj
@classmethod
def initTypeRegistry(cls):
cls.registerTypeForImport(TOPIC_TREE_FROM_MODULE, TopicDefnDeserialModule)
cls.registerTypeForImport(TOPIC_TREE_FROM_STRING, TopicDefnDeserialString)
cls.registerTypeForImport(TOPIC_TREE_FROM_CLASS, TopicDefnDeserialClass)
TopicDefnProvider.initTypeRegistry()
def _backupIfExists(filename: str, bak: str):
import shutil
from pathlib import Path
if Path(filename).exists():
backupName = '%s.%s' % (filename, bak)
shutil.copy(filename, backupName)
defaultTopicTreeSpecHeader = \
"""
Topic tree for application.
Used via pub.addTopicDefnProvider(thisModuleName).
"""
defaultTopicTreeSpecFooter = \
"""\
# End of topic tree definition. Note that application may load
# more than one definitions provider.
"""
[docs]def exportTopicTreeSpec(moduleName: str = None, rootTopic: Union[Topic, str] = None,
bak: str = 'bak', moduleDoc: str = None):
"""
Using TopicTreeSpecPrinter, exports the topic tree rooted at rootTopic to a
Python module (.py) file. This module will define module-level classes
representing root topics, nested classes for subtopics etc. Returns a string
representing the contents of the file. Parameters:
- If moduleName is given, the topic tree is written to moduleName.py in
os.getcwd(). By default, it is first backed up, it it already exists,
using bak as the filename extension. If bak is None, existing module file
gets overwritten.
- If rootTopic is specified, the export only traverses tree from
corresponding topic. Otherwise, complete tree, using
pub.getDefaultTopicTreeRoot() as starting point.
- The moduleDoc is the doc string for the module ie topic tree.
"""
if rootTopic is None:
from .. import pub
rootTopic = pub.getDefaultTopicMgr().getRootAllTopics()
elif isinstance(rootTopic, str):
from .. import pub
rootTopic = pub.getDefaultTopicMgr().getTopic(rootTopic)
# create exporter
if moduleName is None:
capture = io.StringIO()
TopicTreeSpecPrinter(rootTopic, fileObj=capture, treeDoc=moduleDoc)
return capture.getvalue()
else:
filename = '%s.py' % moduleName
if bak:
_backupIfExists(filename, bak)
moduleFile = open(filename, 'w')
try:
TopicTreeSpecPrinter(rootTopic, fileObj=moduleFile, treeDoc=moduleDoc)
finally:
moduleFile.close()
##############################################################
class TopicTreeSpecPrinter:
"""
Helper class to print the topic tree using the Python class
syntax. The "printout" can be sent to any file object (object that has a
write() method). If printed to a module, the module can be imported and
given to pub.addTopicDefnProvider(module, 'module'). Importing the module
also provides code completion of topic names (rootTopic.subTopic can be
given to any pubsub function requiring a topic name).
"""
INDENT_CH = ' '
# INDENT_CH = '.'
def __init__(self, rootTopic: Union[str, Topic] = None, fileObj: TextIO = None, width: int = 70,
indentStep: int = 4,
treeDoc: str = defaultTopicTreeSpecHeader, footer: str = defaultTopicTreeSpecFooter):
"""
For formatting, can specify the width of output, the indent step, the
header and footer to print to override defaults. The destination is fileObj;
if none is given, then sys.stdout is used. If rootTopic is given, calls
writeAll(rootTopic) at end of __init__.
"""
self.__traverser = TopicTreeTraverser(self)
import sys
fileObj = fileObj or sys.stdout
self.__destination = fileObj
self.__output = []
self.__header = self.__toDocString(treeDoc)
self.__footer = dedent(footer)
self.__lastWasAll = False # True when last topic done was the ALL_TOPICS
self.__width = width
self.__wrapper = TextWrapper(width)
self.__indentStep = indentStep
self.__indent = 0
args = dict(width=width, indentStep=indentStep, treeDoc=treeDoc,
footer=self.__footer, fileObj=fileObj)
def fmItem(argName, argVal):
if isinstance(argVal, str):
MIN_OFFSET = 5
lenAV = width - MIN_OFFSET - len(argName)
if lenAV > 0:
argVal = repr(argVal[:lenAV] + '...')
elif argName == 'fileObj':
argVal = fileObj.__class__.__name__
return '# - %s: %s' % (argName, argVal)
fmtArgs = [fmItem(key, args[key]) for key in sorted(args.keys())]
self.__comment = [
'# Automatically generated by %s(**kwargs).' % self.__class__.__name__,
'# The kwargs were:',
]
self.__comment.extend(fmtArgs)
self.__comment.extend(['']) # two empty line after comment
if rootTopic is not None:
self.writeAll(rootTopic)
def getOutput(self) -> str:
"""
Each line that was sent to fileObj was saved in a list; returns a
string which is ``'\\n'.join(list)``.
"""
return '\n'.join(self.__output)
def writeAll(self, topicObj: Topic):
"""
Traverse each topic of topic tree, starting at topicObj, printing
each topic definition as the tree gets traversed.
"""
self.__traverser.traverse(topicObj)
def _accept(self, topicObj: Topic):
# accept every topic
return True
def _startTraversal(self):
# output comment
self.__wrapper.initial_indent = '# '
self.__wrapper.subsequent_indent = self.__wrapper.initial_indent
self.__output.extend(self.__comment)
# output header:
if self.__header:
self.__output.extend([''])
self.__output.append(self.__header)
self.__output.extend([''])
def _doneTraversal(self):
if self.__footer:
self.__output.append('')
self.__output.append('')
self.__output.append(self.__footer)
if self.__destination is not None:
self.__destination.write(self.getOutput())
def _onTopic(self, topicObj: Topic):
"""This gets called for each topic. Print as per specified content."""
# don't print root of tree, it is the ALL_TOPICS builtin topic
if topicObj.isAll():
self.__lastWasAll = True
return
self.__lastWasAll = False
self.__output.append('') # empty line
# topic name
self.__wrapper.width = self.__width
head = 'class %s:' % topicObj.getNodeName()
self.__formatItem(head)
# each extra content (assume constructor verified that chars are valid)
self.__printTopicDescription(topicObj)
self.__printTopicArgSpec(topicObj)
def _startChildren(self):
"""Increase the indent"""
if not self.__lastWasAll:
self.__indent += self.__indentStep
def _endChildren(self):
"""Decrease the indent"""
if not self.__lastWasAll:
self.__indent -= self.__indentStep
def __toDocString(self, msg: str) -> str:
if not msg:
return msg
if msg.startswith("'''") or msg.startswith('"""'):
return msg
return '"""\n%s\n"""' % msg.strip()
def __printTopicDescription(self, topicObj: Topic):
if topicObj.getDescription():
extraIndent = self.__indentStep
self.__formatItem('"""', extraIndent)
self.__formatItem(topicObj.getDescription(), extraIndent)
self.__formatItem('"""', extraIndent)
def __printTopicArgSpec(self, topicObj: Topic):
extraIndent = self.__indentStep
# generate the message data specification
reqdArgs, optArgs = topicObj.getArgs()
argsStr = []
if reqdArgs:
argsStr.append(", ".join(reqdArgs))
if optArgs:
optStr = ', '.join([('%s=None' % arg) for arg in optArgs])
argsStr.append(optStr)
argsStr = ', '.join(argsStr)
# print it only if there are args; ie if listener() don't print it
if argsStr:
# output a blank line and protocol
self.__formatItem('\n', extraIndent)
protoListener = 'def %s(%s):' % (SPEC_METHOD_NAME, argsStr)
self.__formatItem(protoListener, extraIndent)
# and finally, the args docs
extraIndent += self.__indentStep
self.__formatItem('"""', extraIndent)
# but ignore the arg keys that are in parent args docs:
parentMsgKeys = ()
if topicObj.getParent() is not None:
parentMsgKeys = topicObj.getParent().getArgDescriptions().keys() # keys iter ok
argsDocs = topicObj.getArgDescriptions()
for key in sorted(argsDocs.keys()):
if key not in parentMsgKeys:
argDesc = argsDocs[key]
msg = "- %s: %s" % (key, argDesc)
self.__formatItem(msg, extraIndent)
self.__formatItem('"""', extraIndent)
def __formatItem(self, item: str, extraIndent: int = 0):
indent = extraIndent + self.__indent
indentStr = self.INDENT_CH * indent
lines = item.splitlines()
for line in lines:
self.__output.append('%s%s' % (indentStr, line))
def __formatBlock(self, text: str, extraIndent: int = 0):
self.__wrapper.initial_indent = self.INDENT_CH * (self.__indent + extraIndent)
self.__wrapper.subsequent_indent = self.__wrapper.initial_indent
self.__output.append(self.__wrapper.fill(text))