Geometry 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_model_definitions(
    self,
    projectsite: Union[str, None] = None,
) -> dict[str, Union[pd.DataFrame, bool, np.int64, None]]:
    """
    Get all relevant model definitions.

    Parameters
    ----------
    projectsite : str, optional
        Title of the projectsite.

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

        - "data": Pandas dataframe with the model definitions
        - "exists": Boolean indicating whether matching records
          are found

    Examples
    --------
            >>> from unittest import mock
            >>> api = GeometryAPI(
            ...     api_root="https://example",
            ...     header={"Authorization": "Token test"},
            ... )
            >>> df = pd.DataFrame({"id": [1]})
            >>> with mock.patch.object(
            ...     GeometryAPI,
            ...     "process_data",
            ...     return_value=(df, {"existance": True}),
            ... ):
            ...     out = api.get_model_definitions(projectsite="Site")
            >>> out["exists"]
            True
    """
    url_params = {}
    if projectsite is not None:
        url_params["site"] = projectsite
    url_data_type = "modeldefinitions"
    output_type = "list"
    with self._temp_api_root(self.api_root.replace("userroutes", "routes")):
        df, df_add = self.process_data(url_data_type, url_params, output_type)
    return {"data": df, "exists": df_add["existance"]}

def get_modeldefinition_id(
    self,
    assetlocation: Union[str, None] = None,
    projectsite: Union[str, None] = None,
    model_definition: Union[str, None] = None,
) -> dict[str, Union[int, np.int64, bool, None]]:
    """
    Get the ID of a model definition.

    Either the asset location or the project site must be specified.

    Parameters
    ----------
    assetlocation : str, optional
        Title of the asset location.
    projectsite : str, optional
        Title of the projectsite.
    model_definition : str, optional
        Title of the model definition.

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

        - "id": ID of the specified model definition
        - "multiple_modeldef": Boolean indicating whether there
          are multiple model definitions for the asset location
          in general

    Raises
    ------
    ValueError
        If at least one of assetlocation or projectsite is not
        specified, if no location found, if no model definitions
        found, if multiple model definitions found without
        specification, or if specified model definition not found.

    Examples
    --------
    >>> api = GeometryAPI(
    ...     api_root="https://example",
    ...     header={"Authorization": "Token test"},
    ... )
    >>> api.get_modeldefinition_id()  # doctest: +ELLIPSIS
    Traceback (most recent call last):
        ...
    ValueError: At least either of the related ... must be specified!
    """
    if assetlocation is None and projectsite is None:
        raise ValueError("At least either of the related `assetlocation` or `projectsite` must be specified!")

    result = {}
    if projectsite is None:
        if assetlocation is None:
            raise ValueError("Asset location must be specified when projectsite is None.")
        location_data = self.loc_api.get_assetlocation_detail(assetlocation=assetlocation)
        if location_data["exists"]:
            location = cast(pd.DataFrame, location_data["data"])
        else:
            raise ValueError(f"No location found for asset {assetlocation}.")
        projectsite = location["projectsite_name"].loc[0]
    model_definitions_data = self.get_model_definitions(projectsite=projectsite)
    if model_definitions_data["exists"]:
        model_definitions = cast(pd.DataFrame, model_definitions_data["data"])
    else:
        raise ValueError(f"No model definitions found for project site {projectsite}.")
    if model_definition is None and len(model_definitions) > 1:
        raise ValueError(
            f"Multiple model definitions found for project site {projectsite}. Please specify which one to use."
        )
    if model_definition is None:
        model_definition_id = model_definitions["id"].values[0]
        result["id"] = model_definition_id
        result["multiple_modeldef"] = False
    else:
        matching_definitions = model_definitions[model_definitions["title"] == model_definition]
        if matching_definitions.empty:
            raise ValueError(f"Model definition '{model_definition}' not found for project site {projectsite}.")
        if len(matching_definitions) > 1:
            raise ValueError(
                f"Multiple model definitions found for '{model_definition}' in project site {projectsite}.\n"
                f"Please check the data consistency."
            )
        model_definition_id = matching_definitions["id"].values[0]
        result["id"] = model_definition_id
        result["multiple_modeldef"] = len(model_definitions) > 1
    return result

def get_subassemblies(
    self,
    projectsite: Union[str, None] = None,
    assetlocation: Union[str, None] = None,
    subassembly_type: Union[str, None] = None,
    model_definition: Union[str, None] = None,
) -> dict[str, Union[pd.DataFrame, bool, np.int64, None]]:
    """
    Get all relevant structure subassemblies.

    If you specify a model definition, you also must specify either
    the projectsite or the asset location.

    Parameters
    ----------
    projectsite : str, optional
        Title of the projectsite.
    assetlocation : str, optional
        Title of the asset location.
    subassembly_type : str, optional
        Type of the subassembly.
    model_definition : str, optional
        Title of the model definition.

    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

    Raises
    ------
    ValueError
        If model definition specified without projectsite or
        assetlocation, or if specified model definition not found.

    Examples
    --------
    >>> from unittest import mock
    >>> api = GeometryAPI(
    ...     api_root="https://example",
    ...     header={"Authorization": "Token test"},
    ... )
    >>> df = pd.DataFrame({"id": [1]})
    >>> with mock.patch.object(
    ...     GeometryAPI,
    ...     "process_data",
    ...     return_value=(df, {"existance": True}),
    ... ):
    ...     out = api.get_subassemblies(projectsite="Site")
    >>> out["exists"]
    True
    """
    url_params = {}
    func_args = {}
    if projectsite is not None:
        url_params["asset__projectsite__title"] = projectsite
        func_args["projectsite"] = projectsite
    if assetlocation is not None:
        url_params["asset__title"] = assetlocation
        func_args["assetlocation"] = assetlocation
    if subassembly_type is not None:
        url_params["subassembly_type"] = subassembly_type
    if model_definition is not None:
        if projectsite is not None or assetlocation is not None:
            func_args["model_definition"] = model_definition
            modeldef_data = self.get_modeldefinition_id(**func_args)
            if modeldef_data["id"] is not None:
                url_params["model_definition"] = str(modeldef_data["id"])
            else:
                raise ValueError(
                    f"No model definition {model_definition} found for project site {projectsite} "
                    f"or asset location {assetlocation}."
                )
        else:
            raise ValueError(
                "If you specify a model definition, you also must specify either "
                "the projectsite or the asset location!"
            )
    url_data_type = "subassemblies"
    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_buildingblocks(
    self,
    projectsite: Union[str, None] = None,
    assetlocation: Union[str, None] = None,
    subassembly_type: Union[str, None] = None,
    subassembly_id: Union[int, np.int64, None] = None,
) -> dict[str, Union[pd.DataFrame, bool, np.int64, None]]:
    """
    Get all relevant building blocks.

    Parameters
    ----------
    projectsite : str, optional
        Title of the projectsite.
    assetlocation : str, optional
        Title of the asset location.
    subassembly_type : str, optional
        Type of the subassemblies (e.g. 'MP', 'TW', 'TP').
    subassembly_id : int or np.int64, optional
        ID of the subassembly.

    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 = GeometryAPI(
            ...     api_root="https://example",
            ...     header={"Authorization": "Token test"},
            ... )
            >>> df = pd.DataFrame({"id": [1]})
            >>> with mock.patch.object(
            ...     GeometryAPI,
            ...     "process_data",
            ...     return_value=(df, {"existance": True}),
            ... ):
            ...     out = api.get_buildingblocks(projectsite="Site")
            >>> out["exists"]
            True
    """
    url_params = {}
    if projectsite is not None:
        url_params["sub_assembly__asset__projectsite__title"] = projectsite
    if assetlocation is not None:
        url_params["sub_assembly__asset__title"] = assetlocation
    if subassembly_type is not None:
        url_params["sub_assembly__subassembly_type"] = subassembly_type
    if subassembly_id is not None:
        url_params["sub_assembly__id"] = str(subassembly_id)
    url_data_type = "buildingblocks"
    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_materials(
    self,
) -> dict[str, Union[pd.DataFrame, bool, np.int64, None]]:
    """
    Get all the materials of building blocks.

    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 = GeometryAPI(
            ...     api_root="https://example",
            ...     header={"Authorization": "Token test"},
            ... )
            >>> df = pd.DataFrame({"id": [1]})
            >>> with mock.patch.object(
            ...     GeometryAPI,
            ...     "process_data",
            ...     return_value=(df, {"existance": True}),
            ... ):
            ...     out = api.get_materials()
            >>> out["exists"]
            True
    """
    url_params = {}  # type: dict[str, str]
    url_data_type = "materials"
    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_subassembly_objects(
    self,
    turbine: str,
    subassembly: Union[str, None] = None,
    model_definition_id: Union[int, np.int64, None] = None,
) -> dict[str, SubAssembly]:
    """
    Get all subassemblies for a given turbine, divided by type.

    Parameters
    ----------
    turbine : str
        Turbine title.
    subassembly : str, optional
        Sub-assembly type (e.g. 'MP', 'TW', 'TP').
    model_definition_id : int or np.int64, optional
        ID of the model definition to filter the subassemblies.

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

        - "TW": SubAssembly object for the tower
        - "TP": SubAssembly object for the transition piece
        - "MP": SubAssembly object for the monopile

    Raises
    ------
    ValueError
        If no subassemblies found for the turbine or if no
        materials found in the database.

    Examples
    --------
    >>> from types import SimpleNamespace
    >>> from unittest import mock
    >>> materials = pd.DataFrame(
    ...     [
    ...         {
    ...             "title": "Steel",
    ...             "slug": "steel",
    ...             "id": np.int64(1),
    ...             "description": "",
    ...             "young_modulus": np.float64(210000.0),
    ...             "density": np.float64(7850.0),
    ...             "poisson_ratio": np.float64(0.3),
    ...         }
    ...     ]
    ... )
    >>> item = {
    ...     "id": np.int64(1),
    ...     "title": "SA_1",
    ...     "description": "",
    ...     "slug": "sa",
    ...     "x_position": np.float64(0),
    ...     "y_position": np.float64(0),
    ...     "z_position": np.float64(0),
    ...     "vertical_position_reference_system": "LAT",
    ...     "subassembly_type": "TW",
    ...     "source": "api",
    ...     "asset": np.int64(1),
    ...     "model_definition": np.int64(1),
    ... }
    >>> response = SimpleNamespace(
    ...     status_code=200,
    ...     reason="OK",
    ...     json=lambda: [item],
    ... )
    >>> api = GeometryAPI(
    ...     api_root="https://example",
    ...     header={"Authorization": "Token test"},
    ... )
    >>> with mock.patch.object(
    ...     GeometryAPI,
    ...     "send_request",
    ...     return_value=response,
    ... ), mock.patch.object(
    ...     GeometryAPI,
    ...     "get_materials",
    ...     return_value={"exists": True, "data": materials},
    ... ):
    ...     out = api.get_subassembly_objects("T01")
    >>> sorted(out.keys())
    ['TW']
    """
    url_data_type = "subassemblies"
    url_params = {"asset__title": turbine}
    if subassembly is not None:
        url_params["subassembly_type"] = subassembly
    if model_definition_id is not None:
        url_params["model_definition"] = str(model_definition_id)
    resp = self.send_request(url_data_type, url_params)
    self.check_request_health(resp)
    if not resp.json():
        raise ValueError("No subassemblies found for " + str(turbine))

    material_data = self.get_materials()
    if material_data["exists"]:
        materials = material_data["data"]
    else:
        raise ValueError("No materials found in the database.")

    subassemblies = {}
    for item in resp.json():
        subassembly_type = item["subassembly_type"]
        subassembly_obj = SubAssembly(materials, item, api_object=self)
        if subassembly_type in subassemblies:
            if not isinstance(subassemblies[subassembly_type], list):
                subassemblies[subassembly_type] = [subassemblies[subassembly_type]]
            subassemblies[subassembly_type].append(subassembly_obj)
        else:
            subassemblies[subassembly_type] = subassembly_obj

    return subassemblies