Locations Module¶

def __eq__(self, other: object) -> bool:
    """
    Compare two instances of the API class.

    Parameters
    ----------
    other : object
        Another instance of the API class or a dictionary.

    Returns
    -------
    bool
        True if the instances are equal, False otherwise.

    Raises
    ------
    AssertionError
        If comparison is not possible due to incompatible types.

    Examples
    --------
    >>> api_1 = API(api_root="https://example", token="test")
    >>> api_2 = API(api_root="https://example", token="test")
    >>> api_1 == api_2
    True
    """
    if not isinstance(other, (API, dict)):
        return NotImplemented
    if isinstance(other, type(self)):
        comp = deepcompare(self, other)
        assert comp[0], comp[1]
    elif isinstance(other, dict):
        comp = deepcompare(self.__dict__, other)
        assert comp[0], comp[1]
    else:
        raise AssertionError("Comparison is not possible due to incompatible types!")
    return comp[0]

def send_request(
    self,
    url_data_type: str,
    url_params: Mapping[str, Union[str, float, int, Sequence[Union[str, float, int]], None]],
) -> requests.Response:
    """
    Handle sending appropriate request.

    Handles sending appropriate request according to the type of
    authentication.

    Parameters
    ----------
    url_data_type : str
        Type of the data we want to request (according to database
        model).
    url_params : Mapping
        Parameters to send with the request to the database.

    Returns
    -------
    requests.Response
        An instance of the Response object.

    Raises
    ------
    InvalidParameterError
        If neither header nor username and password are defined.

    Examples
    --------
    >>> from types import SimpleNamespace
    >>> from unittest import mock
    >>> response = SimpleNamespace(status_code=200, reason="OK")
    >>> with mock.patch("requests.get", return_value=response):
    ...     api = API(api_root="https://example", token="test")
    ...     resp = api.send_request("/projects", {})
    >>> resp is response
    True
    """
    if self.header is not None:
        response = requests.get(
            url=self.api_root + url_data_type,
            headers=self.header,
            params=url_params,
        )
    else:
        if self.uname is None or self.password is None:
            e = "Either self.header or self.uname and self.password must be defined."
            raise InvalidParameterError(e)
        else:
            response = requests.get(
                url=self.api_root + url_data_type,
                auth=self.auth,
                params=url_params,
            )
    return response

@staticmethod
def check_request_health(resp: requests.Response) -> None:
    """
    Check status code of the response and provide details.

    Checks status code of the response to request and provides
    details if unexpected.

    Parameters
    ----------
    resp : requests.Response
        Instance of the Response object.

    Raises
    ------
    APIConnectionError
        If response status code is not 200.

    Examples
    --------
    >>> from types import SimpleNamespace
    >>> ok = SimpleNamespace(status_code=200, reason="OK")
    >>> API.check_request_health(ok)
    """
    if resp.status_code != 200:
        message = f"Error {resp.status_code}.\n{resp.reason}\n{resp.text}"
        raise APIConnectionError(message=message, response=resp)

@staticmethod
def output_to_df(response: requests.Response) -> pd.DataFrame:
    """
    Transform output to Pandas dataframe.

    Parameters
    ----------
    response : requests.Response
        Raw output of the sent request.

    Returns
    -------
    pd.DataFrame
        Pandas dataframe of the data from the output.

    Raises
    ------
    DataProcessingError
        If failed to decode JSON from API response.

    Examples
    --------
    >>> from types import SimpleNamespace
    >>> resp = SimpleNamespace(text='[{"a": 1}]')
    >>> int(API.output_to_df(resp)["a"].iloc[0])
    1
    """
    try:
        data = json.loads(response.text)
    except Exception as err:
        raise DataProcessingError("Failed to decode JSON from API response") from err
    return pd.DataFrame(data)

@staticmethod
def postprocess_data(df: pd.DataFrame, output_type: str) -> PostprocessData:
    """
    Process dataframe information to extract additional data.

    Processes dataframe information to extract the necessary
    additional data.

    Parameters
    ----------
    df : pd.DataFrame
        Dataframe of the output data.
    output_type : str
        Expected type (amount) of the data extracted.

    Returns
    -------
    PostprocessData
        Dictionary of the additional data extracted.

    Raises
    ------
    InvalidParameterError
        If more than one record was returned for 'single' output
        type, or if output type is not 'single' or 'list'.

    Examples
    --------
    >>> df = pd.DataFrame({"id": [1]})
    >>> int(API.postprocess_data(df, "single")["id"])
    1
    """
    if output_type == "single":
        if df.__len__() == 0:
            exists = False
            project_id = None
        elif df.__len__() == 1:
            exists = True
            project_id = df["id"].iloc[0]
        else:
            raise InvalidParameterError("More than one project site was returned, check search criteria.")
        data_add: PostprocessData = {
            "existance": exists,
            "id": project_id,
            "response": None,
        }
    elif output_type == "list":
        exists = df.__len__() != 0
        data_add: PostprocessData = {
            "existance": exists,
            "id": None,
            "response": None,
        }
    else:
        raise InvalidParameterError("Output type must be either 'single' or 'list', not " + output_type + ".")
    return data_add

@staticmethod
def validate_data(df: pd.DataFrame, data_type: str) -> pd.DataFrame:
    """
    Validate the data extracted from the database.

    Parameters
    ----------
    df : pd.DataFrame
        Dataframe of the output data.
    data_type : str
        Type of the data we want to request (according to database
        model).

    Returns
    -------
    pd.DataFrame
        Dataframe with corrected data.

    Examples
    --------
    >>> df = pd.DataFrame()
    >>> API.validate_data(df, "subassemblies").empty
    True
    """
    z_sa_mp = {"min": -100000, "max": -10000}
    z_sa_tp = {"min": -20000, "max": -1000}
    z_sa_tw = {"min": 1000, "max": 100000}
    sa_type = ["TW", "TP", "MP"]
    z = [z_sa_tw, z_sa_tp, z_sa_mp]
    if data_type == "subassemblies":
        if df.__len__() == 0:
            return df
        for i, sat in enumerate(sa_type):
            cond_small_units = (df["subassembly_type"] == sat) & (df["z_position"] < z[i]["min"])
            cond_big_units = (df["subassembly_type"] == sat) & (df["z_position"] > z[i]["max"])
            if df[cond_small_units].__len__() > 0:
                df.loc[cond_small_units, "z_position"] = df.loc[cond_small_units, "z_position"] / 1e3
                warnings.warn(
                    f"The value of z location for {df.loc[cond_small_units | cond_big_units, 'title'].values} \
                    might be wrong or in wrong units! There will be an attempt to correct the units.",
                    stacklevel=2,
                )
            if df[cond_big_units].__len__() > 0:
                df.loc[cond_big_units, "z_position"] = df.loc[cond_big_units, "z_position"] * 1e3
                warnings.warn(
                    f"The value of z location for {df.loc[cond_small_units | cond_big_units, 'title'].values} \
                    might be wrong or in wrong units! There will be an attempt to correct the units.",
                    stacklevel=2,
                )
    return df

def process_data(
    self,
    url_data_type: str,
    url_params: Mapping[str, Union[str, float, int, Sequence[Union[str, float, int]], None]],
    output_type: str,
) -> tuple[pd.DataFrame, PostprocessData]:
    """
    Process output data according to specified request parameters.

    Parameters
    ----------
    url_data_type : str
        Type of the data we want to request (according to database
        model).
    url_params : Mapping
        Parameters to send with the request to the database.
    output_type : str
        Expected type (amount) of the data extracted.

    Returns
    -------
    tuple
        A tuple of dataframe with the requested data and
        additional data from postprocessing.

    Examples
    --------
    >>> from types import SimpleNamespace
    >>> from unittest import mock
    >>> response = SimpleNamespace(text="[]", status_code=200, reason="OK")
    >>> api = API(api_root="https://example", token="test")
    >>> with mock.patch.object(API, "send_request", return_value=response):
    ...     df, info = api.process_data("projects", {}, "list")
    >>> df.empty
    True
    >>> info["existance"]
    False
    """
    resp = self.send_request(url_data_type, url_params)
    self.check_request_health(resp)
    df = self.output_to_df(resp)
    df = self.validate_data(df, url_data_type)
    df_add = self.postprocess_data(df, output_type)
    # Add the response object to the returned dictionary so tests can inspect it
    df_add["response"] = resp
    return df, df_add

def get_projectsites(self, **kwargs: Any) -> dict[str, Union[pd.DataFrame, bool, np.int64, None]]:
    """
    Get all available projects.

    Parameters
    ----------
    **kwargs
        Additional parameters to pass to the API.

    Returns
    -------
    dict
        Dictionary with the following keys:

        - "data": Pandas dataframe with the location data for each
          project
        - "exists": Boolean indicating whether matching records
          are found

    Examples
    --------
            >>> from unittest import mock
            >>> api = LocationsAPI(
            ...     api_root="https://example",
            ...     header={"Authorization": "Token test"},
            ... )
            >>> df = pd.DataFrame({"id": [1]})
            >>> with mock.patch.object(
            ...     LocationsAPI,
            ...     "process_data",
            ...     return_value=(df, {"existance": True}),
            ... ):
            ...     out = api.get_projectsites()
            >>> out["exists"]
            True
    """
    url_params = {}  # type: dict[str, str]
    url_params = {**url_params, **kwargs}
    url_data_type = "projectsites"
    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_projectsite_detail(
    self, projectsite: str, **kwargs: Any
) -> dict[str, Union[pd.DataFrame, bool, np.int64, None]]:
    """
    Get details for a specific projectsite.

    Parameters
    ----------
    projectsite : str
        Title of the projectsite.
    **kwargs
        Additional parameters to pass to the API.

    Returns
    -------
    dict
        Dictionary with the following keys:

        - "id": ID of the selected project site.
        - "data": Pandas dataframe with the location data for each
          projectsite.
        - "exists": Boolean indicating whether matching records
          are found.

    Examples
    --------
            >>> from unittest import mock
            >>> api = LocationsAPI(
            ...     api_root="https://example",
            ...     header={"Authorization": "Token test"},
            ... )
            >>> df = pd.DataFrame({"id": [1]})
            >>> with mock.patch.object(
            ...     LocationsAPI,
            ...     "process_data",
            ...     return_value=(df, {"existance": True, "id": 1}),
            ... ):
            ...     out = api.get_projectsite_detail("Site")
            >>> out["id"]
            1
    """
    url_params = {"projectsite": projectsite}
    url_params = {**url_params, **kwargs}
    url_data_type = "projectsites"
    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_assetlocations(
    self, projectsite: Union[str, None] = None, **kwargs: Any
) -> dict[str, Union[pd.DataFrame, bool, list[bool], np.int64, None]]:
    """
    Get all available asset locations.

    Get all available asset locations for all projectsites or a
    specific projectsite.

    Parameters
    ----------
    projectsite : str, optional
        String with the projectsite title (e.g. "Nobelwind").
    **kwargs
        Additional parameters to pass to the API.

    Returns
    -------
    dict
        Dictionary with the following keys:

        - "data": Pandas dataframe with the location data for each
          location in the projectsite.
        - "exists": Boolean indicating whether matching records
          are found.

    Examples
    --------
            >>> from unittest import mock
            >>> api = LocationsAPI(
            ...     api_root="https://example",
            ...     header={"Authorization": "Token test"},
            ... )
            >>> df = pd.DataFrame({"id": [1]})
            >>> with mock.patch.object(
            ...     LocationsAPI,
            ...     "process_data",
            ...     return_value=(df, {"existance": True}),
            ... ):
            ...     out = api.get_assetlocations(projectsite="Site")
            >>> out["exists"]
            True
    """
    url_params = {}  # type: dict[str, str]
    url_params = {**url_params, **kwargs}
    if projectsite:
        url_params["projectsite__title"] = projectsite
    url_data_type = "assetlocations"
    if "assetlocations" in url_params and isinstance(url_params["assetlocations"], list):
        df = []
        df_add = {"existance": []}
        for assetlocation in url_params["assetlocations"]:
            output_type = "single"
            url_params["assetlocation"] = assetlocation
            df_temp, df_add_temp = self.process_data(url_data_type, url_params, output_type)
            df.append(df_temp)
            df_add["existance"].append(df_add_temp["existance"])
        df = pd.concat(df)
    else:
        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_assetlocation_detail(
    self,
    assetlocation: str,
    projectsite: Union[None, str] = None,
    **kwargs: Any,
) -> dict[str, Union[pd.DataFrame, bool, np.int64, None]]:
    """
    Get a selected turbine.

    Parameters
    ----------
    assetlocation : str
        Title of the asset location (e.g. "BBK05").
    projectsite : str, optional
        Name of the projectsite (e.g. "Nobelwind").
    **kwargs
        Additional parameters to pass to the API.

    Returns
    -------
    dict
        Dictionary with the following keys:

        - "id": ID of the selected projectsite site.
        - "data": Pandas dataframe with the location data for the
          individual location.
        - "exists": Boolean indicating whether a matching location
          is found.

    Examples
    --------
            >>> from unittest import mock
            >>> api = LocationsAPI(
            ...     api_root="https://example",
            ...     header={"Authorization": "Token test"},
            ... )
            >>> df = pd.DataFrame({"id": [1]})
            >>> with mock.patch.object(
            ...     LocationsAPI,
            ...     "process_data",
            ...     return_value=(df, {"existance": True, "id": 1}),
            ... ):
            ...     out = api.get_assetlocation_detail("T01")
            >>> out["id"]
            1
    """
    if projectsite is None:
        url_params = {"assetlocation": assetlocation}
    else:
        url_params = {
            "projectsite": projectsite,
            "assetlocation": assetlocation,
        }
    url_params = {**url_params, **kwargs}
    url_data_type = "assetlocations"
    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 plot_assetlocations(self, return_fig: bool = False, **kwargs: Any) -> Union[go.Figure, None]:
    """
    Retrieve asset locations and generate a Plotly plot.

    Retrieves asset locations and generates a Plotly plot to show
    them.

    Parameters
    ----------
    return_fig : bool, optional
        Boolean indicating whether to return the figure, default
        is False.
    **kwargs
        Keyword arguments for the search (see
        ``get_assetlocations``).

    Returns
    -------
    plotly.graph_objects.Figure or None
        Plotly figure object with selected asset locations plotted
        on OpenStreetMap tiles (if requested) or nothing.

    Raises
    ------
    ValueError
        If no asset locations found for the given parameters.

    Examples
    --------
    >>> from unittest import mock
    >>> api = LocationsAPI(
    ...     api_root="https://example",
    ...     header={"Authorization": "Token test"},
    ... )
    >>> data = pd.DataFrame(
    ...     {
    ...         "northing": [51.5],
    ...         "easting": [2.8],
    ...         "title": ["T01"],
    ...         "projectsite_name": ["Site"],
    ...         "description": [""],
    ...     }
    ... )
    >>> with mock.patch.object(
    ...     LocationsAPI,
    ...     "get_assetlocations",
    ...     return_value={"exists": True, "data": data},
    ... ):
    ...     fig = api.plot_assetlocations(return_fig=True)
    >>> fig is not None
    True
    """
    assetlocations_data = self.get_assetlocations(**kwargs)
    if assetlocations_data["exists"]:
        assetlocations = assetlocations_data["data"]
    else:
        raise ValueError(
            f"No asset locations found for the given parameters: {kwargs}. \
            Please check for typos or if it is expected to exists."
        )
    fig = px.scatter_map(
        assetlocations,
        lat="northing",
        lon="easting",
        hover_name="title",
        hover_data=["projectsite_name", "description"],
        zoom=9.6,
        height=500,
    )
    fig.update_layout(map_style="open-street-map")
    fig.update_layout(margin={"r": 0, "t": 0, "l": 0, "b": 0})
    if return_fig:
        return fig
    else:
        fig.show()
        return None

def get_projectsites(self, **kwargs: Any) -> dict[str, Union[pd.DataFrame, bool, np.int64, None]]:
    """
    Get all available projects.

    Parameters
    ----------
    **kwargs
        Additional parameters to pass to the API.

    Returns
    -------
    dict
        Dictionary with the following keys:

        - "data": Pandas dataframe with the location data for each
          project
        - "exists": Boolean indicating whether matching records
          are found

    Examples
    --------
            >>> from unittest import mock
            >>> api = LocationsAPI(
            ...     api_root="https://example",
            ...     header={"Authorization": "Token test"},
            ... )
            >>> df = pd.DataFrame({"id": [1]})
            >>> with mock.patch.object(
            ...     LocationsAPI,
            ...     "process_data",
            ...     return_value=(df, {"existance": True}),
            ... ):
            ...     out = api.get_projectsites()
            >>> out["exists"]
            True
    """
    url_params = {}  # type: dict[str, str]
    url_params = {**url_params, **kwargs}
    url_data_type = "projectsites"
    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_projectsite_detail(
    self, projectsite: str, **kwargs: Any
) -> dict[str, Union[pd.DataFrame, bool, np.int64, None]]:
    """
    Get details for a specific projectsite.

    Parameters
    ----------
    projectsite : str
        Title of the projectsite.
    **kwargs
        Additional parameters to pass to the API.

    Returns
    -------
    dict
        Dictionary with the following keys:

        - "id": ID of the selected project site.
        - "data": Pandas dataframe with the location data for each
          projectsite.
        - "exists": Boolean indicating whether matching records
          are found.

    Examples
    --------
            >>> from unittest import mock
            >>> api = LocationsAPI(
            ...     api_root="https://example",
            ...     header={"Authorization": "Token test"},
            ... )
            >>> df = pd.DataFrame({"id": [1]})
            >>> with mock.patch.object(
            ...     LocationsAPI,
            ...     "process_data",
            ...     return_value=(df, {"existance": True, "id": 1}),
            ... ):
            ...     out = api.get_projectsite_detail("Site")
            >>> out["id"]
            1
    """
    url_params = {"projectsite": projectsite}
    url_params = {**url_params, **kwargs}
    url_data_type = "projectsites"
    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_assetlocations(
    self, projectsite: Union[str, None] = None, **kwargs: Any
) -> dict[str, Union[pd.DataFrame, bool, list[bool], np.int64, None]]:
    """
    Get all available asset locations.

    Get all available asset locations for all projectsites or a
    specific projectsite.

    Parameters
    ----------
    projectsite : str, optional
        String with the projectsite title (e.g. "Nobelwind").
    **kwargs
        Additional parameters to pass to the API.

    Returns
    -------
    dict
        Dictionary with the following keys:

        - "data": Pandas dataframe with the location data for each
          location in the projectsite.
        - "exists": Boolean indicating whether matching records
          are found.

    Examples
    --------
            >>> from unittest import mock
            >>> api = LocationsAPI(
            ...     api_root="https://example",
            ...     header={"Authorization": "Token test"},
            ... )
            >>> df = pd.DataFrame({"id": [1]})
            >>> with mock.patch.object(
            ...     LocationsAPI,
            ...     "process_data",
            ...     return_value=(df, {"existance": True}),
            ... ):
            ...     out = api.get_assetlocations(projectsite="Site")
            >>> out["exists"]
            True
    """
    url_params = {}  # type: dict[str, str]
    url_params = {**url_params, **kwargs}
    if projectsite:
        url_params["projectsite__title"] = projectsite
    url_data_type = "assetlocations"
    if "assetlocations" in url_params and isinstance(url_params["assetlocations"], list):
        df = []
        df_add = {"existance": []}
        for assetlocation in url_params["assetlocations"]:
            output_type = "single"
            url_params["assetlocation"] = assetlocation
            df_temp, df_add_temp = self.process_data(url_data_type, url_params, output_type)
            df.append(df_temp)
            df_add["existance"].append(df_add_temp["existance"])
        df = pd.concat(df)
    else:
        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_assetlocation_detail(
    self,
    assetlocation: str,
    projectsite: Union[None, str] = None,
    **kwargs: Any,
) -> dict[str, Union[pd.DataFrame, bool, np.int64, None]]:
    """
    Get a selected turbine.

    Parameters
    ----------
    assetlocation : str
        Title of the asset location (e.g. "BBK05").
    projectsite : str, optional
        Name of the projectsite (e.g. "Nobelwind").
    **kwargs
        Additional parameters to pass to the API.

    Returns
    -------
    dict
        Dictionary with the following keys:

        - "id": ID of the selected projectsite site.
        - "data": Pandas dataframe with the location data for the
          individual location.
        - "exists": Boolean indicating whether a matching location
          is found.

    Examples
    --------
            >>> from unittest import mock
            >>> api = LocationsAPI(
            ...     api_root="https://example",
            ...     header={"Authorization": "Token test"},
            ... )
            >>> df = pd.DataFrame({"id": [1]})
            >>> with mock.patch.object(
            ...     LocationsAPI,
            ...     "process_data",
            ...     return_value=(df, {"existance": True, "id": 1}),
            ... ):
            ...     out = api.get_assetlocation_detail("T01")
            >>> out["id"]
            1
    """
    if projectsite is None:
        url_params = {"assetlocation": assetlocation}
    else:
        url_params = {
            "projectsite": projectsite,
            "assetlocation": assetlocation,
        }
    url_params = {**url_params, **kwargs}
    url_data_type = "assetlocations"
    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"]}