Ménage dans cranslib, 1er passage
This commit is contained in:
Vincent Le Gallic
parent
b0ae7b5589
commit
0ec65b2036
13 changed files with 0 additions and 1 deletions
File diff suppressed because it is too large
Load diff
|
|
@ -1,93 +0,0 @@
|
|||
#!/usr/bin/env python
|
||||
#
|
||||
# DIALOGWIZARD.PY--
|
||||
#
|
||||
# Copyright (C) 2009 Antoine Durand-Gasselin
|
||||
# Author: Antoine Durand-Gasselin <adg@crans.org>
|
||||
#
|
||||
|
||||
from wizard import Step, PreviousStep, EndScenario, TryAgain
|
||||
from itertools import izip
|
||||
import dialog, time, sys, commands
|
||||
|
||||
def really_quit(dico):
|
||||
raise EndScenario("game over!", data = dico)
|
||||
|
||||
def error_handler(exc):
|
||||
dialog.Dialog().msgbox(str(exc), widht=0, height =0, title = 'Erreur :')
|
||||
raise TryAgain()
|
||||
|
||||
class DialogStepGenerator:
|
||||
u"""This class defines a step, that will prompt the user for various
|
||||
fields."""
|
||||
|
||||
def __init__(self, backtitle):
|
||||
self.d = dialog.Dialog()
|
||||
self.d.add_persistent_args(["--backtitle", backtitle])
|
||||
|
||||
def _check_rc(self, rc):
|
||||
if rc == 2: really_quit(dico)
|
||||
if rc == 1: raise PreviousStep
|
||||
|
||||
def _skim_choices(self, choices, dico):
|
||||
return [ f for f in choices if f.get('show_cond', lambda x : True)(dico) ]
|
||||
|
||||
def form_step(self, title, enonce, form, **kw):
|
||||
def fn(dico, default):
|
||||
sform = self._skim_choices(form, dico)
|
||||
for field in sform:
|
||||
item = default.get(field['var'], dico.get(field['var'], ''))
|
||||
field['item'] = item
|
||||
rc, res = self.d.form(enonce % dico, fields = sform, title=title % dico, **kw)
|
||||
self._check_rc(rc)
|
||||
for field, val in izip (sform, res):
|
||||
dico[field['var']] = val
|
||||
return dico
|
||||
return Step(fn)
|
||||
|
||||
def select_step(self, title, enonce, var, choix, **kw):
|
||||
def fn(dico, default):
|
||||
schoix = []
|
||||
for c in choix:
|
||||
try:
|
||||
schoix.append((c[0], c[1]))
|
||||
except Exception,e:
|
||||
if c.get('show_cond', lambda x : True)(dico):
|
||||
schoix.append((c['label'], c['item']))
|
||||
print schoix
|
||||
rc, res = self.d.menu(enonce % dico, choices = schoix, title = title % dico, **kw)
|
||||
self._check_rc(rc)
|
||||
dico[var] = res
|
||||
return dico
|
||||
return Step(fn)
|
||||
|
||||
def input_step(self, title, enonce, var, **kw):
|
||||
def fn(dico, default):
|
||||
rc, res = self.d.inputbox(enonce % dico, title = title % dico, init = dico.get(var, ''), **kw)
|
||||
self._check_rc(rc)
|
||||
dico[var] = res
|
||||
return dico
|
||||
return Step(fn)
|
||||
|
||||
def checklist_step(self, title, enonce, liste, **kw):
|
||||
def fn(dico, default):
|
||||
sliste = []
|
||||
for c in liste:
|
||||
try:
|
||||
if dico.get(c[0], False): checked = 'on'
|
||||
else: checked = 'off'
|
||||
sliste.append((c[0], c[1], checked))
|
||||
except Exception,e:
|
||||
if c.get('show_cond', lambda x : True)(dico):
|
||||
if dico.get(c['var'], False): checked = 'on'
|
||||
else: checked = 'off'
|
||||
sliste.append((c['var'], c['item'], checked))
|
||||
rc, res = self.d.checklist(enonce % dico, title = title % dico, choices = sliste, **kw)
|
||||
self._check_rc(rc)
|
||||
for tag, item, status in liste:
|
||||
if tag in res:
|
||||
dico[tag] = True
|
||||
else:
|
||||
dico[tag] = False
|
||||
return dico
|
||||
return Step(fn)
|
||||
|
|
@ -1,220 +0,0 @@
|
|||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2009 Antoine Durand-Gasselin
|
||||
# Author: Antoine Durand-Gasselin <adg@crans.org>
|
||||
#
|
||||
|
||||
class TryAgain(Exception):
|
||||
"""Exception raised when the step should be taken again."""
|
||||
|
||||
class PreviousStep(Exception):
|
||||
"""Exception raised when should be backtracked to previous step."""
|
||||
|
||||
class EndScenario(Exception):
|
||||
"""Exception raised when the scenario should be halted.
|
||||
'msg' is the error message
|
||||
'data' is the updated state object"""
|
||||
def __init__(self, msg, data=None):
|
||||
self.msg = msg
|
||||
self.data = data
|
||||
|
||||
class Step:
|
||||
u"""This class defines a step. A step is defined by providing a
|
||||
function that, given an environment and an expected answer will
|
||||
return an update of the environment (as dictionnary)."""
|
||||
def __init__(self, update_fn):
|
||||
self.update = update_fn
|
||||
|
||||
def run(self, env, default):
|
||||
"""This function makes the call"""
|
||||
return self.update(env, default)
|
||||
|
||||
class Scenario:
|
||||
u"""This class allows us to define scenarios."""
|
||||
|
||||
def __init__(self):
|
||||
u"""empty scenario, with an error handler (that takes an exception as
|
||||
argument)"""
|
||||
self.steps = None
|
||||
|
||||
def nest(self, step):
|
||||
u"""Adds a step to a scenario"""
|
||||
if not isinstance(step, Step):
|
||||
raise TypeError("Can only bind steps")
|
||||
self.steps = ('NEST', step, self.steps)
|
||||
|
||||
def case(self, switch, cases, fallback = None): #revoir le fallback
|
||||
u"""Calls a function (passing to it the environment), and will
|
||||
call the corresponding scenario in the second arg."""
|
||||
if not callable(switch):
|
||||
raise TypeError("switch must be callable")
|
||||
if not isinstance(cases, dict):
|
||||
raise TypeError("cases must be a dict")
|
||||
for case in cases.values():
|
||||
if not isinstance(case, Scenario):
|
||||
raise TypeError("cases must all be Scenarios")
|
||||
self.steps = ('CASE', (switch, cases, fallback), self.steps)
|
||||
|
||||
def branch(self, cond, plan_A, plan_B):
|
||||
u"""Makes a test (will call it passing the environnement), and
|
||||
depending on the result, will process one of the two scenarios"""
|
||||
if not callable(cond):
|
||||
raise TypeError("cond must be callable")
|
||||
if not isinstance(plan_A, Scenario) or not isinstance(plan_B, Scenario):
|
||||
raise TypeError("Can only branch on scenarios")
|
||||
# self.steps = ('BRANCH', (cond, plan_A, plan_B) , self.steps)
|
||||
self.case(cond, { True: plan_A, False: plan_B })
|
||||
|
||||
def quote(self, scenario):
|
||||
u"""Runs a scenario as a single scenario step"""
|
||||
|
||||
if not isinstance(scenario, Scenario):
|
||||
raise TypeError("scenario must be a scenario")
|
||||
|
||||
def quote_scenar (dict1, dict2):
|
||||
try:
|
||||
return Running(scenario).run()
|
||||
except EndScenario:
|
||||
raise PreviousStep
|
||||
|
||||
self.nest(Step(quote_scenar))
|
||||
|
||||
def step_scenario(step):
|
||||
s = Scenario()
|
||||
s.nest(step)
|
||||
return s
|
||||
|
||||
def unit_scenario():
|
||||
return ( Scenario())
|
||||
|
||||
def _relance(exc): raise exc
|
||||
|
||||
class Running:
|
||||
u"""To run scenarios"""
|
||||
|
||||
defaults = {}
|
||||
steps = None
|
||||
stack = None
|
||||
|
||||
def __init__(self, scenario, env = {}, handle = _relance):
|
||||
if not isinstance(scenario, Scenario):
|
||||
raise TypeError("Can only run Scenarios")
|
||||
accu = scenario.steps
|
||||
self.env = env
|
||||
self.handle = handle
|
||||
|
||||
# To avoid brain spots on the walls, we shall reverse the list
|
||||
# of steps.
|
||||
while accu:
|
||||
self.steps = accu[0], accu[1], self.steps
|
||||
accu = accu[2]
|
||||
|
||||
def step(self):
|
||||
if self.steps:
|
||||
# Case of a Case, as a list of possible choices depending on
|
||||
# the result of a function call, if it is not handled, then
|
||||
# it does nothing.
|
||||
if self.steps[0] == 'CASE' :
|
||||
switch, cases, fallback = self.steps[1]
|
||||
self.steps = self.steps[2]
|
||||
scenar = cases.get(switch(self.env), fallback)
|
||||
if scenar: plan_steps = scenar.steps
|
||||
else: plan_steps = None
|
||||
while plan_steps:
|
||||
self.steps = plan_steps[0], plan_steps[1], self.steps
|
||||
plan_steps = plan_steps[2]
|
||||
|
||||
## # Case of a Branching
|
||||
## if self.steps[0] == 'BRANCH' :
|
||||
## # As it is (should be) an epsilon-test we won't
|
||||
## # backtrack it.
|
||||
## cond, plan_A, plan_B = self.steps[1]
|
||||
## self.steps = self.steps[2]
|
||||
## if cond(self.env):
|
||||
## plan_steps = plan_A.steps
|
||||
## else:
|
||||
## plan_steps = plan_B.steps
|
||||
## # Let's not forget we need to reverse the steps lists.
|
||||
## while plan_steps:
|
||||
## self.steps = plan_steps[0], plan_steps[1], self.steps
|
||||
## plan_steps = plan_steps[2]
|
||||
|
||||
# Case of nesting
|
||||
elif self.steps[0] == 'NEST':
|
||||
try:
|
||||
this_step = self.steps[1]
|
||||
new_env = this_step.run(self.env, self.defaults)
|
||||
# Should we perform sanity checks on new_env ? and raise
|
||||
# TryAgain if it fails ? After updating defaults ?
|
||||
|
||||
self.stack = (self.env.copy(), new_env, self.steps, self.stack)
|
||||
self.env.update(new_env)
|
||||
self.defaults = {}
|
||||
self.steps = self.steps[2]
|
||||
|
||||
except PreviousStep:
|
||||
if self.stack:
|
||||
self.env, self.defaults, self.steps, self.stack = self.stack
|
||||
else:
|
||||
raise EndScenario("No previous step", self.env)
|
||||
|
||||
except TryAgain:
|
||||
# We can update defaults
|
||||
pass
|
||||
|
||||
except Exception, e:
|
||||
try:
|
||||
self.handle(e)
|
||||
|
||||
except PreviousStep:
|
||||
if self.stack:
|
||||
self.env, self.defaults, self.steps, self.stack = self.stack
|
||||
else:
|
||||
raise EndScenario("No previous step", self.env)
|
||||
|
||||
except TryAgain:
|
||||
# We can update defaults
|
||||
pass
|
||||
|
||||
else:
|
||||
# Should not be called
|
||||
raise "invalid step"
|
||||
|
||||
def run(self):
|
||||
while self.steps:
|
||||
self.step()
|
||||
return (self.env)
|
||||
|
||||
|
||||
# For testing issues
|
||||
def prompt(var):
|
||||
def fn(dict, default):
|
||||
a = raw_input(u"%s (%s):= " % (var, default.get(var, '<no default>')))
|
||||
if a == 'n':
|
||||
raise TryAgain
|
||||
elif a == 'b':
|
||||
raise PreviousStep
|
||||
else:
|
||||
return { var : a }
|
||||
return Step(fn)
|
||||
|
||||
if __name__ == "__main__":
|
||||
s = Scenario()
|
||||
t = Scenario()
|
||||
u = Scenario()
|
||||
|
||||
for i in ['toto', 'tata', 'titi', 'tutu']:
|
||||
s.nest(prompt(i))
|
||||
t.nest(prompt(i[1]))
|
||||
|
||||
for i in range(6,9):
|
||||
u.nest(prompt(str(i)))
|
||||
|
||||
u.quote(t)
|
||||
|
||||
for i in range(12,15):
|
||||
u.nest(prompt(str(i)))
|
||||
|
||||
|
||||
print (Running(u).run())
|
||||
|
|
@ -1 +0,0 @@
|
|||
/usr/scripts/lib/
|
||||
File diff suppressed because it is too large
Load diff
|
|
@ -1,12 +0,0 @@
|
|||
|
||||
import logging
|
||||
|
||||
import sys
|
||||
sys.path.append('/usr/scripts/')
|
||||
from cranslib.utils.logs import getFileLogger
|
||||
LOGGER = getFileLogger("helloworld")
|
||||
LOGGER.setLevel(logging.INFO)
|
||||
LOGGER.addHandler(logging.StreamHandler())
|
||||
|
||||
if __name__ == "__main__":
|
||||
LOGGER.info(u"Hello World")
|
||||
|
|
@ -1,12 +0,0 @@
|
|||
SAP_FILE_URL = "http://tv/sap.txt"
|
||||
BASE_IMAGE_URL = "http://tv/images/"
|
||||
IMAGE_SUFFIX = ".jpg"
|
||||
SMALL_IMAGE_SUFFIX = "_petites.jpg"
|
||||
|
||||
|
||||
class Channel:
|
||||
pass
|
||||
|
||||
class ChannelGroup:
|
||||
pass
|
||||
|
||||
|
|
@ -1,83 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
class PagePerso:
|
||||
"""Classe représentant la page perso d'une personne"""
|
||||
|
||||
home = "/home"
|
||||
www = "/www"
|
||||
|
||||
def __init__(self, login):
|
||||
"""Instanciation avec le login' de la personne"""
|
||||
self.login = login
|
||||
self.home = "%s/%s" % (self.home, login)
|
||||
_info = {}
|
||||
self._load_informations()
|
||||
|
||||
def _filename( self ):
|
||||
return "%s/.info" % self.home
|
||||
|
||||
def _load_informations(self):
|
||||
try:
|
||||
lignes = file( self._filename() )
|
||||
except IOError:
|
||||
lignes = []
|
||||
# self._info est un dictionnaire qui reprend le contenu du .info
|
||||
self._info = dict( map( lambda z: (unicode(z[0].lower(),"iso-8859-15"),
|
||||
unicode(z[1],"iso-8859-15")),
|
||||
filter(lambda w: len(w) == 2 and len(w[1]),
|
||||
map(lambda x: map(lambda y: y.strip(),
|
||||
x.split(":")),
|
||||
lignes))))
|
||||
|
||||
def _save_informations( self ):
|
||||
myfile = file(self._filename(), "w")
|
||||
for aKey in self._info.keys():
|
||||
myfile.write("%s:%s\n" % (aKey, self.
|
||||
_info[aKey]) )
|
||||
myfile.write("\n")
|
||||
|
||||
def save( self ):
|
||||
self._save_informations()
|
||||
|
||||
def chemin(self):
|
||||
"""Chemin vers le www"""
|
||||
return u"%s%s" % (self.home, self.www)
|
||||
|
||||
def url(self):
|
||||
"""URL vers la page perso"""
|
||||
return u"http://perso.crans.org/%s/" % self.login
|
||||
|
||||
def nom( self ):
|
||||
return self._info.get("nom", "")
|
||||
|
||||
def setNom( self, nom ):
|
||||
self._info["nom"] = nom
|
||||
|
||||
def slogan( self ):
|
||||
return self._info.get("slogan", "")
|
||||
|
||||
def setSlogan( self, slogan ):
|
||||
self._info["slogan"] = slogan
|
||||
|
||||
def logo(self):
|
||||
"""URL du logo s'il y en a un"""
|
||||
logo = self._info.get("logo", None)
|
||||
if logo:
|
||||
# Le logo peut être en absolu ou en relatif
|
||||
if logo.startswith(self.chemin()):
|
||||
logo = logo.replace("%s/" % self.chemin(), "")
|
||||
if os.path.isfile("%s/%s" % (self.chemin(), logo)):
|
||||
return u"%s%s" % (self.url(), logo)
|
||||
return u"http://perso.crans.org/pageperso.png"
|
||||
|
||||
def __str__(self):
|
||||
"""Renvoie le code HTML correspondant au fichier .info"""
|
||||
html = [ u'<div class="vignetteperso">',
|
||||
u'<a href="%s">' % self.url(),
|
||||
u'<img src="%s" alt="%s">' % (self.logo(), self.login),
|
||||
u'</a><br>',
|
||||
self.info("nom") and u'<b>%s</b><br>' % self.info("nom") or u'%s<br>' % self.login,
|
||||
self.info("devise") and u'<i>%s</i>' % self.info("devise") or u'',
|
||||
u'</div>' ]
|
||||
return u'\n'.join(html)
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue