/
core.py
1729 lines (1361 loc) · 54.7 KB
/
core.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
Core functions to be used in Python scripts.
Usage:
::
from grass.script import core as grass
grass.parser()
(C) 2008-2020 by the GRASS Development Team
This program is free software under the GNU General Public
License (>=v2). Read the file COPYING that comes with GRASS
for details.
.. sectionauthor:: Glynn Clements
.. sectionauthor:: Martin Landa <landa.martin gmail.com>
.. sectionauthor:: Michael Barton <michael.barton asu.edu>
"""
from __future__ import absolute_import, print_function
import os
import sys
import atexit
import subprocess
import shutil
import codecs
import string
import random
import pipes
import types as python_types
from .utils import KeyValue, parse_key_val, basename, encode, decode
from grass.exceptions import ScriptError, CalledModuleError
# PY2/PY3 compat
if sys.version_info.major > 2:
unicode = str
# subprocess wrapper that uses shell on Windows
class Popen(subprocess.Popen):
_builtin_exts = set(['.com', '.exe', '.bat', '.cmd'])
@staticmethod
def _escape_for_shell(arg):
# TODO: what are cmd.exe's parsing rules?
return arg
def __init__(self, args, **kwargs):
if (sys.platform == 'win32'
and isinstance(args, list)
and not kwargs.get('shell', False)
and kwargs.get('executable') is None):
cmd = shutil_which(args[0])
if cmd is None:
raise OSError(_("Cannot find the executable {0}")
.format(args[0]))
args = [cmd] + args[1:]
name, ext = os.path.splitext(cmd)
if ext.lower() not in self._builtin_exts:
kwargs['shell'] = True
args = [self._escape_for_shell(arg) for arg in args]
subprocess.Popen.__init__(self, args, **kwargs)
PIPE = subprocess.PIPE
STDOUT = subprocess.STDOUT
raise_on_error = False # raise exception instead of calling fatal()
_capture_stderr = False # capture stderr of subprocesses if possible
def call(*args, **kwargs):
return Popen(*args, **kwargs).wait()
# GRASS-oriented interface to subprocess module
_popen_args = ["bufsize", "executable", "stdin", "stdout", "stderr",
"preexec_fn", "close_fds", "cwd", "env",
"universal_newlines", "startupinfo", "creationflags"]
def _make_val(val):
"""Convert value to unicode"""
if isinstance(val, (bytes, str, unicode)):
return decode(val)
if isinstance(val, (int, float)):
return unicode(val)
try:
return ",".join(map(_make_val, iter(val)))
except TypeError:
pass
return unicode(val)
def _make_unicode(val, enc):
"""Convert value to unicode with given encoding
:param val: value to be converted
:param enc: encoding to be used
"""
if val is None or enc is None:
return val
else:
if enc == 'default':
return decode(val)
else:
return decode(val, encoding=enc)
def get_commands():
"""Create list of available GRASS commands to use when parsing
string from the command line
:return: list of commands (set) and directory of scripts (collected
by extension - MS Windows only)
>>> cmds = list(get_commands()[0])
>>> cmds.sort()
>>> cmds[:5]
['d.barscale', 'd.colorlist', 'd.colortable', 'd.correlate', 'd.erase']
"""
gisbase = os.environ['GISBASE']
cmd = list()
scripts = {'.py': list()} if sys.platform == 'win32' else {}
def scan(gisbase, directory):
dir_path = os.path.join(gisbase, directory)
if os.path.exists(dir_path):
for fname in os.listdir(os.path.join(gisbase, directory)):
if scripts: # win32
name, ext = os.path.splitext(fname)
if ext != '.manifest':
cmd.append(name)
if ext in scripts.keys():
scripts[ext].append(name)
else:
cmd.append(fname)
for directory in ('bin', 'scripts'):
scan(gisbase, directory)
# scan gui/scripts/
gui_path = os.path.join(gisbase, 'etc', 'gui', 'scripts')
if os.path.exists(gui_path):
os.environ["PATH"] = os.getenv("PATH") + os.pathsep + gui_path
cmd = cmd + os.listdir(gui_path)
return set(cmd), scripts
# TODO: Please replace this function with shutil.which() before 8.0 comes out
# replacement for which function from shutil (not available in all versions)
# from http://hg.python.org/cpython/file/6860263c05b3/Lib/shutil.py#l1068
# added because of Python scripts running Python scripts on MS Windows
# see also ticket #2008 which is unrelated but same function was proposed
def shutil_which(cmd, mode=os.F_OK | os.X_OK, path=None):
"""Given a command, mode, and a PATH string, return the path which
conforms to the given mode on the PATH, or None if there is no such
file.
`mode` defaults to os.F_OK | os.X_OK. `path` defaults to the result
of os.environ.get("PATH"), or can be overridden with a custom search
path.
:param cmd: the command
:param mode:
:param path:
"""
# Check that a given file can be accessed with the correct mode.
# Additionally check that `file` is not a directory, as on Windows
# directories pass the os.access check.
def _access_check(fn, mode):
return (os.path.exists(fn) and os.access(fn, mode)
and not os.path.isdir(fn))
# If we're given a path with a directory part, look it up directly rather
# than referring to PATH directories. This includes checking relative to the
# current directory, e.g. ./script
if os.path.dirname(cmd):
if _access_check(cmd, mode):
return cmd
return None
if path is None:
path = os.environ.get("PATH", os.defpath)
if not path:
return None
path = path.split(os.pathsep)
if sys.platform == "win32":
# The current directory takes precedence on Windows.
if not os.curdir in path:
path.insert(0, os.curdir)
# PATHEXT is necessary to check on Windows (force lowercase)
pathext = list(map(lambda x: x.lower(),
os.environ.get("PATHEXT", "").split(os.pathsep)))
if '.py' not in pathext:
# we assume that PATHEXT contains always '.py'
pathext.insert(0, '.py')
# See if the given file matches any of the expected path extensions.
# This will allow us to short circuit when given "python3.exe".
# If it does match, only test that one, otherwise we have to try
# others.
if any(cmd.lower().endswith(ext) for ext in pathext):
files = [cmd]
else:
files = [cmd + ext for ext in pathext]
else:
# On other platforms you don't have things like PATHEXT to tell you
# what file suffixes are executable, so just pass on cmd as-is.
files = [cmd]
seen = set()
for dir in path:
normdir = os.path.normcase(dir)
if not normdir in seen:
seen.add(normdir)
for thefile in files:
name = os.path.join(dir, thefile)
if _access_check(name, mode):
return name
return None
if sys.version_info.major > 2:
shutil_which = shutil.which
# Added because of scripts calling scripts on MS Windows.
# Module name (here cmd) differs from the file name (does not have extension).
# Additionally, we don't run scripts using system executable mechanism,
# so we need the full path name.
# However, scripts are on the PATH and '.PY' in in PATHEXT, so we can use
# shutil.which to get the full file path. Addons are on PATH too.
# An alternative to which function call would be to check the script path and
# addons path. This is proposed improvement for the future.
# Another alternative is to check some global list of scripts but this list
# needs to be created first. The question is what is less expensive.
# Note that getting the full path is only part of the solution,
# the other part is to use the right Python as an executable and pass the full
# script path as a parameter.
# Nevertheless, it is unclear on which places which extensions are added.
# This function also could skip the check for platform but depends
# how will be used, this is most general but not most effective.
def get_real_command(cmd):
"""Returns the real file command for a module (cmd)
For Python scripts on MS Windows it returns full path to the script
and adds a '.py' extension.
For other cases it just returns a module (name).
So, you can just use this function for all without further check.
>>> get_real_command('g.region')
'g.region'
:param cmd: the command
"""
if sys.platform == 'win32':
# we in fact expect pure module name (without extension)
# so, lets remove extension
if os.path.splitext(cmd)[1] == '.py':
cmd = cmd[:-3]
# PATHEXT is necessary to check on Windows (force lowercase)
pathext = list(map(lambda x: x.lower(),
os.environ['PATHEXT'].split(os.pathsep)))
if '.py' not in pathext:
# we assume that PATHEXT contains always '.py'
os.environ['PATHEXT'] = '.py;' + os.environ['PATHEXT']
full_path = shutil_which(cmd + '.py')
if full_path:
return full_path
return cmd
def make_command(prog, flags="", overwrite=False, quiet=False, verbose=False,
superquiet=False, errors=None, **options):
"""Return a list of strings suitable for use as the args parameter to
Popen() or call(). Example:
>>> make_command("g.message", flags = 'w', message = 'this is a warning')
['g.message', '-w', 'message=this is a warning']
:param str prog: GRASS module
:param str flags: flags to be used (given as a string)
:param bool overwrite: True to enable overwriting the output (<tt>--o</tt>)
:param bool quiet: True to run quietly (<tt>--q</tt>)
:param bool verbose: True to run verbosely (<tt>--v</tt>)
:param options: module's parameters
:return: list of arguments
"""
args = [_make_val(prog)]
if overwrite:
args.append("--o")
if quiet:
args.append("--q")
if verbose:
args.append("--v")
if superquiet:
args.append("--qq")
if flags:
flags = _make_val(flags)
if '-' in flags:
raise ScriptError("'-' is not a valid flag")
args.append("-" + flags)
for opt, val in options.items():
if opt in _popen_args:
continue
# convert string to bytes
if val is not None:
if opt.startswith('_'):
opt = opt[1:]
warning(_("To run the module <%s> add underscore at the end"
" of the option <%s> to avoid conflict with Python"
" keywords. Underscore at the beginning is"
" depreciated in GRASS GIS 7.0 and will be removed"
" in version 7.1.") % (prog, opt))
elif opt.endswith('_'):
opt = opt[:-1]
args.append(opt + '=' + _make_val(val))
return args
def handle_errors(returncode, result, args, kwargs):
if returncode == 0:
return result
handler = kwargs.get('errors', 'raise')
if handler.lower() == 'ignore':
return result
elif handler.lower() == 'status':
return returncode
elif handler.lower() == 'exit':
sys.exit(1)
else:
# TODO: construction of the whole command is far from perfect
args = make_command(*args, **kwargs)
code = ' '.join(args)
raise CalledModuleError(module=None, code=code,
returncode=returncode)
def start_command(prog, flags="", overwrite=False, quiet=False,
verbose=False, superquiet=False, **kwargs):
"""Returns a Popen object with the command created by make_command.
Accepts any of the arguments which Popen() accepts apart from "args"
and "shell".
>>> p = start_command("g.gisenv", stdout=subprocess.PIPE)
>>> print(p) # doctest: +ELLIPSIS
<...Popen object at 0x...>
>>> print(p.communicate()[0]) # doctest: +SKIP
GISDBASE='/opt/grass-data';
LOCATION_NAME='spearfish60';
MAPSET='glynn';
GUI='text';
MONITOR='x0';
If the module parameter is the same as Python keyword, add
underscore at the end of the parameter. For example, use
``lambda_=1.6`` instead of ``lambda=1.6``.
:param str prog: GRASS module
:param str flags: flags to be used (given as a string)
:param bool overwrite: True to enable overwriting the output (<tt>--o</tt>)
:param bool quiet: True to run quietly (<tt>--q</tt>)
:param bool verbose: True to run verbosely (<tt>--v</tt>)
:param kwargs: module's parameters
:return: Popen object
"""
if 'encoding' in kwargs.keys():
encoding = kwargs.pop('encoding')
options = {}
popts = {}
for opt, val in kwargs.items():
if opt in _popen_args:
popts[opt] = val
else:
options[opt] = val
args = make_command(prog, flags, overwrite, quiet, verbose, **options)
if debug_level() > 0:
sys.stderr.write("D1/{}: {}.start_command(): {}\n".format(
debug_level(), __name__,
' '.join(args))
)
sys.stderr.flush()
return Popen(args, **popts)
def run_command(*args, **kwargs):
"""Execute a module synchronously
This function passes all arguments to ``start_command()``,
then waits for the process to complete. It is similar to
``subprocess.check_call()``, but with the ``make_command()``
interface.
For backward compatibility, the function returns exit code
by default but only if it is equal to zero. An exception is raised
in case of an non-zero return code.
>>> run_command('g.region', raster='elevation')
0
See :func:`start_command()` for details about parameters and usage.
..note::
You should ignore the return value of this function unless, you
change the default behavior using *errors* parameter.
:param *args: unnamed arguments passed to ``start_command()``
:param **kwargs: named arguments passed to ``start_command()``
:returns: 0 with default parameters for backward compatibility only
:raises: ``CalledModuleError`` when module returns non-zero return code
"""
encoding = 'default'
if 'encoding' in kwargs:
encoding = kwargs['encoding']
if _capture_stderr and 'stderr' not in kwargs.keys():
kwargs['stderr'] = PIPE
ps = start_command(*args, **kwargs)
if _capture_stderr:
stdout, stderr = ps.communicate()
if encoding is not None:
stdout = _make_unicode(stdout, encoding)
stderr = _make_unicode(stderr, encoding)
returncode = ps.poll()
if returncode:
sys.stderr.write(stderr)
else:
returncode = ps.wait()
return handle_errors(returncode, returncode, args, kwargs)
def pipe_command(*args, **kwargs):
"""Passes all arguments to start_command(), but also adds
"stdout = PIPE". Returns the Popen object.
>>> p = pipe_command("g.gisenv")
>>> print(p) # doctest: +ELLIPSIS
<....Popen object at 0x...>
>>> print(p.communicate()[0]) # doctest: +SKIP
GISDBASE='/opt/grass-data';
LOCATION_NAME='spearfish60';
MAPSET='glynn';
GUI='text';
MONITOR='x0';
:param list args: list of unnamed arguments (see start_command() for details)
:param list kwargs: list of named arguments (see start_command() for details)
:return: Popen object
"""
kwargs['stdout'] = PIPE
return start_command(*args, **kwargs)
def feed_command(*args, **kwargs):
"""Passes all arguments to start_command(), but also adds
"stdin = PIPE". Returns the Popen object.
:param list args: list of unnamed arguments (see start_command() for details)
:param list kwargs: list of named arguments (see start_command() for details)
:return: Popen object
"""
kwargs['stdin'] = PIPE
return start_command(*args, **kwargs)
def read_command(*args, **kwargs):
"""Passes all arguments to pipe_command, then waits for the process to
complete, returning its stdout (i.e. similar to shell `backticks`).
:param list args: list of unnamed arguments (see start_command() for details)
:param list kwargs: list of named arguments (see start_command() for details)
:return: stdout
"""
encoding = 'default'
if 'encoding' in kwargs:
encoding = kwargs['encoding']
if _capture_stderr and 'stderr' not in kwargs.keys():
kwargs['stderr'] = PIPE
process = pipe_command(*args, **kwargs)
stdout, stderr = process.communicate()
if encoding is not None:
stdout = _make_unicode(stdout, encoding)
stderr = _make_unicode(stderr, encoding)
returncode = process.poll()
if _capture_stderr and returncode:
sys.stderr.write(stderr)
return handle_errors(returncode, stdout, args, kwargs)
def parse_command(*args, **kwargs):
"""Passes all arguments to read_command, then parses the output
by parse_key_val().
Parsing function can be optionally given by <em>parse</em> parameter
including its arguments, e.g.
::
parse_command(..., parse = (grass.parse_key_val, { 'sep' : ':' }))
or you can simply define <em>delimiter</em>
::
parse_command(..., delimiter = ':')
:param args: list of unnamed arguments (see start_command() for details)
:param kwargs: list of named arguments (see start_command() for details)
:return: parsed module output
"""
parse = None
parse_args = {}
if 'parse' in kwargs:
if isinstance(kwargs['parse'], tuple):
parse = kwargs['parse'][0]
parse_args = kwargs['parse'][1]
del kwargs['parse']
if 'delimiter' in kwargs:
parse_args = {'sep': kwargs['delimiter']}
del kwargs['delimiter']
if not parse:
parse = parse_key_val # use default fn
res = read_command(*args, **kwargs)
return parse(res, **parse_args)
def write_command(*args, **kwargs):
"""Execute a module with standard input given by *stdin* parameter.
Passes all arguments to ``feed_command()``, with the string specified
by the *stdin* argument fed to the process' standard input.
>>> gscript.write_command(
... 'v.in.ascii', input='-',
... stdin='%s|%s' % (635818.8, 221342.4),
... output='view_point')
0
See ``start_command()`` for details about parameters and usage.
:param *args: unnamed arguments passed to ``start_command()``
:param **kwargs: named arguments passed to ``start_command()``
:returns: 0 with default parameters for backward compatibility only
:raises: ``CalledModuleError`` when module returns non-zero return code
"""
encoding = 'default'
if 'encoding' in kwargs:
encoding = kwargs['encoding']
# TODO: should we delete it from kwargs?
stdin = kwargs['stdin']
if encoding is None or encoding == 'default':
stdin = encode(stdin)
else:
stdin = encode(stdin, encoding=encoding)
if _capture_stderr and 'stderr' not in kwargs.keys():
kwargs['stderr'] = PIPE
process = feed_command(*args, **kwargs)
unused, stderr = process.communicate(stdin)
if encoding is not None:
unused = _make_unicode(unused, encoding)
stderr = _make_unicode(stderr, encoding)
returncode = process.poll()
if _capture_stderr and returncode:
sys.stderr.write(stderr)
return handle_errors(returncode, returncode, args, kwargs)
def exec_command(prog, flags="", overwrite=False, quiet=False, verbose=False,
superquiet=False, env=None, **kwargs):
"""Interface to os.execvpe(), but with the make_command() interface.
:param str prog: GRASS module
:param str flags: flags to be used (given as a string)
:param bool overwrite: True to enable overwriting the output (<tt>--o</tt>)
:param bool quiet: True to run quietly (<tt>--q</tt>)
:param bool verbose: True to run verbosely (<tt>--v</tt>)
:param env: directory with environmental variables
:param list kwargs: module's parameters
"""
args = make_command(prog, flags, overwrite, quiet, verbose, **kwargs)
if env is None:
env = os.environ
os.execvpe(prog, args, env)
# interface to g.message
def message(msg, flag=None):
"""Display a message using `g.message`
:param str msg: message to be displayed
:param str flag: flags (given as string)
"""
run_command("g.message", flags=flag, message=msg, errors='ignore')
def debug(msg, debug=1):
"""Display a debugging message using `g.message -d`
:param str msg: debugging message to be displayed
:param str debug: debug level (0-5)
"""
if debug_level() >= debug:
# TODO: quite a random hack here, do we need it somewhere else too?
if sys.platform == "win32":
msg = msg.replace('&', '^&')
run_command("g.message", flags='d', message=msg, debug=debug)
def verbose(msg):
"""Display a verbose message using `g.message -v`
:param str msg: verbose message to be displayed
"""
message(msg, flag='v')
def info(msg):
"""Display an informational message using `g.message -i`
:param str msg: informational message to be displayed
"""
message(msg, flag='i')
def percent(i, n, s):
"""Display a progress info message using `g.message -p`
::
message(_("Percent complete..."))
n = 100
for i in range(n):
percent(i, n, 1)
percent(1, 1, 1)
:param int i: current item
:param int n: total number of items
:param int s: increment size
"""
message("%d %d %d" % (i, n, s), flag='p')
def warning(msg):
"""Display a warning message using `g.message -w`
:param str msg: warning message to be displayed
"""
message(msg, flag='w')
def error(msg):
"""Display an error message using `g.message -e`
This function does not end the execution of the program.
The right action after the error is up to the caller.
For error handling using the standard mechanism use :func:`fatal()`.
:param str msg: error message to be displayed
"""
message(msg, flag='e')
def fatal(msg):
"""Display an error message using `g.message -e`, then abort or raise
Raises exception when module global raise_on_error is 'True', abort
(calls exit) otherwise.
Use :func:`set_raise_on_error()` to set the behavior.
:param str msg: error message to be displayed
"""
global raise_on_error
if raise_on_error:
raise ScriptError(msg)
error(msg)
sys.exit(1)
def set_raise_on_error(raise_exp=True):
"""Define behaviour on fatal error (fatal() called)
:param bool raise_exp: True to raise ScriptError instead of calling
sys.exit(1) in fatal()
:return: current status
"""
global raise_on_error
tmp_raise = raise_on_error
raise_on_error = raise_exp
return tmp_raise
def get_raise_on_error():
"""Return True if a ScriptError exception is raised instead of calling
sys.exit(1) in case a fatal error was invoked with fatal()
"""
global raise_on_error
return raise_on_error
# TODO: solve also warnings (not printed now)
def set_capture_stderr(capture=True):
"""Enable capturing standard error output of modules and print it.
By default, standard error output (stderr) of child processes shows
in the same place as output of the parent process. This may not
always be the same place as ``sys.stderr`` is written.
After calling this function, functions in the ``grass.script``
package will capture the stderr of child processes and pass it
to ``sys.stderr`` if there is an error.
.. note::
This is advantages for interactive shells such as the one in GUI
and interactive notebooks such as Jupyer Notebook.
The capturing can be applied only in certain cases, for example
in case of run_command() it is applied because run_command() nor
its callers do not handle the streams, however feed_command()
cannot do capturing because its callers handle the streams.
The previous state is returned. Passing ``False`` disables the
capturing.
.. versionadded:: 7.4
"""
global _capture_stderr
tmp = _capture_stderr
_capture_stderr = capture
return tmp
def get_capture_stderr():
"""Return True if stderr is captured, False otherwise.
See set_capture_stderr().
"""
global _capture_stderr
return _capture_stderr
# interface to g.parser
def _parse_opts(lines):
options = {}
flags = {}
for line in lines:
if not line:
break
try:
[var, val] = line.split(b'=', 1)
[var, val] = [decode(var), decode(val)]
except:
raise SyntaxError("invalid output from g.parser: %s" % line)
if var.startswith('flag_'):
flags[var[5:]] = bool(int(val))
elif var.startswith('opt_'):
options[var[4:]] = val
elif var in ['GRASS_OVERWRITE', 'GRASS_VERBOSE']:
os.environ[var] = val
else:
raise SyntaxError("invalid output from g.parser: %s" % line)
return (options, flags)
def parser():
"""Interface to g.parser, intended to be run from the top-level, e.g.:
::
if __name__ == "__main__":
options, flags = grass.parser()
main()
Thereafter, the global variables "options" and "flags" will be
dictionaries containing option/flag values, keyed by lower-case
option/flag names. The values in "options" are strings, those in
"flags" are Python booleans.
Overview table of parser standard options:
https://grass.osgeo.org/grass78/manuals/parser_standard_options.html
"""
if not os.getenv("GISBASE"):
print("You must be in GRASS GIS to run this program.", file=sys.stderr)
sys.exit(1)
cmdline = [basename(sys.argv[0])]
cmdline += [pipes.quote(a) for a in sys.argv[1:]]
os.environ['CMDLINE'] = ' '.join(cmdline)
argv = sys.argv[:]
name = argv[0]
if not os.path.isabs(name):
if os.sep in name or (os.altsep and os.altsep in name):
argv[0] = os.path.abspath(name)
else:
argv[0] = os.path.join(sys.path[0], name)
prog = "g.parser.exe" if sys.platform == "win32" else "g.parser"
p = subprocess.Popen([prog, '-n'] + argv, stdout=subprocess.PIPE)
s = p.communicate()[0]
lines = s.split(b'\0')
if not lines or lines[0] != b"@ARGS_PARSED@":
stdout = os.fdopen(sys.stdout.fileno(), 'wb')
stdout.write(s)
sys.exit(p.returncode)
return _parse_opts(lines[1:])
# interface to g.tempfile
def tempfile(create=True):
"""Returns the name of a temporary file, created with g.tempfile.
:param bool create: True to create a file
:return: path to a tmp file
"""
flags = ''
if not create:
flags += 'd'
return read_command("g.tempfile", flags=flags, pid=os.getpid()).strip()
def tempdir():
"""Returns the name of a temporary dir, created with g.tempfile."""
tmp = tempfile(create=False)
os.mkdir(tmp)
return tmp
def tempname(length, lowercase=False):
"""Generate a GRASS and SQL compliant random name starting with tmp_
followed by a random part of length "length"
:param int length: length of the random part of the name to generate
:param bool lowercase: use only lowercase characters to generate name
:returns: String with a random name of length "length" starting with a letter
:rtype: str
:Example:
>>> tempname(12)
'tmp_MxMa1kAS13s9'
"""
chars = string.ascii_lowercase + string.digits
if not lowercase:
chars += string.ascii_uppercase
random_part = ''.join(random.choice(chars) for _ in range(length))
randomname = 'tmp_' + random_part
return randomname
def _compare_projection(dic):
"""Check if projection has some possibility of duplicate names like
Universal Transverse Mercator and Universe Transverse Mercator and
unify them
:param dic: The dictionary containing information about projection
:return: The dictionary with the new values if needed
"""
# the lookup variable is a list of list, each list contains all the
# possible name for a projection system
lookup = [['Universal Transverse Mercator', 'Universe Transverse Mercator']]
for lo in lookup:
for n in range(len(dic['name'])):
if dic['name'][n] in lo:
dic['name'][n] = lo[0]
return dic
def _compare_units(dic):
"""Check if units has some possibility of duplicate names like
meter and metre and unify them
:param dic: The dictionary containing information about units
:return: The dictionary with the new values if needed
"""
# the lookup variable is a list of list, each list contains all the
# possible name for a units
lookup = [['meter', 'metre'], ['meters', 'metres'], ['kilometer',
'kilometre'], ['kilometers', 'kilometres']]
for l in lookup:
for n in range(len(dic['unit'])):
if dic['unit'][n].lower() in l:
dic['unit'][n] = l[0]
for n in range(len(dic['units'])):
if dic['units'][n].lower() in l:
dic['units'][n] = l[0]
return dic
def _text_to_key_value_dict(filename, sep=":", val_sep=",", checkproj=False,
checkunits=False):
"""Convert a key-value text file, where entries are separated by newlines
and the key and value are separated by `sep', into a key-value dictionary
and discover/use the correct data types (float, int or string) for values.
:param str filename: The name or name and path of the text file to convert
:param str sep: The character that separates the keys and values, default
is ":"
:param str val_sep: The character that separates the values of a single
key, default is ","
:param bool checkproj: True if it has to check some information about
projection system
:param bool checkproj: True if it has to check some information about units
:return: The dictionary
A text file with this content:
::
a: Hello
b: 1.0
c: 1,2,3,4,5
d : hello,8,0.1
Will be represented as this dictionary:
::
{'a': ['Hello'], 'c': [1, 2, 3, 4, 5], 'b': [1.0], 'd': ['hello', 8, 0.1]}
"""
text = open(filename, "r").readlines()
kvdict = KeyValue()
for line in text:
if line.find(sep) >= 0:
key, value = line.split(sep)
key = key.strip()
value = value.strip()
else:
# Jump over empty values
continue
values = value.split(val_sep)
value_list = []
for value in values:
not_float = False
not_int = False
# Convert values into correct types
# We first try integer then float
try:
value_converted = int(value)
except:
not_int = True
if not_int:
try:
value_converted = float(value)
except:
not_float = True
if not_int and not_float:
value_converted = value.strip()
value_list.append(value_converted)
kvdict[key] = value_list
if checkproj:
kvdict = _compare_projection(kvdict)
if checkunits: