1. Olemis Lang
  2. bloodhound-rpc


bloodhound-rpc / trunk / tracrpc / timeline.py

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# Copyright 2009-2011 Olemis Lang <olemis at gmail.com>
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#       http://www.apache.org/licenses/LICENSE-2.0
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.

"""RPC handlers for timeline RPC.

Copyright 2009-2013 Olemis Lang <olemis at gmail.com>
Licensed under the Apache License, Version 2.0 
from __future__ import with_statement

__author__ = 'Olemis Lang'

__all__ = 'TimelineRPC',

from trac.core import implements, ExtensionPoint, Component
from trac.timeline.api import ITimelineEventProvider
from trac.timeline.web_ui import TimelineModule
from trac.util.translation import _

from datetime import datetime, date, time
import xmlrpclib

from tracrpc.api import IXMLRPCHandler

__metaclass__ = type

#   Event Timeline RPC

class TimelineRPC(Component):
    r""" An interface to Trac's timeline module.
    sources = ExtensionPoint(ITimelineEventProvider)

    def __init__(self):
        self._event_data = TimelineModule(self.env)._event_data
    # IXMLRPCHandler methods
    def xmlrpc_namespace(self):
        return 'timeline'
    def xmlrpc_methods(self):
        yield ('TIMELINE_VIEW', 
                ((list, datetime, datetime, list),
                 (list, datetime, datetime), 
                 (list, datetime), 
                 (list, )), 
        yield ('TIMELINE_VIEW', ((list,),), self.getEventFilters)

    STMT = "(kind, author, date, dateuid, " \
            "unicode(render('url', ctx)).encode('utf-8', 'replace'), " \
            "unicode(render('title', ctx)).encode('utf-8', 'replace'), " \
            "unicode(render('description', ctx)).encode('utf-8', 'replace'))"
    STMT = compile(STMT, '<string>', 'eval')
    # Exported methods
    def getEvents(self, req, start=None, stop=None, filters=None):
        r"""Retrieve events taking place in a time window.
        @param start    initial date in time interval
        @param stop     last date in time interval
        @param filter   is a list of the enabled filters, 
                        each item being the name (first item) in the 
                        tuples returned by 
                        If none is specified then all available 
                        filters will be used.
        @return         a list of events in the time range given by 
                        the `start` and `stop` parameters. Each item 
                        is represented as a tuple of the form 
                        (kind, author, date, timestamp, url, title, desc) 
                        representing information about each event.
        Note: In order to retrieve all events related to ticket 
              changes (e.g. attachments) you need to set
              timeline.ticket_show_details option in trac.ini to true.
        if filters is not None :
          if not filters:
            # No filter selected
            return []
          # Filter unwanted filters ... 8$
          filters = [f for f, _ in self.getEventFilters(req) \
                      if f in filters]
          if not filters:
            # No filter selected
            return []
        else :
          filters = list(f for f, _ in self.getEventFilters(req))
        if start is None:
            start = _epoc
        elif not isinstance(start, datetime):
            start = rpc_to_datetime(start, req)
        if stop is None:
            stop = datetime.now(req.tz)
        elif not isinstance(stop, datetime):
            stop = rpc_to_datetime(stop, req)
        ctx = Context.from_request(req, absurls=True)
        globs = dict(ctx=ctx)
        return list(eval(self.STMT, globs, self._event_data(p, e)) \
                                for p in self.sources \
                                for e in p.get_timeline_events(req, \
                                            start, stop, filters))
    def getEventFilters(self, req):
        r"""Return a list of the filters available to retrieve events 
        provided  by the timeline module. The data returned for each 
        filter is a binary tuple containing the filter 
        name as well as its display name.
        fdata = dict(fi[:2] for p in self.sources \
                        for fi in p.get_timeline_filters(req))
        return fdata.iteritems()