masq/i3pystatus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #4 from enkore/py3k

Browse Source 
Py3k
This commit is contained in:
Jan Oliver Oelerich 2013-02-15 15:53:29 -08:00
commit 1548209222
13 changed files with 405 additions and 361 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@ -1,2 +1,4 @@
*__pycache__*
*.pyc
wrapper.py
i3pystatus/__main__.py

1
MIT-LICENSE
View File

@ -1,4 +1,5 @@
Copyright (c) 2012 Jan Oliver Oelerich, http://www.oelerich.org
Copyright (c) 2013 mabe, http://enkore.de
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the

34
README.md
View File

@ -9,8 +9,8 @@ To install it, follow these steps:
cd ~/.config/i3status/
git clone git@github.com:janoliver/i3pystatus contrib
cd contrib
cp wrapper.py.dist wrapper.py
cd contrib/i3pystatus
cp __main__.py.dist __main__.py
Add the following to `~/.config/i3status/config`:
@ -24,14 +24,37 @@ Change your i3wm config to the following:
# i3bar
bar {
status_command i3status | python2 ~/.config/i3status/contrib/wrapper.py
status_command cd ~/.config/i3status/contrib ; i3status | python -m i3pystatus
position top
workspace_buttons yes
}
And finally adjust the settings in `~/.config/i3status/contrib/wrapper.py`
And finally adjust the settings in `~/.config/i3status/contrib/i3pystatus/__main__.py`
as you like.
## Modules
### thunderbirdnewmail
Requires
* python-dbus
* python-gobject2
Settings
* format
### modsde
Settings
* username
* password
* pause (delay between updates)
* offset (subtract number of posts before output)
* format
## Contribute
To contribute a script, make sure it has a function `output()` that outputs
@ -39,4 +62,5 @@ valid json code that can be interpreted by i3status. The protocol is documented
here: [i3status Protocol](http://i3wm.org/docs/i3bar-protocol.html).
Please add an example for how to configure it to `wrapper.py.dist`. It should be
a python class that can be registered with the `I3StatusHandler` class.
a python class that can be registered with the `I3StatusHandler` class.

95
i3pystatus/__init__.py Normal file
View File

@ -0,0 +1,95 @@
#!/usr/bin/env python
import sys
import json
import urllib.request, urllib.error, urllib.parse
from threading import Thread
import time
class BaseModule:
output = None
def registered(self, status_handler):
"""Called when this module is registered with a status handler"""
def tick(self):
"""Called once per tick"""
class Module(BaseModule):
pass
class AsyncModule(BaseModule):
def registered(self, status_handler):
self.thread = Thread(target=self.mainloop)
self.thread.daemon = True
self.thread.start()
def mainloop(self):
"""This is run in a separate daemon-thread"""
class IntervalModule(AsyncModule):
interval = 5 # seconds
def run(self):
"""Called every self.interval seconds"""
def mainloop(self):
while True:
self.run()
time.sleep(self.interval)
class I3statusHandler:
modules = []
def __init__(self):
pass
def register(self, module):
"""Register a new module."""
self.modules.append(module)
module.registered(self)
def print_line(self, message):
"""Unbuffered printing to stdout."""
sys.stdout.write(message + "\n")
sys.stdout.flush()
def read_line(self):
"""Interrupted respecting reader for stdin."""
# try reading a line, removing any extra whitespace
try:
line = sys.stdin.readline().strip()
# i3status sends EOF, or an empty line
if not line:
sys.exit(3)
return line
# exit on ctrl-c
except KeyboardInterrupt:
sys.exit()
def run(self):
self.print_line(self.read_line())
self.print_line(self.read_line())
while True:
line, prefix = self.read_line(), ""
# ignore comma at start of lines
if line.startswith(","):
line, prefix = line[1:], ","
j = json.loads(line)
for module in self.modules:
module.tick()
output = module.output
if output:
j.insert(0, output)
# and echo back new encoded json
self.print_line(prefix+json.dumps(j))

57
i3pystatus/__main__.py.dist Executable file
View File

@ -0,0 +1,57 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from i3pystatus import (
I3statusHandler,
mailchecker,
modsde,
notmuchmailchecker,
thunderbird,
)
status = I3statusHandler()
# The imap checker module
mailsettings = {
"color": "#ff0000",
"servers": [
{
"host": "www.testhost1.com",
"port": "993",
"ssl" : True,
"username": "your_username",
"password": "your_password",
"pause": 20
},
{
"host": "www.testhost2.net",
"port": "993",
"ssl" : True,
"username": "your_username",
"password": "your_password",
"pause": 20
}
]
}
mailchecker = mailchecker.MailChecker(mailsettings)
status.register_module(mailchecker)
# the mods.de forum new bookmarks module
mdesettings = {
"username": "your_username",
"password": "your_password"
}
mde = modsde.ModsDeChecker(mdesettings)
status.register_module(mde)
# the notmuch mail checker module
db_path = "path_to_your_notmuch_database"
notmuch = notmuchmailchecker.NotmuchMailChecker(db_path)
status.register_module(notmuch)
# the thunderbird dbus new mail checker module
tb = thunderbirdnewmail.ThunderbirdMailChecker()
status.register_module(tb)
# start the handler
status.run()

91
i3pystatus/mailchecker.py Normal file
View File

@ -0,0 +1,91 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
import json
from datetime import datetime,timedelta
import imaplib
from i3pystatus import IntervalModule
class MailChecker(IntervalModule):
"""
This class handles mailservers and outputs i3status compatible
json data for the accumulated unread count. The mail server
functionality is implemented in the subclass MailChecker.MailServer
"""
settings = {
"color": "#ff0000",
"servers": []
}
servers = []
def __init__(self, settings = None):
self.settings.update(settings)
for server in settings["servers"]:
srv = MailChecker.MailServer(server)
self.servers.append(srv)
def run(self):
unread = sum([server.get_unread_count() for server in self.servers])
if not unread:
return None
self.output = {
"full_text" : "%d new email%s" % (unread, ("s" if unread > 1 else "")),
"name" : "newmail",
"urgent" : "true",
"color" : self.settings["color"]
}
class MailServer:
"""
This class provides the functionality to connect
to a mail server and fetch the count of unread emails.
When the server connection is lost, it returns 0 and
tries to reconnect. It checks every "pause" seconds.
"""
host = ""
port = ""
imap_class = imaplib.IMAP4
username = ""
password = ""
connection = None
def __init__(self, settings_dict):
self.host = settings_dict["host"]
self.port = settings_dict["port"]
self.username = settings_dict["username"]
self.password = settings_dict["password"]
if settings_dict["ssl"]:
self.imap_class = imaplib.IMAP4_SSL
def get_connection(self):
if not self.connection:
try:
self.connection = self.imap_class(self.host, self.port)
self.connection.login(self.username, self.password)
self.connection.select()
except Exception:
self.connection = None
try:
self.connection.select()
except Exception as e:
self.connection = None
return self.connection
def get_unread_count(self):
unread = 0
conn = self.get_connection()
if conn:
unread += len(conn.search(None,"UnSeen")[1][0].split())
return unread

81
i3pystatus/modsde.py Normal file
View File

@ -0,0 +1,81 @@
#!/usr/bin/env python
import sys
import json
import time
import threading
import urllib.request, urllib.parse, urllib.error, urllib.request, urllib.error, urllib.parse
import re
import http.cookiejar
import xml.etree.ElementTree as ET
from i3pystatus import IntervalModule
class ModsDeChecker(IntervalModule):
"""
This class returns i3status parsable output of the number of
unread posts in any bookmark in the mods.de forums.
"""
login_url = "http://login.mods.de/"
bookmark_url = "http://forum.mods.de/bb/xml/bookmarks.php"
opener = None
cj = None
logged_in = False
settings = {
"color": "#7181fe",
"offset": 0,
"format": "%d new posts in bookmarks"
}
def __init__(self, settings = None):
self.settings.update(settings)
self.cj = http.cookiejar.CookieJar()
self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.cj))
def run(self):
unread = self.get_unread_count()
if not unread:
self.output = None
else:
self.output = {
"full_text" : self.settings["format"] % unread,
"name" : "modsde",
"urgent" : "true",
"color" : self.settings["color"]
}
def get_unread_count(self):
if not self.logged_in:
self.login()
try:
f = self.opener.open(self.bookmark_url)
root = ET.fromstring(f.read())
return int(root.attrib["newposts"]) - self.settings["offset"]
except Exception:
self.cj.clear()
self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.cj))
self.logged_in = False
def login(self):
data = urllib.parse.urlencode({
"login_username": self.settings["username"],
"login_password": self.settings["password"],
"login_lifetime": "31536000"
})
response = self.opener.open(self.login_url, data.encode("ascii"))
m = re.search("http://forum.mods.de/SSO.php[^']*", response.read().decode("ISO-8859-15"))
self.cj.clear()
if m and m.group(0):
# get the cookie
response = self.opener.open(m.group(0))
for cookie in self.cj:
self.cj.clear
self.logged_in = True
self.opener.addheaders.append(("Cookie", "{}={}".format(cookie.name, cookie.value)))
return True

28
notmuchmailchecker.py → i3pystatus/notmuch.py
View File

@ -6,30 +6,34 @@
import notmuch
import json
class NotmuchMailChecker(object):
from i3pystatus import IntervalModule
class NotmuchMailChecker(IntervalModule):
"""
This class uses the notmuch python bindings to check for the
number of messages in the notmuch database with the tags "inbox"
and "unread"
"""
db_path = ''
db_path = ""
def __init__(self, db_path):
self.db_path = db_path
def output(self):
def run(self):
db = notmuch.Database(self.db_path)
unread = notmuch.Query(db, 'tag:unread and tag:inbox').count_messages()
unread = notmuch.Query(db, "tag:unread and tag:inbox").count_messages()
if (unread == 0):
color = '#00FF00'
urgent = 'false'
color = "#00FF00"
urgent = "false"
else:
color = '#ff0000'
urgent = 'true'
color = "#ff0000"
urgent = "true"
return {'full_text' : '%d new email%s' % (unread, ('s' if unread > 1 else '')),
'name' : 'newmail',
'urgent' : urgent,
'color' : color }
self.output = {
"full_text" : "%d new email%s" % (unread, ("s" if unread > 1 else "")),
"name" : "newmail",
"urgent" : urgent,
"color" : color
}

42
thunderbirdnewmail.py → i3pystatus/thunderbird.py
View File

@ -10,16 +10,27 @@
import dbus, gobject
from dbus.mainloop.glib import DBusGMainLoop
import json
import threading
import time
class ThunderbirdMailChecker(object):
from i3pystatus import AsyncModule
class ThunderbirdMailChecker(AsyncModule):
"""
This class listens for dbus signals emitted by
the dbus-sender extension for thunderbird.
"""
unread = []
settings = {
"format": "%d new email"
}
unread = set()
def __init__(self, settings=None):
if settings is not None:
self.settings.update(settings)
def __init__(self):
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus = dbus.SessionBus()
bus.add_signal_receiver(self.new_msg,
@ -32,20 +43,31 @@ class ThunderbirdMailChecker(object):
dbus.mainloop.glib.threads_init()
self.context = loop.get_context()
def mainloop(self):
while True:
self.context.iteration(False)
time.sleep(1)
def new_msg(self, id, author, subject):
if id not in self.unread:
self.unread.append(id)
self.unread.add(id)
self._output()
def changed_msg(self, id, event):
if event == "read" and id in self.unread:
self.unread.remove(id)
self._output()
def output(self):
def _output(self):
self.context.iteration(False)
unread = len(self.unread)
return {'full_text' : '%d new email' % unread,
'name' : 'newmail-tb',
'urgent' : True,
'color' : '#ff0000' } if unread else None
if unread:
self.output = {
"full_text": self.settings["format"] % unread,
"name": "newmail-tb",
"urgent": True,
"color": "#ff0000",
}
else:
self.output = None

107
mailchecker.py
View File

@ -1,107 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
import json
from datetime import datetime,timedelta
import imaplib
from statushandler import has_internet_connection
class MailChecker(object):
"""
This class handles mailservers and outputs i3status compatible
json data for the accumulated unread count. The mail server
functionality is implemented in the subclass MailChecker.MailServer
"""
settings = {
'color': '#ff0000',
'servers': []
}
servers = []
def __init__(self, settings = None):
self.settings.update(settings)
for server in settings['servers']:
srv = MailChecker.MailServer(server)
self.servers.append(srv)
def output(self):
unread = 0
for srv in self.servers:
unread += srv.get_unread_count()
if not unread:
return None
return {'full_text' : '%d new email%s' % (unread, ('s' if unread > 1 else '')),
'name' : 'newmail',
'urgent' : 'true',
'color' : self.settings['color']}
class MailServer:
"""
This class provides the functionality to connect
to a mail server and fetch the count of unread emails.
When the server connection is lost, it returns 0 and
tries to reconnect. It checks every 'pause' seconds.
"""
host = ""
port = ""
imap_class = imaplib.IMAP4
username = ""
password = ""
connection = None
pause = 30
unread_cache = 0
last_checked = datetime.now()
def __init__(self, settings_dict):
self.host = settings_dict['host']
self.port = settings_dict['port']
self.username = settings_dict['username']
self.password = settings_dict['password']
self.pause = settings_dict['pause']
if settings_dict['ssl']:
self.imap_class = imaplib.IMAP4_SSL
self.last_checked = \
datetime.now() - timedelta(seconds=self.pause)
def get_connection(self):
if not has_internet_connection():
self.connection = None
else:
if not self.connection:
try:
self.connection = self.imap_class(self.host, self.port)
self.connection.login(self.username, self.password)
self.connection.select()
except Exception:
self.connection = None
try:
self.connection.select()
except Exception,e:
self.connection = None
return self.connection
def get_unread_count(self):
delta = datetime.now() - self.last_checked
if delta.total_seconds() > self.pause:
unread = 0
conn = self.get_connection()
if conn:
unread += len(conn.search(None,'UnSeen')[1][0].split())
self.unread_cache = unread
self.last_checked = datetime.now()
return self.unread_cache

98
modsde.py
View File

@ -1,98 +0,0 @@
#!/usr/bin/env python
import sys
import json
from datetime import datetime,timedelta
import urllib, urllib2
import re
import cookielib
import xml.etree.ElementTree as ET
class ModsDeChecker:
"""
This class returns i3status parsable output of the number of
unread posts in any bookmark in the mods.de forums.
"""
last_checked = datetime.now()
unread_cache = 0
login_url = 'http://login.mods.de/'
bookmark_url = "http://forum.mods.de/bb/xml/bookmarks.php"
opener = None
cj = None
logged_in = False
settings = {
'color': '#7181fe',
'pause': 20,
'username': "",
'password': ""
}
def __init__(self, settings = None):
self.settings.update(settings)
self.cj = cookielib.CookieJar()
self.last_checked = \
datetime.now() - timedelta(seconds=self.settings['pause'])
self.opener = urllib2.build_opener(
urllib2.HTTPCookieProcessor(self.cj))
def get_unread_count(self):
delta = datetime.now() - self.last_checked
if delta.total_seconds() > self.settings['pause']:
if not self.logged_in:
try:
self.login()
except Exception:
pass
try:
f = self.opener.open(self.bookmark_url)
root = ET.fromstring(f.read())
self.last_checked = datetime.now()
self.unread_cache = int(root.attrib['newposts'])
except Exception:
self.cj.clear()
self.opener = urllib2.build_opener(
urllib2.HTTPCookieProcessor(self.cj))
self.logged_in = False
return self.unread_cache
def login(self):
data = urllib.urlencode({
"login_username": self.settings["username"],
"login_password": self.settings["password"],
"login_lifetime": "31536000"
})
response = self.opener.open(self.login_url, data)
m = re.search("http://forum.mods.de/SSO.php[^']*", response.read())
self.cj.clear()
if m and m.group(0):
# get the cookie
response = self.opener.open(m.group(0))
for cookie in self.cj:
self.cj.clear
self.logged_in = True
self.opener.addheaders.append(('Cookie',
'{}={}'.format(cookie.name, cookie.value)))
return True
return False
def output(self):
unread = self.get_unread_count()
if not unread:
return None
return {'full_text' : '%d new posts in bookmarks' % unread,
'name' : 'modsde',
'urgent' : 'true',
'color' : self.settings['color']}

72
statushandler.py
View File

@ -1,72 +0,0 @@
#!/usr/bin/env python
import sys
import json
import urllib2
class I3statusHandler:
modules = []
def __init__(self):
pass
def register_module(self, module):
""" Register a new module. """
# check if module implemented the
# correct functions
if not hasattr(module, 'output'):
raise Exception("Module %s does not implement \
all the needed functions!".format(module))
self.modules.append(module)
def print_line(self, message):
""" Non-buffered printing to stdout. """
sys.stdout.write(message + '\n')
sys.stdout.flush()
def read_line(self):
""" Interrupted respecting reader for stdin. """
# try reading a line, removing any extra whitespace
try:
line = sys.stdin.readline().strip()
# i3status sends EOF, or an empty line
if not line:
sys.exit(3)
return line
# exit on ctrl-c
except KeyboardInterrupt:
sys.exit()
def run(self):
self.print_line(self.read_line())
self.print_line(self.read_line())
while True:
line, prefix = self.read_line(), ''
# ignore comma at start of lines
if line.startswith(','):
line, prefix = line[1:], ','
j = json.loads(line)
for module in self.modules:
output = module.output()
if output:
j.insert(0, module.output())
# and echo back new encoded json
self.print_line(prefix+json.dumps(j))
def has_internet_connection():
try:
response=urllib2.urlopen('http://74.125.113.99',timeout=1)
return True
except urllib2.URLError as err: pass
return False

56
wrapper.py.dist
View File

@ -1,56 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import mailchecker
import modsde
import notmuchmailchecker
from statushandler import I3statusHandler
if __name__ == '__main__':
status = I3statusHandler()
# The imap checker module
mailsettings = {
'color': '#ff0000',
'servers': [
{
'host': 'www.testhost1.com',
'port': '993',
'ssl' : True,
'username': 'your_username',
'password': 'your_password',
'pause': 20
},
{
'host': 'www.testhost2.net',
'port': '993',
'ssl' : True,
'username': 'your_username',
'password': 'your_password',
'pause': 20
}
]
}
mailchecker = mailchecker.MailChecker(mailsettings)
status.register_module(mailchecker)
# the mods.de forum new bookmarks module
mdesettings = {
'username': "your_username",
'password': "your_password"
}
mde = modsde.ModsDeChecker(mdesettings)
status.register_module(mde)
# the notmuch mail checker module
db_path = 'path_to_your_notmuch_database'
notmuch = notmuchmailchecker.NotmuchMailChecker(db_path)
status.register_module(notmuch)
# the thunderbird dbus new mail checker module
tb = thunderbirdnewmail.ThunderbirdMailChecker()
status.register_module(tb)
# start the handler
status.run()