Commit 70eb6a61 authored by Martin Matějek's avatar Martin Matějek

WIP - proof of concept

parent 555aea02
#!/usr/bin/env python3
import argparse
from notifylib.api import Api
"""
Interface:
* list <id>
* list (all)
* create <template, message, **params>
* dismiss <id>
* action <id, action_name>
* available notifications
* available media types <template_name>??
* available actions of plugin <name>
* show complete message as media type (type)
"""
def create_argparser():
"""Create new argument parser"""
parser = argparse.ArgumentParser()
parser.add_argument("-c", "--config", help="Specify config file")
subparsers = parser.add_subparsers(help="sub-command help", dest='command')
parser_action = subparsers.add_parser("add", help="Add new notification")
parser_action.add_argument("message", help="Notification message")
parser_action.add_argument("-t", "--template", help="Notification type / template", default='simple')
parser_list = subparsers.add_parser("list", help="List notification")
# parser_list.add_argument("--id", help="ID of notification")
parser_list.add_argument("plugins", help="List available plugins")
return parser
def process_args(parser, api, args):
"""Call module interface based on args"""
# not working yet
if args.command == 'add':
api.create(**vars(args))
if args.command == 'list':
if args.plugins:
ret = api.get_plugins()
print("Available plugins:")
for k, v in ret.items():
print(v)
else:
parser.print_usage()
def main():
parser = create_argparser()
args = parser.parse_args()
api = Api()
process_args(parser, api, args)
if __name__ == '__main__':
main()
from .config import config
from .pluginstorage import PluginStorage
from .notificationstorage import NotificationStorage
class Api:
"""Public interface of module"""
def __init__(self, conf=None):
print("Creating instance of API")
self.init_logger()
if conf: # override default config
config.load_config(conf)
self.plugins = PluginStorage(config.get('settings', 'plugin_dir'))
self.notifications = NotificationStorage(
config.get('settings', 'persistent_dir'),
config.get('settings', 'volatile_dir'),
self.plugins.get_notification_types()
)
print("Debug: Available notification_types: {}".format(self.notifications.notification_types))
def init_logger(self):
"""Init new logger instance"""
pass
def get_plugins(self):
return self.plugins.get_all()
def get_actions(self, plug_name):
"""Get actions of specified plugin"""
return self.plugins.get_plugin(plug_name).get_actions()
def get_notifications(self):
return self.notifications.get_all()
def get_notification(self, msgid):
"""Show notification of one specific by id"""
if msgid:
return self.notifications.get_notification(msgid)
# data manipulation
def create(self, **user_opts):
"""Create new notification"""
# get pre-filled skeleton of class Notification
print(user_opts)
notif = self.notifications.get_skeleton(user_opts['template'])
self.notifications.store(notif)
# TODO: refactor
def call_action(self, mgsid, name, **kwargs):
"""Call action on notification"""
pass
# storage.actions[name](**kwargs)
def dismiss(self, msgid):
"""Dismiss specific notification"""
self.call_action(msgid, 'dismiss')
import json
class Notification:
def __init__(self, name, template, actions, timestamp=None, **opts):
self.name = name
self.template = template
self.actions = actions
self.timestamp = timestamp
self.opts = opts
# TODO: parse opts into metadata
@classmethod
def from_file(cls, f):
"""Load notification from it's file"""
try:
dict = json.load(f)
return cls(**dict)
except Exception:
pass
# TODO: log failure
@classmethod
def create_instance(cls, **args):
return cls(args)
def valid(self, timestamp):
"""If notification is still valid"""
pass
def __str__(self):
pass
class NotificationStorage:
"""In-memory notification storage that serialize and deserialize them"""
def __init__(self, volatile_dir, persistent_dir, notification_types):
print("Constructing new NotifyStorage")
self.init_logger()
self.load(volatile_dir)
self.load(persistent_dir)
self.notification_types = notification_types # notification data types/templates
self.notifications = []
# self.cached = {}
def init_logger(self):
pass
def store(self, n):
"""Store in memory and serializate to disk"""
self.notifications.append(n)
self.serialize(n)
# TODO: WIP
#def store_persistent(self, n):
# self.serialize(n, self.storage_dirs['persistent'])
#def store_volatile(self, n):
# self.serialize(n, self.storage_dirs['volatile'])
#def serialize(self, n, destination):
# """Serialize notification to disk"""
# # save metadata to FS
# # render fallback form
# pass
def load(self, storage_dir):
"""Deserialize from FS"""
# for f in ls storage_dir:
# n = NotificationType.from_file(f)
# if not n.valid():
# delete_from_fs()
# TODO: find better key to identify notification instance in dict
def get_skeleton(self, name):
"""Return notification instance with filled in mandatory attributes"""
return self.notification_types[name].create_instance()
# TODO: find better key to identify notification instance in dict
def get_notification(self, name):
"""Return notification either cached or if missing, cache it and return"""
if not self.cached[name]:
# do something to create/cache
# do it inside serialization
pass
return self.cached[name]
def get_notification_types(self):
"""Return all notification types"""
return self.notification_types
# TODO: WIP helper fce
def render_one(self, notif):
pass
def render_all(self):
"""Render all notifications"""
for n in self.notifications:
self.render_one(n)
import yaml
class Plugin:
def __init__(self, name, actions, templates, notifications):
self.name = name
self.actions = []
self.templates = []
self.notifications = []
for a in actions:
self.actions.append(a)
for t in templates:
self.templates.append(t)
for n in notifications:
self.notifications.append(n)
@classmethod
def from_file(cls, filename):
with open(filename, 'r') as f:
data = yaml.load(f)
# TODO: better filename spliting handling
name = filename.split('.')[0].split('/')[1]
# print("YML data {}".format(data))
return cls(name, **data)
def get_actions(self):
return self.actions
def get_notification_types(self):
nt = []
for n in self.notifications:
nt.append(n)
return nt
def __str__(self):
"""For debug purposes"""
out = "{\n"
out += "\tname: {}\n".format(self.name)
for a in self.actions:
out += "\tActions: {}\n".format(a)
for t in self.templates:
out += "\tTemplates: {}\n".format(t)
for n in self.notifications:
out += "\tNotifications: {}\n".format(n)
out += "}"
return out
import os
from .plugin import Plugin
class PluginStorage:
"""Storage for plugins"""
def __init__(self, plugin_dir):
print("Constructing new PluginStorage")
self.plugin_dir = plugin_dir
self.init_logger()
self.plugins = {}
self.load()
def init_logger(self):
pass
def plugin_file_path(self, f):
return os.path.join(self.plugin_dir, f)
def load(self):
"""Load plugins from FS"""
for root, dirs, files in os.walk(self.plugin_dir):
for f in files:
if f.startswith('.'): # ignore dotfiles
continue
p = Plugin.from_file(self.plugin_file_path(f))
self.plugins[p.name] = p
# don't walk into lower levels
break
def get_plugin(self, name):
return self.plugins[name]
def get_all(self):
"""Return all plugins"""
return self.plugins
def get_notification_types(self):
ret = {}
for k, v in self.plugins.items():
ret[k] = v.get_notification_types()
return ret
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment