Mercurial > libervia-backend
view tools/memory.py @ 0:c4bc297b82f0
sat:
- first public release, initial commit
author | goffi@necton2 |
---|---|
date | Sat, 29 Aug 2009 13:34:59 +0200 |
parents | |
children | bd9e9997d540 |
line wrap: on
line source
#!/usr/bin/python # -*- coding: utf-8 -*- """ SAT: a jabber client Copyright (C) 2009 Jérôme Poisson (goffi@goffi.org) This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program. If not, see <http://www.gnu.org/licenses/>. """ from __future__ import with_statement import os.path import time import pickle from logging import debug, info, error const_SAVEFILE_PARAM=os.path.expanduser("~/.sat_param.save") const_SAVEFILE_HISTORY=os.path.expanduser("~/.sat_history.save") class Memory: """This class manage all persistent informations""" def __init__(self): info ("Memory manager init") self.contact={} self.presenceStatus={} self.params={} self.history={} self.disco={} #XXX: maybe best in a separate class self.features={} self.load() def load_default_params(self): """Load default value, in case of no save file or error.""" self.createParam("JabberID", "goffi@jabber.goffi.int/TestScript", "string", "Connection") self.createParam("Password", "toto", "password", "Connection") self.createParam("Server", "jabber.goffi.int", "string", "Connection") self.createParam("IP", "192.168.0.2", "string", "File Transfert") self.createParam("Port", "28915", "string", "File Transfert") self.createParam("Watched", "test@jabber.goffi.int", "string", "Misc") #TODO: a mettre dans un plugin def load(self): """Load parameters and all memory things from file/db""" #first parameters if os.path.exists(const_SAVEFILE_PARAM): try: with open(const_SAVEFILE_PARAM, 'r') as params_pickle: self.params=pickle.load(params_pickle) debug("params loaded") except: error ("Can't load params !") self.load_default_params() else: self.load_default_params() #history if os.path.exists(const_SAVEFILE_HISTORY): try: with open(const_SAVEFILE_HISTORY, 'r') as history_pickle: self.history=pickle.load(history_pickle) debug("history loaded") except: error ("Can't load history !") def save(self): """Save parameters and all memory things to file/db""" with open(const_SAVEFILE_PARAM, 'w') as params_pickle: pickle.dump(self.params, params_pickle) with open(const_SAVEFILE_HISTORY, 'w') as history_pickle: pickle.dump(self.history, history_pickle) def addToHistory(self, me_jid, from_jid, to_jid, type, message): me_short=me_jid.userhost() from_short=from_jid.userhost() to_short=to_jid.userhost() if from_jid==me_jid: key=to_short else: key=from_short if not self.history.has_key(me_short): self.history[me_short]={} if not self.history[me_short].has_key(key): self.history[me_short][key]={} self.history[me_short][key][int(time.time())] = (from_short, message) def getHistory(self, from_jid, to_jid, size): ret={} if not self.history.has_key(from_jid): error("source JID not found !") #TODO: throw an error here return {} if not self.history[from_jid].has_key(to_jid): error("dest JID not found !") #TODO: throw an error here return {} stamps=self.history[from_jid][to_jid].keys() stamps.sort() for stamp in stamps[-size:]: ret[stamp]=self.history[from_jid][to_jid][stamp] return ret def addContact(self, JID, attributes, groups): debug("Memory addContact: %s",JID) assert(isinstance(attributes,dict)) assert(isinstance(groups,list)) self.contact[JID]=[attributes, groups] def addPresenceStatus(self, jid, type, show, status, priority): self.presenceStatus[jid]=[type, show, status, priority] def getContacts(self): debug ("Memory getContact OK (%s)", self.contact) ret=[] for contact in self.contact: ret.append([contact] + [self.contact[contact][0]] + [self.contact[contact][1]]) #very ugly I know ! return ret def getPresenceStatus(self): status=[] for contact, contactStatus in self.presenceStatus.items(): status.append([contact]+contactStatus) debug ("Memory getPresenceStatus (%s)", status) return status def getParamV(self, name, namespace): """Helper method to get the value in the good type getParamV stands for GetParamValue""" if not self.params.has_key(namespace) or not self.params[namespace].has_key(name): error("Requested param %s in namespace %s doesn't exist !" , name, namespace) return None return self.params[namespace][name][0] def getParam(self, name, namespace): if self.params.has_key(namespace) and self.params[namespace].has_key(name): return self.params[namespace][name] return ["",""] def getParams(self, namespace): if self.params.has_key(namespace): ret=[] for name in self.params[namespace]: ret.append([name] + self.params[namespace][name]) return ret return [[]] def getParamsCategories(self): """return the namespaces availables""" return self.params.keys() def setParam(self, name, value, namespace): if not self.params.has_key(namespace) or not self.params[namespace].has_key(name): return #TODO: throw an error self.params[namespace][name][0]=value; def createParam (self, name, value, type, namespace): ### TODO: add desciption in params if not self.params.has_key(namespace): self.params[namespace]={} self.params[namespace][name]=[value,type]; def registerFeature(self, feature, callback=None): self.features[feature]=callback