1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126 | from typing import Optional, List
import os
import requests
ACCESS_TOKEN = os.environ["WHITSON_API_TOKEN"]
CLIENT_NAME = os.environ["CLIENT_NAME"]
def get_well_data_casing(well_id: int, use_from_date: Optional[str] = None):
"""
Get well data casing based on well_id and use_from_date.
"""
base_url = f"http://{CLIENT_NAME}.whitson.com/api-external/v1/wells/{well_id}/bhp_input/well_data/well_data_casing"
response = requests.get(
base_url,
headers={
"content-type": "application/json",
"Authorization": f"Bearer {ACCESS_TOKEN}",
},
params={"use_from_date": use_from_date},
)
if not response.json():
raise Exception("no existing wells")
return response.json()
def upload_well_data_casing_to_well(
well_id: int, payload: List[dict], use_from_date: Optional[str] = None
):
"""
Upload a well data casing to a well.
"""
base_url = f"http://{CLIENT_NAME}.whitson.com/api-external/v1/wells/{well_id}/bhp_input/well_data/well_data_casing"
response = requests.post(
base_url,
headers={
"content-type": "application/json",
"Authorization": f"Bearer {ACCESS_TOKEN}",
},
json=payload,
params={"use_from_date": use_from_date},
)
if response.status_code >= 200 and response.status_code < 300:
print(f"success on well_id {well_id}")
else:
print(response.text)
def edit_well_data_casing(well_data_id: int, pipe_number: int, payload: dict):
"""
Edit a well data casing to a well data.
"""
base_url = f"http://{CLIENT_NAME}.whitson.com/api-external/v1/wells/bhp_input/well_data/{well_data_id}/well_data_casing/{pipe_number}"
response = requests.put(
base_url,
headers={
"content-type": "application/json",
"Authorization": f"Bearer {ACCESS_TOKEN}",
},
json=payload,
)
if response.status_code >= 200 and response.status_code < 300:
print(f"success on well_id {well_id}")
else:
print(response.text)
if __name__ == "__main__":
# First get the well id
well_name = "SPE-DATA-REPOSITORY-DATASET-1-WELL-1-OSPREY"
well_json = get_well(project_id=1, name=well_name)[0]
well_id = well_json["id"]
# Create a new well data
use_from_date = "2022-03-01"
well_data_payload = [
{
"use_from_date": use_from_date,
"flow_path": "casing",
"gas_lift_config": "poorboy",
"lift_method": "none",
"well_data_casing": [
{
"bottom_md": 7029,
"d_casing_inner": 6.276,
"k_casing": 0.0006,
"pipe_number": 1,
"top_md": 0,
}
],
}
]
print("uploading well_data to well: ", well_id, ", well name", well_name)
upload_well_data_to_well(well_id, well_data_payload)
# Upload a new casing pipe to the well data
well_data_casing_payload = [
{
"bottom_md": 11900,
"d_casing_inner": 4.0,
"k_casing": 0.0006,
"pipe_number": 2,
"top_md": 6936,
}
]
print("uploading well_data_casing to well: ", well_id, ", well name", well_name)
upload_well_data_casing_to_well(well_id, well_data_casing_payload, use_from_date)
# Get the well data id for the new well data we created
well_data_json = get_well_data(well_id, use_from_date=use_from_date)[0]
well_data_id = well_data_json["id"]
# Alter well data casing data
print(f"Edit well_data_casing to well data {well_data_id}")
edit_well_data_casing(
well_data_id=well_data_id,
pipe_number=2,
payload={
"bottom_md": 11999,
},
)
# Last query the casing data
print(f"Get well_data_casing to well {well_name}")
well_data_casing_res = get_well_data_casing(well_id, use_from_date)
print(well_data_casing_res)
|
