本文整理汇总了Python中zeitgeist.datamodel.Event类的典型用法代码示例。如果您正苦于以下问题:Python Event类的具体用法?Python Event怎么用?Python Event使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Event类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: testFindEventTwoInterpretations
def testFindEventTwoInterpretations(self):
import_events("test/data/twenty_events.js", self)
result = self.findEventIdsAndWait([
Event.new_for_values(interpretation="stfu:OpenEvent"),
Event.new_for_values(interpretation="stfu:EvilEvent")],
timerange = (102, 117), num_events=0, result_type=0)
self.assertEquals(15, len(result))
开发者ID:freedesktop-unofficial-mirror,项目名称:zeitgeist__zeitgeist,代码行数:7,代码来源:engine-test.py
示例2: __init__
def __init__(self, date):
next_date = date + datetime.timedelta(days=1)
# time.mktime coverts from date to secs since epoch
day_start = time.mktime(date.timetuple())
# day_end really is the start of the next day
day_end = time.mktime(next_date.timetuple())
DayWidget.__init__(self, day_start, day_end)
day_part_widgets = self.view.get_children()
day_part = day_part_widgets[1]
day_part.get_events()
print day_part.events
#self.vbox.remove(self.daylabel)
self.event_timerange = [day_start * 1000, day_end * 1000]
self.event_templates = (
Event.new_for_values(interpretation=Interpretation.VISIT_EVENT.uri),
Event.new_for_values(interpretation=Interpretation.MODIFY_EVENT.uri),
Event.new_for_values(interpretation=Interpretation.CREATE_EVENT.uri),
Event.new_for_values(interpretation=Interpretation.OPEN_EVENT.uri),
)
CLIENT.install_monitor(self.event_timerange, self.event_templates,
self.notify_insert_handler, self.notify_delete_handler)
self.get_events()
开发者ID:tomka,项目名称:rednotebook,代码行数:28,代码来源:journalgeist.py
示例3: _on_uri_changed
def _on_uri_changed(self, webview, load_status):
if self.webview.get_property("load-status") ==\
WebKit.LoadStatus.FINISHED:
if self.current_uri != self.webview.get_uri():
if self.current_uri:
event = Event()
event.interpretation = Interpretation.LEAVE_EVENT
event.manifestation = Manifestation.USER_ACTIVITY
event.actor = "application://web.desktop"
subject = Subject()
subject.uri = self.current_uri
subject.mimetype = "text/html"
subject.text = self.current_title
subject.interpretation = Interpretation.WEBSITE
subject.manifestation = Manifestation.REMOTE_PORT_ADDRESS
event.subjects = [subject]
ZG.insert_event(event)
self.current_uri = self.webview.get_uri()
self.current_title = self.webview.get_title()
print "===>", self.current_title
event = Event()
event.interpretation = Interpretation.ACCESS_EVENT
event.manifestation = Manifestation.USER_ACTIVITY
event.actor = "application://web.desktop"
subject = Subject()
subject.uri = self.current_uri
subject.mimetype = "text/html"
subject.text = self.current_title
subject.interpretation = Interpretation.WEBSITE
subject.manifestation = Manifestation.REMOTE_PORT_ADDRESS
event.subjects = [subject]
ZG.insert_event(event)
开发者ID:gnome-design-team,项目名称:gnome-prototypes,代码行数:32,代码来源:page_widget.py
示例4: __init__
def __init__(self):
self[0] = all_events
template2 = Event.new_for_values(interpretation = \
Interpretation.ACCESS_EVENT)
self[1] = ["Access Events", \
"Fetch all the access events", \
template2, \
TimeRange.always(), \
ResultType.MostRecentEvents]
template3 = Event.new_for_values(interpretation = \
Interpretation.LEAVE_EVENT)
self[2] = ["Leave Events", \
"Fetch all the leave events", \
template3, \
TimeRange.always(), \
ResultType.MostRecentEvents]
template4 = Event.new_for_values(interpretation = \
Interpretation.MODIFY_EVENT)
self[3] = ["Modify Events", \
"Fetch all the modify events", \
template4, \
TimeRange.always(), \
ResultType.MostRecentEvents]
template5 = Event.new_for_values(actor = "application://banshee.desktop")
self[4] = ["Banshee events", \
"All the Banshee related events", \
template5, \
TimeRange.always(), \
ResultType.MostRecentEvents]
开发者ID:dhananjaysathe,项目名称:zeitgeist-explorer,代码行数:33,代码来源:templates.py
示例5: testMoving
def testMoving(self):
import_events("test/data/five_events.js", self)
import_events("test/data/five_events_ext_move.js", self)
template = Event.new_for_values(subject_current_uri='file:///*')
ids = self.findEventIdsAndWait([template],
num_events = 0,
result_type = ResultType.MostRecentCurrentUri)
self.assertEquals(2, len(ids))
ids = self.findEventIdsAndWait([template],
num_events = 0,
result_type = ResultType.MostRecentCurrentUri)
self.assertEquals(2, len(ids))
ids = self.findEventIdsAndWait([template],
num_events = 0,
result_type = ResultType.MostRecentEvents)
self.assertEquals(5, len(ids))
template = Event.new_for_values(subject_current_origin='file:///*')
ids = self.findEventIdsAndWait([template],
num_events = 0,
result_type = ResultType.MostRecentEvents)
self.assertEquals(4, len(ids))
开发者ID:freedesktop-unofficial-mirror,项目名称:zeitgeist__zeitgeist,代码行数:25,代码来源:engine-test.py
示例6: _on_view_pages
def _on_view_pages(self, widget):
self.home.reset()
uris = []
for pIndex in xrange(self.pages.get_n_pages()):
self.pages.set_current_page(pIndex)
page = self.pages.get_nth_page(pIndex)
overviewPage = PageOverviewWidget(page)
overviewPage.connect("clicked", self._on_page_clicked)
self.home.add_page(overviewPage)
uris.append(page.webview.get_uri())
self.home.show_all()
self.box.set_current_page(0)
def callback(events):
self.home.recentDashboard.recentPages.reset()
for event in events:
page = PageButton(event.subjects[0])
print event.subjects[0].uri
self.home.recentDashboard.recentPages.add_page(page)
self.home.show_all()
event = Event()
event.actor = "application://web.desktop"
event.subjects = []
for uri in uris:
if uri:
subject = Subject()
subject.uri = "!"+uri
event.subjects.append(subject)
print event
ZG.find_events_for_template(event, callback, result_type=2)
开发者ID:gnome-design-team,项目名称:gnome-prototypes,代码行数:30,代码来源:web.py
示例7: _get_base_template
def _get_base_template(self):
base_template = Event()
if self.has_uri:
subj = Subject()
subj.set_uri(self.uri+"/*")
base_template.set_subjects([subj])
else:
base_template.set_actor(self.uri)
return base_template
开发者ID:genesi,项目名称:dockmanager,代码行数:9,代码来源:zeitgeist_menus.py
示例8: testFindMultipleEvents
def testFindMultipleEvents(self):
import_events("test/data/five_events.js", self)
subj1 = Subject.new_for_values(uri="file:///home/foo.txt")
event_template1 = Event.new_for_values(subjects=[subj1])
subj2 = Subject.new_for_values(uri="file:///tmp/foo.txt")
event_template2 = Event.new_for_values(subjects=[subj2])
result = self.findEventIdsAndWait([event_template1, event_template2], num_events=0, result_type=4)
self.assertEquals(2, len(result))
events = self.getEventsAndWait(result)
开发者ID:freedesktop-unofficial-mirror,项目名称:zeitgeist__zeitgeist,代码行数:9,代码来源:engine-test.py
示例9: testFindWithSubjectText
def testFindWithSubjectText(self):
import_events("test/data/five_events.js", self)
result = self.findEventIdsAndWait([Event.new_for_values(subject_text='this is not real')])
self.assertEquals(0, len(result))
result = self.findEventIdsAndWait([Event.new_for_values(subject_text='some text')])
self.assertEquals(1, len(result))
result = self.findEventIdsAndWait([Event.new_for_values(subject_text='this *')])
self.assertEquals(0, len(result)) # We don't support wildcards for text
result = self.findEventIdsAndWait([Event.new_for_values(subject_text='this item *')])
self.assertEquals(1, len(result))
开发者ID:freedesktop-unofficial-mirror,项目名称:zeitgeist__zeitgeist,代码行数:10,代码来源:engine-test.py
示例10: testUnicodeEventInsert
def testUnicodeEventInsert(self):
# Insert and get a unicode event
ids = import_events("test/data/unicode_event.js", self)
self.assertEquals(len(ids), 1)
result = self.getEventsAndWait(ids)
self.assertEquals(1, len(result))
event = result[0]
self.assertEquals(1, len(event.subjects))
self.assertEquals(u"hällö, I'm gürmen - åge drikker øl - ☠", event.subjects[0].text)
self.assertEquals(u"http://live.gnome.org/☠", event.subjects[0].uri)
# Update the event we got from the DB's timestamp and insert
# it again, we want to to test some ping-pong back and forth
event[0][Event.Id] = ""
event.timestamp = "243"
ids = self.insertEventsAndWait([event])
result = self.getEventsAndWait(ids)
self.assertEquals(1, len(result))
event = result[0]
self.assertEquals(1, len(event.subjects))
self.assertEquals(u"hällö, I'm gürmen - åge drikker øl - ☠", event.subjects[0].text)
self.assertEquals(u"http://live.gnome.org/☠", event.subjects[0].uri)
# Try and find a unicode event
subj = Subject.new_for_values(text="hällö, I'm gürmen - åge drikker øl - ☠",
origin="file:///åges_øl í", uri=u"http://live.gnome.org/☠")
event_template = Event.new_for_values(subjects=[subj,])
result = self.findEventIdsAndWait([event_template],
timerange=(0,200), num_events=100, result_type=0)
self.assertEquals(len(result), 1)
开发者ID:freedesktop-unofficial-mirror,项目名称:zeitgeist__zeitgeist,代码行数:31,代码来源:engine-test.py
示例11: post_commit
def post_commit(local, master, old_revno, old_revid, new_revno, new_revid):
client = get_client()
if client is None:
return
from zeitgeist.datamodel import Event, Interpretation, Manifestation
revision = master.repository.get_revision(new_revid)
if new_revno == 1:
interpretation = Interpretation.CREATE_EVENT
else:
interpretation = Interpretation.MODIFY_EVENT
_text = _("Revision: ")
_text += str(new_revno) + "\n"
_text += revision.message.rstrip()
subjects = subjects_for_branch(master, new_revno, old_revno,
None, revision.message.rstrip())
#subjects.set_text(_text)
event = Event.new_for_values(
timestamp=int(revision.timestamp*1000),
interpretation=unicode(interpretation),
manifestation=unicode(Manifestation.USER_ACTIVITY),
actor=get_actor(),
subjects=subjects,
)
client.insert_event(event)
开发者ID:dolanor,项目名称:zeitgeist-datasources,代码行数:25,代码来源:hooks.py
示例12: onNewTrack
def onNewTrack(self, track):
""" Send track information to Zeitgeist """
import mimetypes, os.path
from zeitgeist.datamodel import Event, Subject, Interpretation, Manifestation
mime, encoding = mimetypes.guess_type(track.getFilePath(), strict=False)
subject = Subject.new_for_values(
uri = os.path.dirname(track.getURI()),
text = track.getTitle() + ' - ' + track.getArtist() + ' - ' + track.getExtendedAlbum(),
mimetype = mime,
manifestation = unicode(Manifestation.FILE),
interpretation = unicode(Interpretation.AUDIO),
)
if hasattr(Interpretation, 'ACCESS_EVENT'):
eventType = Interpretation.ACCESS_EVENT
else:
eventType = Interpretation.OPEN_EVENT
event = Event.new_for_values(
actor = "application://decibel-audio-player.desktop",
subjects = [subject,],
interpretation = eventType,
)
self.client.insert_event(event)
开发者ID:Nenuphar,项目名称:decibel-audio-player,代码行数:28,代码来源:Zeitgeist.py
示例13: main
def main(argv):
interpretation = Interpretation.MODIFY_EVENT.uri
# FIXME: I'd be great if zeitgeist would allow for more detail:
# * branch
# * log summary (git log -1 --format=%s HEAD)
curdir = os.path.abspath(os.curdir).decode(sys.getfilesystemencoding())
uri = u"file://%s" % curdir
repo, origin = get_repo()
if not repo:
repo = unicode(curdir.rsplit('/', 1)[1])
origin = uri
subject = Subject.new_for_values(
uri=uri,
interpretation=Interpretation.DOCUMENT.TEXT_DOCUMENT.PLAIN_TEXT_DOCUMENT.SOURCE_CODE.uri,
manifestation=Manifestation.FILE_DATA_OBJECT.uri,
text=repo,
origin=origin)
event = Event.new_for_values(
timestamp=int(time.time() * 1000),
interpretation=interpretation,
manifestation=Manifestation.USER_ACTIVITY.uri,
actor="application://gitg.desktop",
subjects=[subject])
CLIENT.insert_event(event)
开发者ID:agx,项目名称:git-buildpackage,代码行数:27,代码来源:zeitgeist-git.py
示例14: testMonitorInstallRemoval
def testMonitorInstallRemoval(self):
result = []
mainloop = self.create_mainloop()
tmpl = Event.new_for_values(interpretation="stfu:OpenEvent")
@asyncTestMethod(mainloop)
def notify_insert_handler(notification_type, events):
pass
@asyncTestMethod(mainloop)
def notify_delete_handler(time_range, event_ids):
mainloop.quit()
self.fail("Unexpected delete notification")
mon = self.client.install_monitor(TimeRange.always(), [tmpl],
notify_insert_handler, notify_delete_handler)
@asyncTestMethod(mainloop)
def removed_handler(result_state):
result.append(result_state)
mainloop.quit()
self.client.remove_monitor(mon, removed_handler)
mainloop.run()
self.assertEquals(1, len(result))
self.assertEquals(1, result.pop())
开发者ID:freedesktop-unofficial-mirror,项目名称:zeitgeist__zeitgeist,代码行数:26,代码来源:monitor-test.py
示例15: insert_event_for_values
def insert_event_for_values (self, **values):
"""
Send an event to the Zeitgeist event log. The keyword arguments
must match those as provided to Event.new_for_values().
The insertion will be done via an asynchronous DBus call and
this method will return immediately. This means that the
Zeitgeist engine will most likely not have inserted the event
when this method returns. There will be a short delay.
If the ids_reply_handler argument is set to a callable it will
be invoked with a list containing the ids of the inserted events
when they have been registered in Zeitgeist.
In case of errors a message will be printed on stderr, and
an empty result passed to ids_reply_handler (if set).
To override this default set the error_handler named argument
to a callable that takes a single exception as its sole
argument.
In order to use this method there needs to be a mainloop
runnning. Both Qt and GLib mainloops are supported.
"""
ev = Event.new_for_values(**values)
self.insert_events([ev],
values.get("ids_reply_handler", None),
values.get("error_handler", None))
开发者ID:thnguyn2,项目名称:ECE_527_MP,代码行数:27,代码来源:client.py
示例16: send_to_zeitgeist
def send_to_zeitgeist(self, db, entry, event_type):
song = self.get_song_info(db, entry)
if self.__manual_switch:
manifest = Manifestation.USER_ACTIVITY
else:
manifest = Manifestation.SCHEDULED_ACTIVITY
subject = Subject.new_for_values(
uri=song["location"],
interpretation=unicode(Interpretation.AUDIO),
manifestation=unicode(Manifestation.FILE_DATA_OBJECT),
#~ origin="", #TBD
mimetype=song["mimetype"],
text=" - ".join([song["title"], song["artist"], song["album"]])
)
event = Event.new_for_values(
timestamp=int(time.time()*1000),
interpretation=unicode(event_type),
manifestation=unicode(manifest),
actor="application://rhythmbox.desktop",
subjects=[subject,]
)
#print event
IFACE.insert_event(event)
开发者ID:dignan,项目名称:control,代码行数:25,代码来源:__init__.py
示例17: testFindFilteredByEventButNotSubject
def testFindFilteredByEventButNotSubject(self):
# revision [email protected] (1185)
# Fix _build_templates so that it works when the Subject is empty.
self.testSingleInsertGet()
result = self.findEventIdsAndWait([Event.new_for_values(
interpretation=Interpretation.LEAVE_EVENT)])
self.assertEquals(0, len(result))
开发者ID:freedesktop-unofficial-mirror,项目名称:zeitgeist__zeitgeist,代码行数:7,代码来源:engine-test.py
示例18: handle_AQ_UNLINK_OK
def handle_AQ_UNLINK_OK(self, share_id, parent_id, node_id,
new_generation, was_dir, old_path):
"""A file or folder was deleted on the server by Syncdaemon,"""
if was_dir:
mime, interp = DIRECTORY_MIMETYPE, Interpretation.FOLDER
else:
mime, interp = self.get_mime_and_interpretation_for_filepath(
old_path)
file_subject = Subject.new_for_values(
uri=URI_PROTOCOL_U1 + str(node_id),
interpretation=interp,
manifestation=Manifestation.DELETED_RESOURCE,
origin="file:///" + old_path,
text=basename(old_path),
mimetype=mime,
storage=STORAGE_DELETED)
event = Event.new_for_values(
interpretation=Interpretation.DELETE_EVENT,
manifestation=Manifestation.SCHEDULED_ACTIVITY,
actor=ACTOR_UBUNTUONE,
subjects=[file_subject])
self.zg.log(event)
开发者ID:thnguyn2,项目名称:ECE_527_MP,代码行数:25,代码来源:zg_listener.py
示例19: log_folder_shared
def log_folder_shared(self, share, share_id):
"""Log the 'directory shared' event."""
fullpath = share.path
folder_name = basename(fullpath)
folder = Subject.new_for_values(
uri=URI_PROTOCOL_U1 + str(share.node_id),
interpretation=Interpretation.FOLDER,
manifestation=Manifestation.REMOTE_DATA_OBJECT,
origin="file:///" + fullpath,
text=folder_name,
mimetype=DIRECTORY_MIMETYPE,
storage=STORAGE_NETWORK)
other_username = share.other_username
other_user = Subject.new_for_values(
uri="mailto:" + other_username,
interpretation=INTERPRETATION_U1_CONTACT,
text=other_username,
manifestation=MANIFESTATION_U1_CONTACT_DATA_OBJECT)
event = Event.new_for_values(
interpretation=EVENT_INTERPRETATION_U1_FOLDER_SHARED,
manifestation=Manifestation.USER_ACTIVITY,
actor=ACTOR_UBUNTUONE,
subjects=[folder, other_user])
self.zg.log(event)
开发者ID:thnguyn2,项目名称:ECE_527_MP,代码行数:28,代码来源:zg_listener.py
示例20: handle_VM_SHARE_CREATED
def handle_VM_SHARE_CREATED(self, share_id):
"""Log the share accepted event."""
share = self.vm.shares[share_id]
folder = Subject.new_for_values(
uri=URI_PROTOCOL_U1 + str(share.node_id),
interpretation=Interpretation.FOLDER,
manifestation=Manifestation.REMOTE_DATA_OBJECT,
origin="file:///" + share.path,
text=basename(share.path),
mimetype=DIRECTORY_MIMETYPE,
storage=STORAGE_NETWORK)
other_username = share.other_username
other_user = Subject.new_for_values(
uri="mailto:" + other_username,
interpretation=INTERPRETATION_U1_CONTACT,
text=other_username,
manifestation=MANIFESTATION_U1_CONTACT_DATA_OBJECT)
event = Event.new_for_values(
interpretation=EVENT_INTERPRETATION_U1_SHARE_ACCEPTED,
manifestation=Manifestation.USER_ACTIVITY,
actor=ACTOR_UBUNTUONE,
subjects=[folder, other_user])
self.zg.log(event)
开发者ID:thnguyn2,项目名称:ECE_527_MP,代码行数:27,代码来源:zg_listener.py
注:本文中的zeitgeist.datamodel.Event类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论