grass.temporal: support fully qualified names on read

python/grass/temporal/core.py

@@ -640,6 +640,7 @@ def init(
     # Set defaults for the TGIS DB (overwritten later if initialized)
     tgis_backend = "sqlite"
     tgis_database_string = None
+    tgis_dbmi_paramstyle = "qmark"
 
     # Set the global variable for faster access
     current_mapset = new_mapset
@@ -1169,6 +1170,31 @@ def close(self) -> None:
 
         self.connected = False
 
+    def add_mapset(self, mapset):
+        """Ensure the temporal database of the given mapset is part of this
+        connection, so datasets can be accessed by fully qualified name
+        (name@mapset) even when the mapset is not on the current search path.
+
+        The method is idempotent: mapsets already known to the connection are
+        left untouched. If the mapset has no accessible temporal database, it
+        is not added and a later query against it fails with the usual error.
+        """
+        mapset = decode(mapset)
+        if mapset in self.tgis_mapsets:
+            return
+        new_mapsets = get_available_temporal_mapsets(mapset)
+        if mapset not in new_mapsets:
+            return
+        driver, dbstring = new_mapsets[mapset]
+        self.tgis_mapsets[mapset] = (driver, dbstring)
+        if dbstring not in self.unique_connections:
+            self.unique_connections[dbstring] = DBConnection(
+                backend=driver, dbstring=dbstring
+            )
+        self.connections[mapset] = self.unique_connections[dbstring]
+        if self.connected and not self.connections[mapset].is_connected():
+            self.connections[mapset].connect(dbstring)
+
     def mogrify_sql_statement(self, content, mapset=None):
         """Return the SQL statement and arguments as executable SQL string
 

python/grass/temporal/open_stds.py

@@ -1,5 +1,4 @@
-"""
-Functions to open or create space time datasets
+"""Functions to open or create space time datasets
 
 Usage:
 
@@ -9,20 +8,55 @@
 
     tgis.register_maps_in_space_time_dataset(type, name, maps)
 
-
-SPDX-FileCopyrightText: 2012-2014 GRASS Development Team
+SPDX-FileCopyrightText: 2012-2026 GRASS Development Team
 SPDX-License-Identifier: GPL-2.0-or-later
 
 :authors: Soeren Gebbert
 """
 
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .core import SQLDatabaseInterfaceConnection
+    from .abstract_space_time_dataset import AbstractSpaceTimeDataset
+import contextlib
+
+from grass.exceptions import FatalError
+
 from .abstract_map_dataset import AbstractMapDataset
 from .core import get_current_mapset, get_tgis_message_interface, init_dbif
 from .factory import dataset_factory
 
 ###############################################################################
 
 
+def _parse_id(
+    ident: str, stds_type: str = "strds"
+) -> tuple[str, str | None, str | None]:
+    """Parse parts of a user given dataset name.
+
+    :param ident: The id of a space time dataset
+
+    :return: Tuple with ID components: name, mapset, semantic_label
+    """
+    msgr = get_tgis_message_interface()
+    name = ident
+
+    mapset = None
+    if "@" in ident:
+        name, mapset = ident.split("@", 1)
+
+    semantic_label = None
+    if "." in name and stds_type == "strds":
+        name, semantic_label = name.split(".", 1)
+        if "." in semantic_label:
+            msgr.fatal(_("Invalid semantic_label <%s>") % semantic_label)
+
+    return name, mapset, semantic_label
+
+
 def open_old_stds(name, type, dbif=None):
     """This function opens an existing space time dataset and return the
     created and initialized object of the specified type.
@@ -32,8 +66,10 @@ def open_old_stds(name, type, dbif=None):
     or the space time dataset was not found.
 
     :param name: The name of the space time dataset, if the name does not
-                 contain the mapset (name@mapset) then the current mapset
-                 will be used to identify the space time dataset
+                 contain the mapset (name@mapset) then the space time dataset
+                 is searched first in the current mapset, then on the search
+                 path. The mapset of the first match will be used to identify
+                 the space time dataset.
     :param type: The type of the space time dataset (strd, str3ds, stvds,
                  raster, vector, raster3d)
     :param dbif: The optional database interface to be used
@@ -42,38 +78,78 @@ def open_old_stds(name, type, dbif=None):
 
     """
     msgr = get_tgis_message_interface()
+    stds_type = type
+    if stds_type not in {
+        "strds",
+        "str3ds",
+        "stvds",
+        "raster",
+        "rast",
+        "raster3d",
+        "rast3d",
+        "raster_3d",
+        "vector",
+        "vect",
+    }:
+        msgr.fatal(_("Unknown type: %s") % (stds_type))
+
+    def try_get_stds(
+        ds_id: str,
+        stds_type: str,
+        semantic_label: str | None,
+        dbif: SQLDatabaseInterfaceConnection,
+    ) -> AbstractSpaceTimeDataset | None:
+        if stds_type in {"str3ds", "raster3d", "rast3d", "raster_3d"}:
+            sp = dataset_factory("str3ds", ds_id)
+        elif stds_type in {"stvds", "vect", "vector"}:
+            sp = dataset_factory("stvds", ds_id)
+        else:
+            sp = dataset_factory("strds", ds_id)
+            # Set the semantic label if it was given
+            if semantic_label:
+                sp.set_semantic_label(semantic_label)
+
+        with contextlib.suppress(SystemExit, FatalError):
+            if sp.is_in_db(dbif):
+                return sp
+        return None
 
     # Check if the dataset name contains the mapset and the semantic label as well
-    if name.find("@") < 0:
-        mapset = get_current_mapset()
-    else:
-        name, mapset = name.split("@")
-    semantic_label = None
-    if name.find(".") > -1:
-        try:
-            name, semantic_label = name.split(".")
-        except ValueError:
-            msgr.fatal("Invalid name of the space time dataset. Only one dot allowed.")
-    id = name + "@" + mapset
-
-    if type in {"strds", "rast", "raster"}:
-        sp = dataset_factory("strds", id)
-        if semantic_label:
-            sp.set_semantic_label(semantic_label)
-    elif type in {"str3ds", "raster3d", "rast3d", "raster_3d"}:
-        sp = dataset_factory("str3ds", id)
-    elif type in {"stvds", "vect", "vector"}:
-        sp = dataset_factory("stvds", id)
-    else:
-        msgr.fatal(_("Unknown type: %s") % (type))
+    name, mapset, semantic_label = _parse_id(name, stds_type=stds_type)
 
     dbif, connection_state_changed = init_dbif(dbif)
 
-    if not sp.is_in_db(dbif):
-        dbif.close()
+    # Check user given ID
+    sp = None
+    if mapset:
+        dbif.add_mapset(mapset)
+        sp = try_get_stds(f"{name}@{mapset}", stds_type, semantic_label, dbif)
+    else:
+        # Check current mapset first
+        sp = try_get_stds(
+            f"{name}@{get_current_mapset()}",
+            stds_type,
+            semantic_label,
+            dbif,
+        )
+        if not sp:
+            for tgis_mapset in dbif.tgis_mapsets:
+                if tgis_mapset == get_current_mapset():
+                    continue
+                sp = try_get_stds(
+                    f"{name}@{tgis_mapset}",
+                    stds_type,
+                    semantic_label,
+                    dbif,
+                )
+                if sp:
+                    break
+    if not sp:
+        if connection_state_changed:
+            dbif.close()
         msgr.fatal(
-            _("Space time %(sp)s dataset <%(id)s> not found")
-            % {"sp": sp.get_new_map_instance(None).get_type(), "id": id}
+            _("Space time dataset <%(name)s> of type <%(sp)s> not found")
+            % {"name": name, "sp": stds_type},
         )
     # Read content from temporal database
     sp.select(dbif)
@@ -100,35 +176,29 @@ def check_new_stds(name, type, dbif=None, overwrite: bool = False):
 
     This function will raise a FatalError in case of an error.
     """
-
+    stds_type = type
     # Get the current mapset to create the id of the space time dataset
 
-    mapset = get_current_mapset()
     msgr = get_tgis_message_interface()
 
-    if name.find("@") < 0:
-        id = name + "@" + mapset
-    else:
-        n, m = name.split("@")
-        if mapset != m:
+    name, mapset, _semantic_label = _parse_id(name, stds_type=stds_type)
+    if mapset:
+        if mapset != get_current_mapset():
             msgr.fatal(
-                _("Space time datasets can only be created in the current mapset")
+                _("Space time datasets can only be created in the current mapset"),
             )
-        id = name
+        id = f"{name}@{mapset}"
+    else:
+        id = f"{name}@{get_current_mapset()}"
 
-    if type in {"strds", "rast", "raster"}:
-        if name.find(".") > -1:
-            # a dot is used as a separator for semantic label filtering
-            msgr.fatal(
-                _("Illegal dataset name <{}>. Character '.' not allowed.").format(name)
-            )
+    if stds_type in {"strds", "rast", "raster"}:
         sp = dataset_factory("strds", id)
-    elif type in {"str3ds", "raster3d", "rast3d ", "raster_3d"}:
+    elif stds_type in {"str3ds", "raster3d", "rast3d ", "raster_3d"}:
         sp = dataset_factory("str3ds", id)
-    elif type in {"stvds", "vect", "vector"}:
+    elif stds_type in {"stvds", "vect", "vector"}:
         sp = dataset_factory("stvds", id)
     else:
-        msgr.error(_("Unknown type: %s") % (type))
+        msgr.error(_("Unknown type: %s") % (stds_type))
         return None
 
     dbif, connection_state_changed = init_dbif(dbif)
@@ -137,9 +207,9 @@ def check_new_stds(name, type, dbif=None, overwrite: bool = False):
         msgr.fatal(
             _(
                 "Space time %(sp)s dataset <%(name)s> is already in the"
-                " database. Use the overwrite flag."
+                " database. Use the overwrite flag.",
             )
-            % {"sp": sp.get_new_map_instance(None).get_type(), "name": name}
+            % {"sp": sp.get_new_map_instance(None).get_type(), "name": name},
         )
     if connection_state_changed:
         dbif.close()
@@ -151,7 +221,14 @@ def check_new_stds(name, type, dbif=None, overwrite: bool = False):
 
 
 def open_new_stds(
-    name, type, temporaltype, title, descr, semantic, dbif=None, overwrite: bool = False
+    name,
+    type,
+    temporaltype,
+    title,
+    descr,
+    semantic,
+    dbif=None,
+    overwrite: bool = False,
 ):
     """Create a new space time dataset of a specific type
 
@@ -177,17 +254,17 @@ def open_new_stds(
         msgr.warning(
             _(
                 "Overwriting space time %(sp)s dataset <%(name)s> and "
-                "unregistering all maps"
+                "unregistering all maps",
             )
-            % {"sp": sp.get_new_map_instance(None).get_type(), "name": name}
+            % {"sp": sp.get_new_map_instance(None).get_type(), "name": name},
         )
         id = sp.get_id()
         sp.delete(dbif)
         sp = sp.get_new_instance(id)
 
     msgr.verbose(
         _("Creating a new space time %s dataset")
-        % sp.get_new_map_instance(None).get_type()
+        % sp.get_new_map_instance(None).get_type(),
     )
 
     sp.set_initial_values(
@@ -209,7 +286,11 @@ def open_new_stds(
 
 
 def check_new_map_dataset(
-    name, layer=None, type="raster", overwrite: bool = False, dbif=None
+    name,
+    layer=None,
+    type="raster",
+    overwrite: bool = False,
+    dbif=None,
 ):
     """Check if a new map dataset of a specific type can be created in
      the temporal database
@@ -232,17 +313,16 @@ def check_new_map_dataset(
 
     new_map = dataset_factory(type, map_id)
     # Check if new map is in the temporal database
-    if new_map.is_in_db(dbif):
-        if not overwrite:
-            if connection_state_changed:
-                dbif.close()
-            msgr.fatal(
-                _(
-                    "Map <%s> is already in temporal database,"
-                    " use overwrite flag to overwrite"
-                )
-                % (map_id)
+    if new_map.is_in_db(dbif) and not overwrite:
+        if connection_state_changed:
+            dbif.close()
+        msgr.fatal(
+            _(
+                "Map <%s> is already in temporal database,"
+                " use overwrite flag to overwrite",
             )
+            % (map_id),
+        )
 
     if connection_state_changed:
         dbif.close()
@@ -273,7 +353,6 @@ def open_new_map_dataset(
     :return: A map dataset object
 
     """
-
     dbif, connection_state_changed = init_dbif(dbif)
     new_map = check_new_map_dataset(name, layer, type, overwrite, dbif)