Source code for autopilot.emulators.zeitgeist

# -*- Mode: Python; coding: utf-8; indent-tabs-mode: nil; tab-width: 4 -*-
# Copyright 2012 Canonical
# Author: Brandon Schaefer
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 3, as published
# by the Free Software Foundation.

"""Provide ability to register text files with the file lens."""

from __future__ import absolute_import

import logging
import os.path
from zeitgeist.client import ZeitgeistClient
from zeitgeist.datamodel import Event, Interpretation, Manifestation, ResultType


[docs]class Zeitgeist(object): """Provide access zeitgeist.""" def __init__(self): self.zg = ZeitgeistClient() self.logger = logging.getLogger(__name__)
[docs] def add_existing_file(self, path): """Registers *file* with zeitgeist. :param string file: full path to an existing text file to register. :raises: **RuntimeError** if *file* does not exist. """ if os.path.exists(path): self.__add_text_file(path) else: raise RuntimeError("File not found on path: %s." % (path))
def __add_text_file(self, path): """Takes a path to a file and creates an event for it then querys it.""" file_lens = "file://" dir_path = os.path.dirname(path) name = os.path.basename(path) event = Event.new_for_values (interpretation=Interpretation.ACCESS_EVENT, manifestation=Manifestation.USER_ACTIVITY, subject_uri=file_lens + path, subject_interpretation=Interpretation.TEXT_DOCUMENT, subject_manifestation=Manifestation.FILE_DATA_OBJECT, subject_origin=file_lens + dir_path, subject_text=name) self.zg.insert_event(event) template = Event.new_for_values (interpretation=Interpretation.ACCESS_EVENT, manifestation=Manifestation.USER_ACTIVITY) self.zg.find_events_for_template (template, self.__log_events_cb, num_events=1, result_type=ResultType.MostRecentSubjects) def __log_events_cb(self, events): """Callback to recive events, we are just using it to log each event.""" self.logger.info("Found Events") for event in events: for subject in event.subjects: self.logger.info(" * %s" % (subject.uri))