fix type deprecation error when parsing metadata

mapchete/commands/_index.py

@@ -8,8 +8,10 @@
 
 import mapchete
 from mapchete.commands.observer import ObserverProtocol, Observers
+from mapchete.commands.parser import InputInfo
 from mapchete.config import MapcheteConfig
-from mapchete.config.parse import bounds_from_opts, raw_conf, raw_conf_process_pyramid
+from mapchete.config.parse import bounds_from_opts, raw_conf
+from mapchete.enums import InputType
 from mapchete.index import zoom_index_gen
 from mapchete.path import MPath
 from mapchete.types import MPathLike, Progress
@@ -85,7 +87,6 @@ def index(
     fs_opts : dict
         Configuration options for fsspec filesystem.
     """
-
     if not any([geojson, gpkg, shp, txt, vrt]):
         raise ValueError(
             """At least one of '--geojson', '--gpkg', '--shp', '--vrt' or '--txt'"""
@@ -96,21 +97,21 @@ def index(
 
     all_observers.notify(message=f"create index(es) for {some_input}")
 
+    input_info = InputInfo.from_inp(some_input)
     if tile:
-        tile = raw_conf_process_pyramid(raw_conf(some_input)).tile(*tile)
+        tile = input_info.output_pyramid.tile(*tile)
         bounds = tile.bounds
         zoom = tile.zoom
+    elif input_info.input_type == InputType.mapchete:
+        bounds = bounds_from_opts(
+            point=point,
+            point_crs=point_crs,
+            bounds=bounds,
+            bounds_crs=bounds_crs,
+            raw_conf=raw_conf(some_input),
+        )
     else:
-        try:
-            bounds = bounds_from_opts(
-                point=point,
-                point_crs=point_crs,
-                bounds=bounds,
-                bounds_crs=bounds_crs,
-                raw_conf=raw_conf(some_input),
-            )
-        except IsADirectoryError:
-            pass
+        bounds = bounds or input_info.bounds
 
     with mapchete.open(
         some_input,

mapchete/commands/parser.py

@@ -0,0 +1,191 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Optional, Union
+
+from rasterio.crs import CRS
+from rasterio.vrt import WarpedVRT
+
+from mapchete.config.base import MapcheteConfig
+from mapchete.config.parse import raw_conf, raw_conf_output_pyramid
+from mapchete.enums import DataType, InputType, OutputType
+from mapchete.formats import (
+    available_input_formats,
+    available_output_formats,
+    driver_from_file,
+)
+from mapchete.io import fiona_open, rasterio_open
+from mapchete.path import MPath
+from mapchete.tile import BufferedTilePyramid
+from mapchete.types import Bounds, CRSLike, MPathLike, ZoomLevels
+
+logger = logging.getLogger(__name__)
+OUTPUT_FORMATS = available_output_formats()
+
+
+@dataclass
+class InputInfo:
+    output_params: dict
+    crs: CRSLike
+    data_type: DataType
+    input_type: InputType
+    bounds: Optional[Bounds] = None
+    zoom_levels: Optional[ZoomLevels] = None
+    output_pyramid: Optional[BufferedTilePyramid] = None
+    pixel_size: Optional[int] = None
+
+    @staticmethod
+    def from_inp(inp: Union[MPathLike, dict, MapcheteConfig]) -> InputInfo:
+        try:
+            path = MPath.from_inp(inp)
+
+        except Exception:
+            if isinstance(inp, dict):
+                return InputInfo.from_config_dict(inp)
+            elif isinstance(inp, MapcheteConfig):  # pragma: no cover
+                return InputInfo.from_mapchete_config(inp)
+
+            raise TypeError(f"cannot create InputInfo from {inp}")  # pragma: no cover
+
+        return InputInfo.from_path(path)
+
+    @staticmethod
+    def from_config_dict(conf: dict) -> InputInfo:
+        output_params = conf["output"]
+        output_pyramid = raw_conf_output_pyramid(conf)
+        return InputInfo(
+            input_type=InputType.mapchete,
+            output_params=output_params,
+            output_pyramid=output_pyramid,
+            crs=output_pyramid.crs,
+            zoom_levels=ZoomLevels.from_inp(conf["zoom_levels"]),
+            data_type=DataType[OUTPUT_FORMATS[output_params["format"]]["data_type"]],
+            bounds=Bounds.from_inp(conf.get("bounds")) if conf.get("bounds") else None,
+        )
+
+    @staticmethod
+    def from_mapchete_config(
+        mapchete_config: MapcheteConfig,
+    ) -> InputInfo:  # pragma: no cover
+        return InputInfo(
+            input_type=InputType.mapchete,
+            output_params=mapchete_config.output.params,
+            output_pyramid=mapchete_config.output_pyramid,
+            crs=mapchete_config.output_pyramid.crs,
+            zoom_levels=mapchete_config.zoom_levels,
+            data_type=DataType[
+                OUTPUT_FORMATS[mapchete_config.output.params["format"]]["data_type"]
+            ],
+            bounds=mapchete_config.bounds,
+        )
+
+    @staticmethod
+    def from_path(path: MPath) -> InputInfo:
+        # assuming single file if path has a file extension
+        if path.suffix:
+            logger.debug("assuming single file")
+            driver = driver_from_file(path)
+
+            # single file input can be a mapchete file or a rasterio/fiona file
+            if driver == "Mapchete":
+                logger.debug("input is mapchete file")
+                return InputInfo.from_mapchete_file(path)
+
+            elif driver == "raster_file":
+                # this should be readable by rasterio
+                logger.debug("input is raster_file")
+                return InputInfo.from_rasterio_file(path)
+
+            elif driver == "vector_file":
+                # this should be readable by Fiona
+                return InputInfo.from_fiona_file(path)
+
+            else:  # pragma: no cover
+                raise NotImplementedError(f"driver {driver} is not supported")
+
+        # assuming tile directory
+        else:
+            logger.debug("input is maybe a tile directory")
+            return InputInfo.from_tile_directory(path)
+
+    @staticmethod
+    def from_mapchete_file(path: MPath) -> InputInfo:
+        return InputInfo.from_config_dict(raw_conf(path))
+
+    @staticmethod
+    def from_rasterio_file(path: MPath) -> InputInfo:
+        with rasterio_open(path) as src:
+            if src.transform.is_identity:
+                if src.gcps[1] is not None:
+                    with WarpedVRT(src) as dst:
+                        bounds = dst.bounds
+                        crs = src.gcps[1]
+                elif src.rpcs:  # pragma: no cover
+                    with WarpedVRT(src) as dst:
+                        bounds = dst.bounds
+                        crs = CRS.from_string("EPSG:4326")
+                else:  # pragma: no cover
+                    raise TypeError("cannot determine georeference")
+            else:
+                crs = src.crs
+                bounds = src.bounds
+            return InputInfo(
+                input_type=InputType.single_file,
+                output_params=dict(
+                    bands=src.meta["count"],
+                    dtype=src.meta["dtype"],
+                    format=src.driver
+                    if src.driver in available_input_formats()
+                    else None,
+                ),
+                crs=crs,
+                pixel_size=src.transform[0],
+                data_type=DataType.raster,
+                bounds=Bounds(*bounds),
+            )
+
+    @staticmethod
+    def from_fiona_file(path: MPath) -> InputInfo:
+        with fiona_open(path) as src:
+            return InputInfo(
+                input_type=InputType.single_file,
+                output_params=dict(
+                    schema=src.schema,
+                    format=src.driver
+                    if src.driver in available_input_formats()
+                    else None,
+                ),
+                crs=src.crs,
+                data_type=DataType.vector,
+                bounds=Bounds(*src.bounds) if len(src) else None,
+            )
+
+    @staticmethod
+    def from_tile_directory(path) -> InputInfo:
+        conf = (path / "metadata.json").read_json()
+        pyramid = BufferedTilePyramid.from_dict(conf["pyramid"])
+        return InputInfo(
+            input_type=InputType.tile_directory,
+            output_params=conf["driver"],
+            output_pyramid=pyramid,
+            crs=pyramid.crs,
+            data_type=DataType[OUTPUT_FORMATS[conf["driver"]["format"]]["data_type"]],
+        )
+
+
+@dataclass
+class OutputInfo:
+    type: OutputType
+    driver: Optional[str]
+
+    @staticmethod
+    def from_path(path: MPath) -> OutputInfo:
+        if path.suffix:
+            if path.suffix == ".tif":
+                return OutputInfo(type=OutputType.single_file, driver="GTiff")
+            else:
+                raise ValueError("currently only single file GeoTIFFs are allowed")
+
+        if not path.suffix:
+            return OutputInfo(type=OutputType.tile_directory, driver=None)

mapchete/config/parse.py

@@ -4,7 +4,6 @@
 from collections import OrderedDict
 from typing import Any, Iterable, Optional, Tuple, Union
 
-import oyaml as yaml
 from rasterio.crs import CRS
 from shapely import wkt
 from shapely.geometry import Point, box, shape
@@ -48,7 +47,7 @@ def _config_to_dict(input_config: Union[dict, MPathLike]) -> dict:
             return OrderedDict(_include_env(input_config), mapchete_file=None)
         # from Mapchete file
         elif input_config.suffix == ".mapchete":
-            config_dict = _include_env(yaml.safe_load(input_config.read_text()))
+            config_dict = _include_env(input_config.read_yaml())
             return OrderedDict(
                 config_dict,
                 config_dir=config_dict.get(
@@ -84,9 +83,7 @@ def raw_conf(mapchete_file: Union[dict, MPathLike]) -> dict:
     if isinstance(mapchete_file, dict):
         return map_to_new_config_dict(mapchete_file)
     else:
-        return map_to_new_config_dict(
-            yaml.safe_load(MPath.from_inp(mapchete_file).read_text())
-        )
+        return map_to_new_config_dict(MPath.from_inp(mapchete_file).read_yaml())
 
 
 def raw_conf_process_pyramid(

mapchete/enums.py

@@ -15,6 +15,22 @@ class Concurrency(str, Enum):
     dask = "dask"
 
 
+class DataType(str, Enum):
+    raster = "raster"
+    vector = "vector"
+
+
+class OutputType(str, Enum):
+    single_file = "single_file"
+    tile_directory = "tile_directory"
+
+
+class InputType(str, Enum):
+    single_file = "single_file"
+    tile_directory = "tile_directory"
+    mapchete = "mapchete"
+
+
 class Status(str, Enum):
     r"""
     Status describin life cycle of a Job.

mapchete/formats/models.py

@@ -0,0 +1,12 @@
+from typing import List
+
+from pydantic import BaseModel, Field
+
+from mapchete.enums import DataType
+
+
+class DriverMetadata(BaseModel):
+    driver_name: str
+    data_type: DataType
+    mode: str
+    file_extensions: List[str] = Field(default_factory=list)

mapchete/formats/tools.py

@@ -15,9 +15,11 @@
 from shapely.geometry.base import BaseGeometry
 
 from mapchete.errors import MapcheteConfigError, MapcheteDriverError
-from mapchete.io import MPath, fiona_open, rasterio_open, read_json, write_json
+from mapchete.formats.models import DriverMetadata
+from mapchete.io import MPath, fiona_open, rasterio_open
 from mapchete.registered import drivers
 from mapchete.tile import BufferedTilePyramid
+from mapchete.types import MPathLike
 
 logger = logging.getLogger(__name__)
 
@@ -80,7 +82,7 @@ def driver_metadata(driver_name: str) -> Dict:
         raise ValueError(f"driver '{driver_name}' not found")
 
 
-def driver_from_file(input_file: str, quick: bool = True) -> str:
+def driver_from_file(input_file: MPathLike, quick: bool = True) -> str:
     """
     Guess driver from file by opening it.
 
@@ -236,7 +238,7 @@ def _geometry_to_none(value):
     return _unparse_dict(out, strategies=strategies) if parse_datetime else out
 
 
-def read_output_metadata(metadata_json: str, **kwargs: str) -> Dict:
+def read_output_metadata(metadata_json: MPathLike, **kwargs: str) -> Dict:
     """
     Read and parse metadata.json.
 
@@ -250,7 +252,7 @@ def read_output_metadata(metadata_json: str, **kwargs: str) -> Dict:
         Parsed output metadata.
     """
 
-    return load_metadata(read_json(metadata_json, **kwargs))
+    return load_metadata(MPath.from_inp(metadata_json).read_json())
 
 
 def load_metadata(params: Dict, parse_known_types=True) -> Dict:
@@ -274,7 +276,12 @@ def load_metadata(params: Dict, parse_known_types=True) -> Dict:
         raise TypeError(f"metadata parameters must be a dictionary, not {params}")
     out = params.copy()
     grid = out["pyramid"]["grid"]
-    if grid["type"] == "geodetic" and grid["shape"] == [2, 1]:  # pragma: no cover
+    if "type" in grid:  # pragma: no cover
+        warnings.warn(DeprecationWarning("'type' is deprecated and should be 'grid'"))
+        if "grid" not in grid:
+            grid["grid"] = grid.pop("type")
+
+    if grid["grid"] == "geodetic" and grid["shape"] == [2, 1]:  # pragma: no cover
         warnings.warn(
             DeprecationWarning(
                 "Deprecated grid shape ordering found. "
@@ -332,7 +339,7 @@ def write_output_metadata(output_params: Dict) -> None:
             logger.debug("%s does not exist", metadata_path)
             dump_params = dump_metadata(output_params)
             # dump output metadata
-            write_json(metadata_path, dump_params)
+            metadata_path.write_json(dump_params)
 
 
 def compare_metadata_params(params1: Dict, params2: Dict) -> None: