-
Notifications
You must be signed in to change notification settings - Fork 1
/
nsls2api_lib.py
131 lines (90 loc) · 3.97 KB
/
nsls2api_lib.py
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
127
128
129
130
131
import httpx
import time
import logging
logger = logging.getLogger(__name__)
base_url = "https://api.nsls2.bnl.gov/v1"
ispyb_instruments = ("AMX", "FMX", "NYX")
def is_ispyb_instrument(instruments):
for instrument in instruments:
if instrument in ispyb_instruments:
return True
return False
def get_from_api(url):
if url:
response = httpx.get(f"{base_url}/{url}")
if response.status_code == httpx.codes.OK:
return response.json()
raise RuntimeError(
f"failed to get value from {url}. response code: {response.status_code}"
)
else:
raise ValueError("url cannot be empty")
def get_all_cycles():
return get_from_api(f"facility/nsls2/cycles")
def get_proposals_from_cycle(cycle):
return get_from_api(f"facility/nsls2/cycle/{cycle}/proposals")["proposals"]
# return true if any of check_instruments are in proposal_instruments
def check_instruments_in_proposal(proposal_instruments, check_instruments):
for instrument in check_instruments:
if instrument.upper() in proposal_instruments:
return True
return False
def get_proposal_ids_from_proposals_endpoint(proposal_query, items_per_page=50):
proposal_ids_to_return = set()
page = 1
logger.debug(f"Proposal query: {proposal_query}")
while 1:
proposals = get_from_api(
f"{proposal_query}&page_size={items_per_page}&page={page}"
)
logger.debug(
f"Page: {page} Number of unique proposals: {len(proposal_ids_to_return)}"
)
for proposal in proposals["proposals"]:
proposal_ids_to_return.add(proposal["proposal_id"])
if proposals["count"] != items_per_page:
break
page += 1
return sorted(proposal_ids_to_return)
def get_proposal_ids_for_cycle_instruments(cycle, instruments):
instruments_string = "beamline=" + "&beamline=".join(instruments)
proposal_query = f"proposals/?{instruments_string}&cycle={cycle}&facility=nsls2"
return get_proposal_ids_from_proposals_endpoint(proposal_query)
def get_all_proposal_ids_for_instruments(instruments):
instruments_string = "beamline=" + "&beamline=".join(instruments)
cycles_string = "cycle=" + "&cycle=".join(get_all_cycles()["cycles"])
proposal_query = f"proposals/?{instruments_string}&{cycles_string}&facility=nsls2"
return get_proposal_ids_from_proposals_endpoint(proposal_query)
def get_usernames_from_proposal(proposal_id):
return set(get_from_api(f"proposal/{proposal_id}/usernames")["usernames"])
def get_users_from_proposal(proposal_id):
return get_from_api(f"proposal/{proposal_id}/users")
def get_directories_from_proposal(proposal_id):
return get_from_api(f"proposal/{proposal_id}/directories")
def get_proposal_info(proposal_id):
return get_from_api(f"proposal/{proposal_id}")["proposal"]
def get_active_safs_for_proposal(proposal_id): # currently unused
safs = get_all_proposals(proposal_id)["safs"]
def get_all_active_safs_in_current_cycle(cycle="2023-1"): # currently unused
proposals = get_proposals_from_cycle(cycle)
for proposal in proposals:
safs = get_all_proposals(proposal.id)["safs"]
def get_proposals_for_instrument(cycle="2023-1", instrument="FMX"):
proposals_on_instrument = []
proposals = get_proposals_from_cycle(cycle)
for proposal_num in proposals:
proposal = get_proposal_info(proposal_num)
if instrument in proposal["instruments"]:
proposals_on_instrument.append(proposal_num)
return proposals_on_instrument
def get_data_sessions_for_user(username):
sessions = get_from_api(f"data-session/{username}")
return sessions
def get_cycles_from_facility(facilityname):
cycles = get_from_api(f"facility/{facilityname}/cycles")
return cycles
def get_beamline(beamline):
beamline = get_from_api(f"beamline/{beamline}")
return beamline
def get_current_cycle():
return get_from_api(f"facility/nsls2/cycles/current")["cycle"]