Source code for shapelib.core.handlers.shapefile_handler

from pathlib import Path
from typing import TYPE_CHECKING, Any

import fiona
import geopandas as gpd

from shapelib.core.handlers.geo_file_handler import GeoFileHandler

if TYPE_CHECKING:
    from shapelib.core.geometries.base import BaseShapes
from shapelib.factories.registry import register_handler




@register_handler(file_extension=".shp")
class ShapefileHandler(GeoFileHandler):
    """
    A handler for ESRI Shapefile (.shp) format.

    This class provides functionality for reading, validating, and managing shapefiles
    using the Geopandas library. It attempts to load shapefiles with different encodings
    and handles exceptions related to file decoding.

    Attributes
    ----------
    driver : str
        The driver used for handling shapefiles ("ESRI Shapefile").
    extension : str
        The file extension associated with shapefiles (".shp").
    encoding_list : list of str
        A list of possible encodings to try when loading the shapefile.

    Methods
    -------
    load_data()
        Loads the shapefile into a GeoDataFrame, trying different encodings.
    validate_file()
        Validates the shapefile to ensure it can be read.
    create_obj()
        Creates and returns a geometry object based on the shapefile's contents.
    """

    driver = "ESRI Shapefile"
    extension = ".shp"
    encoding_list = ["utf-8", "iso8859_15", "latin1", "cp1252", "ascii"]
    engines = ["pyogrio", "fiona"]

    def __init__(self, path: Path) -> None:
        """
        Initialize the ShapefileHandler with a given shapefile path.

        This method sets the file path, validates the shapefile, and loads the data
        using the `load_data` method.

        Parameters
        ----------
        path : Path
            The path to the shapefile (.shp) to be processed.

        Raises
        ------
        FileNotFoundError
            If the provided shapefile does not exist or is not a valid file.
        """
        super().__init__(path)
        self.validate_file()



    def load_data(self, **kwargs: Any) -> gpd.GeoDataFrame:
        """
        Attempt to load the shapefile using different encodings.

        This method iterates over the encodings in `encoding_list` and attempts to read
        the shapefile using each encoding. If successful, the file is returned as a
        GeoDataFrame.

        Returns
        -------
        gpd.GeoDataFrame
            The loaded shapefile as a GeoDataFrame.

        Raises
        ------
        ValueError
            If none of the encodings are able to decode the file.

        Examples
        --------
        >>> handler = ShapefileHandler(path=Path("/path/to/shapefile.shp"))
        >>> gdf = handler.load_data()
        """
        for engine in self.engines:
            for encoding in self.encoding_list:
                try:
                    return gpd.read_file(self.path, encoding=encoding, engine=engine, **kwargs)
                except UnicodeDecodeError:
                    # If the current encoding fails, try the next one
                    continue
        # If all encodings fail and all engines fail, raise a ValueError
        raise ValueError(
            f"Failed to read the shapefile with the encodings: {self.encoding_list} andengines: {self.engines}"
        )




    def validate_file(self) -> None:
        """
        Validate the shapefile to ensure it exists and is readable.

        This method checks whether the shapefile and its associated files (e.g., .shx,
        .dbf) exist and can be read by GDAL.

        Raises
        ------
        FileNotFoundError
            If the shapefile or one of its associated files (.shx, .dbf) is missing.
        GDALException
            If GDAL cannot read the shapefile or if the shapefile is invalid.
        """
        self.check_empty_file()
        self.check_missing_parts()




    def check_empty_file(self) -> None:
        """
        Check if the shapefile is empty based on its file size.

        This method checks if the shapefile is empty by evaluating its file size.
        Shapefiles that are exactly 100 bytes in size are considered empty.

        Returns
        -------
        None
            The method prints a message if the file is empty.

        Examples
        --------
        >>> handler.check_empty_file()
        Empty Shapefile!
        """
        if self.path.stat().st_size == 100:  # The size of an empty shapefile
            print("Empty Shapefile!")




    def check_missing_parts(self) -> None:
        """
        Check for the existence of essential auxiliary files associated with the
        shapefile.

        This method checks if the following auxiliary files are missing:
        - .shx (shape index file)
        - .cpg (code page file for character encoding)
        - .dbf (attribute data file)
        - .prj (projection file)

        If any of these files are missing, the corresponding handler method is invoked
        to handle the missing file.

        Returns
        -------
        None
            The method invokes the handler methods if files are missing.

        Examples
        --------
        >>> handler.check_missing_parts()
        """
        if not self.path.with_suffix(suffix=".shx").exists():
            self.handle_missing_shx()

        if not self.path.with_suffix(suffix=".cpg").exists():
            self.handle_missing_cpg()

        if not self.path.with_suffix(suffix=".dbf").exists():
            self.handle_missing_dbf()

        if not self.path.with_suffix(suffix=".prj").exists():
            self.handle_missing_prj()




    def handle_missing_shx(self) -> None:
        """
        Attempt to recreate the missing .shx file using Fiona.

        This method uses GDAL's (via Fiona) to recreate the missing .shx file
        associated with the shapefile. GDAL's ESRI Shapefile driver can regenerate
        the .shx when the configuration option SHAPE_RESTORE_SHX=YES is set.

        Raises
        ------
        RuntimeError
            If the .shx file cannot be recreated, a RuntimeError is raised.

        Examples
        --------
        >>> handler.handle_missing_shx()
        Failed to recreate .shx file: [error message]
        """
        target_shx = self.path.with_suffix(".shx")
        try:
            with fiona.Env(SHAPE_RESTORE_SHX="YES"), fiona.open(self.path, "r") as _src:  # noqa: F841
                pass

            if not target_shx.exists():
                raise RuntimeError("GDAL failed to regenerate the .shx index.")
        except Exception as e:
            raise RuntimeError(f"Failed to recreate .shx file: {e}") from e




    def handle_missing_cpg(self) -> None:
        """
        Handle the missing .cpg file by notifying the user.

        The .cpg file specifies character encoding for the shapefile. While it is useful
        for ensuring correct character display, it is not mandatory for the core
        functionality of the shapefile. This method simply warns the user if the .cpg
        file is missing.

        Returns
        -------
        None
            The method prints a warning message if the .cpg file is missing.

        Examples
        --------
        >>> handler.handle_missing_cpg()
        Missing .cpg. Ignoring it since it's not mandatory for the functionality of the
        Shapefile.
        """
        print(
            "Missing .cpg. Ignoring it since it's not mandatory for the functionality of \
                the Shapefile."
        )




    def handle_missing_dbf(self) -> None:
        """
        Handle the missing .dbf file by notifying the user.

        The .dbf file holds attribute data (e.g., table data associated with the
        shapefile). While this information is useful, its absence does not prevent the
        shapefile from functioning. This method prints a warning message if the .dbf file
        is missing.

        Returns
        -------
        None
            The method prints a warning message if the .dbf file is missing.

        Examples
        --------
        >>> handler.handle_missing_dbf()
        Missing .dbf. Ignoring it since it's not mandatory for the functionality of the
        Shapefile.
        """
        print("Missing .dbf. Ignoring it since it's not mandatory for the functionality of the Shapefile.")




    def handle_missing_prj(self) -> None:
        """
        Handle the missing .prj file by notifying the user.

        The .prj file holds the shapefile's CRS. Some producers leave it out (e.g.
        prescription maps exported for agricultural machinery), so its absence is not
        fatal here: the data is read without a CRS, and `create_geometry_instance`
        assumes WGS84 when the coordinates are lon/lat or raises a `ValueError`
        otherwise.

        Returns
        -------
        None
            The method prints a notice if the .prj file is missing.

        Examples
        --------
        >>> handler.handle_missing_prj()
        Missing .prj. The CRS will be inferred from the coordinates (WGS84 if they are lon/lat).
        """
        print("Missing .prj. The CRS will be inferred from the coordinates (WGS84 if they are lon/lat).")




    def create_obj(self, **kwargs: Any) -> "BaseShapes":
        """
        Create an object representing the shapefile's geometries based on the most common
        type.

        This method analyzes the most common geometry type in the shapefile and returns an
        appropriate object (`Polygons`, `Lines`, `Points`, or `Empty`). If the geometry
        type is not recognized, a `TypeError` may be raised.

        Returns
        -------
        object
            A geometry object corresponding to the shapefile's data, such as `Polygons`,
            `Lines`, `Points`, or `Empty`.

        Raises
        ------
        TypeError
            If an unexpected geometry type is encountered.

        Examples
        --------
        >>> geom_obj = handler.create_obj()
        >>> isinstance(geom_obj, Polygons)
        True
        """
        data = self.load_data(**kwargs)
        return self.create_geometry_instance(data=data, driver=self.driver)