Skip to content

Commit

Permalink
feat(api): added new links retrieval exception
Browse files Browse the repository at this point in the history
  • Loading branch information
mehdiwahada committed Jan 8, 2025
1 parent b264ced commit 382fdf9
Show file tree
Hide file tree
Showing 2 changed files with 45 additions and 36 deletions.
6 changes: 6 additions & 0 deletions src/antares/craft/exceptions/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,12 @@ def __init__(self, link_name: str, message: str) -> None:
super().__init__(self.message)


class LinksRetrievalError(Exception):
def __init__(self, study_id: str, message: str) -> None:
self.message = f"Could not retrieve links from study {study_id} : {message}"
super().__init__(self.message)


class ThermalCreationError(Exception):
def __init__(self, thermal_name: str, area_id: str, message: str) -> None:
self.message = f"Could not create the thermal cluster {thermal_name} inside area {area_id}: " + message
Expand Down
75 changes: 39 additions & 36 deletions src/antares/craft/service/api_services/link_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
#
# This file is part of the Antares project.

from typing import Optional
from typing import Any, Optional

import pandas as pd

Expand All @@ -22,6 +22,7 @@
LinkDeletionError,
LinkDownloadError,
LinkPropertiesUpdateError,
LinksRetrievalError,
LinkUiUpdateError,
LinkUploadError,
)
Expand Down Expand Up @@ -214,47 +215,49 @@ def create_capacity_indirect(self, series: pd.DataFrame, area_from: str, area_to
raise LinkUploadError(area_from, area_to, "indirectcapacity", e.message) from e

def read_links(self) -> list[Link]:
url = f"{self._base_url}/studies/{self.study_id}/links"
json_links = self._wrapper.get(url).json()
links = self.read_study_links(json_links)

links.sort(key=lambda link_obj: link_obj.area_from_id)
try:
url = f"{self._base_url}/studies/{self.study_id}/links"
json_links = self._wrapper.get(url).json()
links = self.read_study_links(json_links)

links.sort(key=lambda link_obj: link_obj.area_from_id)
except APIError as e:
raise LinksRetrievalError(self.study_id, e.message) from e
return links

def read_link(self, link: dict[str, Any]) -> Link:
link_area_from_id = link.pop("area1")
link_area_to_id = link.pop("area2")

link_style = link.pop("linkStyle")
link_width = link.pop("linkWidth")
color_r = link.pop("colorr")
color_g = link.pop("colorg")
color_b = link.pop("colorb")

link_ui = LinkUi(link_style=link_style, link_width=link_width, colorr=color_r, colorg=color_g, colorb=color_b)

mapping = {
"hurdlesCost": "hurdles-cost",
"loopFlow": "loop-flow",
"usePhaseShifter": "use-phase-shifter",
"transmissionCapacities": "transmission-capacities",
"displayComments": "display-comments",
"filterSynthesis": "filter-synthesis",
"filterYearByYear": "filter-year-by-year",
"assetType": "asset-type",
}

link = {mapping.get(k, k): v for k, v in link.items()}
link["filter-synthesis"] = set(link["filter-synthesis"].split(", "))
link["filter-year-by-year"] = set(link["filter-year-by-year"].split(", "))
link_properties = LinkProperties(**link)
return Link(link_area_from_id, link_area_to_id, self, link_properties, link_ui)

def read_study_links(self, link_list: list) -> list[Link]:
links = []
for link in link_list:
link_area_from_id = link.pop("area1")
link_area_to_id = link.pop("area2")

link_style = link.pop("linkStyle")
link_width = link.pop("linkWidth")
color_r = link.pop("colorr")
color_g = link.pop("colorg")
color_b = link.pop("colorb")

link_ui = LinkUi(
link_style=link_style, link_width=link_width, colorr=color_r, colorg=color_g, colorb=color_b
)

mapping = {
"hurdlesCost": "hurdles-cost",
"loopFlow": "loop-flow",
"usePhaseShifter": "use-phase-shifter",
"transmissionCapacities": "transmission-capacities",
"displayComments": "display-comments",
"filterSynthesis": "filter-synthesis",
"filterYearByYear": "filter-year-by-year",
"assetType": "asset-type",
}

link = {mapping.get(k, k): v for k, v in link.items()}
link["filter-synthesis"] = set(link["filter-synthesis"].split(", "))
link["filter-year-by-year"] = set(link["filter-year-by-year"].split(", "))
link_properties = LinkProperties(**link)
link_object = Link(link_area_from_id, link_area_to_id, self, link_properties, link_ui)

link_object = self.read_link(link)
links.append(link_object)

return links
Expand Down

0 comments on commit 382fdf9

Please sign in to comment.