Skip to content

Commit 015e761

Browse files
ninsblninsbl
committed
add STRDS class
1 parent c483f7b commit 015e761

File tree

1 file changed

+194
-31
lines changed

1 file changed

+194
-31
lines changed

‎src/actinia/strds.py‎

Lines changed: 194 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,36 +1,199 @@
11
#!/usr/bin/env python
2-
# -*- coding: utf-8 -*-
3-
#######
4-
# actinia-python-client is a python client for actinia - an open source REST
5-
# API for scalable, distributed, high performance processing of geographical
6-
# data that uses GRASS GIS for computational tasks.
7-
#
8-
# Copyright (c) 2022 mundialis GmbH & Co. KG
9-
#
10-
# This program is free software: you can redistribute it and/or modify
11-
# it under the terms of the GNU General Public License as published by
12-
# the Free Software Foundation, either version 3 of the License, or
13-
# (at your option) any later version.
14-
#
15-
# This program is distributed in the hope that it will be useful,
16-
# but WITHOUT ANY WARRANTY; without even the implied warranty of
17-
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18-
# GNU General Public License for more details.
19-
#
20-
# You should have received a copy of the GNU General Public License
21-
# along with this program. If not, see <https://www.gnu.org/licenses/>.
22-
#
23-
#######
2+
3+
"""The strds module provides a class for SpaceTimeRasterDataSet (STRDS) operations.
4+
5+
actinia-python-client is a python client for actinia - an open source REST
6+
API for scalable, distributed, high performance processing of geographical
7+
data that uses GRASS GIS for computational tasks.
8+
9+
Copyright (c) 2024 mundialis GmbH & Co. KG
10+
11+
This program is free software: you can redistribute it and/or modify
12+
it under the terms of the GNU General Public License as published by
13+
the Free Software Foundation, either version 3 of the License, or
14+
(at your option) any later version.
15+
16+
This program is distributed in the hope that it will be useful,
17+
but WITHOUT ANY WARRANTY; without even the implied warranty of
18+
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19+
GNU General Public License for more details.
20+
21+
You should have received a copy of the GNU General Public License
22+
along with this program. If not, see <https://www.gnu.org/licenses/>.
23+
"""
24+
25+
from __future__ import annotations
2426

2527
__license__ = "GPLv3"
26-
__author__ = "Anika Weinmann"
27-
__copyright__ = "Copyright 2022, mundialis GmbH & Co. KG"
28+
__author__ = "Anika Weinmann, Stefan Blumentrath"
29+
__copyright__ = "Copyright 2022-2024, mundialis GmbH & Co. KG"
2830
__maintainer__ = "Anika Weinmann"
2931

30-
# TODO:
31-
# * /locations/{location_name}/mapsets/{mapset_name}/strds/{strds_name}
32-
# - GET, DELETE, POST
33-
# * /locations/{location_name}/mapsets/{mapset_name}/strds/{strds_name}/
34-
# raster_layers - DELETE, GET, PUT
35-
# * /locations/{location_name}/mapsets/{mapset_name}/strds/{strds_name}/
36-
# render - GET
32+
import json
33+
from datetime import datetime
34+
from typing import TYPE_CHECKING, Optional
35+
36+
from actinia.utils import request_and_check
37+
38+
if TYPE_CHECKING:
39+
from actinia import Actinia
40+
41+
42+
class SpaceTimeRasterDataSet:
43+
"""Class for SpaceTimeRasterDataSet (STRDS) operations."""
44+
45+
def __init__(
46+
self,
47+
name: str,
48+
location_name: str,
49+
mapset_name: str,
50+
actinia: Actinia,
51+
auth: tuple,
52+
) -> None:
53+
"""Initialize the SpaceTimeRasterDataSet (STRDS) object."""
54+
self.name = name
55+
self.__location_name = location_name
56+
self.__mapset_name = mapset_name
57+
self.__actinia = actinia
58+
self.__auth = auth
59+
self.raster_layers = None
60+
self.info = None
61+
62+
def get_info(self, *, force: Optional(bool) = False) -> dict:
63+
"""Return the information of the SpaceTimeRasterDataSet (STRDS).
64+
65+
Parameters
66+
----------
67+
force: bool
68+
Force uptating STRDS info
69+
70+
Returns
71+
-------
72+
info: dict
73+
dict with information about the SpaceTimeRasterDataset
74+
75+
"""
76+
if self.info is None or force is True:
77+
url = (
78+
f"{self.__actinia.url}/locations/{self.__location_name}/"
79+
f"mapsets/{self.__mapset_name}/strds/{self.name}"
80+
)
81+
resp = request_and_check("GET", url, auth=self.__auth)
82+
strds_info = resp["process_results"]
83+
self.info = strds_info
84+
return self.info
85+
86+
def get_strds_raster_layers(
87+
self,
88+
where: Optional(str) = None,
89+
*,
90+
force: Optional(bool) = False,
91+
) -> dict:
92+
"""Return a list of Raster Layers from a SpaceTimeRasterDataSet.
93+
94+
Parameters
95+
----------
96+
where: str
97+
String with WHERE-clause for a STRDS-where-query without
98+
WHERE keyword
99+
force: bool
100+
Force uptating STRDS info
101+
102+
Returns
103+
-------
104+
info: dict
105+
dict with information about the SpaceTimeRasterDataset
106+
107+
"""
108+
if self.info is None or force is True:
109+
url = (
110+
f"{self.__actinia.url}/locations/{self.__location_name}/"
111+
f"mapsets/{self.__mapset_name}/strds/{self.name}"
112+
)
113+
if where:
114+
url += f"?where={where}"
115+
resp = request_and_check("GET", url, auth=self.__auth)
116+
raster_layers = resp["process_results"]
117+
self.raster_layers = raster_layers
118+
return self.info
119+
120+
def register_raster_layer(
121+
self,
122+
name: str,
123+
start_time: str | datetime,
124+
end_time: str | datetime,
125+
) -> None:
126+
"""Register a Raster Layer in a SpaceTimeRasterDataSet (STRDS)."""
127+
if isinstance(start_time, datetime):
128+
start_time = start_time.strftime("%Y-%m-%d %H:%M:%S")
129+
if isinstance(end_time, datetime):
130+
end_time = end_time.strftime("%Y-%m-%d %H:%M:%S")
131+
putkwargs = {
132+
"headers": self.__actinia.headers,
133+
"auth": self.__auth,
134+
"timeout": self.__actinia.timeout,
135+
"data": json.dumps(
136+
{"name": name, "start_time": start_time, "end_time": end_time},
137+
),
138+
}
139+
url = (
140+
f"{self.__actinia.url}/locations/{self.__location_name}/"
141+
f"mapsets/{self.__mapset_name}/strds/{self.name}/raster_layers"
142+
)
143+
request_and_check("PUT", url, **putkwargs)
144+
145+
def unregister_raster_layers(self, raster_layers: list[str]) -> None:
146+
"""Unregister Raster Layers from a SpaceTimeRasterDataSet (STRDS)."""
147+
delkwargs = {
148+
"headers": self.__actinia.headers,
149+
"auth": self.__auth,
150+
"timeout": self.__actinia.timeout,
151+
"data": json.dumps(raster_layers),
152+
}
153+
url = (
154+
f"{self.__actinia.url}/locations/{self.__location_name}/"
155+
f"mapsets/{self.__mapset_name}/strds/{self.name}/raster_layers"
156+
)
157+
request_and_check("DEL", url, **delkwargs)
158+
159+
def render(self, render_dict: dict) -> dict:
160+
"""Render Raster layers in a SpaceTimeRasterDataSet (STRDS).
161+
162+
Returns
163+
-------
164+
render: dict
165+
dict with render response.
166+
167+
Raises
168+
------
169+
ValueError:
170+
ValueError if dict does not containt required keys.
171+
172+
"""
173+
if set(render_dict.keys()) != {
174+
"n",
175+
"s",
176+
"e",
177+
"w",
178+
"width",
179+
"height",
180+
"start_time",
181+
"end_time",
182+
}:
183+
msg = (
184+
"render_dict must contain the keys"
185+
" 'n', 's', 'e', 'w', 'width', 'height',"
186+
" 'start_time', 'end_time'"
187+
)
188+
raise ValueError(msg)
189+
getkwargs = {
190+
"headers": self.__actinia.headers,
191+
"auth": self.__auth,
192+
"timeout": self.__actinia.timeout,
193+
"data": json.dumps(render_dict),
194+
}
195+
url = (
196+
f"{self.__actinia.url}/locations/{self.__location_name}/"
197+
f"mapsets/{self.__mapset_name}/strds/{self.name}/render"
198+
)
199+
return request_and_check("GET", url, **getkwargs)

0 commit comments

Comments
 (0)