Commit 84254789 authored by Gregory Harris's avatar Gregory Harris
Browse files

Adding capability to search for a wellbore by name and return matching Ids.

parent a44f2e17
......@@ -48,10 +48,14 @@ class LasCommandLoader(CLICommandsLoader):
group.command('wellbore', 'wellbore')
group.command('welllog', 'welllog')
group.command('curves', 'welllog_data')
group.command('wellbore_search', 'wellbore_search')
with CommandGroup(self, 'update', 'lasloader.commands.update#{}') as group:
group.command('welllog', 'welllog')
with CommandGroup(self, 'search', 'lasloader.commands.search#{}') as group:
group.command('wellbore', 'wellbore_search')
with CommandGroup(self, 'download', 'lasloader.commands.download#{}') as group:
group.command('welllog', 'download_las')
......@@ -141,6 +145,19 @@ class LasCommandLoader(CLICommandsLoader):
help='If specified or set to True LASCLI will attempt to recognize and update the curve families.')
# 'download'
with ArgumentsContext(self, 'download') as arg_context:
with ArgumentsContext(self, 'search') as arg_context:
arg_context.argument('url', type=str, options_list=('-u', '--url'),
help='The base url of the OSDU instance.')
with ArgumentsContext(self, 'search') as arg_context:
arg_context.argument('token', type=str, options_list=('-t', '--token'),
help='A valid bearer token used to authenticate with the OSDU instance.')
with ArgumentsContext(self, 'search wellbore') as arg_context:
arg_context.argument('wellbore_name', type=str, options_list=('--name'),
help='The facility name of wellbore ids to retrieve.')
with ArgumentsContext(self, 'download') as arg_context:
arg_context.argument('url', type=str, options_list=('-u', '--url'),
help='Url for the OSDU instance.')
......
......@@ -3,6 +3,7 @@ from knack.log import get_logger
from lasloader.osdu_client import OsduClient
from lasloader.configuration import Configuration
from lasloader.file_loader import LocalFileLoader
from lasloader.well_service import WellBoreService, WellLogService
logger = get_logger(__name__)
......@@ -61,7 +62,7 @@ def welllog_data(url: str,
:param str url: The base URL of the OSDU instance
:param str token: a valid bearer token that is used to authenticate against the OSDU instance
:param str config_path: Path to the las metadata file
:param str wellbore_id: The well bore id of the record to retrieve
:param str welllog_id: The welllog id of the record to retrieve
:param list[str] curves: The curves to retrieve, use None to get all curves
"""
config = Configuration(LocalFileLoader(), config_path)
......
import json
from knack.log import get_logger
from lasloader.osdu_client import OsduClient
from lasloader.configuration import Configuration
from lasloader.file_loader import LocalFileLoader
from lasloader.well_service import WellBoreService, WellLogService
logger = get_logger(__name__)
def wellbore_search(url: str,
token: str,
config_path: str,
wellbore_name: str):
"""
Retrieve and print the ids of wellbores that match the specified name
:param str url: The base URL of the OSDU instance
:param str token: a valid bearer token that is used to authenticate against the OSDU instance
:param str config_path: Path to the las metadata file
:param str wellbore_name: The well bore name to search for
"""
config = Configuration(LocalFileLoader(), config_path)
client = OsduClient(url, token, config.data_partition_id)
service = WellBoreService(client, WellLogService(client))
data = service.search_for_wellbore(wellbore_name)
if data is None or len(data) < 1:
logger.warning("No records found")
else:
print("Matching Wellbore ids:")
print(json.dumps(data, indent=4, sort_keys=True))
......@@ -146,6 +146,28 @@ class OsduClient:
payload = {"label": mnemonic, "log_unit": unit}
return self._send_request_json_response("POST", recognize_family_url, None, payload)
def search_for_wellbore(self, wellbore_name: str) -> list[str]:
"""
Search an OSDU instance to find all wellbores with the specified name and return their ids.
:param str wellbore_name: The well name
:return: A list that contains the matching wellbore ids
:rtype: list[str]
"""
url = f"{self._base_url}/api/search/v2/query"
payload = {
"kind": "*:*:master-data--Wellbore:*",
"query": f'data.FacilityName:"{wellbore_name}"',
"returnedFields": [ "id" ],
"limit": 1000
}
result = self._send_request_json_response("POST", url, None, payload)
if result is None or result.get("totalCount") is None or result.get("totalCount") < 1:
return []
else:
return [r["id"] for r in result.get("results")]
def _post_data_with_id_response(self, path: str, record: Record) -> list[any]:
payload = [vars(record)]
url = f"{self._base_url}{path}"
......
......@@ -239,3 +239,12 @@ class WellBoreService:
:rtype: any
"""
return array[0] if array is not None and len(array) > 0 else None
def search_for_wellbore(self, wellbore_name: str):
"""
Get the first element from a list or return None if the list is None or empty
:param list array: An OSDU client wrapper
:return: The first element of the list or None
:rtype: any
"""
return self._client.search_for_wellbore(wellbore_name)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment