Generators Module¶

class PoiViewGenerator(ABC):
    """
    Base class for generating views from Points of Interest (POI) datasets.

    This class provides the structure for processing downloaded data sources
    and mapping them to POI data. Concrete implementations should extend this
    class for specific data sources.
    """

    def __init__(
        self,
        data_config: Optional[Any] = None,
        generator_config: Optional["PoiViewGeneratorConfig"] = None,
        data_store: Optional["DataStore"] = None,
    ):
        """
        Initialize the POI View Generator.

        Args:
            generator_config: Configuration for the view generator
            data_store: Data store for reading/writing data
        """
        self.data_config = data_config
        self.generator_config = generator_config or PoiViewGeneratorConfig()
        self.data_store = data_store or LocalDataStore()
        self.logger = config.get_logger(self.__class__.__name__)

    def resolve_source_paths(
        self,
        poi_data: Union[pd.DataFrame, gpd.GeoDataFrame],
        explicit_paths: Optional[Union[Path, str, List[Union[str, Path]]]] = None,
        **kwargs,
    ) -> List[Union[str, Path]]:
        """
        Resolve source data paths based on POI data or explicit paths.

        This method allows generators to dynamically determine source paths
        based on the POI data (e.g., by geographic intersection).

        Args:
            poi_data: POI data that may be used to determine relevant source paths
            explicit_paths: Explicitly provided source paths, if any
            **kwargs: Additional parameters for path resolution

        Returns:
            List of resolved source paths

        Notes:
            Default implementation returns explicit_paths if provided.
            Subclasses should override this to implement dynamic path resolution.
        """
        if explicit_paths is not None:
            if isinstance(explicit_paths, (str, Path)):
                return [explicit_paths]
            return list(explicit_paths)

        # Raises NotImplementedError if no explicit paths
        # and subclass hasn't overridden this method
        raise NotImplementedError(
            "This generator requires explicit source paths or a subclass "
            "implementation of resolve_source_paths()"
        )

    def _pre_load_hook(self, source_data_path, **kwargs) -> Any:
        """Hook called before loading data"""
        return source_data_path

    def _post_load_hook(self, data, **kwargs) -> Any:
        """Hook called after loading data"""
        return data

    @abstractmethod
    def load_data(self, source_data_path: List[Union[str, Path]], **kwargs) -> Any:
        """
        Load source data for POI processing. This method handles diverse source data formats.

        Args:
            source_data_path: List of source paths
            **kwargs: Additional parameters for data loading

        Returns:
            Data in its source format (DataFrame, GeoDataFrame, TifProcessor, etc.)
        """
        pass

    def process_data(self, data: Any, **kwargs) -> Any:
        """Process the source data to prepare it for POI view generation."""
        return data
        raise NotImplementedError("Subclasses must implement this method...")

    @abstractmethod
    def map_to_poi(
        self,
        processed_data: Any,
        poi_data: Union[pd.DataFrame, gpd.GeoDataFrame],
        **kwargs,
    ) -> Union[pd.DataFrame, gpd.GeoDataFrame]:
        """
        Map processed data to POI data.

        Args:
            processed_data: Processed source data as a GeoDataFrame
            poi_data: POI data to map to
            **kwargs: Additional mapping parameters

        Returns:
            (Geo)DataFrame with POI data mapped to source data
        """
        pass

    def generate_poi_view(
        self,
        poi_data: Union[pd.DataFrame, gpd.GeoDataFrame],
        source_data_path: Optional[Union[Path, str, List[Union[str, Path]]]] = None,
        custom_pipeline: bool = False,
        **kwargs,
    ) -> Union[pd.DataFrame, gpd.GeoDataFrame]:
        """
        Generate a POI view by running the complete pipeline.

        This method has been updated to make source_data_path optional.
        If not provided, it will use resolve_source_paths() to determine paths.

        Args:
            poi_data: POI data to map to
            source_data_path: Optional explicit path(s) to the source data
            **kwargs: Additional parameters for the pipeline

        Returns:
            DataFrame with the generated POI view
        """
        if custom_pipeline:
            return self._custom_pipeline(source_data_path, poi_data, **kwargs)

        self.logger.info("Starting POI view generation pipeline")

        # Resolve source paths if not explicitly provided
        resolved_paths = self.resolve_source_paths(poi_data, source_data_path, **kwargs)

        if not resolved_paths:
            self.logger.warning(
                "No source data paths resolved. Returning original POI data."
            )
            return poi_data

        # load data from resolved sources
        source_data = self.load_data(resolved_paths, **kwargs)
        self.logger.info("Source data loaded successfully")

        # process the data
        processed_data = self.process_data(source_data, **kwargs)
        self.logger.info("Data processing completed")

        # map to POI
        poi_view = self.map_to_poi(processed_data, poi_data, **kwargs)
        self.logger.info("POI mapping completed")

        return poi_view

    def save_poi_view(
        self,
        poi_view: Union[pd.DataFrame, gpd.GeoDataFrame],
        output_path: Union[Path, str],
        **kwargs,
    ) -> None:
        """
        Save the generated POI view to the data store.

        Args:
            poi_view: The POI view DataFrame to save
            output_path: Path where the POI view will be saved in DataStore
            **kwargs: Additional parameters for saving
        """
        self.logger.info(f"Saving POI view to {output_path}")
        write_dataset(poi_view, self.data_store, output_path, **kwargs)

@dataclass
class PoiViewGeneratorConfig:
    """Configuration for POI view generation."""

    base_path: Path = Field(default=config.get_path("poi", "views"))
    n_workers: int = 4

class GhslBuiltSurfacePoiViewGenerator(PoiViewGenerator):
    """Generate POI views from GHSL Built Surface."""

    def __init__(
        self,
        data_config: Optional[GHSLDataConfig] = None,
        generator_config: Optional[PoiViewGeneratorConfig] = None,
        data_store: Optional[DataStore] = None,
    ):
        super().__init__(generator_config=generator_config, data_store=data_store)
        self.data_config = data_config or GHSLDataConfig(
            product="GHS_BUILT_S", year=2020, resolution=100, coord_system=4326
        )

    def _pre_load_hook(self, source_data_path, **kwargs) -> Any:
        """Pre-processing before loading data files."""

        # Convert single path to list for consistent handling
        if isinstance(source_data_path, (Path, str)):
            source_data_path = [source_data_path]

        # change source suffix, .zip, to .tif and paths to string
        source_data_path = [
            str(file_path.with_suffix(".tif")) for file_path in source_data_path
        ]

        # Validate all paths exist
        for file_path in source_data_path:
            if not self.data_store.file_exists(file_path):
                raise RuntimeError(
                    f"Source raster does not exist in the data store: {file_path}"
                )

        self.logger.info(
            f"Pre-loading validation complete for {len(source_data_path)} files"
        )
        return source_data_path

    def _post_load_hook(self, data, **kwargs) -> Any:
        """Post-processing after loading data files."""
        if not data:
            self.logger.warning("No data was loaded from the source files")
            return data

        self.logger.info(
            f"Post-load processing complete. {len(data)} valid TifProcessors."
        )
        return data

    def resolve_source_paths(
        self,
        poi_data: Union[pd.DataFrame, gpd.GeoDataFrame],
        explicit_paths: Optional[Union[Path, str, List[Union[str, Path]]]] = None,
        **kwargs,
    ) -> List[Union[str, Path]]:
        """
        For GHSL Built Surface rasters, resolve source data paths based on POI data geography.

        Returns:
            List of paths to relevant GHSL BUILT S tile rasters
        """
        if explicit_paths is not None:
            if isinstance(explicit_paths, (str, Path)):
                return [explicit_paths]
            return list(explicit_paths)

        # Convert to GeoDataFrame if needed
        if isinstance(poi_data, pd.DataFrame):
            poi_data = convert_to_geodataframe(poi_data)

        # Find intersecting tiles
        intersection_tiles = self.data_config.get_intersecting_tiles(
            geometry=poi_data, crs=poi_data.crs
        )

        if not intersection_tiles:
            self.logger.warning("There are no matching GHSL tiles for the POI data")
            return []

        # Generate paths for each intersecting tile
        source_data_paths = [
            self.data_config.get_tile_path(tile_id=tile) for tile in intersection_tiles
        ]

        self.logger.info(f"Resolved {len(source_data_paths)} tile paths for POI data")
        return source_data_paths

    def load_data(
        self, source_data_path: Union[Path, List[Path]], **kwargs
    ) -> List[TifProcessor]:
        """
        Load GHSL Built Surface rasters into TifProcessors from paths.

        Args:
            source_data_path: Path(s) to the source data
            **kwargs: Additional loading parameters

        Returns:
            List of TifProcessors with built surface data
        """

        processed_paths = self._pre_load_hook(source_data_path, **kwargs)

        tif_processors = [
            TifProcessor(data_path, self.data_store, mode="single")
            for data_path in processed_paths
        ]

        return self._post_load_hook(tif_processors)

    def map_to_poi(
        self,
        processed_data: List[TifProcessor],
        poi_data: Union[pd.DataFrame, gpd.GeoDataFrame],
        map_radius_meters: float,
        **kwargs,
    ) -> pd.DataFrame:
        """
        Map from TifProcessors to POI data.

        Args:
            processed_data: TifProcessors
            poi_data: POI data to map to
            **kwargs: Additional mapping parameters

        Returns:
            DataFrame with POI data and built surface information
        """

        # Convert to GeoDataFrame if needed
        if not isinstance(poi_data, gpd.GeoDataFrame):
            gdf_points = convert_to_geodataframe(poi_data)
        else:
            gdf_points = poi_data

        gdf_points = gdf_points.to_crs(self.data_config.crs)

        polygon_list = buffer_geodataframe(
            gdf_points, buffer_distance_meters=map_radius_meters, cap_style="round"
        ).geometry

        sampled_values = sample_multiple_tifs_by_polygons(
            tif_processors=processed_data, polygon_list=polygon_list, stat="sum"
        )

        poi_data["built_surface_m2"] = sampled_values

        return poi_data

    def generate_view(
        self,
        poi_data: Union[pd.DataFrame, gpd.GeoDataFrame],
        source_data_path: Optional[Union[Path, List[Path]]] = None,
        map_radius_meters: float = 150,
        **kwargs,
    ) -> pd.DataFrame:
        """
        Generate POI view from GHSL Built Surface.

        Returns:
            Enhanced POI data with Built Surface information
        """
        self.logger.info("Generating GHSL Built Surface POI view")

        return self.generate_poi_view(
            poi_data=poi_data,
            source_data_path=source_data_path,
            map_radius_meters=map_radius_meters,
            **kwargs,
        )

class GhslSmodPoiViewGenerator(PoiViewGenerator):
    """Generate POI views from GHSL Settlement Model (SMOD)."""

    def __init__(
        self,
        data_config: Optional[GHSLDataConfig] = None,
        generator_config: Optional[PoiViewGeneratorConfig] = None,
        data_store: Optional[DataStore] = None,
    ):
        super().__init__(generator_config=generator_config, data_store=data_store)
        self.data_config = data_config or GHSLDataConfig(
            product="GHS_SMOD", year=2020, resolution=1000, coord_system=54009
        )

    def _pre_load_hook(self, source_data_path, **kwargs) -> Any:
        """Pre-processing before loading data files."""

        # Convert single path to list for consistent handling
        if isinstance(source_data_path, (Path, str)):
            source_data_path = [source_data_path]

        # change source suffix, .zip, to .tif and paths to string
        source_data_path = [
            str(file_path.with_suffix(".tif")) for file_path in source_data_path
        ]

        # Validate all paths exist
        for file_path in source_data_path:
            if not self.data_store.file_exists(file_path):
                raise RuntimeError(
                    f"Source raster does not exist in the data store: {file_path}"
                )

        self.logger.info(
            f"Pre-loading validation complete for {len(source_data_path)} files"
        )
        return source_data_path

    def _post_load_hook(self, data, **kwargs) -> Any:
        """Post-processing after loading data files."""
        if not data:
            self.logger.warning("No data was loaded from the source files")
            return data

        self.logger.info(
            f"Post-load processing complete. {len(data)} valid TifProcessors."
        )
        return data

    def resolve_source_paths(
        self,
        poi_data: Union[pd.DataFrame, gpd.GeoDataFrame],
        explicit_paths: Optional[Union[Path, str, List[Union[str, Path]]]] = None,
        **kwargs,
    ) -> List[Union[str, Path]]:
        """
        For GHSL SMOD rasters, resolve source data paths based on POI data geography.

        Returns:
            List of paths to relevant GHSL SMOD tile rasters
        """
        if explicit_paths is not None:
            if isinstance(explicit_paths, (str, Path)):
                return [explicit_paths]
            return list(explicit_paths)

        # Convert to GeoDataFrame if needed
        if isinstance(poi_data, pd.DataFrame):
            poi_data = convert_to_geodataframe(poi_data)

        # Find intersecting tiles
        intersection_tiles = self.data_config.get_intersecting_tiles(
            geometry=poi_data, crs=poi_data.crs
        )

        if not intersection_tiles:
            self.logger.warning("There are no matching GHSL tiles for the POI data")
            return []

        # Generate paths for each intersecting tile
        source_data_paths = [
            self.data_config.get_tile_path(tile_id=tile) for tile in intersection_tiles
        ]

        self.logger.info(f"Resolved {len(source_data_paths)} tile paths for POI data")
        return source_data_paths

    def load_data(
        self, source_data_path: Union[Path, List[Path]], **kwargs
    ) -> List[TifProcessor]:
        """
        Load GHSL SMOD rasters into TifProcessors from paths.

        Args:
            source_data_path: Path(s) to the source data
            **kwargs: Additional loading parameters

        Returns:
            List of TifProcessors with settlement model data
        """

        processed_paths = self._pre_load_hook(source_data_path, **kwargs)

        tif_processors = [
            TifProcessor(data_path, self.data_store, mode="single")
            for data_path in processed_paths
        ]

        return self._post_load_hook(tif_processors)

    def map_to_poi(
        self,
        processed_data: List[TifProcessor],
        poi_data: Union[pd.DataFrame, gpd.GeoDataFrame],
        **kwargs,
    ) -> pd.DataFrame:
        """
        Map from TifProcessors to POI data.

        Args:
            processed_data: TifProcessors
            poi_data: POI data to map to
            **kwargs: Additional mapping parameters

        Returns:
            DataFrame with POI data and SMOD classification information
        """

        # Convert to GeoDataFrame if needed
        if not isinstance(poi_data, gpd.GeoDataFrame):
            gdf_points = convert_to_geodataframe(poi_data)
        else:
            gdf_points = poi_data

        gdf_points = gdf_points.to_crs(self.data_config.crs)

        coord_list = [
            (x, y) for x, y in zip(gdf_points["geometry"].x, gdf_points["geometry"].y)
        ]

        sampled_values = sample_multiple_tifs_by_coordinates(
            tif_processors=processed_data, coordinate_list=coord_list
        )

        poi_data["smod_class"] = sampled_values

        return poi_data

    def generate_view(
        self,
        poi_data: Union[pd.DataFrame, gpd.GeoDataFrame],
        source_data_path: Optional[Union[Path, List[Path]]] = None,
        **kwargs,
    ) -> pd.DataFrame:
        """
        Generate POI view from GHSL Settlement Model.

        Returns:
            Enhanced POI data with SMOD classification
        """
        self.logger.info("Generating GHSL Settlement Model POI view")

        return self.generate_poi_view(
            poi_data=poi_data,
            source_data_path=source_data_path,
            **kwargs,
        )