module.py 34 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017
  1. # -*- coding: utf-8 -*-
  2. from __future__ import (nested_scopes, generators, division, absolute_import,
  3. with_statement, print_function, unicode_literals)
  4. import sys
  5. from multiprocessing import cpu_count, Process, Queue
  6. import time
  7. from xml.etree.ElementTree import fromstring
  8. from grass.exceptions import CalledModuleError, GrassError, ParameterError
  9. from grass.script.core import Popen, PIPE, use_temp_region, del_temp_region
  10. from grass.script.utils import encode, decode
  11. from .docstring import docstring_property
  12. from .parameter import Parameter
  13. from .flag import Flag
  14. from .typedict import TypeDict
  15. from .read import GETFROMTAG, DOC
  16. from .env import G_debug
  17. if sys.version_info[0] == 2:
  18. from itertools import izip_longest as zip_longest
  19. else:
  20. from itertools import zip_longest
  21. unicode = str
  22. def _get_bash(self, *args, **kargs):
  23. return self.get_bash()
  24. class ParallelModuleQueue(object):
  25. """This class is designed to run an arbitrary number of pygrass Module or MultiModule
  26. processes in parallel.
  27. Objects of type grass.pygrass.modules.Module or
  28. grass.pygrass.modules.MultiModule can be put into the
  29. queue using put() method. When the queue is full with the maximum
  30. number of parallel processes it will wait for all processes to finish,
  31. sets the stdout and stderr of the Module object and removes it
  32. from the queue when its finished.
  33. To finish the queue before the maximum number of parallel
  34. processes was reached call wait() .
  35. This class will raise a GrassError in case a Module process exits
  36. with a return code other than 0.
  37. Processes that were run asynchronously with the MultiModule class
  38. will not raise a GrassError in case of failure. This must be manually checked
  39. by accessing finished modules by calling get_finished_modules().
  40. Usage:
  41. Check with a queue size of 3 and 5 processes
  42. >>> import copy
  43. >>> from grass.pygrass.modules import Module, ParallelModuleQueue
  44. >>> mapcalc_list = []
  45. Setting run_ to False is important, otherwise a parallel processing is not possible
  46. >>> mapcalc = Module("r.mapcalc", overwrite=True, run_=False)
  47. >>> queue = ParallelModuleQueue(nprocs=3)
  48. >>> for i in range(5):
  49. ... new_mapcalc = copy.deepcopy(mapcalc)
  50. ... mapcalc_list.append(new_mapcalc)
  51. ... m = new_mapcalc(expression="test_pygrass_%i = %i"%(i, i))
  52. ... queue.put(m)
  53. >>> queue.wait()
  54. >>> mapcalc_list = queue.get_finished_modules()
  55. >>> queue.get_num_run_procs()
  56. 0
  57. >>> queue.get_max_num_procs()
  58. 3
  59. >>> for mapcalc in mapcalc_list:
  60. ... print(mapcalc.popen.returncode)
  61. 0
  62. 0
  63. 0
  64. 0
  65. 0
  66. Check with a queue size of 8 and 5 processes
  67. >>> queue = ParallelModuleQueue(nprocs=8)
  68. >>> mapcalc_list = []
  69. >>> for i in range(5):
  70. ... new_mapcalc = copy.deepcopy(mapcalc)
  71. ... mapcalc_list.append(new_mapcalc)
  72. ... m = new_mapcalc(expression="test_pygrass_%i = %i"%(i, i))
  73. ... queue.put(m)
  74. >>> queue.wait()
  75. >>> mapcalc_list = queue.get_finished_modules()
  76. >>> queue.get_num_run_procs()
  77. 0
  78. >>> queue.get_max_num_procs()
  79. 8
  80. >>> for mapcalc in mapcalc_list:
  81. ... print(mapcalc.popen.returncode)
  82. 0
  83. 0
  84. 0
  85. 0
  86. 0
  87. Check MultiModule approach with three by two processes running in a background process
  88. >>> gregion = Module("g.region", flags="p", run_=False)
  89. >>> queue = ParallelModuleQueue(nprocs=3)
  90. >>> proc_list = []
  91. >>> for i in range(3):
  92. ... new_gregion = copy.deepcopy(gregion)
  93. ... proc_list.append(new_gregion)
  94. ... new_mapcalc = copy.deepcopy(mapcalc)
  95. ... m = new_mapcalc(expression="test_pygrass_%i = %i"%(i, i))
  96. ... proc_list.append(new_mapcalc)
  97. ... mm = MultiModule(module_list=[new_gregion, new_mapcalc], sync=False, set_temp_region=True)
  98. ... queue.put(mm)
  99. >>> queue.wait()
  100. >>> proc_list = queue.get_finished_modules()
  101. >>> queue.get_num_run_procs()
  102. 0
  103. >>> queue.get_max_num_procs()
  104. 3
  105. >>> for proc in proc_list:
  106. ... print(proc.popen.returncode)
  107. 0
  108. 0
  109. 0
  110. 0
  111. 0
  112. 0
  113. Check with a queue size of 8 and 4 processes
  114. >>> queue = ParallelModuleQueue(nprocs=8)
  115. >>> mapcalc_list = []
  116. >>> new_mapcalc = copy.deepcopy(mapcalc)
  117. >>> mapcalc_list.append(new_mapcalc)
  118. >>> m = new_mapcalc(expression="test_pygrass_1 =1")
  119. >>> queue.put(m)
  120. >>> queue.get_num_run_procs()
  121. 1
  122. >>> new_mapcalc = copy.deepcopy(mapcalc)
  123. >>> mapcalc_list.append(new_mapcalc)
  124. >>> m = new_mapcalc(expression="test_pygrass_2 =2")
  125. >>> queue.put(m)
  126. >>> queue.get_num_run_procs()
  127. 2
  128. >>> new_mapcalc = copy.deepcopy(mapcalc)
  129. >>> mapcalc_list.append(new_mapcalc)
  130. >>> m = new_mapcalc(expression="test_pygrass_3 =3")
  131. >>> queue.put(m)
  132. >>> queue.get_num_run_procs()
  133. 3
  134. >>> new_mapcalc = copy.deepcopy(mapcalc)
  135. >>> mapcalc_list.append(new_mapcalc)
  136. >>> m = new_mapcalc(expression="test_pygrass_4 =4")
  137. >>> queue.put(m)
  138. >>> queue.get_num_run_procs()
  139. 4
  140. >>> queue.wait()
  141. >>> mapcalc_list = queue.get_finished_modules()
  142. >>> queue.get_num_run_procs()
  143. 0
  144. >>> queue.get_max_num_procs()
  145. 8
  146. >>> for mapcalc in mapcalc_list:
  147. ... print(mapcalc.popen.returncode)
  148. 0
  149. 0
  150. 0
  151. 0
  152. Check with a queue size of 3 and 4 processes
  153. >>> queue = ParallelModuleQueue(nprocs=3)
  154. >>> mapcalc_list = []
  155. >>> new_mapcalc = copy.deepcopy(mapcalc)
  156. >>> mapcalc_list.append(new_mapcalc)
  157. >>> m = new_mapcalc(expression="test_pygrass_1 =1")
  158. >>> queue.put(m)
  159. >>> queue.get_num_run_procs()
  160. 1
  161. >>> new_mapcalc = copy.deepcopy(mapcalc)
  162. >>> mapcalc_list.append(new_mapcalc)
  163. >>> m = new_mapcalc(expression="test_pygrass_2 =2")
  164. >>> queue.put(m)
  165. >>> queue.get_num_run_procs()
  166. 2
  167. >>> new_mapcalc = copy.deepcopy(mapcalc)
  168. >>> mapcalc_list.append(new_mapcalc)
  169. >>> m = new_mapcalc(expression="test_pygrass_3 =3")
  170. >>> queue.put(m) # Now it will wait until all procs finish and set the counter back to 0
  171. >>> queue.get_num_run_procs()
  172. 0
  173. >>> new_mapcalc = copy.deepcopy(mapcalc)
  174. >>> mapcalc_list.append(new_mapcalc)
  175. >>> m = new_mapcalc(expression="test_pygrass_%i = %i"%(i, i))
  176. >>> queue.put(m)
  177. >>> queue.get_num_run_procs()
  178. 1
  179. >>> queue.wait()
  180. >>> mapcalc_list = queue.get_finished_modules()
  181. >>> queue.get_num_run_procs()
  182. 0
  183. >>> queue.get_max_num_procs()
  184. 3
  185. >>> for mapcalc in mapcalc_list:
  186. ... print(mapcalc.popen.returncode)
  187. 0
  188. 0
  189. 0
  190. 0
  191. """
  192. def __init__(self, nprocs=1):
  193. """Constructor
  194. :param nprocs: The maximum number of Module processes that
  195. can be run in parallel, default is 1, if None
  196. then use all the available CPUs.
  197. :type nprocs: int
  198. """
  199. nprocs = int(nprocs) if nprocs else cpu_count()
  200. self._num_procs = nprocs
  201. self._list = nprocs * [None]
  202. self._proc_count = 0
  203. self._finished_modules = [] # Store all processed modules in a list
  204. def put(self, module):
  205. """Put the next Module or MultiModule object in the queue
  206. To run the Module objects in parallel the run\_ and finish\_ options
  207. of the Module must be set to False.
  208. :param module: a preconfigured Module or MultiModule object that were configured
  209. with run\_ and finish\_ set to False,
  210. :type module: Module or MultiModule object
  211. """
  212. self._list[self._proc_count] = module
  213. # Force that finish is False, otherwise the execution
  214. # will not be parallel
  215. self._list[self._proc_count].finish_ = False
  216. self._list[self._proc_count].run()
  217. self._proc_count += 1
  218. if self._proc_count == self._num_procs:
  219. self.wait()
  220. def get(self, num):
  221. """Get a Module object or list of Module objects from the queue
  222. :param num: the number of the object in queue
  223. :type num: int
  224. :returns: the Module object or list of Module objects or None if num is not in the queue
  225. """
  226. if num < self._num_procs:
  227. return self._list[num]
  228. return None
  229. def get_num_run_procs(self):
  230. """Get the number of Module processes that are in the queue running
  231. or finished
  232. :returns: the number fo Module processes running/finished in the queue
  233. """
  234. return self._proc_count
  235. def get_max_num_procs(self):
  236. """Return the maximum number of parallel Module processes
  237. :returns: the maximum number of parallel Module processes
  238. """
  239. return self._num_procs
  240. def set_max_num_procs(self, nprocs):
  241. """Set the maximum number of Module processes that should run
  242. in parallel
  243. :param nprocs: The maximum number of Module processes that can be
  244. run in parallel
  245. :type nprocs: int
  246. """
  247. self._num_procs = int(nprocs)
  248. self.wait()
  249. def get_finished_modules(self):
  250. """Return all finished processes that were run by this queue
  251. :return: A list of Module objects
  252. """
  253. return self._finished_modules
  254. def wait(self):
  255. """Wait for all Module processes that are in the list to finish
  256. and set the modules stdout and stderr output options
  257. :return: A list of modules that were run
  258. """
  259. for proc in self._list:
  260. if proc:
  261. if isinstance(proc, Module):
  262. self._finished_modules.extend([proc.wait(),])
  263. else:
  264. self._finished_modules.extend(proc.wait())
  265. self._list = self._num_procs * [None]
  266. self._proc_count = 0
  267. class Module(object):
  268. """This class is design to wrap/run/interact with the GRASS modules.
  269. The class during the init phase read the XML description generate using
  270. the ``--interface-description`` in order to understand which parameters
  271. are required which optionals. ::
  272. >>> from grass.pygrass.modules import Module
  273. >>> from subprocess import PIPE
  274. >>> import copy
  275. >>> region = Module("g.region")
  276. >>> region.flags.p = True # set flags
  277. >>> region.flags.u = True
  278. >>> region.flags["3"].value = True # set numeric flags
  279. >>> region.get_bash()
  280. 'g.region -p -3 -u'
  281. >>> new_region = copy.deepcopy(region)
  282. >>> new_region.inputs.res = "10"
  283. >>> new_region.get_bash()
  284. 'g.region res=10 -p -3 -u'
  285. >>> neighbors = Module("r.neighbors")
  286. >>> neighbors.inputs.input = "mapA"
  287. >>> neighbors.outputs.output = "mapB"
  288. >>> neighbors.inputs.size = 5
  289. >>> neighbors.inputs.quantile = 0.5
  290. >>> neighbors.get_bash()
  291. 'r.neighbors input=mapA method=average size=5 quantile=0.5 output=mapB'
  292. >>> new_neighbors1 = copy.deepcopy(neighbors)
  293. >>> new_neighbors1.inputs.input = "mapD"
  294. >>> new_neighbors1.inputs.size = 3
  295. >>> new_neighbors1.inputs.quantile = 0.5
  296. >>> new_neighbors1.get_bash()
  297. 'r.neighbors input=mapD method=average size=3 quantile=0.5 output=mapB'
  298. >>> new_neighbors2 = copy.deepcopy(neighbors)
  299. >>> new_neighbors2(input="mapD", size=3, run_=False)
  300. Module('r.neighbors')
  301. >>> new_neighbors2.get_bash()
  302. 'r.neighbors input=mapD method=average size=3 quantile=0.5 output=mapB'
  303. >>> neighbors = Module("r.neighbors")
  304. >>> neighbors.get_bash()
  305. 'r.neighbors method=average size=3'
  306. >>> new_neighbors3 = copy.deepcopy(neighbors)
  307. >>> new_neighbors3(input="mapA", size=3, output="mapB", run_=False)
  308. Module('r.neighbors')
  309. >>> new_neighbors3.get_bash()
  310. 'r.neighbors input=mapA method=average size=3 output=mapB'
  311. >>> mapcalc = Module("r.mapcalc", expression="test_a = 1",
  312. ... overwrite=True, run_=False)
  313. >>> mapcalc.run()
  314. Module('r.mapcalc')
  315. >>> mapcalc.popen.returncode
  316. 0
  317. >>> mapcalc = Module("r.mapcalc", expression="test_a = 1",
  318. ... overwrite=True, run_=False, finish_=False)
  319. >>> mapcalc.run()
  320. Module('r.mapcalc')
  321. >>> p = mapcalc.wait()
  322. >>> p.popen.returncode
  323. 0
  324. >>> mapcalc.run()
  325. Module('r.mapcalc')
  326. >>> p = mapcalc.wait()
  327. >>> p.popen.returncode
  328. 0
  329. >>> colors = Module("r.colors", map="test_a", rules="-",
  330. ... run_=False, stdout_=PIPE,
  331. ... stderr_=PIPE, stdin_="1 red")
  332. >>> colors.run()
  333. Module('r.colors')
  334. >>> p = mapcalc.wait()
  335. >>> p.popen.returncode
  336. 0
  337. >>> colors.inputs["stdin"].value
  338. '1 red'
  339. >>> colors.outputs["stdout"].value
  340. ''
  341. >>> colors.outputs["stderr"].value.strip()
  342. "Color table for raster map <test_a> set to 'rules'"
  343. >>> colors = Module("r.colors", map="test_a", rules="-",
  344. ... run_=False, finish_=False, stdin_=PIPE)
  345. >>> colors.run()
  346. Module('r.colors')
  347. >>> stdout, stderr = colors.popen.communicate(input=b"1 red")
  348. >>> colors.popen.returncode
  349. 0
  350. >>> stdout
  351. >>> stderr
  352. >>> colors = Module("r.colors", map="test_a", rules="-",
  353. ... run_=False, finish_=False,
  354. ... stdin_=PIPE, stderr_=PIPE)
  355. >>> colors.run()
  356. Module('r.colors')
  357. >>> stdout, stderr = colors.popen.communicate(input=b"1 red")
  358. >>> colors.popen.returncode
  359. 0
  360. >>> stdout
  361. >>> stderr.strip()
  362. b"Color table for raster map <test_a> set to 'rules'"
  363. Run a second time
  364. >>> colors.run()
  365. Module('r.colors')
  366. >>> stdout, stderr = colors.popen.communicate(input=b"1 blue")
  367. >>> colors.popen.returncode
  368. 0
  369. >>> stdout
  370. >>> stderr.strip()
  371. b"Color table for raster map <test_a> set to 'rules'"
  372. Multiple run test
  373. >>> colors = Module("r.colors", map="test_a",
  374. ... color="ryb", run_=False)
  375. >>> colors.get_bash()
  376. 'r.colors map=test_a color=ryb'
  377. >>> colors.run()
  378. Module('r.colors')
  379. >>> colors(color="gyr")
  380. Module('r.colors')
  381. >>> colors.run()
  382. Module('r.colors')
  383. >>> colors(color="ryg")
  384. Module('r.colors')
  385. >>> colors(stderr_=PIPE)
  386. Module('r.colors')
  387. >>> colors.run()
  388. Module('r.colors')
  389. >>> print(colors.outputs["stderr"].value.strip())
  390. Color table for raster map <test_a> set to 'ryg'
  391. >>> colors(color="byg")
  392. Module('r.colors')
  393. >>> colors(stdout_=PIPE)
  394. Module('r.colors')
  395. >>> colors.run()
  396. Module('r.colors')
  397. >>> print(colors.outputs["stderr"].value.strip())
  398. Color table for raster map <test_a> set to 'byg'
  399. Often in the Module class you can find ``*args`` and ``kwargs`` annotation
  400. in methods, like in the __call__ method.
  401. Python allow developers to not specify all the arguments and
  402. keyword arguments of a method or function. ::
  403. def f(*args):
  404. for arg in args:
  405. print arg
  406. therefore if we call the function like:
  407. >>> f('grass', 'gis', 'modules') # doctest: +SKIP
  408. grass
  409. gis
  410. modules
  411. or we can define a new list:
  412. >>> words = ['grass', 'gis', 'modules'] # doctest: +SKIP
  413. >>> f(*words) # doctest: +SKIP
  414. grass
  415. gis
  416. modules
  417. we can do the same with keyword arguments, rewrite the above function: ::
  418. def f(*args, **kargs):
  419. for arg in args:
  420. print arg
  421. for key, value in kargs.items():
  422. print "%s = %r" % (key, value)
  423. now we can use the new function, with:
  424. >>> f('grass', 'gis', 'modules', os = 'linux', language = 'python')
  425. ... # doctest: +SKIP
  426. grass
  427. gis
  428. modules
  429. os = 'linux'
  430. language = 'python'
  431. or, as before we can, define a dictionary and give the dictionary to
  432. the function, like:
  433. >>> keywords = {'os' : 'linux', 'language' : 'python'} # doctest: +SKIP
  434. >>> f(*words, **keywords) # doctest: +SKIP
  435. grass
  436. gis
  437. modules
  438. os = 'linux'
  439. language = 'python'
  440. In the Module class we heavily use this language feature to pass arguments
  441. and keyword arguments to the grass module.
  442. """
  443. def __init__(self, cmd, *args, **kargs):
  444. if isinstance(cmd, unicode):
  445. self.name = str(cmd)
  446. elif isinstance(cmd, str):
  447. self.name = cmd
  448. else:
  449. raise GrassError("Problem initializing the module {s}".format(s=cmd))
  450. try:
  451. # call the command with --interface-description
  452. get_cmd_xml = Popen([cmd, "--interface-description"], stdout=PIPE)
  453. except OSError as e:
  454. print("OSError error({0}): {1}".format(e.errno, e.strerror))
  455. str_err = "Error running: `%s --interface-description`."
  456. raise GrassError(str_err % self.name)
  457. # get the xml of the module
  458. self.xml = get_cmd_xml.communicate()[0]
  459. # transform and parse the xml into an Element class:
  460. # http://docs.python.org/library/xml.etree.elementtree.html
  461. tree = fromstring(self.xml)
  462. for e in tree:
  463. if e.tag not in ('parameter', 'flag'):
  464. self.__setattr__(e.tag, GETFROMTAG[e.tag](e))
  465. #
  466. # extract parameters from the xml
  467. #
  468. self.params_list = [Parameter(p) for p in tree.findall("parameter")]
  469. self.inputs = TypeDict(Parameter)
  470. self.outputs = TypeDict(Parameter)
  471. self.required = []
  472. # Insert parameters into input/output and required
  473. for par in self.params_list:
  474. if par.input:
  475. self.inputs[par.name] = par
  476. else:
  477. self.outputs[par.name] = par
  478. if par.required:
  479. self.required.append(par.name)
  480. #
  481. # extract flags from the xml
  482. #
  483. flags_list = [Flag(f) for f in tree.findall("flag")]
  484. self.flags = TypeDict(Flag)
  485. for flag in flags_list:
  486. self.flags[flag.name] = flag
  487. #
  488. # Add new attributes to the class
  489. #
  490. self.run_ = True
  491. self.finish_ = True
  492. self.check_ = True
  493. self.env_ = None
  494. self.stdin_ = None
  495. self.stdin = None
  496. self.stdout_ = None
  497. self.stderr_ = None
  498. diz = {'name': 'stdin', 'required': False,
  499. 'multiple': False, 'type': 'all',
  500. 'value': None}
  501. self.inputs['stdin'] = Parameter(diz=diz)
  502. diz['name'] = 'stdout'
  503. self.outputs['stdout'] = Parameter(diz=diz)
  504. diz['name'] = 'stderr'
  505. self.outputs['stderr'] = Parameter(diz=diz)
  506. self.popen = None
  507. self.time = None
  508. self.start_time = None # This variable will be set in the run() function
  509. self._finished = False # This variable is set True if wait() was successfully called
  510. if args or kargs:
  511. self.__call__(*args, **kargs)
  512. self.__call__.__func__.__doc__ = self.__doc__
  513. def __call__(self, *args, **kargs):
  514. """Set module parameters to the class and, if run_ is True execute the
  515. module, therefore valid parameters are all the module parameters
  516. plus some extra parameters that are: run_, stdin_, stdout_, stderr_,
  517. env_ and finish_.
  518. """
  519. if not args and not kargs:
  520. self.run()
  521. return self
  522. #
  523. # check for extra kargs, set attribute and remove from dictionary
  524. #
  525. if 'flags' in kargs:
  526. for flg in kargs['flags']:
  527. self.flags[flg].value = True
  528. del(kargs['flags'])
  529. # set attributs
  530. for key in ('run_', 'env_', 'finish_', 'stdout_', 'stderr_', 'check_'):
  531. if key in kargs:
  532. setattr(self, key, kargs.pop(key))
  533. # set inputs
  534. for key in ('stdin_', ):
  535. if key in kargs:
  536. self.inputs[key[:-1]].value = kargs.pop(key)
  537. #
  538. # set/update args
  539. #
  540. for param, arg in zip(self.params_list, args):
  541. param.value = arg
  542. for key, val in kargs.items():
  543. key = key.strip('_')
  544. if key in self.inputs:
  545. self.inputs[key].value = val
  546. elif key in self.outputs:
  547. self.outputs[key].value = val
  548. elif key in self.flags:
  549. # we need to add this, because some parameters (overwrite,
  550. # verbose and quiet) work like parameters
  551. self.flags[key].value = val
  552. else:
  553. raise ParameterError('%s is not a valid parameter.' % key)
  554. #
  555. # check if execute
  556. #
  557. if self.run_:
  558. #
  559. # check reqire parameters
  560. #
  561. if self.check_:
  562. self.check()
  563. return self.run()
  564. return self
  565. def get_bash(self):
  566. """Return a BASH representation of the Module."""
  567. return ' '.join(self.make_cmd())
  568. def get_python(self):
  569. """Return a Python representation of the Module."""
  570. prefix = self.name.split('.')[0]
  571. name = '_'.join(self.name.split('.')[1:])
  572. params = ', '.join([par.get_python() for par in self.params_list
  573. if par.get_python() != ''])
  574. flags = ''.join([flg.get_python()
  575. for flg in self.flags.values()
  576. if not flg.special and flg.get_python() != ''])
  577. special = ', '.join([flg.get_python()
  578. for flg in self.flags.values()
  579. if flg.special and flg.get_python() != ''])
  580. # pre name par flg special
  581. if flags and special:
  582. return "%s.%s(%s, flags=%r, %s)" % (prefix, name, params,
  583. flags, special)
  584. elif flags:
  585. return "%s.%s(%s, flags=%r)" % (prefix, name, params, flags)
  586. elif special:
  587. return "%s.%s(%s, %s)" % (prefix, name, params, special)
  588. else:
  589. return "%s.%s(%s)" % (prefix, name, params)
  590. def __str__(self):
  591. """Return the command string that can be executed in a shell"""
  592. return ' '.join(self.make_cmd())
  593. def __repr__(self):
  594. return "Module(%r)" % self.name
  595. @docstring_property(__doc__)
  596. def __doc__(self):
  597. """{cmd_name}({cmd_params})
  598. """
  599. head = DOC['head'].format(cmd_name=self.name,
  600. cmd_params=('\n' + # go to a new line
  601. # give space under the function name
  602. (' ' * (len(self.name) + 1))).join([', '.join(
  603. # transform each parameter in string
  604. [str(param) for param in line if param is not None])
  605. # make a list of parameters with only 3 param per line
  606. for line in zip_longest(*[iter(self.params_list)] * 3)]),)
  607. params = '\n'.join([par.__doc__ for par in self.params_list])
  608. flags = self.flags.__doc__
  609. return '\n'.join([head, params, DOC['flag_head'], flags, DOC['foot']])
  610. def check(self):
  611. """Check the correctness of the provide parameters"""
  612. required = True
  613. for flg in self.flags.values():
  614. if flg and flg.suppress_required:
  615. required = False
  616. if required:
  617. for k in self.required:
  618. if ((k in self.inputs and self.inputs[k].value is None) or
  619. (k in self.outputs and self.outputs[k].value is None)):
  620. msg = "Required parameter <%s> not set."
  621. raise ParameterError(msg % k)
  622. def get_dict(self):
  623. """Return a dictionary that includes the name, all valid
  624. inputs, outputs and flags
  625. """
  626. dic = {}
  627. dic['name'] = self.name
  628. dic['inputs'] = [(k, v.value) for k, v in self.inputs.items()
  629. if v.value]
  630. dic['outputs'] = [(k, v.value) for k, v in self.outputs.items()
  631. if v.value]
  632. dic['flags'] = [flg for flg in self.flags if self.flags[flg].value]
  633. return dic
  634. def make_cmd(self):
  635. """Create the command string that can be executed in a shell
  636. :returns: the command string
  637. """
  638. skip = ['stdin', 'stdout', 'stderr']
  639. args = [self.name, ]
  640. for key in self.inputs:
  641. if key not in skip and self.inputs[key].value is not None and self.inputs[key].value != '':
  642. args.append(self.inputs[key].get_bash())
  643. for key in self.outputs:
  644. if key not in skip and self.outputs[key].value is not None and self.outputs[key].value != '':
  645. args.append(self.outputs[key].get_bash())
  646. for flg in self.flags:
  647. if self.flags[flg].value:
  648. args.append(str(self.flags[flg]))
  649. return args
  650. def run(self):
  651. """Run the module
  652. This function will wait for the process to terminate in case
  653. finish_==True and sets up stdout and stderr. If finish_==False this
  654. function will return after starting the process. Use wait() to wait for
  655. the started process
  656. :return: A reference to this object
  657. """
  658. G_debug(1, self.get_bash())
  659. self._finished = False
  660. if self.inputs['stdin'].value:
  661. self.stdin = self.inputs['stdin'].value
  662. self.stdin_ = PIPE
  663. cmd = self.make_cmd()
  664. self.start_time = time.time()
  665. self.popen = Popen(cmd,
  666. stdin=self.stdin_,
  667. stdout=self.stdout_,
  668. stderr=self.stderr_,
  669. env=self.env_)
  670. if self.finish_ is True:
  671. self.wait()
  672. return self
  673. def wait(self):
  674. """Wait for the module to finish. Call this method if
  675. the run() call was performed with self.false_ = False.
  676. :return: A reference to this object
  677. """
  678. if self._finished is False:
  679. if self.stdin:
  680. self.stdin = encode(self.stdin)
  681. stdout, stderr = self.popen.communicate(input=self.stdin)
  682. self.outputs['stdout'].value = decode(stdout) if stdout else ''
  683. self.outputs['stderr'].value = decode(stderr) if stderr else ''
  684. self.time = time.time() - self.start_time
  685. self._finished = True
  686. if self.popen.poll():
  687. raise CalledModuleError(returncode=self.popen.returncode,
  688. code=self.get_bash(),
  689. module=self.name, errors=stderr)
  690. return self
  691. class MultiModule(object):
  692. """This class is designed to run a list of modules in serial in the provided order
  693. within a temporary region environment.
  694. Module can be run in serial synchronously or asynchronously.
  695. - Synchronously: When calling run() all modules will run in serial order
  696. until they are finished, The run() method will return until all modules finished.
  697. The modules objects can be accessed by calling get_modules() to check their return
  698. values.
  699. - Asynchronously: When calling run() all modules will run in serial order in a background process.
  700. Method run() will return after starting the modules without waiting for them to finish.
  701. The user must call the wait() method to wait for the modules to finish.
  702. Asynchronously called module can be optionally run in a temporary region
  703. environment, hence invokeing g.region will not alter the current
  704. region or the region of other MultiModule runs.
  705. Note:
  706. Modules run in asynchronous mode can only be accessed via the wait() method.
  707. The wait() method will return all finished module objects as list.
  708. Objects of this class can be passed to the ParallelModuleQueue to run serial stacks
  709. of modules in parallel. This is meaningful if region settings must be applied
  710. to each parallel module run.
  711. >>> from grass.pygrass.modules import Module
  712. >>> from grass.pygrass.modules import MultiModule
  713. >>> from multiprocessing import Process
  714. >>> import copy
  715. Synchronous module run
  716. >>> region_1 = Module("g.region", run_=False)
  717. >>> region_1.flags.p = True
  718. >>> region_2 = copy.deepcopy(region_1)
  719. >>> region_2.flags.p = True
  720. >>> mm = MultiModule(module_list=[region_1, region_2])
  721. >>> mm.run()
  722. >>> m_list = mm.get_modules()
  723. >>> m_list[0].popen.returncode
  724. 0
  725. >>> m_list[1].popen.returncode
  726. 0
  727. Asynchronous module run, setting finish = False
  728. >>> region_1 = Module("g.region", run_=False) # doctest: +SKIP
  729. >>> region_1.flags.p = True # doctest: +SKIP
  730. >>> region_2 = copy.deepcopy(region_1) # doctest: +SKIP
  731. >>> region_2.flags.p = True # doctest: +SKIP
  732. >>> region_3 = copy.deepcopy(region_1) # doctest: +SKIP
  733. >>> region_3.flags.p = True # doctest: +SKIP
  734. >>> region_4 = copy.deepcopy(region_1) # doctest: +SKIP
  735. >>> region_4.flags.p = True # doctest: +SKIP
  736. >>> region_5 = copy.deepcopy(region_1) # doctest: +SKIP
  737. >>> region_5.flags.p = True # doctest: +SKIP
  738. >>> mm = MultiModule(module_list=[region_1, region_2, region_3, region_4, region_5],
  739. ... sync=False) # doctest: +SKIP
  740. >>> t = mm.run() # doctest: +SKIP
  741. >>> isinstance(t, Process) # doctest: +SKIP
  742. True
  743. >>> m_list = mm.wait() # doctest: +SKIP
  744. >>> m_list[0].popen.returncode # doctest: +SKIP
  745. 0
  746. >>> m_list[1].popen.returncode # doctest: +SKIP
  747. 0
  748. >>> m_list[2].popen.returncode # doctest: +SKIP
  749. 0
  750. >>> m_list[3].popen.returncode # doctest: +SKIP
  751. 0
  752. >>> m_list[4].popen.returncode # doctest: +SKIP
  753. 0
  754. Asynchronous module run, setting finish = False and using temporary region
  755. >>> mm = MultiModule(module_list=[region_1, region_2, region_3, region_4, region_5],
  756. ... sync=False, set_temp_region=True)
  757. >>> str(mm)
  758. 'g.region -p ; g.region -p ; g.region -p ; g.region -p ; g.region -p'
  759. >>> t = mm.run()
  760. >>> isinstance(t, Process)
  761. True
  762. >>> m_list = mm.wait()
  763. >>> m_list[0].popen.returncode
  764. 0
  765. >>> m_list[1].popen.returncode
  766. 0
  767. >>> m_list[2].popen.returncode
  768. 0
  769. >>> m_list[3].popen.returncode
  770. 0
  771. >>> m_list[4].popen.returncode
  772. 0
  773. """
  774. def __init__(self, module_list, sync=True, set_temp_region=False):
  775. """Constructor of the multi module class
  776. :param module_list: A list of pre-configured Module objects that should be run
  777. :param sync: If set True the run() method will wait for all processes to finish -> synchronously run.
  778. If set False, the run() method will return after starting the processes -> asynchronously run.
  779. The wait() method must be called to finish the modules.
  780. :param set_temp_region: Set a temporary region in which the modules should be run, hence
  781. region settings in the process list will not affect the current
  782. computation region.
  783. Note:
  784. This flag is only available in asynchronous mode!
  785. :return:
  786. """
  787. self.module_list = module_list
  788. self.set_temp_region = set_temp_region
  789. self.finish_ = sync # We use the same variable name a Module
  790. self.p = None
  791. self.q = Queue()
  792. def __str__(self):
  793. """Return the command string that can be executed in a shell"""
  794. return ' ; '.join(str(string) for string in self.module_list)
  795. def get_modules(self):
  796. """Return the list of modules that have been run in synchronous mode
  797. Note: Asynchronously run module can only be accessed via the wait() method.
  798. :return: The list of modules
  799. """
  800. return self.module_list
  801. def run(self):
  802. """Start the modules in the list. If self.finished_ is set True
  803. this method will return after all processes finished.
  804. If self.finish_ is set False, this method will return
  805. after the process list was started for execution.
  806. In a background process, the processes in the list will
  807. be run one after the another.
  808. :return: None in case of self.finish_ is True,
  809. otherwise a multiprocessing.Process object that invokes the modules
  810. """
  811. if self.finish_ is True:
  812. for module in self.module_list:
  813. module.finish_ = True
  814. module.run()
  815. return None
  816. else:
  817. if self.set_temp_region is True:
  818. self.p = Process(target=run_modules_in_temp_region,
  819. args=[self.module_list, self.q])
  820. else:
  821. self.p = Process(target=run_modules,
  822. args=[self.module_list, self.q])
  823. self.p.start()
  824. return self.p
  825. def wait(self):
  826. """Wait for all processes to finish. Call this method
  827. in asynchronous mode, hence if finished was set False.
  828. :return: The process list with finished processes to check their return states
  829. """
  830. if self.p:
  831. proc_list = self.q.get()
  832. self.p.join()
  833. return proc_list
  834. def run_modules_in_temp_region(module_list, q):
  835. """Run the modules in a temporary region environment
  836. This function is the argument for multiprocessing.Process class
  837. in the MultiModule asynchronous execution.
  838. :param module_list: The list of modules to run in serial
  839. :param q: The process queue to put the finished process list
  840. """
  841. use_temp_region()
  842. try:
  843. for proc in module_list:
  844. proc.run()
  845. proc.wait()
  846. except:
  847. raise
  848. finally:
  849. q.put(module_list)
  850. del_temp_region()
  851. def run_modules(module_list, q):
  852. """Run the modules
  853. This function is the argument for multiprocessing.Process class
  854. in the MultiModule asynchronous execution.
  855. :param module_list: The list of modules to run in serial
  856. :param q: The process queue to put the finished process list
  857. """
  858. try:
  859. for proc in module_list:
  860. proc.run()
  861. proc.wait()
  862. except:
  863. raise
  864. finally:
  865. q.put(module_list)
  866. ###############################################################################
  867. if __name__ == "__main__":
  868. import doctest
  869. doctest.testmod()