Source code for serge.common
import sys
import traceback
import logging
import re
version = '0.6'
#
# Number of channels of audio - pygame default is 8 but
# it looks like this can easily be reached and causes
# some sounds to cut out. The following number is large
# enough to avoid that.
NUM_AUDIO_CHANNELS = 32
#
# Look for Pymunk
try:
import pymunk
#
# Looks like the following is not needed on pymunk 2.1
if hasattr(pymunk, 'init_pymunk'):
pymunk.init_pymunk()
PYMUNK_OK = True
except ImportError:
import simplevecs as pymunk
PYMUNK_OK = False
# The following needed for pygame and py2exe
try:
import pygame._view
except ImportError:
pass
DETAIL = 5
class Filtering(logging.Filter):
"""A nice filtering formatter than can show certain types of log"""
not_allowed = set([
])
def filter(self, record):
"""Format the record"""
return record.name not in self.not_allowed
filterer = Filtering()
log = logger = logging.getLogger('serge')
hdlr = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter('[%(relativeCreated)6d] :: %(levelname)7s %(name)20s :: %(message)s')
hdlr.setFormatter(formatter)
logger.addHandler(hdlr)
logger.addFilter(filterer)
def addFileLogging():
"""Add file logging"""
global LOG_TO_FILE, fhdlr
LOG_TO_FILE = True
fhdlr = logging.FileHandler('log.txt', 'w')
fhdlr.setFormatter(formatter)
#logger.setLevel(logging.DEBUG)
logger.setLevel(DETAIL)
#logger.setLevel(logging.ERROR)
def tb():
"""Return the traceback as a string"""
exc_type, exc_value, exc_traceback = sys.exc_info()
return traceback.format_tb(exc_traceback)
def info(type, value, tb):
if hasattr(sys, 'ps1') or not sys.stderr.isatty():
# we are in interactive mode or we don't have a tty-like
# device, so we call the default hook
sys.__excepthook__(type, value, tb)
else:
import traceback, pdb
# we are NOT in interactive mode, print the exception...
traceback.print_exception(type, value, tb)
print
# ...then start the debugger in post-mortem mode.
pdb.pm()
def installDebugHook():
sys.excepthook = info
class BaseError(Exception):
"""A useful base class for errors"""
def __init__(self, text):
"""Initialialise and add traceback"""
super(BaseError, self).__init__(text + '\n' + ''.join(tb()))
LOG_TO_FILE = False
def getLogger(name):
"""Return a new logger with the name"""
l = logging.getLogger(name)
l.addHandler(hdlr)
l.setLevel(logger.level)
l.addFilter(filterer)
if LOG_TO_FILE:
l.addHandler(fhdlr)
l.propagate = False
return l
[docs]class Loggable(object):
"""A helper class that adds a logger to a class
Each instance of the class will have a *log* attribute and can
use this to log output. The `log` attrbute is a logger with the
usual *debug*, *warn*, *info*, and *error* methods.
"""
log = None
[docs] def addLogger(self):
"""Add a logger"""
if not 'log' in self.__class__.__dict__:
self.__class__.log = getLogger(self.__class__.__name__)
class EventNotLinked(Exception): """The event was not linked to a callback"""
class EventNotFound(Exception): """The event was not a registered event"""
class DuplicateEvent(Exception): """An event was registered twice"""
[docs]class EventAware(object):
"""A mixin class that allows objects to respond to events"""
# Legacy flag - set to True to enforce only registered events
strict = False
[docs] def initEvents(self):
"""Initialise the events system"""
self._event_handlers = {}
self._registered_events = set()
[docs] def registerEvent(self, event):
"""Register an event"""
if event in self._registered_events:
raise DuplicateEvent('The event "%s" is already registered' % event)
self._registered_events.add(event)
[docs] def registerEvents(self, events):
"""Register a number of events"""
for event in events:
self.registerEvent(event)
[docs] def registerEventsFromModule(self, module):
"""Register all events found in the module
Events must be strings and their name must be of the
form E_THE_NAME
ie: Begins with an 'E' and is all uppercase
"""
finder = re.compile('E_[A-Z_]+$')
for name, obj in module.__dict__.iteritems():
if finder.match(name) and isinstance(obj, str):
self.registerEvent(obj)
[docs] def processEvent(self, event):
"""Process an incoming event"""
name, obj = event
if self.strict and name not in self._registered_events:
raise EventNotFound('The event "%s" was not registered' % name)
#
# Try to pass this off to a handler
inhibits = set()
try:
links = self._event_handlers[name]
except KeyError:
new_inhibits = self.handleEvent(event)
# Watch for new events to inhibit
if new_inhibits:
inhibits.add(new_inhibits)
else:
#
# Process all the handler functions
for callback, arg in links:
new_inhibits = callback(obj, arg)
# Watch for new events to inhibit
if new_inhibits:
inhibits.add(new_inhibits)
return inhibits
[docs] def handleEvent(self, event):
"""Handle an incoming event"""
pass
[docs] def linkEvent(self, name, callback, arg=None):
"""Link an event to a callback"""
if self.strict and name not in self._registered_events:
raise EventNotFound('The event "%s" was not registered' % name)
self._event_handlers.setdefault(name, []).append((callback, arg))
[docs] def unlinkEvent(self, name, callback=None):
"""Unlink an event from a callback"""
if self.strict and name not in self._registered_events:
raise EventNotFound('The event "%s" was not registered' % name)
if callback is None:
try:
del(self._event_handlers[name])
except KeyError:
raise EventNotLinked('No links to event "%s"' % name)
else:
#
# Look for items with the same name and callback
try:
old_items, new_items = self._event_handlers[name], []
except KeyError:
raise EventNotLinked('No links to event "%s"' % name)
#
for the_callback, arg in old_items:
if the_callback != callback:
# This one is ok
new_items.append((the_callback, arg))
#
# Were any changed?
if old_items == new_items:
raise EventNotLinked('No links for event "%s" with callback "%s"' % (name, callback))
#
# Reset events
self._event_handlers[name] = new_items