Soil API¶

def get_proximity_entities_2d(
    self, api_url: str, latitude: float, longitude: float, radius: float, **kwargs
) -> SoilApiResult:
    """Find entities within a 2D radius around a geographic point.

    Parameters
    ----------
    api_url : str
        Endpoint suffix for the proximity query.
    latitude : float
        Latitude in decimal degrees.
    longitude : float
        Longitude in decimal degrees.
    radius : float
        Search radius in kilometers.
    **kwargs
        Additional query filters.

    Returns
    -------
    SoilApiResult
        Dictionary with ``data`` and ``exists`` keys.

    Raises
    ------
    InvalidParameterError
        If invalid query values are provided.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> import pandas as pd
    >>> api = SoilAPI(token="dummy")
    >>> with patch.object(SoilAPI, 'process_data', return_value=(pd.DataFrame({'id': [1]}), {'existance': True})):
    ...     out = api.get_proximity_entities_2d('testlocationproximity', 50.0, 2.0, 1.0)
    >>> out['exists']
    True
    """
    geosearch_params = {"latitude": latitude, "longitude": longitude, "offset": radius}
    url_params = {**geosearch_params, **kwargs}
    url_data_type = api_url
    output_type = "list"
    try:
        df, df_add = self.process_data(url_data_type, url_params, output_type)
    except ValueError as err:
        raise InvalidParameterError(str(err)) from err
    for col in df.columns:
        dtype = df[col].dtype
        if isinstance(dtype, pd.Int64Dtype):
            df[col] = df[col].astype("int64")
        elif isinstance(dtype, pd.Float64Dtype):
            df[col] = df[col].astype("float64")
        elif isinstance(dtype, pd.BooleanDtype):
            df[col] = df[col].astype("bool")
        elif isinstance(dtype, pd.StringDtype):
            df[col] = df[col].astype(object)
    return {"data": df, "exists": df_add["existance"]}

def get_closest_entity_2d(
    self,
    api_url: str,
    latitude: float,
    longitude: float,
    radius_init: float = 1.0,
    target_srid: str = "25831",
    **kwargs,
) -> SoilApiResult:
    """Return the nearest entity in 2D for a given endpoint.

    Parameters
    ----------
    api_url : str
        Endpoint suffix for the proximity query.
    latitude : float
        Latitude in decimal degrees.
    longitude : float
        Longitude in decimal degrees.
    radius_init : float, default=1.0
        Initial search radius in kilometers.
    target_srid : str, default="25831"
        SRID used for metric distance calculations.
    **kwargs
        Additional query filters.

    Returns
    -------
    SoilApiResult
        Dictionary with nearest entity metadata and full data table.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> import pandas as pd
    >>> api = SoilAPI(token="dummy")
    >>> sample = pd.DataFrame({'id': [1], 'title': ['A'], 'easting': [2.0], 'northing': [50.0]})
    >>> with patch.object(SoilAPI, '_search_any_entity', return_value=sample):
    ...     out = api.get_closest_entity_2d('test', 50.0, 2.0)
    >>> int(out['id'])
    1
    """
    geosearch_params = {"latitude": latitude, "longitude": longitude}
    url_params = {**geosearch_params, **kwargs}
    df = self._search_any_entity(api_url, radius_init, url_params)
    df, point_east, point_north = SoilDataProcessor.transform_coord(df, longitude, latitude, target_srid)
    df["offset [m]"] = np.sqrt((df["easting [m]"] - point_east) ** 2 + (df["northing [m]"] - point_north) ** 2)
    return cast(SoilApiResult, SoilDataProcessor.gather_data_entity(df))

def get_closest_entity_3d(
    self,
    api_url: str,
    latitude: float,
    longitude: float,
    depth: float,
    radius_init: float = 1.0,
    target_srid: str = "25831",
    sampletest: bool = True,
    **kwargs,
) -> SoilApiResult:
    """Return the nearest entity in 3D for a given endpoint.

    Parameters
    ----------
    api_url : str
        Endpoint suffix for the proximity query.
    latitude : float
        Latitude in decimal degrees.
    longitude : float
        Longitude in decimal degrees.
    depth : float
        Reference depth in meters below seabed.
    radius_init : float, default=1.0
        Initial search radius in kilometers.
    target_srid : str, default="25831"
        SRID used for metric distance calculations.
    sampletest : bool, default=True
        Whether rows are sample tests (uses ``depth`` directly) or samples
        (depth inferred from ``top_depth`` and ``bottom_depth``).
    **kwargs
        Additional query filters.

    Returns
    -------
    SoilApiResult
        Dictionary with nearest entity metadata and full data table.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> import pandas as pd
    >>> api = SoilAPI(token="dummy")
    >>> sample = pd.DataFrame({'id': [1], 'title': ['A'], 'easting': [2.0], 'northing': [50.0], 'depth': [5.0]})
    >>> with patch.object(SoilAPI, '_search_any_entity', return_value=sample):
    ...     out = api.get_closest_entity_3d('test', 50.0, 2.0, depth=5.0)
    >>> int(out['id'])
    1
    """
    geosearch_params = {"latitude": latitude, "longitude": longitude}
    url_params = {**geosearch_params, **kwargs}
    df = self._search_any_entity(api_url, radius_init, url_params)
    df, point_east, point_north = SoilDataProcessor.transform_coord(df, longitude, latitude, target_srid)
    if not sampletest:
        df["depth"] = 0.5 * (df["top_depth"] + df["bottom_depth"])
    df["offset [m]"] = np.sqrt(
        (df["easting [m]"] - point_east) ** 2 + (df["northing [m]"] - point_north) ** 2 + (df["depth"] - depth) ** 2
    )
    return cast(SoilApiResult, SoilDataProcessor.gather_data_entity(df))

def get_surveycampaigns(self, projectsite: Union[str, None] = None, **kwargs) -> SoilApiResult:
    """Retrieve survey campaigns, optionally filtered by project site.

    Parameters
    ----------
    projectsite : str or None, default=None
        Project site name.
    **kwargs
        Additional query filters.

    Returns
    -------
    SoilApiResult
        Dictionary with ``data`` and ``exists`` keys.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> import pandas as pd
    >>> api = SoilAPI(token="dummy")
    >>> payload = (
    ...     pd.DataFrame({'title': ['C1']}),
    ...     {'existance': True, 'id': None, 'response': None},
    ... )
    >>> with patch.object(SoilAPI, 'process_data', return_value=payload):
    ...     out = api.get_surveycampaigns(projectsite='Demo')
    >>> out['exists']
    True
    """
    url_params = {"projectsite": projectsite}
    url_params = {**url_params, **kwargs}
    url_data_type = "surveycampaign"
    output_type = "list"
    df, df_add = self.process_data(url_data_type, url_params, output_type)
    return {"data": df, "exists": df_add["existance"]}

def get_surveycampaign_detail(
    self,
    projectsite: str,
    campaign: str,
    **kwargs,
) -> SoilApiResult:
    """Retrieve details for a single survey campaign.

    Parameters
    ----------
    projectsite : str
        Project site name.
    campaign : str
        Survey campaign title.
    **kwargs
        Additional query filters.

    Returns
    -------
    SoilApiResult
        Dictionary with ``id``, ``data`` and ``exists`` keys.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> import pandas as pd
    >>> api = SoilAPI(token="dummy")
    >>> payload = (pd.DataFrame({'title': ['C1']}), {'existance': True, 'id': 3, 'response': None})
    >>> with patch.object(SoilAPI, 'process_data', return_value=payload):
    ...     out = api.get_surveycampaign_detail('P', 'C1')
    >>> int(out['id'])
    3
    """
    url_params = {"projectsite": projectsite, "campaign": campaign}
    url_params = {**url_params, **kwargs}
    url_data_type = "surveycampaign"
    output_type = "single"
    df, df_add = self.process_data(url_data_type, url_params, output_type)
    return {"id": df_add["id"], "data": df, "exists": df_add["existance"]}

def get_proximity_testlocations(self, latitude: float, longitude: float, radius: float, **kwargs) -> SoilApiResult:
    """Retrieve test locations within a radius around a coordinate.

    Parameters
    ----------
    latitude : float
        Latitude in decimal degrees.
    longitude : float
        Longitude in decimal degrees.
    radius : float
        Radius in kilometers.
    **kwargs
        Additional query filters.

    Returns
    -------
    SoilApiResult
        Dictionary with ``data`` and ``exists`` keys.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> import pandas as pd
    >>> api = SoilAPI(token="dummy")
    >>> prox = {'data': pd.DataFrame(), 'exists': False}
    >>> with patch.object(SoilAPI, 'get_proximity_entities_2d', return_value=prox):
    ...     out = api.get_proximity_testlocations(50.0, 2.0, 1.0)
    >>> out['exists']
    False
    """
    return self.get_proximity_entities_2d(
        api_url="testlocationproximity",
        latitude=latitude,
        longitude=longitude,
        radius=radius,
        **kwargs,
    )

def get_closest_testlocation(
    self,
    latitude: float,
    longitude: float,
    radius: float = 1.0,
    target_srid: str = "25831",
    **kwargs,
) -> SoilApiResult:
    """Retrieve the closest test location around a coordinate.

    Parameters
    ----------
    latitude : float
        Latitude in decimal degrees.
    longitude : float
        Longitude in decimal degrees.
    radius : float, default=1.0
        Initial search radius in kilometers.
    target_srid : str, default="25831"
        SRID for metric distance calculations.
    **kwargs
        Additional query filters.

    Returns
    -------
    SoilApiResult
        Dictionary including selected ``id``, ``title`` and ``offset [m]``.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> import pandas as pd
    >>> api = SoilAPI(token="dummy")
    >>> closest = {'data': pd.DataFrame(), 'id': 1, 'title': 'A', 'offset [m]': 0.0}
    >>> with patch.object(SoilAPI, 'get_closest_entity_2d', return_value=closest):
    ...     out = api.get_closest_testlocation(50.0, 2.0)
    >>> out['id']
    1
    """
    return self.get_closest_entity_2d(
        api_url="testlocationproximity",
        latitude=latitude,
        longitude=longitude,
        radius_init=radius,
        target_srid=target_srid,
        **kwargs,
    )

def get_testlocations(
    self,
    projectsite: Union[str, None] = None,
    campaign: Union[str, None] = None,
    **kwargs,
) -> SoilApiResult:
    """Retrieve geotechnical test locations matching search filters.

    Parameters
    ----------
    projectsite : str or None, default=None
        Project site filter.
    campaign : str or None, default=None
        Survey campaign filter.
    **kwargs
        Additional queryset filters.

    Returns
    -------
    SoilApiResult
        Dictionary with ``data`` and ``exists`` keys.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> import pandas as pd
    >>> api = SoilAPI(token="dummy")
    >>> payload = (pd.DataFrame({'title': ['L1']}), {'existance': True, 'id': None, 'response': None})
    >>> with patch.object(SoilAPI, 'process_data', return_value=payload):
    ...     out = api.get_testlocations(projectsite='P')
    >>> out['exists']
    True
    """
    url_params = {
        "projectsite": projectsite,
        "campaign": campaign,
    }
    url_params = {**url_params, **kwargs}
    url_data_type = "testlocation"
    output_type = "list"
    df, df_add = self.process_data(url_data_type, url_params, output_type)
    return {"data": df, "exists": df_add["existance"]}

def get_testlocation_detail(
    self,
    location: str,
    projectsite: Union[str, None] = None,
    campaign: Union[str, None] = None,
    **kwargs,
) -> SoilApiResult:
    """Retrieve details for a specific geotechnical test location.

    Parameters
    ----------
    location : str
        Location title.
    projectsite : str or None, default=None
        Project site filter.
    campaign : str or None, default=None
        Campaign filter.
    **kwargs
        Additional queryset filters.

    Returns
    -------
    SoilApiResult
        Dictionary with ``id``, ``data`` and ``exists`` keys.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> import pandas as pd
    >>> api = SoilAPI(token="dummy")
    >>> payload = (pd.DataFrame({'title': ['L1']}), {'existance': True, 'id': 10, 'response': None})
    >>> with patch.object(SoilAPI, 'process_data', return_value=payload):
    ...     out = api.get_testlocation_detail('L1')
    >>> int(out['id'])
    10
    """
    url_params = {
        "projectsite": projectsite,
        "campaign": campaign,
        "location": location,
    }
    url_params = {**url_params, **kwargs}
    url_data_type = "testlocation"
    output_type = "single"
    df, df_add = self.process_data(url_data_type, url_params, output_type)
    return {"id": df_add["id"], "data": df, "exists": df_add["existance"]}

def testlocation_exists(
    self,
    location: str,
    projectsite: Union[str, None] = None,
    campaign: Union[str, None] = None,
    **kwargs,
) -> SoilIdResult:
    """Check whether a test location exists and return its id.

    Parameters
    ----------
    location : str
        Location title.
    projectsite : str or None, default=None
        Project site filter.
    campaign : str or None, default=None
        Campaign filter.
    **kwargs
        Additional queryset filters.

    Returns
    -------
    int | numpy.int64 | bool | None
        Location id when found; otherwise ``False``.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> api = SoilAPI(token="dummy")
    >>> payload = (None, {'existance': True, 'id': 4, 'response': None})
    >>> with patch.object(SoilAPI, 'process_data', return_value=payload):
    ...     out = api.testlocation_exists('L1')
    >>> int(out)
    4
    """
    url_params = {
        "projectsite": projectsite,
        "campaign": campaign,
        "location": location,
    }
    url_params = {**url_params, **kwargs}
    url_data_type = "testlocation"
    output_type = "single"
    _, df_add = self.process_data(url_data_type, url_params, output_type)
    return df_add["id"] if df_add["existance"] else False

def get_insitutest_types(self, **kwargs):
    """Retrieve available in-situ test types.

    Parameters
    ----------
    **kwargs
        Additional queryset filters.

    Returns
    -------
    SoilApiResult
        Dictionary with available test type rows and existence flag.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> import pandas as pd
    >>> api = SoilAPI(token="dummy")
    >>> payload = (
    ...     pd.DataFrame({'title': ['PCPT']}),
    ...     {'existance': True, 'id': None, 'response': None},
    ... )
    >>> with patch.object(SoilAPI, 'process_data', return_value=payload):
    ...     out = api.get_insitutest_types()
    >>> out['exists']
    True
    """
    url_data_type = "insitutesttype"
    output_type = "list"
    df, df_add = self.process_data(url_data_type, {}, output_type)
    return {"data": df, "exists": df_add["existance"]}

def insitutest_type_exists(self, testtype: str, **kwargs) -> SoilIdResult:
    """Check whether an in-situ test type exists and return its id.

    Parameters
    ----------
    testtype : str
        In-situ test type title.
    **kwargs
        Additional queryset filters.

    Returns
    -------
    int | numpy.int64 | bool | None
        Type id when found; otherwise ``False``.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> api = SoilAPI(token="dummy")
    >>> payload = (None, {'existance': False, 'id': None, 'response': None})
    >>> with patch.object(SoilAPI, 'process_data', return_value=payload):
    ...     out = api.insitutest_type_exists('PCPT')
    >>> out
    False
    """
    url_params = {"testtype": testtype}
    url_params = {**url_params, **kwargs}
    url_data_type = "insitutesttype"
    output_type = "single"
    _, df_add = self.process_data(url_data_type, url_params, output_type)
    return df_add["id"] if df_add["existance"] else False

def get_insitutests(
    self,
    projectsite: Union[str, None] = None,
    location: Union[str, None] = None,
    testtype: Union[str, None] = None,
    insitutest: Union[str, None] = None,
    **kwargs,
) -> SoilApiResult:
    """Retrieve in-situ tests matching provided filters.

    Parameters
    ----------
    projectsite : str or None, default=None
        Project site filter.
    location : str or None, default=None
        Test location filter.
    testtype : str or None, default=None
        In-situ test type filter.
    insitutest : str or None, default=None
        In-situ test title filter.
    **kwargs
        Additional queryset filters.

    Returns
    -------
    SoilApiResult
        Dictionary with in-situ test summary rows and existence flag.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> import pandas as pd
    >>> api = SoilAPI(token="dummy")
    >>> payload = (pd.DataFrame({'title': ['CPT-1']}), {'existance': True, 'id': None, 'response': None})
    >>> with patch.object(SoilAPI, 'process_data', return_value=payload):
    ...     out = api.get_insitutests(projectsite='P')
    >>> out['exists']
    True
    """
    url_params = {
        "projectsite": projectsite,
        "location": location,
        "testtype": testtype,
        "insitutest": insitutest,
    }
    url_params = {**url_params, **kwargs}
    url_data_type = "insitutestsummary"
    output_type = "list"
    df, df_add = self.process_data(url_data_type, url_params, output_type)
    return {"data": df, "exists": df_add["existance"]}

def get_proximity_insitutests(self, latitude: float, longitude: float, radius: float, **kwargs) -> SoilApiResult:
    """Retrieve in-situ tests in a radius around a coordinate.

    Parameters
    ----------
    latitude : float
        Latitude in decimal degrees.
    longitude : float
        Longitude in decimal degrees.
    radius : float
        Radius in kilometers.
    **kwargs
        Additional queryset filters.

    Returns
    -------
    SoilApiResult
        Dictionary with in-situ test rows and existence flag.

    Examples
    --------
    >>> from unittest.mock import patch
    >>> import pandas as pd
    >>> api = SoilAPI(token="dummy")
    >>> prox = {'data': pd.DataFrame(), 'exists': False}
    >>> with patch.object(SoilAPI, 'get_proximity_entities_2d', return_value=prox):
    ...     out = api.get_proximity_insitutests(50.0, 2.0, 1.0)
    >>> out['exists']
    False
    """
    return self.get_proximity_entities_2d(
        api_url="insitutestproximity",
        latitude=latitude,
        longitude=longitude,
        radius=radius,
        **kwargs,
    )

def get_closest_insitutest(
    self,
    latitude: float,
    longitude: float,
    radius: float = 1.0,
    target_srid: str = "25831",
    **kwargs,
):
    """Return the closest in-situ test for a coordinate.

    Parameters
    ----------
    latitude : float
        Latitude in decimal degrees.
    longitude : float
        Longitude in decimal degrees.
    radius : float, default=1.0
        Initial search radius in kilometers.
    target_srid : str, default="25831"
        SRID for metric distance calculations.
    **kwargs
        Additional queryset filters.

    Returns
    -------
    SoilApiResult
        Dictionary with selected test metadata and candidate data.

    Examples
    --------
    >>> api = SoilAPI(token="dummy")
    >>> api.get_closest_insitutest(50.0, 2.0)  # doctest: +SKIP
    """
    return self.get_closest_entity_2d(
        api_url="insitutestproximity",
        latitude=latitude,
        longitude=longitude,
        radius_init=radius,
        target_srid=target_srid,
        **kwargs,
    )