Skip to content

GitLab

H
Hackspace Device Info
Commit 3b08d312 authored by Markus's avatar Markus
Browse files
Options

Merge branch 'master' into 'master' 

Dicker Re-Faktor

# Re-Faktor

- Split main-file into separate modules.

Also:

- Tidy some un-funky code ;)
- Make files PEP-8 compliant
parents 22484ef7 a99ed4bf
Hide whitespace changes
Inline Side-by-side
Showing with 428 additions and 205 deletions
.gitignore 0 → 100644
View file @ 3b08d312
# ignore object-directory and contents
__pycache__
hack_count
View file @ 3b08d312
#!/usr/bin/env python3
import dbus
import urllib.request
import json
import argparse
import os
from time import sleep
import sys
import itertools
USER_HOME = os.path.expanduser("~")
CACHE_FOLDER = os.path.join(str(USER_HOME),".cache/hack_count")
CACHE_PICTURE_FOLDER = os.path.join(str(USER_HOME),".cache/hack_count/pics")
DEFAULT_CACHE_PICTURE = os.path.join(str(USER_HOME),".cache/hack_count/pics/default.png")
CACHE_FILE = "hack_count.json"
CACHE_PATH = os.path.join(str(CACHE_FOLDER),str(CACHE_FILE))
count_changed=True
import argparse
item = ('org.freedesktop.Notifications')
path = ('/org/freedesktop/Notifications')
interface = ('org.freedesktop.Notifications')
icon = ''
array = ''
hint = ''
time = 10000 # Use seconds x 1000
app_name = ('Hack-Space User Count')
import hc_daemon
import hc_main
###############################################################################
# main entry point
###############################################################################
if __name__ == '__main__':
###parser
parser = argparse.ArgumentParser(description='Access peoplecounter in \
Hackspace and create notification or print the information ')
parser.add_argument("-n","--notify",action="store_true",help="Create a \
notification using dbus (default=False) enable with -n",default=False)
parser.add_argument("-p","--printing",action="store_false",help="Enable or disable\
printing (default=enabled) disable printing with -p",default=True)
parser.add_argument("-u","--show_user",action="store_false",help="Show count\
of users or devices in the hackspace (default=user) toggle with -u)",
default=True)
parser.add_argument("-d","--daemonize",action="store_true",help="Daemonize the\
script (default=False) enable with -d set interval with -i",default=False)
parser.add_argument("-i","--interval",type=int,help="Interval for\
update in seconds (default=30)",default=30)
#### parse command line arguments
# create parser
parser = argparse.ArgumentParser(description='''Access peoplecounter in
Hackspace and create notification or print the information''')
## add arguments to be parsed
# notify option
parser.add_argument(
"-n",
"--notify",
action="store_true",
help="Create a notification using dbus (default=False) enable with -n",
default=False)
# print option
parser.add_argument(
"-p",
"--printing",
action="store_false",
help="""Enable or disable printing (default=enabled)
disable printing with -p""",
default=True)
# show user option
parser.add_argument(
"-u",
"--show_user",
action="store_false",
help="""Show count of users or devices in the hackspace (default=user)
toggle with -u""",
default=True)
# deamon option
parser.add_argument(
"-d",
"--daemonize",
action="store_true",
help="""Daemonize the script (default=False) enable with -d
set interval with -i""",
default=False)
# interval option
parser.add_argument(
"-i",
"--interval",
type=int,
help="Interval for update in seconds (default=30)",
default=30)
# parse arguments from commandline
args = parser.parse_args()
if args.daemonize == True :
print("Printing is disabled because you started hack_count in deamon mode.")
args.printing = False
try:
pid = os.fork()
if pid > 0:
# exit first parent
sys.exit(0)
except OSError:
sys.stderr.write("fork failed")
sys.exit(1)
while True :
##fetch usercount
var = urllib.request.urlopen("http://hack-hro.de/api/users")
json_string = var.read()
json_dict = json.loads(json_string.decode('ascii'))
device_count = json_dict['deviceCount']
users = json_dict['users']
#####create cache folder if not exsits####
if not os.path.exists(CACHE_PICTURE_FOLDER) :
os.makedirs(CACHE_PICTURE_FOLDER)
if not os.path.isfile(DEFAULT_CACHE_PICTURE):
urllib.request.urlretrieve("http://hack-hro.de/avatar/default.png",DEFAULT_CACHE_PICTURE)
for user in users :
user_image_path = os.path.join(CACHE_PICTURE_FOLDER,str(user['id'])+".jpg")
if not os.path.isfile(user_image_path) :
try :
user['avatar']
except :
pass
else :
urllib.request.urlretrieve("http://hack-hro.de"+user['avatar'],user_image_path)
####fetch cached vals
try:
cached_data_dict=json.loads(open(CACHE_PATH).read())
count_changed=False
except:
open(CACHE_PATH, 'a').close()
cached_data_dict={
"device_count" : 0,
"users" : users
}
#compare saved and fetched value
if args.show_user == True and (cached_data_dict['device_count'] != device_count ) :
count_changed=True
#diff saved and new recieved list
user_diff_fetched = list(itertools.filterfalse(lambda x: x in
cached_data_dict['users'],users))
user_diff_saved = list(itertools.filterfalse(lambda x: x in
users, cached_data_dict['users']))
#####prepare to save value
save_cache_dict= {
"device_count" : device_count,
"users" : users
}
if user_diff_fetched.__len__() > 0 :
count_changed = True
user_new_list = user_diff_fetched
if user_diff_saved.__len__() > 0 :
count_changed = True
user_left_list = user_diff_saved
### save to cachefile ####
with open(os.path.join(str(CACHE_FOLDER),str(CACHE_FILE)),"w+" ) as iofile:
json.dump(save_cache_dict,iofile)
iofile.close()
##print count
if args.printing == True :
if args.show_user == False :
print ("HackHro Users: %s " % users.__len__(),end="")
try:
user_left_list
except NameError:
pass
else:
print ("Users left: ",end="",flush = True)
for user in user_left_list :
print (user['name']+" " ,end="",flush = True )
try:
user_new_list
except NameError :
pass
else :
print ("Users new: ",end="",flush = True )
for user in user_new_list :
print (user['name']+" ",end="",flush = True )
else :
print ("HackHro Devcies: ",device_count,flush = True )
##notification
if args.notify and count_changed == True :
###init dbus notifications
bus = dbus.SessionBus()
notif = bus.get_object(item, path)
notify = dbus.Interface(notif, interface)
notification_string = ""
if args.show_user == False :
title = "HackSpace Users Changed"
try:
user_left_list
except NameError:
pass
else:
for user in user_left_list :
icon = os.path.join(CACHE_PICTURE_FOLDER,str(user['id'])+".jpg")
if not os.path.isfile(icon):
icon = os.path.join(CACHE_PICTURE_FOLDER,"default.png")
notification_string = "the user "+str(user['name'])+" left."
#create notification
notify.Notify(app_name, 0, icon, title, notification_string, array, hint, time)
try:
user_new_list
except NameError :
pass
else :
for user in user_new_list :
icon = os.path.join(CACHE_PICTURE_FOLDER,str(user['id'])+".jpg")
if not os.path.isfile(icon):
icon = os.path.join(CACHE_PICTURE_FOLDER,"default.png")
notification_string = "the user "+str(user['name'])+" just came in."
#create notification
notify.Notify(app_name, 0, icon, title, notification_string, array, hint, time)
else :
title = "HackSpace Devices Changed"
notification_string = str(device_count)
#create notification
notify.Notify(app_name, 0, icon, title, notification_string, array, hint, time)
count_changed = False
try:
del user_new_list
del user_left_list
except:
pass
if not args.daemonize == True :
sys.exit(0)
sleep(args.interval)
if args.daemonize:
hc_daemon.Daemon(args).start()
try:
# create Main object and start it
hc_main.Main(args).start()
except KeyboardInterrupt:
print('\rExiting on Keyboard-Interrupt...')
exit(0)
hc_daemon.py 0 → 100644
View file @ 3b08d312
# -*- coding: utf-8 -*-
import os
import sys
class Daemon(object):
"""Daemon class for usage in background."""
def __init__(self, args):
""""""
self.args = args
def start(self):
"""Start daemonizing this program."""
print("hack_count started in daemon-mode (printing disabled)...")
self.args.printing = False
try:
pid = os.fork()
if pid > 0:
# exit first parent
sys.exit(0)
except OSError:
sys.stderr.write("fork failed")
sys.exit(1)
\ No newline at end of file
hc_data.py 0 → 100644
View file @ 3b08d312
# -*- coding: utf-8 -*-
import os
import json
from urllib import request
import hc_head
class Data(object):
"""Class representing a data-interface for hack_count."""
def __init__(self, var=request.urlopen("http://hack-hro.de/api/users")):
"""Initialize values and create necessary dirs."""
#### fetch usercount
# fetch data from server
self.var = var
# create data-objects by decoding the json
self.json_string = self.var.read()
self.json_dict = json.loads(self.json_string.decode('ascii'))
self.device_count = self.json_dict['deviceCount']
self.users = self.json_dict['users']
# create picture folder, if not existant
if not os.path.exists(hc_head.CACHE_PICTURE_FOLDER):
os.makedirs(hc_head.CACHE_PICTURE_FOLDER)
# fetch avatar picture, if not existant
if not os.path.isfile(hc_head.DEFAULT_CACHE_PICTURE):
request.urlretrieve("http://hack-hro.de/avatar/default.png",
hc_head.DEFAULT_CACHE_PICTURE)
def update(self,
var=request.urlopen("http://hack-hro.de/api/users")):
"""Update values, if required."""
# replace var, if not already matching
if var != self.var:
self.var = var
# create data-objects by decoding the json
self.json_string = self.var.read()
self.json_dict = json.loads(self.json_string.decode('ascii'))
self.device_count = self.json_dict['deviceCount']
self.users = self.json_dict['users']
\ No newline at end of file
hc_head.py 0 → 100644
View file @ 3b08d312
# -*- coding: utf-8 -*-
import os
###############################################################################
# global constants
###############################################################################
USER_HOME = os.path.expanduser("~")
CACHE_FOLDER = os.path.join(str(USER_HOME),
".cache/hack_count")
CACHE_PICTURE_FOLDER = os.path.join(str(USER_HOME),
".cache/hack_count/pics")
DEFAULT_CACHE_PICTURE = os.path.join(str(USER_HOME),
".cache/hack_count/pics/default.png")
CACHE_FILE = "hack_count.json"
CACHE_PATH = os.path.join(str(CACHE_FOLDER),
str(CACHE_FILE))
###############################################################################
# end of global constants
###############################################################################
class Variables(object):
"""Use this class as a sort of struct of variables.
See https://docs.python.org/3/tutorial/classes.html#odds-and-ends
for the main idea of this... ehm... idea. :D"""
def __init__(self):
"""Create and initialize the variables used in hack_count."""
####
self.count_changed = True
self.item = ('org.freedesktop.Notifications')
self.path = ('/org/freedesktop/Notifications')
self.interface = ('org.freedesktop.Notifications')
self.icon = ''
self.array = ''
self.hint = ''
# Use seconds x 1000
self.time = 10000
self.app_name = ('Hack-Space User Count')
hc_main.py 0 → 100644
View file @ 3b08d312
# -*- coding: utf-8 -*-
import os
import sys
import dbus
import json
import itertools
import urllib.error
import urllib.request
from time import sleep
import hc_head
import hc_data
class Main(object):
"""Class representing the main procedures."""
def __init__(self, args):
""""""
# store command line arguments
self.args = args
# create variable-object
self.header = hc_head.Variables()
# create date-interface
try:
# import hc_data here, to save one try-except clause :D
import hc_data
# create data_interface
self.data_interface = hc_data.Data()
except urllib.error.HTTPError as e:
print('Could not connect to url "http://hack-hro.de/api/users":')
print(e)
exit(1)
# initiate lists
self.user_left_list = None
self.user_new_list = None
def caching_routine(self):
""""""
#######################################################################
# fetch-routine
#### fetch cached vals
#TODO: check, when data from server is available
try:
cached_data_dict = json.loads(open(hc_head.CACHE_PATH).read())
self.header.count_changed = False
except:
open(hc_head.CACHE_PATH, 'a').close()
cached_data_dict = {
"device_count": 0,
"users": self.data_interface.users
}
#compare saved and fetched value
if (self.args.show_user and
cached_data_dict['device_count'] != self.data_interface.device_count):
self.header.count_changed = True
#diff saved and new recieved list
user_diff_fetched = list(
itertools.filterfalse(lambda x: x in cached_data_dict['users'],
self.data_interface.users))
user_diff_saved = list(
itertools.filterfalse(lambda x: x in self.data_interface.users,
cached_data_dict['users']))
#####prepare to save value
save_cache_dict = {
"device_count": self.data_interface.device_count,
"users": self.data_interface.users
}
if len(user_diff_fetched) > 0:
self.header.count_changed = True
self.user_new_list = user_diff_fetched
if len(user_diff_saved) > 0:
self.header.count_changed = True
self.user_left_list = user_diff_saved
### save to cachefile
with open(os.path.join(str(hc_head.CACHE_FOLDER),
str(hc_head.CACHE_FILE)),
"w+") as iofile:
json.dump(save_cache_dict, iofile)
iofile.close()
# end of fetch-routine
#######################################################################
def printing_routine(self):
""""""
#######################################################################
# printing routine
## print count
if self.args.printing:
if self.args.show_user:
print("HackHro Users: {} ".format(len(self.data_interface.users)))
if self.user_left_list is not None:
print("Users left: ", flush=True, end="")
for user in user_left_list:
print(user['name'] + " ", flush=True)
if self.user_new_list is not None:
print("Users new: ", flush=True, end="")
for user in user_new_list:
print(user['name'] + " ", end="", flush=True)
else:
print("HackHro Devcies: ", self.data_interface.device_count, flush=True)
# end of printing routine
#######################################################################
def notification_routine(self):
""""""
#######################################################################
# notification routine
if self.args.notify and self.header.count_changed:
###init dbus notifications
bus = dbus.SessionBus()
notif = bus.get_object(self.header.item, self.header.path)
notify = dbus.Interface(notif, self.header.interface)
notification_string = ""
if not self.args.show_user:
title = "HackSpace Users Changed"
if self.user_left_list is not None:
for user in self.user_left_list:
self.header.icon = os.path.join(
hc_head.CACHE_PICTURE_FOLDER,
str(user['id']) + ".jpg")
if not os.path.isfile(self.header.icon):
self.header.icon = os.path.join(
hc_head.CACHE_PICTURE_FOLDER,
"default.png")
notification_string = "the user " +\
str(user['name']) +\
" left."
#create notification
notify.Notify(self.header.app_name,
0,
self.header.icon,
title,
notification_string,
self.header.array,
self.header.hint,
self.header.time)
if self.user_new_list is not None:
for user in self.user_new_list:
self.header.icon = os.path.join(hc_head.
hc_head.CACHE_PICTURE_FOLDER,
str(user['id']) + ".jpg")
if not os.path.isfile(self.header.icon):
self.header.icon = os.path.join(
hc_head.CACHE_PICTURE_FOLDER,
"default.png")
notification_string = "the user " +\
str(user['name']) +\
" just came in."
#create notification
notify.Notify(self.header.app_name,
0,
self.header.icon,
title,
notification_string,
self.header.array,
self.header.hint,
self.header.time)
else:
title = "HackSpace Devices Changed"
notification_string = str(device_count)
#create notification
notify.Notify(self.header.app_name,
0,
self.header.icon,
title,
notification_string,
self.header.array,
self.header.hint,
self.header.time)
self.header.count_changed = False
self.user_left_list = None
self.user_new_list = None
# end of notification routine
###################################################################
def start(self):
""""""
while True:
# update variables from url
try:
self.data_interface.update()
# server might be down, so catch HTTPErrors
except urllib.error.HTTPError as e:
print('Could not connect to url: \
"http://hack-hro.de/api/users":')
print(e)
exit(1)
# for each user logged in
for user in self.data_interface.users:
# create a path of its avatar
user_image_path = os.path.join(hc_head.CACHE_PICTURE_FOLDER,
str(user['id']) + ".jpg")
# check, whether this avatar exists
if not os.path.isfile(user_image_path):
# if not, check whether there are resources available
if 'avatar' in user:
# and retrieve them
urllib.request.urlretrieve(
"http://hack-hro.de" + user['avatar'],
user_image_path)
# fetch data from cache/write data into cache
self.caching_routine()
# print user/device infos
self.printing_routine()
# notify user/device infos
self.notification_routine()
# break/continue loop depending on daemonize-value
if not self.args.daemonize:
sys.exit(0)
sleep(self.args.interval)