# !/usr/bin/python
"""Map production built on QGIS print layouts.
Replaces the former ArcGIS Pro implementation (arcpy.mp.ArcGISProject / .aprx projects,
.lyrx layer files, PDFDocumentCreate page stitching). Nothing here needs an Esri licence.
What changed conceptually
-------------------------
* Project file: .aprx -> .qgz (QgsProject, readable and editable in QGIS)
* Layer style: .lyrx -> .qml (QgsRasterLayer.loadNamedStyle)
* Page series: PDFDocumentCreate + appendPages -> QGIS Atlas
The old code exported one temporary PDF per map centre point and stitched them together.
QGIS drives a multi-page series natively: a coverage layer holds one rectangle per page and
QgsLayoutExporter.exportToPdf(atlas, ...) writes a single multi-page PDF in one call.
* Layout: a layout stored inside the proprietary template -> composed in code
The layout (map frame, legend, scale bar, north arrow, title, footer) is built
programmatically, so no binary template is required. An existing .qgz template is used
instead when one is present in 02_Maps/templates/.
The public API -- ``Mapper(condition, map_type, ...)``, ``prepare_layout()``,
``make_pdf_maps()`` and ``.error`` -- is unchanged from the ArcGIS implementation, so
existing callers keep working without modification.
Running headless requires QT_QPA_PLATFORM=offscreen, which this module sets itself.
"""
import glob
import logging
import os
import sys
from . import config
# QGIS must run offscreen when there is no display attached; set before importing qgis.
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
#: Where a QGIS installation keeps its Python bindings, per platform, as
#: ``(bindings glob, prefix template, DLL directory globs)``. A distribution installs the
#: bindings for the *system* interpreter, so a conda environment does not see them even
#: though QGIS is installed and working - which is the case this exists to handle. ``{b}``
#: in a prefix template is replaced by the matched bindings directory.
#:
#: Globs matter on Windows, where the standalone installer puts the version in the folder
#: name (``C:\\Program Files\\QGIS 3.34.5``), and on macOS, where the bundle may be
#: ``QGIS.app`` or ``QGIS-LTR.app``.
QGIS_LAYOUTS = {
"linux": (
# Debian, Ubuntu: python3-qgis. The bindings sit among the distribution's own
# numpy/pandas/scipy, which is why the directory is only ever appended.
("/usr/lib/python3/dist-packages", "/usr", ()),
("/usr/lib/python3*/dist-packages", "/usr", ()),
("/usr/lib/python3*/site-packages", "/usr", ()), # Fedora, Arch
("/usr/share/qgis/python", "/usr", ()),
("/usr/local/share/qgis/python", "/usr/local", ()),
# <prefix>/share/qgis/python -> up three to reach <prefix>
("/opt/qgis*/share/qgis/python", "{b}/../../..", ()),
),
"darwin": (
("/Applications/QGIS*.app/Contents/Resources/python",
"{b}/../../MacOS", ()),
(os.path.expanduser("~/Applications/QGIS*.app/Contents/Resources/python"),
"{b}/../../MacOS", ()),
("/opt/homebrew/opt/qgis/QGIS.app/Contents/Resources/python",
"{b}/../../MacOS", ()),
),
# Written with forward slashes on purpose: `glob` accepts them on Windows and
# `os.path.normpath` turns them into backslashes there, so the patterns stay readable
# and stay testable from any platform.
"win32": (
# OSGeo4W. The DLL directories are not optional on Windows: without them the
# extension modules import but their Qt, GDAL and PROJ libraries fail to load, with
# a bare "DLL load failed" that says nothing about QGIS.
("C:/OSGeo4W*/apps/qgis*/python", "{b}/..",
("{b}/../bin", "{b}/../../../bin")),
("C:/Program Files/QGIS */apps/qgis*/python", "{b}/..",
("{b}/../bin", "{b}/../../../bin")),
("C:/Program Files*/QGIS*/apps/qgis*/python", "{b}/..",
("{b}/../bin", "{b}/../../../bin")),
),
}
#: Directories that held a ``qgis`` package but could not be imported, for the message.
_qgis_rejected = []
#: The directory the bindings were found in, when they were not already importable.
QGIS_BINDINGS_PATH = None
#: The QGIS prefix that goes to ``QgsApplication.setPrefixPath``. ``None`` means the
#: environment's :envvar:`QGIS_PREFIX_PATH`, or the platform default, is used.
QGIS_PREFIX = None
def _qgis_platform():
"""``"win32"``, ``"darwin"`` or ``"linux"`` - the key into :data:`QGIS_LAYOUTS`."""
if sys.platform.startswith("win"):
return "win32"
if sys.platform == "darwin":
return "darwin"
return "linux"
[docs]
def qgis_candidates():
"""Yield ``(bindings, prefix, dll_dirs)`` triples to try, most specific first.
:envvar:`RIVERARCHITECT_QGIS_PATH` (a directory containing the ``qgis`` package) and
:envvar:`QGIS_PREFIX_PATH` come first, so a non-standard installation can be pointed at
explicitly and nothing needs guessing.
"""
explicit = os.environ.get("RIVERARCHITECT_QGIS_PATH")
if explicit:
yield explicit, os.environ.get("QGIS_PREFIX_PATH"), ()
prefix = os.environ.get("QGIS_PREFIX_PATH")
if prefix:
for relative in ("python", os.path.join("share", "qgis", "python"),
os.path.join("..", "Resources", "python")):
yield os.path.join(prefix, relative), prefix, (os.path.join(prefix, "bin"),)
for pattern, prefix_template, dll_patterns in QGIS_LAYOUTS[_qgis_platform()]:
# sorted(reverse=True) puts the highest version first when a glob matches several
# installations, which is what a user with 3.34 and 3.40 side by side expects.
for bindings in sorted(glob.glob(pattern), reverse=True):
resolved_prefix = os.path.normpath(prefix_template.format(b=bindings))
dll_dirs = tuple(os.path.normpath(p.format(b=bindings)) for p in dll_patterns)
yield bindings, resolved_prefix, dll_dirs
def _locate_qgis_bindings():
"""Make ``qgis`` importable, adding a discovered bindings directory if need be.
Returns:
str or None: the directory added to :data:`sys.path`, or ``None`` when nothing
usable was found.
The discovered directory is **appended** to :data:`sys.path`, never prepended, and
:envvar:`PYTHONPATH` is deliberately not used for it. On Debian and Ubuntu the QGIS
bindings sit in ``/usr/lib/python3/dist-packages`` beside the distribution's own numpy,
pandas and scipy; putting that directory *ahead* of a conda environment silently
replaces numpy 2.x with 1.26 and pandas 3.x with 2.1. That is a worse failure than no
mapping, because it produces results rather than an error. Appending leaves the
environment's own stack in charge and lets only the modules it does not have - chiefly
``qgis`` and ``PyQt5`` - come from the system.
"""
global QGIS_BINDINGS_PATH, QGIS_PREFIX
for bindings, prefix, dll_dirs in qgis_candidates():
bindings = os.path.normpath(bindings)
if not os.path.isdir(os.path.join(bindings, "qgis")):
continue
# Windows resolves a extension module's dependent DLLs through an explicit list
# since Python 3.8, not through PATH. Without this the import fails with a bare
# "DLL load failed" that says nothing about QGIS.
opened = []
for directory in dll_dirs:
if os.path.isdir(directory) and hasattr(os, "add_dll_directory"):
try:
opened.append(os.add_dll_directory(directory))
except OSError:
pass
sys.path.append(bindings)
try:
import qgis.core # noqa: F401
except Exception as exc: # ImportError, but also DLL and Qt loading failures
# Built for a different Python, or its libraries are unreachable. Leaving the
# path on would only let its other packages shadow ours for nothing.
sys.path.remove(bindings)
for handle in opened:
handle.close()
_qgis_rejected.append((bindings, str(exc)))
continue
QGIS_BINDINGS_PATH = bindings
QGIS_PREFIX = prefix
return bindings
return None
try:
import qgis.core # noqa: F401
except ImportError:
_locate_qgis_bindings()
try:
from qgis.core import (QgsApplication, QgsProject, QgsRasterLayer, QgsVectorLayer,
QgsPrintLayout, QgsLayoutItemMap, QgsLayoutItemLegend,
QgsLayoutItemScaleBar, QgsLayoutItemLabel, QgsLayoutItemPicture,
QgsLayoutPoint, QgsLayoutSize, QgsUnitTypes, QgsLayoutExporter,
QgsRectangle, QgsFeature, QgsGeometry, QgsLayerTreeGroup,
QgsCoordinateReferenceSystem)
from qgis.PyQt.QtGui import QFont
QGIS_AVAILABLE = True
if QGIS_BINDINGS_PATH:
logging.getLogger("riverarchitect").info(
"QGIS bindings found in %s and added to sys.path - mapping is enabled.",
QGIS_BINDINGS_PATH)
except ImportError:
QGIS_AVAILABLE = False
# A library must not print on import: importing riverarchitect.mapping to *check*
# QGIS_AVAILABLE is a normal thing to do, and the GUI does exactly that before it
# decides what to show. Log it instead, so callers choose whether it is visible.
if _qgis_rejected:
_detail = ("Found QGIS bindings but could not load them, most often because they "
"are built for a different Python than this one (%d.%d): %s"
% (sys.version_info[0], sys.version_info[1],
"; ".join("%s (%s)" % pair for pair in _qgis_rejected)))
else:
_install = {
"linux": "sudo apt install qgis python3-qgis (Debian, Ubuntu)",
"darwin": "install QGIS from qgis.org into /Applications",
"win32": "install QGIS with the OSGeo4W installer, or the standalone installer",
}[_qgis_platform()]
_detail = ("No QGIS bindings were found. %s. If QGIS is installed somewhere "
"unusual, set RIVERARCHITECT_QGIS_PATH to the directory that contains "
"the 'qgis' package, or QGIS_PREFIX_PATH to the installation prefix."
% _install)
logging.getLogger("riverarchitect").info(
"QGIS (qgis.core) is not available - mapping is disabled. %s", _detail)
__all__ = ["QGIS_AVAILABLE", "QGIS_BINDINGS_PATH", "QGIS_PREFIX", "QGIS_LAYOUTS",
"qgis_candidates", "qgis_status", "QgisSession", "Mapper"]
# Page geometry defaults, in millimetres. ANSI E landscape reproduces the former arcpy layout.
ANSI_E_LANDSCAPE = (1117.6, 863.6)
DEFAULT_DPI = 96
SINGLE_MAP_DPI = 192
[docs]
def qgis_status():
"""Whether mapping can run here, and what to do about it if it cannot.
One implementation for both front ends, so the Qt and the tkinter tab cannot give
different advice.
Returns:
tuple: ``(available, message)``.
"""
if QGIS_AVAILABLE:
lines = ["QGIS bindings found. Select a raster directory to begin.", ""]
try:
from qgis.core import Qgis
lines.append("QGIS version : %s" % Qgis.QGIS_VERSION)
except Exception:
pass
if QGIS_BINDINGS_PATH:
lines.append("bindings : %s" % QGIS_BINDINGS_PATH)
lines.append("prefix : %s" % (QGIS_PREFIX or "(default)"))
lines.append("")
lines.append("These were discovered outside this Python environment and added "
"to the end of")
lines.append("its module search path, so the environment's own numpy, pandas "
"and rasterio")
lines.append("keep priority.")
return True, "\n".join(lines)
lines = [
"QGIS is not available in this Python environment, so mapping is disabled.",
"Everything else in River Architect works normally.",
"",
]
if _qgis_rejected:
lines.append("QGIS bindings were found but could not be loaded. That almost always")
lines.append("means they are built for a different Python than this one (%d.%d):"
% (sys.version_info[0], sys.version_info[1]))
lines.append("")
for path, error in _qgis_rejected:
lines.append(" %s" % path)
lines.append(" %s" % error)
lines.append("")
lines.append("Start River Architect with the interpreter QGIS was installed for:")
lines.append("")
lines.append(" RA_PYTHON=/usr/bin/python3 ./runRiverArchitectLinux.sh")
else:
lines.append("No QGIS installation was found in any of the usual places. Install "
"QGIS:")
lines.append("")
lines.append(" %s" % {
"linux": "sudo apt install qgis python3-qgis (Debian, Ubuntu)",
"darwin": "download QGIS from qgis.org and drag it into /Applications",
"win32": "install QGIS with the OSGeo4W installer, or the standalone installer",
}[_qgis_platform()])
lines.append("")
lines.append("If it is installed somewhere unusual, point River Architect at it:")
lines.append("")
lines.append(" RIVERARCHITECT_QGIS_PATH=<directory containing the 'qgis' package>")
lines.append(" QGIS_PREFIX_PATH=<installation prefix>")
return False, "\n".join(lines)
def _default_prefix():
"""Installation prefix to fall back on when nothing else identified one.
Only reached when ``qgis`` was importable without any help from
:func:`_locate_qgis_bindings` - a QGIS-provided interpreter, or an OSGeo4W shell - where
QGIS usually knows its own prefix already and this is a last resort.
"""
if QGIS_BINDINGS_PATH:
return os.path.normpath(os.path.join(QGIS_BINDINGS_PATH, os.pardir))
try:
import qgis
# <prefix>/share/qgis/python/qgis or <prefix>/python/qgis, depending on the layout
root = os.path.dirname(os.path.dirname(os.path.abspath(qgis.__file__)))
for _ in range(3):
if os.path.isdir(os.path.join(root, "resources")):
return root
root = os.path.dirname(root)
except Exception:
pass
return "/usr"
def _p(*parts):
"""Join path fragments and normalise the mixed \\ and / separators used across config.py."""
joined = os.path.join(*[str(x) for x in parts if str(x)])
return os.path.normpath(joined.replace("\\", os.sep).replace("/", os.sep))
[docs]
class QgisSession(object):
"""Process-wide QgsApplication. QGIS may only be initialised once per process."""
_app = None
[docs]
@classmethod
def start(cls):
if not QGIS_AVAILABLE:
return None
if cls._app is None:
# The prefix that goes with the bindings actually loaded wins over the
# environment: on macOS and Windows "/usr" is meaningless, and QGIS then fails
# to find its providers rather than saying the prefix is wrong.
prefix = QGIS_PREFIX or os.environ.get("QGIS_PREFIX_PATH") or _default_prefix()
QgsApplication.setPrefixPath(prefix, True)
cls._app = QgsApplication([], False)
cls._app.initQgis()
return cls._app
[docs]
@classmethod
def stop(cls):
if cls._app is not None:
cls._app.exitQgis()
cls._app = None
[docs]
class Mapper:
def __init__(self, condition, map_type, *args):
# condition = [str] state of planning situation, e.g., "2008"
# map_type = [str] options: "lf", "ds", "mlf", "mt"
# args[0] alternative raster input directory - if empty: uses standard output
# args[1] alternative output directory - if empty: 02_Maps/CONDITION/
self.logger = logging.getLogger("riverarchitect")
self.condition = condition
self.dir_map_shp = ""
self.error = False
self.map_type = map_type
self.map_string = ""
self.map_list = []
self.ras4map_list = []
self.raster_extent = None # QgsRectangle
self.resolution = DEFAULT_DPI
self.xy_center_points = []
self.dx = None
self.dy = None
self.page_size = ANSI_E_LANDSCAPE
self.project = None
self.layout = None
self.map_item = None
self.legend = None
if not QGIS_AVAILABLE:
self.error = True
self.logger.info("ERROR: QGIS is not available - cannot produce maps.")
# args[0]: raster directory, args[1]: output directory. Both optional; the original
# expressed that with `try: args[0] / except:`, which also swallowed KeyboardInterrupt
# and hid genuine errors from makedirs.
requested = str(args[0]) if len(args) > 0 and str(args[0]).strip() else ""
try:
self.dir_map_ras = requested or str(self.get_input_ras_dir(map_type))
os.makedirs(self.dir_map_ras, exist_ok=True)
except OSError:
self.logger.info("WARNING: The provided path to rasters for mapping is invalid. "
"Using templates instead.")
self.dir_map_ras = _p(config.templates_dir(), "rasters")
os.makedirs(self.dir_map_ras, exist_ok=True)
if len(args) > 1 and str(args[1]).strip():
self.output_dir = str(args[1])
else:
self.output_dir = _p(config.dir_maps(), self.condition) + os.sep
os.makedirs(self.output_dir, exist_ok=True)
os.makedirs(_p(self.output_dir, "layers"), exist_ok=True)
if QGIS_AVAILABLE:
QgisSession.start()
self.project = self.open_project()
# ------------------------------------------------------------------ project
[docs]
def project_path(self):
return _p(self.output_dir, "maps_%s_design.qgz" % self.condition)
[docs]
def open_project(self):
"""Create (or reopen) the QGIS project that carries this condition's maps.
Replaces copy_template_project(), which duplicated river_template.aprx. If a QGIS
template exists at 02_Maps/templates/river_template.qgz it is used as the starting
point; otherwise an empty project is created and the layout is composed in code.
"""
project = QgsProject.instance()
project.clear()
target = self.project_path()
if os.path.isfile(target):
if project.read(target):
self.logger.info(" >> Using existing project: " + target)
return project
self.logger.info("WARNING: Could not read %s - starting a new project." % target)
template = _p(config.templates_dir(), "river_template.qgz")
if os.path.isfile(template):
if project.read(template):
self.logger.info(" >> Started from QGIS template: " + template)
else:
self.logger.info("WARNING: Could not read %s - starting an empty project." % template)
else:
self.logger.info(" >> No river_template.qgz found - composing layout in code.")
project.setFileName(target)
return project
[docs]
def save_project(self):
try:
self.project.write(self.project_path())
except Exception as e:
self.logger.info("WARNING: Could not save QGIS project (%s)." % e)
# ------------------------------------------------------------------ styles
[docs]
def choose_ref_layer(self, raster_type):
# returns the base name of the .qml style for a raster type (formerly a .lyrx layer)
try:
if 'lf' in str(raster_type):
return "lf_sym"
if 'ds' in str(raster_type):
return "ds_sym"
except ValueError:
return ""
return ""
[docs]
def choose_ref_layout(self, map_lyr_name=str()):
if "lf" in map_lyr_name:
if not ("mlf" in map_lyr_name):
return "layout_lf"
else:
if "bio" in map_lyr_name.lower():
return "layout_mlf_bioeng"
if "maint" in map_lyr_name.lower():
return "layout_mlf_maintenance"
if "plant" in map_lyr_name.lower():
return "layout_mlf_plants"
if "terra" in map_lyr_name.lower():
return "layout_mlf_terraforming"
if "ds" in map_lyr_name:
if "bio" in map_lyr_name.lower():
return "layout_ds_bio"
if ("dcf" in map_lyr_name.lower()) or ("dcr" in map_lyr_name.lower()) or ("dst" in map_lyr_name.lower()) or ("grav" in map_lyr_name.lower()) or ("rocks" in map_lyr_name.lower()):
return "layout_ds_Dcr"
if "fines" in map_lyr_name.lower():
if "15" in map_lyr_name.lower():
return "layout_ds_FS_D15"
if "85" in map_lyr_name.lower():
return "layout_ds_FS_D85"
if "sidech" in map_lyr_name.lower():
return "layout_ds_SeS0"
if "sideca" in map_lyr_name.lower():
return "layout_ds_sideca"
if "widen" in map_lyr_name.lower():
return "layout_ds_widen"
if ("wood" in map_lyr_name.lower()) or ("Dw" in map_lyr_name.lower()):
return "layout_ds_Dw"
if "volume_" in map_lyr_name:
return map_lyr_name
if "exc" in map_lyr_name:
return "volume_cust_neg"
if "fil" in map_lyr_name:
return "volume_cust_pos"
return "layout_lf"
[docs]
def choose_map_layer(self, map_name=str()):
if "lf" in map_name:
if not ("mlf" in map_name):
return "layer_lf"
else:
if "bio" in map_name.lower():
return "layer_mlf_bioeng"
if "maint" in map_name.lower():
return "layer_mlf_connectivity"
if "plant" in map_name.lower():
return "layer_mlf_plants"
if "terra" in map_name.lower():
return "layer_mlf_terraforming"
if "ds_" in map_name:
if "bio" in map_name.lower():
return "layer_ds_bio"
if ("dcf" in map_name.lower()) or ("dcr" in map_name.lower()) or ("dst" in map_name.lower()) or ("grav" in map_name.lower()) or ("grain" in map_name.lower()):
return "layer_ds_Dcr"
if "fines" in map_name.lower():
if "15" in map_name.lower():
return "layer_ds_FS_D15"
if "85" in map_name.lower():
return "layer_ds_FS_D85"
if "sidech" in map_name.lower():
return "layer_ds_SeS0"
if "sideca" in map_name.lower():
return "layer_ds_sideca"
if "widen" in map_name.lower():
return "layer_ds_widen"
if "wood" in map_name.lower():
return "layer_ds_Dw"
if "volume_" in map_name:
return map_name
if "exc" in map_name:
return "volume_cust_neg"
if "fil" in map_name:
return "volume_cust_pos"
return "layer_lf"
[docs]
def style_file(self, style_name):
"""Full path of a .qml style, or an empty string when it does not exist."""
for candidate in (style_name, style_name + ".qml"):
path = _p(config.symbology_dir(), candidate)
if os.path.isfile(path):
return path
return ""
[docs]
def apply_style(self, layer, style_name):
qml = self.style_file(style_name)
if not qml:
self.logger.info(" * no .qml style '%s' found - using QGIS defaults." % style_name)
return False
try:
message, ok = layer.loadNamedStyle(qml)
if ok:
layer.triggerRepaint()
self.logger.info(" * applied style: " + os.path.basename(qml))
return True
self.logger.info("WARNING: Could not apply %s (%s)." % (qml, message))
except Exception as e:
self.logger.info("WARNING: Could not apply %s (%s)." % (qml, e))
return False
# ------------------------------------------------------------------ layers
[docs]
def get_raster_extent(self, path2raster):
# path2raster = STR of full raster directory (e.g., D:/temp/raster.tif)
layer = QgsRasterLayer(_p(path2raster), os.path.basename(str(path2raster)))
if not layer.isValid():
self.logger.info("ERROR: Invalid raster: " + str(path2raster))
return None
return layer.extent()
[docs]
def add_background_layers(self):
"""Add any background layers shipped with the template directory."""
added = []
bg_dir = _p(config.templates_dir(), "rasters")
if not os.path.isdir(bg_dir):
return added
for name in sorted(os.listdir(bg_dir)):
if not name.lower().endswith((".tif", ".tiff")):
continue
if "background" not in name.lower():
continue
layer = QgsRasterLayer(_p(bg_dir, name), os.path.splitext(name)[0])
if layer.isValid():
self.project.addMapLayer(layer)
added.append(layer)
return added
[docs]
def add_map_raster(self, raster_path, layer_name, style_name):
layer = QgsRasterLayer(_p(raster_path), layer_name)
if not layer.isValid():
self.logger.info("ERROR: Could not load raster " + str(raster_path))
self.error = True
return None
self.apply_style(layer, style_name)
self.project.addMapLayer(layer)
if not self.project.crs().isValid() and layer.crs().isValid():
self.project.setCrs(layer.crs())
self.raster_extent = layer.extent()
return layer
[docs]
def add_shapefiles(self):
"""Add MaxLifespan shapefiles (map_type 'mlf') alongside the rasters."""
added = []
if not self.dir_map_shp or not os.path.isdir(self.dir_map_shp):
return added
for name in sorted(os.listdir(self.dir_map_shp)):
if not name.lower().endswith(".shp"):
continue
layer = QgsVectorLayer(_p(self.dir_map_shp, name), os.path.splitext(name)[0], "ogr")
if layer.isValid():
self.apply_style(layer, os.path.splitext(name)[0])
self.project.addMapLayer(layer)
added.append(layer)
else:
self.logger.info("WARNING: Invalid shapefile: " + name)
return added
# ------------------------------------------------------------------ layout
[docs]
def build_layout(self, layout_name, title_text):
"""Compose a print layout: map frame, legend, scale bar, north arrow, title, footer.
Replaces the layouts that used to live inside river_template.aprx.
"""
manager = self.project.layoutManager()
existing = manager.layoutByName(layout_name)
if existing is not None:
manager.removeLayout(existing)
layout = QgsPrintLayout(self.project)
layout.initializeDefaults()
layout.setName(layout_name)
width, height = self.page_size
page = layout.pageCollection().page(0)
page.setPageSize(QgsLayoutSize(width, height, QgsUnitTypes.LayoutMillimeters))
margin = 0.02 * width
legend_width = 0.16 * width
header = 0.05 * height
footer = 0.04 * height
# --- map frame ---
map_item = QgsLayoutItemMap(layout)
map_item.setId("river_map")
layout.addLayoutItem(map_item)
map_item.attemptMove(QgsLayoutPoint(margin, header, QgsUnitTypes.LayoutMillimeters))
map_item.attemptResize(QgsLayoutSize(width - 3 * margin - legend_width,
height - header - footer,
QgsUnitTypes.LayoutMillimeters))
map_item.setFrameEnabled(True)
if self.raster_extent is not None:
map_item.zoomToExtent(self.raster_extent)
# --- title ---
title = QgsLayoutItemLabel(layout)
title.setText(title_text)
font = QFont()
font.setPointSize(22)
font.setBold(True)
try:
title.setFont(font)
except (AttributeError, TypeError):
pass
title.adjustSizeToText()
layout.addLayoutItem(title)
title.attemptMove(QgsLayoutPoint(margin, 0.3 * header, QgsUnitTypes.LayoutMillimeters))
# --- legend ---
legend = QgsLayoutItemLegend(layout)
legend.setTitle("Legend")
legend.setLinkedMap(map_item)
legend.setLegendFilterByMapEnabled(True)
layout.addLayoutItem(legend)
legend.attemptMove(QgsLayoutPoint(width - margin - legend_width, header,
QgsUnitTypes.LayoutMillimeters))
legend.attemptResize(QgsLayoutSize(legend_width, 0.5 * height,
QgsUnitTypes.LayoutMillimeters))
legend.setFrameEnabled(True)
self.legend = legend
# --- scale bar ---
scalebar = QgsLayoutItemScaleBar(layout)
scalebar.setStyle("Single Box")
scalebar.setLinkedMap(map_item)
scalebar.applyDefaultSize()
layout.addLayoutItem(scalebar)
scalebar.attemptMove(QgsLayoutPoint(margin, height - footer + 0.005 * height,
QgsUnitTypes.LayoutMillimeters))
# --- north arrow (ships with QGIS) ---
arrow_svg = _p(QgsApplication.pkgDataPath(), "svg", "arrows", "NorthArrow_02.svg")
if os.path.isfile(arrow_svg):
arrow = QgsLayoutItemPicture(layout)
arrow.setPicturePath(arrow_svg)
layout.addLayoutItem(arrow)
arrow.attemptResize(QgsLayoutSize(0.03 * width, 0.03 * width,
QgsUnitTypes.LayoutMillimeters))
arrow.attemptMove(QgsLayoutPoint(width - margin - legend_width - 0.05 * width, header,
QgsUnitTypes.LayoutMillimeters))
# --- footer ---
footer_label = QgsLayoutItemLabel(layout)
footer_label.setText("River Architect | condition: %s | page [%% @atlas_featurenumber %%] "
"of [%% @atlas_totalfeatures %%]" % self.condition)
footer_label.adjustSizeToText()
layout.addLayoutItem(footer_label)
footer_label.attemptMove(QgsLayoutPoint(margin, height - 0.6 * footer,
QgsUnitTypes.LayoutMillimeters))
manager.addLayout(layout)
self.layout = layout
self.map_item = map_item
return layout
# ------------------------------------------------------------------ atlas
[docs]
def make_xy_centerpoints(self, new_extents):
# shape(new_extents) = [XMin, YMin, XMax, YMax]
self.xy_center_points = []
self.xy_center_points.append(
[float((new_extents[0] + new_extents[2]) / 2.), float((new_extents[1] + new_extents[3]) / 2.)])
self.dx = float(new_extents[2] - new_extents[0])
self.dy = float(new_extents[3] - new_extents[1])
self.logger.info(" - New X center: " + str(self.xy_center_points[0][0]))
self.logger.info(" - New Y center: " + str(self.xy_center_points[0][1]))
self.logger.info(" - New map width: " + str(self.dx))
self.logger.info(" - New map height: " + str(self.dy))
self.resolution = SINGLE_MAP_DPI # improve resolution of single maps
[docs]
def coverage_layer(self):
"""One rectangle per map page, driving the QGIS atlas.
This is the structural replacement for the old loop that zoomed the map frame to each
centre point, exported a temporary PDF and appended it to a growing document.
"""
crs = self.project.crs()
uri = "Polygon?crs=%s&field=page:integer&field=label:string" % (
crs.authid() if crs.isValid() else "EPSG:4326")
layer = QgsVectorLayer(uri, "map_pages", "memory")
if not layer.isValid():
self.logger.info("ERROR: Could not create the atlas coverage layer.")
return None
centers = self.xy_center_points
if not centers and self.raster_extent is not None:
centers = [[self.raster_extent.center().x(), self.raster_extent.center().y()]]
self.dx = self.raster_extent.width()
self.dy = self.raster_extent.height()
if self.dx is None or self.dy is None or self.dx <= 0 or self.dy <= 0:
if self.raster_extent is None:
self.logger.info("ERROR: No extent available for the map series.")
return None
self.dx = self.raster_extent.width()
self.dy = self.raster_extent.height()
features = []
for i, xy in enumerate(centers):
try:
x, y = float(xy[0]), float(xy[1])
except (TypeError, ValueError, IndexError):
self.logger.info("ERROR: Invalid x-y coordinates in extent source "
"[mapping.inp / reaches] - skipping page %d." % (i + 1))
continue
rect = QgsRectangle(x - 0.5 * self.dx, y - 0.5 * self.dy,
x + 0.5 * self.dx, y + 0.5 * self.dy)
feature = QgsFeature(layer.fields())
feature.setGeometry(QgsGeometry.fromRect(rect))
feature.setAttribute("page", i + 1)
feature.setAttribute("label", "fig_%02d" % (i + 1))
features.append(feature)
if not features:
self.logger.info("ERROR: No valid map pages could be defined.")
return None
layer.dataProvider().addFeatures(features)
layer.updateExtents()
self.project.addMapLayer(layer, False) # keep it out of the legend
self.logger.info(" * map series: %d page(s)" % len(features))
return layer
# ------------------------------------------------------------------ export
[docs]
def make_pdf_maps(self, map_name, *args, **kwargs):
# map_name = STR of pdf name
# args[0] = STR of alternative output directory for PDFs
# optional kwarg "extent": overwrite mapping extent ("raster", "MAXOF" or [xmin, ymin, xmax, ymax])
# optional kwarg "map_layout": alternative QgsPrintLayout object
if not QGIS_AVAILABLE:
self.error = True
return
if args and len(str(args[0])) > 3:
self.output_dir = str(args[0])
os.makedirs(self.output_dir, exist_ok=True)
self.logger.info(" >> Alternative output directory provided: "
+ str(self.output_dir))
try:
for k in kwargs.items():
if "extent" in str(k[0]).lower():
if k[1] == "raster":
self.logger.info(" >> Using Raster coordinates for mapping.")
if self.raster_extent is not None:
self.make_xy_centerpoints([self.raster_extent.xMinimum(),
self.raster_extent.yMinimum(),
self.raster_extent.xMaximum(),
self.raster_extent.yMaximum()])
else:
if not (k[1] == "MAXOF"):
self.make_xy_centerpoints(k[1])
self.logger.info(" >> Special reach extents provided.")
self.logger.info(" --> Overwriting mapping.inp center point definitions.")
if "map_layout" in k[0]:
self.logger.info(" >> External map layout provided - using: " + str(k[1]))
self.layout = k[1]
except (AttributeError, TypeError, ValueError) as exc:
self.logger.info("WARNING: Could not apply a keyword argument (%s)." % exc)
if self.layout is None:
self.logger.info("ERROR: No map layout prepared - call prepare_layout() first.")
self.error = True
return
self.logger.info(" >> Starting PDF creation for: " + str(self.layout.name()))
page = self.layout.pageCollection().page(0)
self.logger.info(" * Map format: %0.1f mm x %0.1f mm"
% (page.pageSize().width(), page.pageSize().height()))
map_name = os.path.basename(str(map_name).replace("\\", "/"))
pdf_name = _p(self.output_dir, map_name.split(".pdf")[0] + ".pdf")
try:
if os.path.isfile(pdf_name):
os.remove(pdf_name)
except OSError:
pass
coverage = self.coverage_layer()
if coverage is None:
self.error = True
return
try:
atlas = self.layout.atlas()
atlas.setEnabled(True)
atlas.setCoverageLayer(coverage)
atlas.setFilenameExpression("\"label\"")
self.map_item.setAtlasDriven(True)
self.map_item.setAtlasScalingMode(QgsLayoutItemMap.Auto)
atlas.updateFeatures()
settings = QgsLayoutExporter.PdfExportSettings()
settings.dpi = self.resolution
settings.rasterizeWholeImage = False
settings.forceVectorOutput = False
self.logger.info(" * Creating PDF %s (%d page(s), %d dpi) ..."
% (pdf_name, atlas.count(), self.resolution))
result, message = QgsLayoutExporter.exportToPdf(atlas, pdf_name, settings)
if result == QgsLayoutExporter.Success:
self.logger.info(" * Finished map: " + pdf_name)
else:
self.error = True
self.logger.info("ERROR: PDF export failed (%s)." % (message or result))
except Exception as e:
self.logger.info("ERROR: Mapping failed (%s)." % e)
self.error = True
finally:
try:
self.project.removeMapLayer(coverage.id())
except (AttributeError, RuntimeError):
pass
self.save_project()
# ------------------------------------------------------------------ driver
[docs]
def prepare_layout(self, *args, **kwargs):
# prepares map layouts in the QGIS project and starts mapping if not args[0]
# kwargs: - map_items=LIST of items for mapping (default: all rasters in dir_map_ras folder)
if not QGIS_AVAILABLE:
self.error = True
return
direct_mapping = args[0] if args else True
self.ras4map_list = [os.path.basename(f) for f in
sorted(glob.glob(os.path.join(self.dir_map_ras, "*.tif")))]
try:
for k in kwargs.items():
if "map_items" in str(k[0]).lower():
self.map_list = list(k[1])
except TypeError as exc:
self.logger.info("WARNING: map_items is not a sequence (%s) - ignoring." % exc)
if self.map_list.__len__() < 1:
self.map_list = self.ras4map_list
self.logger.info("\n\nMAPPING")
self.logger.info("----- ----- ----- ----- ----- ----- ----- ----- -----")
for map_item in self.map_list:
map_name = os.path.basename(str(map_item).replace("\\", "/"))
map_name = str(map_name).lower().split(".tif")[0].split("_mlf")[0]
self.logger.info(" >> Preparing map: " + _p(self.output_dir, map_name + ".pdf"))
self.map_string = self.choose_map_layer(str(map_name).lower().split(".tif")[0])
if self.map_string.__len__() < 1:
self.logger.info("ERROR: No reference map found for %s (skipping)." % str(map_item))
continue
self.logger.info(" * source map layer: " + self.map_string)
layout_name = self.choose_ref_layout(self.map_string)
if layout_name.__len__() < 1:
self.logger.info("ERROR: No reference layout found for %s (skipping)." % str(self.map_string))
continue
self.logger.info(" * source map layout: " + layout_name)
try:
# start each map from a clean layer stack so that the legend only shows
# what belongs on this sheet (the old code toggled lyr.visible instead)
self.project.removeAllMapLayers()
self.raster_extent = None
self.add_background_layers()
raster_path = map_item if os.path.isabs(str(map_item)) \
else _p(self.dir_map_ras, str(map_item))
if not str(raster_path).lower().endswith((".tif", ".tiff")):
raster_path = raster_path + ".tif"
if self.map_type == "mlf":
self.add_shapefiles()
style = "LifespanRasterSymbology"
else:
style = self.choose_ref_layer(map_name)
self.logger.info(" * creating new layer ...")
layer = self.add_map_raster(raster_path, map_name, style)
if layer is None:
continue
# keep a QGIS layer definition next to the PDF, replacing the old .lyrx export
try:
layer.saveNamedStyle(_p(self.output_dir, "layers", map_name + ".qml"))
except Exception as e:
self.logger.info("WARNING: Could not save layer style (%s)." % e)
self.build_layout(layout_name, "%s - %s" % (self.condition, map_name))
self.save_project()
if direct_mapping:
self.logger.info(" * Calling PDF map creator ...")
self.make_pdf_maps(map_name, extent="raster")
except Exception as e:
self.logger.info("ERROR: Map layout preparation failed (%s)." % e)
self.error = True
[docs]
def zoom2map(self, xy):
"""Kept for API compatibility. Page extents are now driven by the atlas coverage layer."""
try:
x, y = float(xy[0]), float(xy[1])
except (TypeError, ValueError, IndexError):
self.logger.info("ERROR: Mapping could not assign xy-values. Undefined zoom.")
return
if self.map_item is None or self.dx is None or self.dy is None:
return
self.map_item.zoomToExtent(QgsRectangle(x - 0.5 * self.dx, y - 0.5 * self.dy,
x + 0.5 * self.dx, y + 0.5 * self.dy))
def __call__(self):
pass