kopia lustrzana https://github.com/OpenDroneMap/WebODM
98 wiersze
3.1 KiB
Python
98 wiersze
3.1 KiB
Python
import logging
|
|
import shutil
|
|
import tempfile
|
|
import subprocess
|
|
import os
|
|
from string import Template
|
|
|
|
logger = logging.getLogger('app.logger')
|
|
|
|
class GrassEngine:
|
|
def __init__(self):
|
|
self.grass_binary = shutil.which('grass7') or \
|
|
shutil.which('grass72') or \
|
|
shutil.which('grass74')
|
|
|
|
if self.grass_binary is None:
|
|
logger.warning("Could not find a GRASS 7 executable. GRASS scripts will not work.")
|
|
else:
|
|
logger.info("Initializing GRASS engine using {}".format(self.grass_binary))
|
|
|
|
def create_context(self):
|
|
if self.grass_binary is None: raise GrassEngineException("GRASS engine is unavailable")
|
|
return GrassContext(self.grass_binary)
|
|
|
|
|
|
|
|
class GrassContext:
|
|
def __init__(self, grass_binary):
|
|
self.grass_binary = grass_binary
|
|
self.cwd = tempfile.mkdtemp('_webodm_grass')
|
|
self.template_args = {}
|
|
self.location = None
|
|
|
|
def add_file(self, filename, source, use_as_location=False):
|
|
param = os.path.splitext(filename)[0] # filename without extension
|
|
|
|
dst_path = os.path.abspath(os.path.join(self.cwd, filename))
|
|
with open(dst_path) as f:
|
|
f.write(source)
|
|
self.template_args[param] = dst_path
|
|
|
|
if use_as_location:
|
|
self.set_location(self.template_args[param])
|
|
|
|
return dst_path
|
|
|
|
def add_param(self, param, value):
|
|
self.template_args[param] = value
|
|
|
|
def set_location(self, location):
|
|
"""
|
|
:param location: either a "epsg:XXXXX" string or a path to a geospatial file defining the location
|
|
"""
|
|
if not location.startsWith('epsg:'):
|
|
location = os.path.abspath(location)
|
|
self.location = location
|
|
|
|
def execute(self, script):
|
|
"""
|
|
:param script: path to .grass script
|
|
:return: script output
|
|
"""
|
|
if self.location is None: raise GrassEngineException("Location is not set")
|
|
|
|
script = os.path.abspath(script)
|
|
|
|
# Create grass script via template substitution
|
|
try:
|
|
with open(script) as f:
|
|
script_content = f.read()
|
|
except FileNotFoundError:
|
|
raise GrassEngineException("Script does not exist: {}".format(script))
|
|
|
|
tmpl = Template(script_content)
|
|
|
|
# Write script to disk
|
|
with open(os.path.join(self.cwd, 'script.sh')) as f:
|
|
f.write(tmpl.substitute(self.template_args))
|
|
|
|
# Execute it
|
|
p = subprocess.Popen([self.grass_binary, '-c', self.location, 'location', '--exec', 'sh', 'script.sh'],
|
|
cwd=self.cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
out, err = p.communicate()
|
|
|
|
if p.returncode == 0:
|
|
return out
|
|
else:
|
|
raise GrassEngineException("Could not execute GRASS script {} from {}: {}".format(script, self.cwd, err))
|
|
|
|
def __del__(self):
|
|
# Cleanup
|
|
if os.path.exists(self.cwd):
|
|
shutil.rmtree(self.cwd)
|
|
|
|
class GrassEngineException(Exception):
|
|
pass
|
|
|
|
grass = GrassEngine() |