Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions src/fmu_settings_api/services/smda.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Service for querying and translating results from SMDA."""

import asyncio
import re
from collections.abc import Sequence
from typing import Any, Final
from uuid import NAMESPACE_URL, uuid5
Expand Down Expand Up @@ -36,6 +37,13 @@

logger = get_logger(__name__)


def _drogon_wellbore_identifier(rms_well_name: str) -> str:
"""Convert a Drogon RMS well name to an SMDA-style wellbore identifier."""
name = re.sub(r"(?<=\d)_(?=\d)", "/", rms_well_name, count=1)
return f"NO {name.replace('_', ' ')}"


DROGON_SMDA_MASTERDATA: Final[dict[str, Any]] = DROGON_MASTERDATA["smda"]
DROGON_FIELD: Final[dict[str, Any]] = DROGON_SMDA_MASTERDATA["field"][0]
DROGON_STRAT_COLUMN: Final[dict[str, Any]] = DROGON_SMDA_MASTERDATA[
Expand Down Expand Up @@ -74,8 +82,8 @@
]
DROGON_WELL_HEADERS: Final[list[SmdaWellHeader]] = [
SmdaWellHeader(
unique_well_identifier=f"NO {well['name']}",
unique_wellbore_identifier=f"NO {well['name']}",
unique_well_identifier=_drogon_wellbore_identifier(str(well["name"])),
unique_wellbore_identifier=_drogon_wellbore_identifier(str(well["name"])),
official_wellbore_name=str(well["name"]),
country_identifier=DROGON_SMDA_MASTERDATA["country"][0]["identifier"],
parent_wellbore=None,
Expand Down
12 changes: 8 additions & 4 deletions tests/test_services/test_smda_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -168,11 +168,15 @@ async def test_get_drogon_well_headers_uses_drogon_data() -> None:
assert [header.official_wellbore_name for header in res.well_headers] == [
well["name"] for well in DROGON_RMS_WELLS
]
assert [header.unique_well_identifier for header in res.well_headers] == [
f"NO {well['name']}" for well in DROGON_RMS_WELLS
]
identifiers_by_name = {
header.official_wellbore_name: header.unique_well_identifier
for header in res.well_headers
}
assert identifiers_by_name["55_33-A-1"] == "NO 55/33-A-1"
assert identifiers_by_name["OP5_Y1"] == "NO OP5 Y1"
assert identifiers_by_name["RFT_55_33-A-2"] == "NO RFT 55/33-A-2"
assert [header.unique_wellbore_identifier for header in res.well_headers] == [
f"NO {well['name']}" for well in DROGON_RMS_WELLS
header.unique_well_identifier for header in res.well_headers
]
assert res.well_headers[0].country_identifier == "Norway"
assert res.well_headers[0].projected_coordinate_system == "ST_WGS84_UTM37N_P32637"
Expand Down