ktouch_reporter.py

import json, os, re, sys, time
from lxml import etree
"""
This tool finds and prints xAPI JSON representations of each ktouch activity it finds.

It is WAY OVER-ENGINEERED for what it needs to be, but it was also intended as a way 
to play around with ideas for how this could be abstracted into a more general framework.

Here are some concepts you will need to understand:
        
    Component:
        A source of learning activities (e.g. ktouch, moodle, edx)
        
    Activity:
        Description of a learning activity ( actor + verb + object/extensions )
        
    RecordSource:
        A place where a component might find evidence of activities (e.g. a ktouch log file)
    
    DataStore:
        A place to which activity descriptions should be sent (database, file, screen, etc)
        
    Site: 
        The configuration of a specific classroom (available DataStores, etc)
"""

## Helper functions

def escape_string(s):
    """Helper function to strip spaces and whatnot from a string
    
    Used to turn a lecture title into part of an xAPI object ID
    """
    return re.sub(r'[^a-zA-Z0-9]+','_',s.lower()).rstrip('_')

DEBUG_LEVEL=0
def dbg(s,level=3):
    """Helper function to print debug infos to stderr"""
    if DEBUG_LEVEL >= level:
        sys.stderr.write("DBG: %s\n" % s)


## CLASSES

# This would be stored elsewhere so that other components 
# can use it.
class Site(object):
    """Base class for storing site-specific information"""
    # Descriptor for this site (given when created)
    name = None

    # This example is a bit silly, but imagine a RecordSource
    # that needs to know how to access the local mysql db...
    home = "/home"
    
    # Base URL to be used for all xAPI objects reported
    # from this site
    url = "http://tunapanda.org/swag"

    # Available components at this site 
    # (see get_components() below)
    _components = None

    # Available output targets for activity JSON
    # (see get_datastores() below)
    _datastores = None

    # Customization based on a local config file or something
    # would go here, but again, keeping it simple for now.
    def __init__(self,name):
        self.name = name

    def get_components(self):
        """Yields a list of available components at this site"""
        # This list would be dynamically generated, based on
        # e.g. a local config file, but let's keep it simple 
        # for now ;)
        # 
        # We store the result so it only needs to be fetched
        # once
        if self._components is None:
            self._components = [ Component_Ktouch, ]
        for component in self._components:
            yield component(self)
    
    def get_datastores(self):
        """Yields a list of available datastores at this site"""
        if self._datastores is None:
            self._datastores = [ DataStore_Stdout, ]
        for store in self._datastores:
            yield store(self)

    def sync_all(self):
        """Sends activities from each component to each datastore"""
        activities = []
        for c in self.get_components():
            dbg("Component: %s" % c)
            for a in c.get_activities():
                dbg("Activity: %s" % a)
                activities.append(a)

        for s in self.get_datastores():
            s.report(activities)
            

class DataStore_Stdout(object):
    """Reports activity info to the screen for testing purposes"""
    def __init__(self,site):
        self.site = site

    def report(self,activities):
        """Prints a JSON representation of each activity to the standard output"""
        print json.dumps([ a.as_dict() for a in activities ], indent=2)
        

# In a full implementation, this would inherit from a hierarchy of superclasses 
# like RecordSource_Base, RecordSource_XmlFile, etc.
#
# Related classes might include RecordSource_Mysql, etc
class RecordSource_Ktouch(object):
    """Interface for getting ktouch data"""
    # Stores the component using this source
    component = None

    # Easily extendable list of possible file locations
    log_fn_patterns = []

    def __init__(self,component):
        self.component = component

    def get_records_by_user(self):
        # Not the best way to do this, but good enough for now...
        homedir = self.component.site.home
        for u in os.listdir(homedir):
            fn = homedir + "/" + u + "/.kde/share/apps/ktouch/statistics.xml"
            dbg("Looking file: %s" % fn)
            if os.path.exists(fn):
                dbg("...found!")
                yield (u,etree.parse(fn))
                

class Component_Base(object):
    """Base class for a source of activities (e.g. Ktouch)"""
    # Name must be defined by subclasses!
    name = None 
    recordsource_classes = None
    test_classes = None
    
    # These will be set when instances are created
    component = None

    def __init__(self, site):
        self.site = site

    def __str__(self):
        """String representation. Used to construct activity ID
        
        Returns e.g. http://tunapanda.org/swag/COMPONENT_NAME
        """
        return self.site.url + "/" + self.name

    def get_recordsources(self):
        for rs_class in self.recordsource_classes:
            yield rs_class(self)

    def get_tests(self):
        for test_class in self.test_classes:
            yield test_class(self)

    def get_activities(self):
        """Returns a list of found activities"""
        # Must be set overridden by subclasses
        pass


class Component_Ktouch(Component_Base):
    """Class for gathering ktouch-related activities"""
    name = "ktouch"

    # Note: all instances will share the recordsource and test
    # instances created here. I'm unsure whether that's cool 
    # or a recipe for pain (or both!)
    recordsource_classes = [ RecordSource_Ktouch, ]

    def get_activities(self):
        dbg("Getting activites for %s..." % self)
        for rs in self.get_recordsources():
            dbg("Analyzing recordsource %s" % rs)
            for (user,record) in rs.get_records_by_user():
                dbg("Analyzing data for user %s" % user)
                for lecture in record.xpath("//LectureStats"):
                    # TODO: Don't crash and burn in the unlikely event
                    # of not being able to find a title here.
                    lecture_title = lecture.find("Title").text
                    activity_name = "lecture" + "/" + escape_string(lecture_title)
                    for level in lecture.xpath(".//LevelStats"):
                        level_number = level.get("Number")
                        level_timestamp = level.find("Time").text

                        # This is enough to award an "attempted" lecture...
                        yield Activity(
                            component   = self,
                            actor       = user + "@tunapanda.org",
                            verb        = "attempted",
                            name        = activity_name,
                            timestamp   = level_timestamp,
                            extensions  = { 
                                "level" : level_number, 
                                "site"  : self.site.name, 
                            },
                        )

                        # TODO: Completions and other activities
                        
    
class Activity(object):
    """Base class for accomplishments that a Component may report"""
    def __init__(self, component, actor, verb, name, timestamp=None, extensions={}):
        self.component  = component

        # Naively assume 'mbox' type for now
        self.actor = actor 
        
        self.verb       = verb 
        self.name       = name
        self.extensions = extensions

        # This isn't the format xAPI wants, but 
        # I'm too lazy to do it properly right now. :P
        if timestamp is None:
            self.timestamp = time.time()
        else:
            self.timestamp = timestamp

    def __str__(self):
        """Returns a unique xAPI ID

        e.g. http://tunapanda.org/swag/COMPONENT_NAME/ACTIVITY_NAME
        """
        return str(self.component) + "/" + self.name

    # Don't actually want this when reporting more than one record
    def as_xapi(self):
        return json.dump(self.as_dict(),indent=2)

    def as_dict(self):
        return {
            "timestamp"     : self.timestamp,
            "actor"     : { "mbox" : self.actor },
            "verb"      : { "id" : self.verb  },
            "object"    : {
                "id"         : str(self),
                "extensions" : self.extensions,
            },
        }


if __name__ == "__main__":
    dbg("Starting...")
    site = Site("Swag Central!")
    site.sync_all()