85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
import os, json, requests
|
|
from getpass import getpass
|
|
|
|
def valid_elabfiles(path):
|
|
'''Lookup directory "path" and
|
|
returns list of valid eLabFTW
|
|
Experiment JSON files.'''
|
|
elabfiles = []
|
|
for filename in os.listdir(path):
|
|
if filename.endswith(".json"):
|
|
try:
|
|
with open(os.path.join(path, filename), "r") as f:
|
|
data = json.load(f)
|
|
if data.get("elabid"): # insert specific NeXus requirements here later
|
|
|
|
elabfiles.append(filename)
|
|
f.close()
|
|
except json.decoder.JSONDecodeError as e: # invalid files "masked" as JSON
|
|
#print(f"wait a moment: {e}") # just for debug
|
|
pass
|
|
return elabfiles
|
|
|
|
def call_sample(elabid, SERVER_URL="https://elabftw.fisica.unina.it/"): # TO-DO: rm default server
|
|
'''Queries the Resources (/items) API endpoint
|
|
of eLabFTW instance to request data (JSON)
|
|
on a certain sample given its eLab-ID.
|
|
|
|
Requires an active (RO/RW) API key.
|
|
Defaults to elabftw.fisica.unina.it.'''
|
|
full_elab_url = f"{SERVER_URL}api/v2" # API endpoint root for eLabFTW
|
|
items_url = f"{full_elab_url}/items" # API endpoint /items
|
|
header = {
|
|
"Authorization": apikey,
|
|
"Content-Type": "application/json"
|
|
}
|
|
sample = requests.get(
|
|
headers=header,
|
|
url=f"{items_url}/{elabid}",
|
|
verify=True
|
|
)
|
|
return sample.json()
|
|
|
|
def id2sample(elabid):
|
|
'''Fetches sample data (JSON) from eLabFTW
|
|
instance (using function "call_sample()")
|
|
and extracts significant information.
|
|
|
|
Currently, it only returns the sample's title.'''
|
|
sample_data = call_sample(elabid)
|
|
sample_title = sample_data["title"]
|
|
return sample_title
|
|
|
|
def fetch_and_group(path):
|
|
'''Fetches experiment data from eLabFTW JSON
|
|
files in a given folder, then
|
|
'''
|
|
sample_dict = {}
|
|
for filename in valid_elabfiles(path):
|
|
with open(os.path.join(path, filename), "r") as f:
|
|
layer = json.load(f)
|
|
extra = layer["metadata_decoded"]["extra_fields"]
|
|
sample_id = extra["Sample"]["value"]
|
|
sample_title = id2sample(sample_id)
|
|
lpn = int(extra["Layer Progressive Number"]["value"]) # Layer Progressive Number
|
|
if not sample_dict.get(sample_title): # if not existent yet, initialize
|
|
sample_dict[sample_title] = {
|
|
"instrument": {
|
|
"deposition_chamber": extra["Chamber"]["value"], # ID of associated resource (PLD chamber) - useless as is!
|
|
"laser_system": extra["Laser System"]["value"],
|
|
"rheed_system": extra["RHEED System"]["value"]
|
|
},
|
|
"multilayer": {}
|
|
}
|
|
sample_dict[sample_title]["multilayer"][f"layer_{lpn}"] = {
|
|
"operator": layer["fullname"],
|
|
"created_at": layer["created_at"],
|
|
"sample": extra["Sample"], # ID of associated sample - useless as is!
|
|
"temperature": extra["Heater temperature "], # space at the end is a config error in eLab!
|
|
"target": extra["Target"]
|
|
}
|
|
return sample_dict
|
|
|
|
apikey = getpass("Paste API key here: ")
|
|
sample_dict = fetch_and_group("./tests/objects")
|
|
print(json.dumps(sample_dict, indent=3)) |