#!/usr/bin/env python3
from __future__ import (
nested_scopes,
generators,
division,
absolute_import,
with_statement,
print_function,
unicode_literals,
)
from os import listdir
from os.path import join, isdir
import shutil
import ctypes as ct
import fnmatch
import grass.lib.gis as libgis
from grass.pygrass.errors import GrassError
from grass.script.utils import encode, decode
from grass.pygrass.utils import getenv
from grass.pygrass.gis.region import Region
test_vector_name = "Gis_test_vector"
test_raster_name = "Gis_test_raster"
libgis.G_gisinit("")
ETYPE = {
"raster": libgis.G_ELEMENT_RASTER,
"raster_3d": libgis.G_ELEMENT_RASTER3D,
"vector": libgis.G_ELEMENT_VECTOR,
"label": libgis.G_ELEMENT_LABEL,
"region": libgis.G_ELEMENT_REGION,
"group": libgis.G_ELEMENT_GROUP,
}
CHECK_IS = {
"GISBASE": libgis.G_is_gisbase,
"GISDBASE": lambda x: True,
"LOCATION_NAME": libgis.G_is_location,
"MAPSET": libgis.G_is_mapset,
}
[docs]def is_valid(value, path, type):
"""Private function to check the correctness of a value.
:param value: Name of the directory
:type value: str
:param path: Path where the directory is located
:type path: path
:param type: it is a string defining the type that will e checked,
valid types are: GISBASE, GISDBASE, LOCATION_NAME, MAPSET
:type type: str
:return: True if valid else False
:rtype: str
"""
return bool(CHECK_IS[type](join(path, value)))
def _check_raise(value, path, type):
"""Private function to check the correctness of a value.
:param value: Name of the directory
:type value: str
:param path: Path where the directory is located
:type path: path
:param type: it is a string defining the type that will e checked,
valid types are: GISBASE, GISDBASE, LOCATION_NAME, MAPSET
:type type: str
:return: the value if verify else None and
if value is empty return environmental variable
:rtype: str
"""
if value == "":
from grass.pygrass.utils import getenv
return getenv(type)
if is_valid(value, path, type):
return value
raise GrassError("%s <%s> not found" % (type.title(), join(path, value)))
[docs]def set_current_mapset(mapset, location=None, gisdbase=None):
"""Set the current mapset as working area
:param mapset: Name of the mapset
:type value: str
:param location: Name of the location
:type location: str
:param gisdbase: Name of the gisdbase
:type gisdbase: str
"""
libgis.G_setenv("MAPSET", mapset)
if location:
libgis.G_setenv("LOCATION_NAME", location)
if gisdbase:
libgis.G_setenv("GISDBASE", gisdbase)
[docs]def make_mapset(mapset, location=None, gisdbase=None):
"""Create a new mapset
:param mapset: Name of the mapset
:type value: str
:param location: Name of the location
:type location: str
:param gisdbase: Name of the gisdbase
:type gisdbase: str"""
res = libgis.G_make_mapset(gisdbase, location, mapset)
if res == -1:
raise GrassError("Cannot create new mapset")
elif res == -2:
raise GrassError("Illegal name")
[docs]class Gisdbase(object):
"""Return Gisdbase object. ::
>>> from grass.script.core import gisenv
>>> gisdbase = Gisdbase()
>>> gisdbase.name == gisenv()['GISDBASE']
True
..
"""
def __init__(self, gisdbase=""):
self.name = gisdbase
def _get_name(self):
return self._name
def _set_name(self, name):
self._name = _check_raise(name, "", "GISDBASE")
name = property(
fget=_get_name, fset=_set_name, doc="Set or obtain the name of GISDBASE"
)
def __str__(self):
return self.name
def __repr__(self):
return "Gisdbase(%s)" % self.name
def __getitem__(self, location):
"""Return a Location object. ::
>>> from grass.script.core import gisenv
>>> loc_env = gisenv()['LOCATION_NAME']
>>> gisdbase = Gisdbase()
>>> loc_py = gisdbase[loc_env]
>>> loc_env == loc_py.name
True
..
"""
if location in self.locations():
return Location(location, self.name)
else:
raise KeyError("Location: %s does not exist" % location)
def __iter__(self):
for loc in self.locations():
yield Location(loc, self.name)
# TODO remove or complete this function
[docs] def new_location(self):
if libgis.G_make_location() != 0:
raise GrassError("Cannot create new location")
[docs] def locations(self):
"""Return a list of locations that are available in the gisdbase: ::
>>> gisdbase = Gisdbase()
>>> gisdbase.locations() # doctest: +ELLIPSIS
[...]
..
"""
return sorted(
[
loc
for loc in listdir(self.name)
if libgis.G_is_location(encode(join(self.name, loc)))
]
)
[docs]class Location(object):
"""Location object ::
>>> from grass.script.core import gisenv
>>> location = Location()
>>> location # doctest: +ELLIPSIS
Location(...)
>>> location.gisdbase == gisenv()['GISDBASE']
True
>>> location.name == gisenv()['LOCATION_NAME']
True
..
"""
def __init__(self, location="", gisdbase=""):
self.gisdbase = gisdbase
self.name = location
def _get_gisdb(self):
return self._gisdb
def _set_gisdb(self, gisdb):
self._gisdb = _check_raise(gisdb, "", "GISDBASE")
gisdbase = property(
fget=_get_gisdb, fset=_set_gisdb, doc="Set or obtain the name of GISDBASE"
)
def _get_name(self):
return self._name
def _set_name(self, name):
self._name = _check_raise(name, self._gisdb, "LOCATION_NAME")
name = property(
fget=_get_name, fset=_set_name, doc="Set or obtain the name of LOCATION"
)
def __getitem__(self, mapset):
if mapset in self.mapsets():
return Mapset(mapset)
else:
raise KeyError("Mapset: %s does not exist" % mapset)
def __iter__(self):
lpath = self.path()
return (
m
for m in listdir(lpath)
if (isdir(join(lpath, m)) and is_valid(m, lpath, "MAPSET"))
)
def __len__(self):
return len(self.mapsets())
def __str__(self):
return self.name
def __repr__(self):
return "Location(%r)" % self.name
[docs] def mapsets(self, pattern=None, permissions=True):
"""Return a list of the available mapsets.
:param pattern: the pattern to filter the result
:type pattern: str
:param permissions: check the permission of mapset
:type permissions: bool
:return: a list of mapset's names
:rtype: list of strings
::
>>> location = Location()
>>> sorted(location.mapsets()) # doctest: +ELLIPSIS
[...]
"""
mapsets = [mapset for mapset in self]
if permissions:
mapsets = [
mapset
for mapset in mapsets
if libgis.G_mapset_permissions(encode(mapset))
]
if pattern:
return fnmatch.filter(mapsets, pattern)
return mapsets
[docs] def path(self):
"""Return the complete path of the location"""
return join(self.gisdbase, self.name)
[docs]class Mapset(object):
"""Mapset ::
>>> from grass.script.core import gisenv
>>> genv = gisenv()
>>> mapset = Mapset()
>>> mapset # doctest: +ELLIPSIS
Mapset(...)
>>> mapset.gisdbase == genv['GISDBASE']
True
>>> mapset.location == genv['LOCATION_NAME']
True
>>> mapset.name == genv['MAPSET']
True
..
"""
def __init__(self, mapset="", location="", gisdbase=""):
self.gisdbase = gisdbase
self.location = location
self.name = mapset
self.visible = VisibleMapset(self.name, self.location, self.gisdbase)
def _get_gisdb(self):
return self._gisdb
def _set_gisdb(self, gisdb):
self._gisdb = _check_raise(gisdb, "", "GISDBASE")
gisdbase = property(
fget=_get_gisdb, fset=_set_gisdb, doc="Set or obtain the name of GISDBASE"
)
def _get_loc(self):
return self._loc
def _set_loc(self, loc):
self._loc = _check_raise(loc, self._gisdb, "LOCATION_NAME")
location = property(
fget=_get_loc, fset=_set_loc, doc="Set or obtain the name of LOCATION"
)
def _get_name(self):
return self._name
def _set_name(self, name):
self._name = _check_raise(name, join(self._gisdb, self._loc), "MAPSET")
name = property(
fget=_get_name, fset=_set_name, doc="Set or obtain the name of MAPSET"
)
def __str__(self):
return self.name
def __repr__(self):
return "Mapset(%r)" % self.name
[docs] def glist(self, type, pattern=None):
"""Return a list of grass types like:
* 'group',
* 'label',
* 'raster',
* 'raster_3d',
* 'region',
* 'vector',
:param type: the type of element to query
:type type: str
:param pattern: the pattern to filter the result
:type pattern: str
::
>>> mapset = Mapset()
>>> mapset.current()
>>> rast = mapset.glist('raster')
>>> test_raster_name in rast
True
>>> vect = mapset.glist('vector')
>>> test_vector_name in vect
True
..
"""
if type not in ETYPE:
str_err = "Type %s is not valid, valid types are: %s."
raise TypeError(str_err % (type, ", ".join(ETYPE.keys())))
clist = libgis.G_list(ETYPE[type], self.gisdbase, self.location, self.name)
elist = []
for el in clist:
el_name = ct.cast(el, ct.c_char_p).value
if el_name:
elist.append(decode(el_name))
else:
if pattern:
return fnmatch.filter(elist, pattern)
return elist
[docs] def is_current(self):
"""Check if the MAPSET is the working MAPSET"""
return (
self.name == getenv("MAPSET")
and self.location == getenv("LOCATION_NAME")
and self.gisdbase == getenv("GISDBASE")
)
[docs] def current(self):
"""Set the mapset as current"""
set_current_mapset(self.name, self.location, self.gisdbase)
[docs] def delete(self):
"""Delete the mapset"""
if self.is_current():
raise GrassError("The mapset is in use.")
shutil.rmtree(self.path())
[docs] def path(self):
"""Return the complete path of the mapset"""
return join(self.gisdbase, self.location, self.name)
[docs]class VisibleMapset(object):
"""VisibleMapset object"""
def __init__(self, mapset, location="", gisdbase=""):
self.mapset = mapset
self.location = Location(location, gisdbase)
self._list = []
self.spath = join(self.location.path(), self.mapset, "SEARCH_PATH")
def __repr__(self):
return repr(self.read())
def __iter__(self):
for mapset in self.read():
yield mapset
[docs] def read(self):
"""Return the mapsets in the search path"""
with open(self.spath, "ab+") as f:
lines = f.readlines()
if lines:
return [decode(line.strip()) for line in lines]
lns = [
"PERMANENT",
]
self._write(lns)
return lns
def _write(self, mapsets):
"""Write to SEARCH_PATH file the changes in the search path
:param mapsets: a list of mapset's names
:type mapsets: list
"""
with open(self.spath, "wb+") as f:
ms = [decode(m) for m in self.location.mapsets()]
f.write(b"\n".join([encode(m) for m in mapsets if m in ms]))
[docs] def add(self, mapset):
"""Add a mapset to the search path
:param mapset: a mapset's name
:type mapset: str
"""
if mapset not in self.read() and mapset in self.location:
with open(self.spath, "a+") as f:
f.write("\n%s" % mapset)
else:
raise TypeError("Mapset not found")
[docs] def remove(self, mapset):
"""Remove mapset to the search path
:param mapset: a mapset's name
:type mapset: str
"""
mapsets = self.read()
mapsets.remove(mapset)
self._write(mapsets)
[docs] def extend(self, mapsets):
"""Add more mapsets to the search path
:param mapsets: a list of mapset's names
:type mapsets: list
"""
ms = [decode(m) for m in self.location.mapsets()]
final = [decode(m) for m in self.read()]
mapsets = [decode(m) for m in mapsets]
final.extend([m for m in mapsets if m in ms and m not in final])
self._write(final)
[docs] def reset(self):
"""Reset to the original search path"""
final = [self.mapset, "PERMANENT"]
self._write(final)
if __name__ == "__main__":
import doctest
from grass.pygrass import utils
from grass.script.core import run_command
utils.create_test_vector_map(test_vector_name)
run_command("g.region", n=50, s=0, e=60, w=0, res=1)
run_command("r.mapcalc", expression="%s = 1" % (test_raster_name), overwrite=True)
run_command("g.region", n=40, s=0, e=40, w=0, res=2)
doctest.testmod()
# Remove the generated vector map, if exist
mset = utils.get_mapset_vector(test_vector_name, mapset="")
if mset:
run_command("g.remove", flags="f", type="vector", name=test_vector_name)
mset = utils.get_mapset_raster(test_raster_name, mapset="")
if mset:
run_command("g.remove", flags="f", type="raster", name=test_raster_name)
