"""
Functions to create space time dataset lists
Usage:
.. code-block:: python
import grass.temporal as tgis
tgis.register_maps_in_space_time_dataset(type, name, maps)
(C) 2012-2022 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 GIS
for details.
:authors: Soeren Gebbert
:authors: Vaclav Petras
"""
import os
from contextlib import contextmanager
import sys
import grass.script as gs
from .core import get_tgis_message_interface, get_available_temporal_mapsets, init_dbif
from .datetime_math import time_delta_to_relative_time
from .factory import dataset_factory
from .open_stds import open_old_stds
###############################################################################
[docs]def get_dataset_list(
type, temporal_type, columns=None, where=None, order=None, dbif=None
):
"""Return a list of time stamped maps or space time datasets of a specific
temporal type that are registered in the temporal database
This method returns a dictionary, the keys are the available mapsets,
the values are the rows from the SQL database query.
:param type: The type of the datasets (strds, str3ds, stvds, raster,
raster_3d, vector)
:param temporal_type: The temporal type of the datasets (absolute,
relative)
:param columns: A comma separated list of columns that will be selected
:param where: A where statement for selected listing without "WHERE"
:param order: A comma separated list of columns to order the
datasets by category
:param dbif: The database interface to be used
:return: A dictionary with the rows of the SQL query for each
available mapset
.. code-block:: python
>>> import grass.temporal as tgis
>>> tgis.core.init()
>>> name = "list_stds_test"
>>> sp = tgis.open_stds.open_new_stds(
... name=name,
... type="strds",
... temporaltype="absolute",
... title="title",
... descr="descr",
... semantic="mean",
... dbif=None,
... overwrite=True,
... )
>>> mapset = tgis.get_current_mapset()
>>> stds_list = tgis.list_stds.get_dataset_list(
... "strds", "absolute", columns="name"
... )
>>> rows = stds_list[mapset]
>>> for row in rows:
... if row["name"] == name:
... print(True)
...
True
>>> stds_list = tgis.list_stds.get_dataset_list(
... "strds",
... "absolute",
... columns="name,mapset",
... where="mapset = '%s'" % (mapset),
... )
>>> rows = stds_list[mapset]
>>> for row in rows:
... if row["name"] == name and row["mapset"] == mapset:
... print(True)
...
True
>>> check = sp.delete()
"""
id = None
sp = dataset_factory(type, id)
dbif, connection_state_changed = init_dbif(dbif)
mapsets = get_available_temporal_mapsets()
result = {}
for mapset in mapsets.keys():
if temporal_type == "absolute":
table = sp.get_type() + "_view_abs_time"
else:
table = sp.get_type() + "_view_rel_time"
if columns and columns.find("all") == -1:
sql = "SELECT " + str(columns) + " FROM " + table
else:
sql = "SELECT * FROM " + table
if where:
sql += " WHERE " + where
sql += " AND mapset = '%s'" % (mapset)
else:
sql += " WHERE mapset = '%s'" % (mapset)
if order:
sql += " ORDER BY " + order
dbif.execute(sql, mapset=mapset)
rows = dbif.fetchall(mapset=mapset)
if rows:
result[mapset] = rows
if connection_state_changed:
dbif.close()
return result
###############################################################################
@contextmanager
def _open_output_file(file, encoding="utf-8", **kwargs):
if not file:
yield sys.stdout
elif not isinstance(file, (str, os.PathLike)):
yield file
else:
with open(file, "w", encoding=encoding, **kwargs) as stream:
yield stream
def _write_line(items, separator, file):
if not separator:
separator = ","
output = separator.join([f"{item}" for item in items])
with _open_output_file(file) as stream:
print(f"{output}", file=stream)
def _write_plain(rows, header, separator, file):
def write_plain_row(items, separator, file):
output = separator.join([f"{item}" for item in items])
print(f"{output}", file=file)
with _open_output_file(file) as stream:
# Print the column names if requested
if header:
write_plain_row(items=header, separator=separator, file=stream)
for row in rows:
write_plain_row(items=row, separator=separator, file=stream)
def _write_json(rows, column_names, file):
# Lazy import output format-specific dependencies.
# pylint: disable=import-outside-toplevel
import json
import datetime
class ResultsEncoder(json.JSONEncoder):
"""Results encoder for JSON which handles SimpleNamespace objects"""
def default(self, o):
"""Handle additional types"""
if isinstance(o, datetime.datetime):
return f"{o}"
return super().default(o)
dict_rows = []
for row in rows:
new_row = {}
for key, value in zip(column_names, row):
new_row[key] = value
dict_rows.append(new_row)
meta = {"column_names": column_names}
with _open_output_file(file) as stream:
json.dump({"data": dict_rows, "metadata": meta}, stream, cls=ResultsEncoder)
def _write_yaml(rows, column_names, file=sys.stdout):
# Lazy import output format-specific dependencies.
# pylint: disable=import-outside-toplevel
import yaml
class NoAliasIndentListSafeDumper(yaml.SafeDumper):
"""YAML dumper class which does not create aliases and indents lists
This avoid dates being labeled with &id001 and referenced with *id001.
Instead, same dates are simply repeated.
Lists have their dash-space (- ) indented instead of considering the
dash and space to be a part of indentation. This might be better handled
when https://github.com/yaml/pyyaml/issues/234 is resolved.
"""
def ignore_aliases(self, data):
return True
def increase_indent(self, flow=False, indentless=False):
return super().increase_indent(flow=flow, indentless=False)
dict_rows = []
for row in rows:
new_row = {}
for key, value in zip(column_names, row):
new_row[key] = value
dict_rows.append(new_row)
meta = {"column_names": column_names}
with _open_output_file(file) as stream:
print(
yaml.dump(
{"data": dict_rows, "metadata": meta},
Dumper=NoAliasIndentListSafeDumper,
default_flow_style=False,
),
end="",
file=stream,
)
def _write_csv(rows, column_names, separator, file=sys.stdout):
# Lazy import output format-specific dependencies.
# pylint: disable=import-outside-toplevel
import csv
# Newlines handled by the CSV writer. Set according to the package doc.
with _open_output_file(file, newline="") as stream:
spamwriter = csv.writer(
stream,
delimiter=separator,
quotechar='"',
doublequote=True,
quoting=csv.QUOTE_NONNUMERIC,
lineterminator="\n",
)
if column_names:
spamwriter.writerow(column_names)
for row in rows:
spamwriter.writerow(row)
def _write_table(rows, column_names, output_format, separator, file):
if output_format == "json":
_write_json(rows=rows, column_names=column_names, file=file)
elif output_format == "yaml":
_write_yaml(rows=rows, column_names=column_names, file=file)
elif output_format == "plain":
# No particular reason for this separator except that this is the original
# behavior.
if not separator:
separator = "\t"
_write_plain(rows=rows, header=column_names, separator=separator, file=file)
elif output_format == "csv":
if not separator:
separator = ","
_write_csv(rows=rows, column_names=column_names, separator=separator, file=file)
else:
raise ValueError(f"Unknown value '{output_format}' for output_format")
def _get_get_registered_maps_as_objects_with_method(dataset, where, method, gran, dbif):
if method == "deltagaps":
return dataset.get_registered_maps_as_objects_with_gaps(where=where, dbif=dbif)
if method == "delta":
return dataset.get_registered_maps_as_objects(
where=where, order="start_time", dbif=dbif
)
if method == "gran":
if where:
raise ValueError(
f"The where parameter is not supported with method={method}"
)
if gran is not None and gran != "":
return dataset.get_registered_maps_as_objects_by_granularity(
gran=gran, dbif=dbif
)
return dataset.get_registered_maps_as_objects_by_granularity(dbif=dbif)
raise ValueError(f"Invalid method '{method}'")
def _get_get_registered_maps_as_objects_delta_gran(
dataset, where, method, gran, dbif, msgr
):
maps = _get_get_registered_maps_as_objects_with_method(
dataset=dataset, where=where, method=method, gran=gran, dbif=dbif
)
if not maps:
return []
if isinstance(maps[0], list):
if len(maps[0]) > 0:
first_time, unused = maps[0][0].get_temporal_extent_as_tuple()
else:
msgr.warning(_("Empty map list"))
return []
else:
first_time, unused = maps[0].get_temporal_extent_as_tuple()
records = []
for map_object in maps:
if isinstance(map_object, list):
if len(map_object) > 0:
map_object = map_object[0]
else:
msgr.fatal(_("Empty entry in map list, this should not happen"))
start, end = map_object.get_temporal_extent_as_tuple()
if end:
delta = end - start
else:
delta = None
delta_first = start - first_time
if map_object.is_time_absolute():
if end:
delta = time_delta_to_relative_time(delta)
delta_first = time_delta_to_relative_time(delta_first)
records.append((map_object, start, end, delta, delta_first))
return records
def _get_list_of_maps_delta_gran(dataset, columns, where, method, gran, dbif, msgr):
maps = _get_get_registered_maps_as_objects_delta_gran(
dataset=dataset, where=where, method=method, gran=gran, dbif=dbif, msgr=msgr
)
rows = []
for map_object, start, end, delta, delta_first in maps:
row = []
# Here the names must be the same as in the database
# to make the interface consistent.
for column in columns:
if column == "id":
row.append(map_object.get_id())
elif column == "name":
row.append(map_object.get_name())
elif column == "layer":
row.append(map_object.get_layer())
elif column == "mapset":
row.append(map_object.get_mapset())
elif column == "start_time":
row.append(start)
elif column == "end_time":
row.append(end)
elif column == "interval_length":
row.append(delta)
elif column == "distance_from_begin":
row.append(delta_first)
else:
raise ValueError(f"Unsupported column '{column}'")
rows.append(row)
return rows
def _get_list_of_maps_stds(
element_type,
name,
columns,
order,
where,
method,
output_format,
gran=None,
dbif=None,
):
dbif, connection_state_changed = init_dbif(dbif)
msgr = get_tgis_message_interface()
dataset = open_old_stds(name, element_type, dbif)
def check_columns(column_names, output_format, element_type):
if element_type != "stvds" and "layer" in columns:
raise ValueError(
f"Column 'layer' is not allowed with temporal type '{element_type}'"
)
if output_format == "line" and len(column_names) > 1:
raise ValueError(
f"'{output_format}' output_format can have only 1 column, "
f"not {len(column_names)}"
)
# This method expects a list of objects for gap detection
if method in ["delta", "deltagaps", "gran"]:
if not columns:
if output_format == "list":
# Only one column is needed.
columns = ["id"]
else:
columns = ["id", "name"]
if element_type == "stvds":
columns.append("layer")
columns.extend(
[
"mapset",
"start_time",
"end_time",
"interval_length",
"distance_from_begin",
]
)
check_columns(
column_names=columns,
output_format=output_format,
element_type=element_type,
)
rows = _get_list_of_maps_delta_gran(
dataset=dataset,
columns=columns,
where=where,
method=method,
gran=gran,
dbif=dbif,
msgr=msgr,
)
else:
if columns:
check_columns(
column_names=columns,
output_format=output_format,
element_type=element_type,
)
else:
if output_format == "line":
# For list of values, only one column is needed.
columns = ["id"]
else:
columns = ["name", "mapset", "start_time", "end_time"]
if not order:
order = "start_time"
rows = dataset.get_registered_maps(",".join(columns), where, order, dbif)
# End with error for the old, custom formats. Proper formats simply return
# empty result whatever empty is for each format (e.g., empty list for JSON).
if not rows and (output_format in ["plain", "line"]):
dbif.close()
gs.fatal(
_(
"Nothing found in the database for space time dataset <{name}> "
"(type: {element_type}): {detail}"
).format(
name=dataset.get_id(),
element_type=element_type,
detail=(
_(
"Dataset is empty or where clause is too constrained or "
"incorrect"
)
if where
else _("Dataset is empty")
),
)
)
if connection_state_changed:
dbif.close()
return rows, columns
# The code is compatible with pre-v8.2 versions, but for v9, it needs to be reviewed
# to remove the backwards compatibility which will clean it up.
[docs]def list_maps_of_stds(
type, # pylint: disable=redefined-builtin
input, # pylint: disable=redefined-builtin
columns,
order,
where,
separator,
method,
no_header=False,
gran=None,
dbif=None,
outpath=None,
output_format=None,
):
"""List the maps of a space time dataset using different methods
:param type: The type of the maps raster, raster3d or vector
:param input: Name of a space time raster dataset
:param columns: A comma separated list of columns to be printed to stdout
:param order: A comma separated list of columns to order the
maps by category
:param where: A where statement for selected listing without "WHERE"
e.g: start_time < "2001-01-01" and end_time > "2001-01-01"
:param separator: The field separator character between the columns
:param method: String identifier to select a method out of cols,
comma,delta or deltagaps
:param dbif: The database interface to be used
- "cols" Print preselected columns specified by columns
- "comma" Print the map ids ("name@mapset") as comma separated string
- "delta" Print the map ids ("name@mapset") with start time,
end time, relative length of intervals and the relative
distance to the begin
- "deltagaps" Same as "delta" with additional listing of gaps.
Gaps can be easily identified as the id is "None"
- "gran" List map using the granularity of the space time dataset,
columns are identical to deltagaps
:param no_header: Suppress the printing of column names
:param gran: The user defined granule to be used if method=gran is
set, in case gran=None the granule of the space time
dataset is used
:param outpath: The path to file where to save output
"""
if not output_format:
if method == "comma":
output_format = "line"
output_format = "plain"
if columns:
if isinstance(columns, str):
columns = columns.split(",")
rows, columns = _get_list_of_maps_stds(
element_type=type,
name=input,
columns=columns,
order=order,
where=where,
method=method,
output_format=output_format,
gran=gran,
dbif=dbif,
)
if output_format == "line":
_write_line(
items=[row[0] for row in rows],
separator=separator,
file=outpath,
)
else:
_write_table(
rows=rows,
column_names=None if no_header else columns,
separator=separator,
output_format=output_format,
file=outpath,
)
###############################################################################
if __name__ == "__main__":
import doctest
doctest.testmod()
