...
:: cd ~map/course/201303/ecflow; python course.py
Code Block | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
#!/usr/bin/env python """ suite builder example for ecFlow course """ import sys, os sys.path.append('/home/ma/emos/def/ecflow') import ecf as ec from ecf import * import inc_emos as ie # provides Seed class + system related dependencies # cd ~map/course/201303/ecflow; python course.py # task wrappers underneath # consume: choices for a family matching the producer-consumer pattern # local family + remote family + BEWARE ECF_OUT + log-server example # barber: an example of a "dynamical suite", with a "family producer task" # perl + python: example that task wrapper may not be ksh/bash scripts import time from datetime import date import argparse today = str(date.today()).replace('-', '') ############################################################################ class GenericFamily(object): """ provide structure for derived classes""" def make(self, node): return node def main(self, node): return BaseException # return node def arch(self, node): return node ############################################################################ class NativePerl(GenericFamily): """ ksh is not the only language for task wrappers""" def __init__(self): self.name = "perl" def main(self, node): tsk = Task(self.name).add( Variables(ECF_MICRO= "^", ECF_JOB_CMD= "^ECF_JOB^ > ^ECF_JOBOUT^ 2>&1"), Meter("step", -1, 100), Event("1"), Event("2"), Label("info", "none"), ) node.add(tsk) ############################################################################ class NativePython(NativePerl): """ ksh is not the only language for task wrappers""" def __init__(self): super(NativePerl, self).__init__() self.name = "python" ############################################################################ def _kind(prod=1, cons=1): return Variables(CONSUME= cons, PRODUCE=prod) def _evt(): return (Event("p"), Event("c")) def _leaf(name="produce", init=0, stop=100, step=1): add = None if type(stop) == int: add = Meter("step", -1, int(stop)) return Task("%s" % name).add( _evt(), add, Variables(INIT= init, STOP= stop, STEP= step),) ############################################################################ class FallBack(GenericFamily): """ in some situation, user may want its family to continue, and repeat to increment, even when some tasks abort """ def main(self, node=None): return Family("daily").repeat(name="YMD", start=today, end=DATE_STOP).add( Task("action").add(Time("10:00")), Family("loop").add( Time("11:00"), Task("dummy").add( TriggerImpossible(), Complete("1==1"))), Task("fallback").add( Label("info", "force complete when action is aborted"), Time("10:55"), Trigger("action eq aborted"), Complete("action eq complete"))) class DailyInc(GenericFamily): """ anopther method to have daily repeat increment, with aborted tasks""" def main(self, node=None): return Family("daily_inc").repeat(name="YMD", start=today, end=DATE_STOP).add( Label("info", "requeue will reset repeat attribute!"), Complete("daily_inc/loop/dummy eq complete"), Task("action").add(Time("10:00")), Family("loop").add( Time("11:00"), Task("dummy").add(TriggerImpossible(), Complete("1==1")))) class Consume(GenericFamily): """ producer-consumer pattern can be implemented in many ways""" def __init__(self): self.init = 0 self.stop = 48 self.step = 3 def main(self, node): """ pass the parent node, so that absolute paths can be used with triggers""" path = node.fullname() top = node.family("consume").add( Variables(SLEEP= 10, PRODUCE= 1, # default: tasks will do both CONSUME= 1), Family("limits").add(Defcomplete(), Limit("consume", 7)), Task("leader").add( Label("info", "set event to get produce1 leader"), Event("1"), # set/cleared by user Defcomplete()), # task does both, ie serial ########################### _leaf("produce", self.init, self.stop, self.step).add( Label("info", "both produce and consume in a task")), # meter will report about producer progress ########### Family("produce0").add( Label("info", "only produce"), _kind(1, 0), _leaf("produce", self.init, self.stop, self.step)), # serialy produced, create a new task for each step ### Family("produce1").add( _kind(1, 0), Label("info", "repeat, one job per step"), _leaf("produce", init="%STEP%", stop="%STEP%", step=1).add( Meter("step", -1, 100)))\ .repeat(kind="integer", name="STEP", start=self.init, end =self.stop, step =self.step).add() ) top.defstatus("suspended") fam = Family("produce2").add( # parallel _kind(1, 0), Label("info", "limited, one task per step, step by 3"), Limit("prod", 5), InLimit("produce2:prod")) top.add(fam) for step in xrange(self.init, self.stop, self.step): fam.add(Family("%02d" % step).add( Variables(STEP= step), _leaf("produce", step, step, 1))) ###################### lead = path + "/consume/leader:1" prod = path + "/consume/produce" top.add( ### trigger may be inside a task _leaf("consume", self.init, self.stop, self.step).add( Label("info", "trigger may be inside a task"), _kind(0, 1), InLimit("limits:consume"), Variables(CALL_WAITER= 1, SLEEP= 3, # sleep less than producer TRIGGER_EXPRESSION= prod + ":step ge $step or " + prod + " eq complete",)), Family("consume1").add( Label("info", "explicit trigger, follow faster"), _kind(0, 1), Trigger("(consume1:STEP lt %s1:STEP and %s) or " % (prod, lead) + "(consume1:STEP lt %s0/produce:step and not %s) or " % (prod, lead) + # lt while both are repeat "(%s1 eq complete and %s0 eq complete)" % (prod, prod) ), InLimit("limits:consume"), _leaf("consume", "%STEP%", "%STEP%", 1), ).repeat(kind="integer", name="STEP", start=self.init, end=self.stop, step=self.step)) fam = Family("consume2").add( # parallel Label("info", "one task per step, step by three"), _kind(0, 1), Limit("consume", 5), InLimit("consume2:consume")) top.add(fam) for step in xrange(self.init, self.stop, self.step): fam.add(Family("%02d" % step).add( Variables(STEP = step), Trigger("(%02d:STEP le %s1:STEP and %s) or " % (step, prod, lead) + "(%02d:STEP le %s0/produce:step and not %s)" % (step, prod, lead)), _leaf("consume", init=step, stop=step, step=1))) ############################################################################ class Barber(GenericFamily): """ a 'barber shop' example with families created by a task """ def _passer_by(self): """ generator """ return Task("passby").add( Time("00:00 23:59 00:05"), Variables(ID=0), Label("info", ""), Label("rem", "this task alters its variable ID, " + "aliases won't work natively"), InLimit("limits:passby")) def _client(self, node, position): """ python version of the family created initialy attention: raw definition file is located in passby task wrapper""" path = node.fullname() + "/limits" fam = node.family("list").family("%s" % position).add( AutoCancel(1), Task("cut").inlimit(path + ":barbers"), Task("pay").add( Trigger("cut eq complete"), InLimit(path + ":barbers"), InLimit(path + ":cashiers")), Task("leave").add( Label("info", ""), Trigger(["cut", "pay"]))) fam.defstatus("complete") def _shop(self, node): fam = node.family("shop").defstatus("suspended").add( Variables(NB_CHAIRS= 4), Family("limits").add(Defcomplete(), Limit("passby", 1), Limit("barbers", 2), Limit("cashiers", 1)), self._passer_by(), ) self._client(fam, 1), def main(self, node): self._shop(node) ############################################################################ def user(): return os.getlogin() def locate_scripts(): pwd = os.getcwd() return Variables( ECF_HOME= "/tmp/%s/ecflow/" % user(), # pwd, ECF_FILES= pwd + "/scripts", ECF_INCLUDE= pwd + "/include", ) DATE_STOP = 20300115 class Course(ie.Seed): """ host families together """ def __init__(self): super(Course, self).__init__() self.name = "course" def suite(self): """ define limits (empty) """ node = Suite(user()) node.defstatus("suspended").add( Variables(USER= user()), locate_scripts()) # self.top(node) fp = open("/tmp/%s/" % user() + self.name + ".def", "w") print >> fp, node return node def top(self, node): barber_shop = Barber() perl = NativePerl() python = NativePython() consume = Consume() with node.family(self.name) as node: node.add(FallBack().main(), DailyInc().main()) barber_shop.main(node) perl.main(node) python.main(node) consume.main(node) return node ############################################################################### class Admin(Course): """ host newlog task + logsvr start/stop/check task -- server logs can be renewed with a ecflowview menu command also """ def __init__(self): self.name = "admin" def top(self, node): with node.family("admin") as node: node.add(self.main()).repeat(name="YMD", start=today, end=DATE_STOP) node.defstatus("suspended") return node def main(self): """ return self contained Family/Task, without absolute node references or with relative path triggers""" remote_submit = "rsh -l %USER% %HOST% %ECF_JOB% > %ECF_JOBOUT% 2>&1" logpath = "/home/ma/map/course/201303/ecflow" return ( Task("newlog").add( Label("info", "renew server log-file"), Time("08:00")), Task("logsvr").add( Defcomplete(), Variables(HOST= "pikachu", ECF_LOGPORT=9316, ECF_LOGPATH= logpath, ECF_LOGMAP= logpath + ":" + logpath, ECF_JOB_CMD= remote_submit), Label("info", "(re)start the logsvr on HOST"), Time("08:00")), Family("loop").add( Time("08:30"), Family("dummy").add(# TriggerImpossible(), Complete("1==1")))) ############################################################################### class EcEvents(Admin): """ connecting to third party software as event generator to update a suite variable, and enable daily family run """ def top(self, node): node = node.family("ecevents") node.add( Label("info", "use web... menu"), Defcomplete(), Variables( URL= "http://eccmd.ecmwf.int:8090/#Mainpanel", ECF_URL_CMD= "${BROWSER:=firefox} -remote 'openURL(%URL%")) self.main(node) return node def main(self, node): for mode in ["list", "register", "delete", "register_all", "delete_all"]: added = None if "list" in mode: added = Label("regs", "") fam = Family(mode).add( Variables(MODE= mode), Task("ecjobs").add(Label("info", "none"), added)) node.add(fam) if "_all" in mode: fam.add(Variables(EVENT= "_all_")) elif mode in ("register", "delete"): fam.add(Variables(EVENT= "an12h000"), Label("info", "update EVENT variable")) event = "an00h000" node.family("ref").defstatus("complete").add( Task(event).add(Variables(YMD= today), Label("YMD", today))) node.family("user").family(event).repeat( name="YMD", start=today, end=DATE_STOP).add( Label("info", "extern cannot be used anymore for " + "intra-suite reference triggers"), Variables(SLEEP= 1), Trigger(event + ":YMD le %s/ref/%s:YMD" % (node.fullname(), event)), _leaf("consume")) return node ############################################################################### class SerialTask(object): """ add trigger on the previous task """ def __init__(self): self.prev = None def add(self, name): fam = Family(name).add( Variables(MODE= name), Task("to_ecflow")) if self.prev != None: fam.add(Trigger("./%s eq complete" % self.prev)) self.prev = name return fam class Reload(Admin): """ a simple task to download SMS content and translate it to ecFlow """ def top(self, node): node = node.family("reload") node.add( Label("info", "from sms to ecFlow"), Defcomplete(), Variables( URL= "https://software.ecmwf.int/wiki/display/ECFLOW/Home", ECF_URL_CMD= "${BROWSER:=firefox} -remote 'openURL(%URL%"), self.main()) return node def main(self, node=None): fam = Family("reload") serial = SerialTask() fam.add( serial.add("get").add( Variables( SMS_SUITE= "eras", # SMS_PROG= 314159, SMS_NODE= "localhost", # SMS_PROG= 314199, SMS_NODE= "marhaus", # eras SMS_PROG= 314197, SMS_NODE= "marhaus", # eras2 ),), serial.add("translate"), serial.add("edit"), serial.add("mail"), serial.add("load"), serial.add("bench").add(Defcomplete())) if node is not None: return node.add(fam) return fam ############################################################################ if __name__ == '__main__': import cli_proc, sys parser = argparse.ArgumentParser( description=DESC, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--host", default="localhost", help= "server hostname") parser.add_argument("--port", default="3141", help= "server port number") if len(sys.argv) > 1: suites = { "course": Course(), } argv = cli_proc.process(suites, compare=False) sys.exit(0) args = parser.parse_args() clt = ec.Client(args.host, args.port) try: clt.ping() except: clt = ec.Client("localhost", 1000 + os.geteuid()) try: clt.ping() except: clt = ec.Client("localhost", 31415) try: clt.ping() except: clt = ec.Client("localhost", 3199) defs = ec.ecflow.Defs() course = Course() suite = course.suite().add(Reload().main()) if 1: # enable/disable remote version of 'consume' rem = suite.family("remote").add( Task("mk_remote").add( Defcomplete(), Label("info", "do not forget to create directory structure" + "on remote host for job output creation"), ie.onws(host = "class01")), Variables(ECF_HOME= "/tmp/%s/ecflow" % user()), ie.onws(host = "class02")) course.top(rem) # print clt.stats() Admin().top(suite) EcEvents().top(suite) defs.add_suite(suite) path = "/%s" % user() # path = "/%s/course" % user() # path = "/%s/admin" % user() # path = "/%s/ecevents" % user() # print defs clt.replace(path, defs) sys.exit (0) |
...