module.py 34 KB

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