Merge pull request #4 from tomorrownow/mapsets

Mapsets restructured

.gitignore

@@ -22,3 +22,6 @@ site/
 
 # Per-project virtualenvs
 env/
+
+# vscode settigns
+.vscode/

actinia/location.py

@@ -77,17 +77,12 @@ def __request_mapsets(self):
 
         :return: A list of the mapset names
         """
-        url = f"{self.__actinia.url}/locations/{self.name}/mapsets"
-        resp = requests.get(url, auth=self.__auth)
-        if resp.status_code != 200:
-            raise Exception(f"Error {resp.status_code}: {resp.text}")
-
-        mapset_names = json.loads(resp.text)["process_results"]
-        mapsets = {
-            mname: Mapset(mname, self.name, self.__actinia, self.__auth)
-            for mname in mapset_names
-        }
-        self.mapsets = mapsets
+        self.mapsets = Mapset.list_mapsets_request(
+            self.name,
+            self.__actinia,
+            self.__auth
+        )
+        return self.mapsets
 
     def delete(self):
         """Delete a location via delete request.
@@ -106,13 +101,33 @@ def get_mapsets(self):
             self.__request_mapsets()
         return self.mapsets
 
-    # def create_mapset(self, name, epsgcode):
-    #     """
-    #     Creates a location with a specified projection.
-    #     """
-    #     url = f"{self.url}/locations/{name}"
-    #     # TODO
-    #     # resp = requests.post(url, auth=(self.__auth))
+    def create_mapset(self, name):
+        """
+        Creates a mapset within the location.
+        """
+        mapset = Mapset.create_mapset_request(
+            name,
+            self.name,
+            self.__actinia,
+            self.__auth
+        )
+        # We could also fetch data from the server again
+        # with self.__request_mapsets() to ensure data is stale
+        self.mapsets[name] = mapset
+        return mapset
+
+    def delete_mapset(self, name):
+        """
+        Deletes a mapset and returns an updated mapset list for the location.
+        """
+        Mapset.delete_mapset_request(
+            name,
+            self.name,
+            self.__actinia,
+            self.__auth
+        )
+        del self.mapsets[name]
+        return self.mapsets
 
     def __validate_process_chain(self, pc, type):
         url = f"{self.__actinia.url}/locations/{self.name}/"

actinia/mapset.py

@@ -23,19 +23,31 @@
 #######
 
 __license__ = "GPLv3"
-__author__ = "Anika Weinmann"
+__author__ = "Anika Weinmann and Corey White"
 __copyright__ = "Copyright 2022, mundialis GmbH & Co. KG"
 __maintainer__ = "Anika Weinmann"
 
 import json
 import requests
-
+from actinia.region import Region
+from enum import Enum, unique
 from actinia.raster import Raster
 from actinia.vector import Vector
 from actinia.utils import request_and_check, print_stdout
 from actinia.job import Job
 
 
+@unique
+class MAPSET_TASK(Enum):
+    INFO = "info"
+    LOCK = "lock"
+    RASTER_LAYER = "raster_layer"
+    STRDS = "strds"
+    VECTOR_LAYERS = "vector_layers"
+    PROCESSING = "processing"
+    PROCESSING_ASYNC = "processing_async"
+
+
 class Mapset:
     def __init__(self, name, location_name, actinia, auth):
         self.name = name
@@ -48,6 +60,220 @@ def __init__(self, name, location_name, actinia, auth):
         self.vector_layers = None
         self.strds = None
 
+    def __request_url(
+        actinia_url,
+        location_name,
+        mapset_name=None,
+        task=None
+    ):
+        """
+        Provides the url to an Actinia mapset resource.
+
+        Parameters
+        ----------
+        actinia_url : str
+            The base url to actinia server
+        location_name : str
+            The GRASS location name
+            route: /locations/{location_name}/mapsets
+        mapset_name : str, default=None
+            The mapset name
+            route: /locations/{location_name}/mapsets/{mapset_name}
+        task : Enum(MAPSET_TASK), default=None
+            The requested task
+            (info) route:
+                /locations/{location_name}/mapsets/{mapset_name}/info
+            (lock) route:
+                /locations/{location_name}/mapsets/{mapset_name}/lock
+            (raster_layers) route:
+                /locations/{location_name}/mapsets/{mapset_name}/raster_layers
+            (vector_layers) route:
+                /locations/{location_name}/mapsets/{mapset_name}/vector_layers
+            (strds) route:
+                /locations/{location_name}/mapsets/{mapset_name}/strds
+            (processing) route:
+                /locations/{location_name}/mapsets/{mapset_name}/processing
+            (processing_async) route:
+                /locations/{location_name}/mapsets/{mapset_name}/processing_async
+
+        raster_layers : bool
+        Returns
+        -------
+        base_url : str
+            Return the url scheme for the mapset request
+        """
+        base_url = f"{actinia_url}/locations/{location_name}/mapsets"
+        if mapset_name is not None:
+            base_url = f"{base_url}/{mapset_name}"
+            if isinstance(task, MAPSET_TASK):
+                base_url = f"{base_url}/{task.value}"
+        return base_url
+
+    def info(self):
+        """Get mapset info"""
+        if self.projection is None or self.region is None:
+            proc_res = self.request_info(
+                self.name,
+                self.__location_name,
+                self.__actinia,
+                self.__auth
+            )
+            self.projection = proc_res["projection"]
+            self.region = Region(**proc_res["region"])
+        return {"projection": self.projection, "region": self.region}
+
+    def delete(self):
+        """Deletes the mapset"""
+        self.delete_mapset_request(
+            self.name,
+            self.__location_name,
+            self.__actinia,
+            self.__auth
+        )
+        del self.__actinia.locations[self.__location_name].mapsets[self.name]
+
+    @classmethod
+    def list_mapsets_request(cls, location_name, actinia, auth):
+        """
+        Lists mapsets within a location.
+
+        Parameters
+        ----------
+        location_name : str
+            The name of the location where the mapsets are located.
+        actinia : Actinia
+            An Actinia instance containing the url
+        auth :
+            Actinia authentication
+
+        Returns
+        -------
+        mapsets : dict[mapset_name, Mapset]
+            A dict of with keys equal to the mapset name and
+            values set to the Mapset class instance.
+
+        Raises
+        ------
+        Exception
+            Error string with response status code
+            and text if request fails.
+        """
+        url = cls.__request_url(actinia.url, location_name)
+        resp = requests.get(url, auth=auth)
+        if resp.status_code != 200:
+            raise Exception(f"Error {resp.status_code}: {resp.text}")
+
+        mapset_names = json.loads(resp.text)["process_results"]
+        mapsets = {
+            mname: Mapset(mname, location_name, actinia, auth)
+            for mname in mapset_names
+        }
+        return mapsets
+
+    @classmethod
+    def create_mapset_request(cls, mapset_name, location_name, actinia, auth):
+        """
+        Creates a mapset within a location.
+
+        Parameters
+        ----------
+        mapset_name : str
+            The name of the created mapset.
+        location_name : str
+            The name of the location where the mapset is created
+        actinia : Actinia
+            An Actinia instance containing the url
+        auth :
+            Actinia authentication
+
+        Returns
+        -------
+        Mapset
+            A new mapset instance for the created mapset
+
+        Raises
+        ------
+        Exception
+            Error string with response status code
+            and text if request fails.
+        """
+        url = cls.__request_url(actinia.url, location_name, mapset_name)
+        resp = requests.post(url, auth=(auth))
+        if resp.status_code != 200:
+            raise Exception(f"Error {resp.status_code}: {resp.text}")
+        return Mapset(mapset_name, location_name, actinia, auth)
+
+    @classmethod
+    def delete_mapset_request(cls, mapset_name, location_name, actinia, auth):
+        """
+        Delete a mapset within a location.
+
+        Parameters
+        ----------
+        mapset_name : str
+            The name of the mapset to delete
+        location_name : str
+            The name of the mapset's location
+        actinia : Actinia
+            An Actinia instance containing the url
+        auth :
+            Actinia authentication
+
+        Returns
+        -------
+        None
+
+        Raises
+        ------
+        Exception
+            Error string with response status code
+            and text if request fails.
+        """
+        url = cls.__request_url(actinia.url, location_name, mapset_name)
+        resp = requests.delete(url, auth=(auth))
+        if resp.status_code != 200:
+            raise Exception(f"Error {resp.status_code}: {resp.text}")
+        return None
+
+    @classmethod
+    def request_info(cls, mapset_name, location_name, actinia, auth):
+        """
+        Gets detailed info about a mapset.
+
+        Parameters
+        ----------
+        mapset_name : str
+            The name of mapset.
+        location_name : str
+            The name of the location
+        actinia : Actinia
+            An Actinia instance containing the url
+        auth :
+            Actinia authentication
+
+        Returns
+        -------
+        dict
+        Returns JSON process results if successful.
+
+        Raises
+        ------
+        Exception
+            Error string with response status code
+            and text if request fails.
+        """
+        url = cls.__request_url(
+            actinia.url,
+            location_name,
+            mapset_name,
+            MAPSET_TASK.INFO
+        )
+        resp = requests.get(url, auth=(auth))
+        if resp.status_code != 200:
+            raise Exception(f"Error {resp.status_code}: {resp.text}")
+        proc_res = json.loads(resp.text)["process_results"]
+        return proc_res
+
     def __request_raster_layers(self):
         """
         Requests the raster layers in the mapset.
@@ -196,9 +422,8 @@ def delete_vector(self, layer_name):
             del self.vector_layers[layer_name]
         print_stdout(f"Vector <{layer_name}> successfully deleted")
 
+
 # TODO:
-# * /locations/{location_name}/mapsets/{mapset_name} - DELETE, POST
-# * /locations/{location_name}/mapsets/{mapset_name}/info - GET
 # * (/locations/{location_name}/mapsets/{mapset_name}/lock - GET, DELETE, POST)
 
 # * /locations/{location_name}/mapsets/{mapset_name}/raster_layers