emanuele/parser-eLabFTW-NFFA-DI
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first attempt to refactor the test code into something more elegant

Browse Source
This commit is contained in:
Emanuele D'Amico
2026-02-09 17:50:35 +01:00
parent 352a223d95
commit c49aa23aea
4 changed files with 266 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/APIHandler.py Normal file
View File

@@ -0,0 +1,31 @@
import requests
class APIHandler:
'''
Class to standardize the format of the headers of our http requests.
'''
# TO-DO: remove static url.
def __init__(self, apikey="", ELABFTW_API_URL="https://elabftw.fisica.unina.it/api/v2"):
'''Init method, apikey suggested but not required (empty by default).'''
self.auth = {"Authorization" : apikey}
self.content = {"Content-Type" : "application/json"}
self.dump = {**self.auth, **self.content}
self.elaburl = ELABFTW_API_URL
def get_entry_from_elabid(self, elabid, entryType="items"):
'''
Method which returns a resource's raw data (as dictionary) from its elabid and entry type.
Entry type can be either "experiments" or "items".
'''
# TO-DO: validation and error handling on entryType value.
header = self.dump
response = requests.get(
headers = header,
url = f"{self.elaburl}/{entryType}/{elabid}",
verify=True
)
if response.status_code // 100 in [2,3]:
entry_data = response.json()
return entry_data
else:
raise ConnectionError(f"HTTP request failed with status code: {response.status_code}.")

139
src/classes.py
View File

@@ -1,14 +1,137 @@
class Header:
import os, json, requests
from APIHandler import APIHandler
class Layer:
'''
Class to standardize the format of the headers of our http requests.
Layer(layer_data) - where layer_data is a Python dictionary.
Meant to be used for eLabFTW Experiments of the "PLD Deposition" category.
eLabFTW experiments contain most of the data required by the NeXus file - although every layer is on a different eLab entry;
unfortunately, some data like the target's chemical formula must be retrieved through additional HTTP requests.
Attributes 'target_elabid', 'rheed_system_elabid' and 'laser_system_elabid' contain elabid's for these resources, which are all items.
'''
def __init__(self, apikey=""):
'''Init method, apikey suggested but not required (empty by default).'''
self.auth = {"Authorization" : apikey}
self.content = {"Content-Type" : "application/json"}
self.dump = {**self.auth, **self.content}
def __init__(self, layer_data):
try:
self.extra = layer_data["metadata_decoded"]["extra_fields"]
self.target_elabid = self.extra["Target"]["value"] # elabid
self.rheed_system_elabid = self.extra["RHEED System"]["value"] # elabid
self.laser_system_elabid = self.extra["Laser System"]["value"] # elabid
self.start_time = layer_data.get("created_at")
self.operator = layer_data.get("fullname")
self.description = layer_data.get("body")
self.deposition_time = self.extra["Duration"]["value"]
self.repetition_rate = self.extra["Repetition rate"]["value"]
try:
self.number_of_pulses = (float(self.deposition_time) * float(self.repetition_rate)).__floor__()
except ValueError:
# Since number_of_pulses is required, if it can't be calculated raise error:
raise ValueError("""
Fatal: either Duration or Repetition Rate are empty or invalid.
This has to be an error, since these fields are required by the NeXus standard.
Please edit your eLabFTW entry and retry.
""")
self.temperature = self.extra["Heater temperature"]["value"] # Note: this field used to have a trailing space in its name
self.process_pressure = self.extra["Process pressure"]["value"] # Note: this field used to have a trailing space in its name
self.heating_method = self.extra["Heating Method"]["value"]
self.layer_thickness = self.extra["Thickness"]["value"]
self.buffer_gas = self.extra["Buffer gas"]["value"]
self.heater_target_distance = self.extra["Heater-target distance"]["value"]
self.laser_fluence = self.extra["Laser Intensity"]["value"] # here fluence = intensity
self.laser_spot_area = self.extra["Spot Area"]["value"]
try:
self.laser_energy = (float(self.laser_fluence) * float(self.laser_spot_area)).__round__(3)
except ValueError:
# Since laser_energy is NOT required, if it can't be calculated warn user but allow the software to continue execution:
print("""
Warning: either Laser Intensity or Spot Area are empty or invalid.
If you think this is an error, please edit your eLabFTW entry and retry.
Setting Laser Energy to NoneType.
""")
# Placeholder
self.laser_energy = None
# Laser rasternig section
self.laser_rastering_geometry = self.extra["Laser Rastering Geometry"]["value"]
self.laser_rastering_positions = self.extra["Laser Rastering Position"]["value"]
self.laser_rastering_velocities = self.extra["Laser Rastering Speed"]["value"]
# Pre annealing section
self.pre_annealing_ambient_gas = self.extra["Buffer gas Pre"]["value"]
self.pre_annealing_pressure = self.extra["Process pressure Pre"]["value"]
self.pre_annealing_temperature = self.extra["Heater temperature Pre"]["value"]
self.pre_annealing_duration = self.extra["Duration Pre"]["value"]
# Post annealing section
self.post_annealing_ambient_gas = self.extra["Buffer gas PA"]["value"]
self.post_annealing_pressure = self.extra["Process pressure PA"]["value"]
self.post_annealing_temperature = self.extra["Heater temperature PA"]["value"]
self.post_annealing_duration = self.extra["Duration PA"]["value"]
# Rejected but suggested by the NeXus standard:
#self.laser_rastering_coefficients = None
except KeyError as k:
# Some keys are not required and can be called through the .get() method - which is permissive and allows null values;
# Other keys are required so if they can't be called (invalid or null) raise error and stop execution of the program:
raise KeyError(f"The provided dictionary lacks a \"{k}\" key. Check the deposition layer entry on eLabFTW and make sure you used the correct Experiment template.")
class Entrypoint:
'''
Entrypoint(sample_data) - where sample_data is a Python dictionary.
Meant to be used for eLabFTW Resources of the "Sample" category.
The entrypoint is the starting point of the process of resolving the data chain.
The entrypoint must be a dictionary containing the data of a sample, created directly from the JSON of the item endpoint on eLabFTW - which can be done through the function get_entry_from_elabid.
'''
def __init__(self, sample_data):
try:
self.extra = sample_data["metadata_decoded"]["extra_fields"]
self.linked_items = sample_data["items_links"]
self.batch_elabid = self.extra["Substrate batch"]["value"]
self.linked_experiments = sample_data["related_experiments_links"]
self.linked_experiments_elabid = [ i["entityid"] for i in self.linked_experiments ]
except KeyError as k:
# Some keys are not required and can be called through the .get() method - which is permissive and allows null values;
# Other keys are required so if they can't be called (invalid or null) raise error and stop execution of the program:
raise KeyError(f"The provided dictionary lacks a \"{k}\" key. Check the sample entry on eLabFTW and make sure you used the correct Resource template.")
# Non-required attributes:
self.name = sample_data.get("title") or None # error prevention is more important than preventing empty fields here
class Material:
'''
Material(material_data) - where material_data is a Python dictionary.
Meant to be used for eLabFTW Resources of either the "PLD Target" or the "Substrate" categories.
Both a PLD Target and a Substrate are materials made of a certain compound, of which we want to know:
* Name and formula;
* Shape and dimensions;
* Misc.
'''
def __init__(self, material_data):
try:
self.extra = material_data["metadata_decoded"]["extra_fields"]
self.compound_elabid = self.extra["Compound"]["value"]
except KeyError as k:
# Some keys are not required and can be called through the .get() method - which is permissive and allows null values;
# Other keys are required so if they can't be called (invalid or null) raise error and stop execution of the program:
raise KeyError(f"The provided dictionary lacks a \"{k}\" key. Check the target/substrate entry on eLabFTW and make sure you used the correct Resource template.")
def get_compound_data(self, apikey):
raw_compound_data = APIHandler(apikey).get_entry_from_elabid(self.compound_elabid, entryType="items")
name = raw_compound_data["title"]
extra = raw_compound_data["metadata_decoded"]["extra_fields"]
formula = extra.get("Chemical formula")
cas = extra.get("CAS number ") or { "value": None }
compound_data = {
"name" : name,
"chemical_formula" : formula.get("value"),
"cas_number" : cas.get("value")
}
return compound_data
def get_compound_formula(self, apikey):
formula = self.get_compound_data(apikey).get("chemical_formula")
return formula
if __name__=="__main__":
head = Header("MyApiKey-123456789abcdef")
print(f"Example header: {head.dump()}")
print(f"Example header:\n\t{head.dump}\n")
print("Warning: you're not supposed to be running this as the main program.")

59
src/functions.py Normal file
View File

@@ -0,0 +1,59 @@
import json, requests
from APIHandler import APIHandler
def get_entry_from_elabid(elabid, entryType="items"):
'''
Function which returns entrypoint data (as dictionary) from its elabid.
'''
header = APIHandler(apikey).dump
response = requests.get(
headers = header,
url = f"{ELABFTW_API_URL}/{entryType}/{elabid}",
verify=True
)
if response.status_code // 100 in [2,3]:
entry_data = response.json()
return entry_data
else:
raise ConnectionError(f"HTTP request failed with status code: {response.status_code}.")
def get_sample_layers_data(elabid):
'''
Return the following data from every eLabFTW experiment linked
to a certain sample, identified by elabid.
- Title of the experiment
- Category (should check it's "PLD Deposition")
- Layer number - if present (PLD depositions)
- Deposition time - returns error if not present
- Repetition rate - returns error if not present
'''
# header = {
# "Authorization": apikey,
# "Content-Type": "application/json"
# }
sample_data = requests.get(
headers = header,
url = f"https://elabftw.fisica.unina.it/api/v2/items/{elabid}",
verify=True
).json()
related_experiments = sample_data["related_experiments_links"]
result = []
for exp in related_experiments:
experiment_data = requests.get(
headers = header,
url = f"https://elabftw.fisica.unina.it/api/v2/experiments/{exp.get("entityid")}",
verify=True
).json()
extra = experiment_data["metadata_decoded"]["extra_fields"]
result.append(
{"title": exp.get("title"),
"layer_number": extra.get("Layer Progressive Number").get("value"),
"category": exp.get("category_title"),
"deposition_time": extra.get("Duration").get("value"),
"repetition_rate": extra.get("Repetition rate").get("value")}
)
return result
if __name__=="__main__":
print("Warning: you're not supposed to be running this as the main program.")

87
src/main.py
View File

@@ -1,46 +1,49 @@
import os, json, requests
from getpass import getpass
from classes import Header
from APIHandler import APIHandler
from classes import *
def get_sample_layers_data(elabid):
'''
Return the following data from every eLabFTW experiment linked
to a certain sample, identified by elabid.
- Title of the experiment
- Category (should check it's "PLD Deposition")
- Layer number - if present (PLD depositions)
- Deposition time - returns error if not present
- Repetition rate - returns error if not present
'''
# header = {
# "Authorization": apikey,
# "Content-Type": "application/json"
# }
sample_data = requests.get(
headers = header,
url = f"https://elabftw.fisica.unina.it/api/v2/items/{elabid}",
verify=True
).json()
related_experiments = sample_data["related_experiments_links"]
result = []
for exp in related_experiments:
experiment_data = requests.get(
headers = header,
url = f"https://elabftw.fisica.unina.it/api/v2/experiments/{exp.get("entityid")}",
verify=True
).json()
extra = experiment_data["metadata_decoded"]["extra_fields"]
result.append(
{"title": exp.get("title"),
"layer_number": extra.get("Layer Progressive Number").get("value"),
"category": exp.get("category_title"),
"deposition_time": extra.get("Duration").get("value"),
"repetition_rate": extra.get("Repetition rate").get("value")}
)
return result
apikey = getpass("Paste API key here: ") # consider replacing with .env file
header = Header(apikey).dump
result = get_sample_layers_data(1108) # edit id at will in case of deletion of remote source
print(json.dumps(result))
if __name__=="__main__":
print(f"=======================\n===== DEBUG MODE! =====\n=======================\n")
ELABFTW_API_URL = "https://elabftw.fisica.unina.it/api/v2"
apikey = getpass("Paste API key here: ")
# TEST. In production the entryType will probably just be "items" since the entrypoint is an item (sample).
entryType = None
while entryType not in ["items", "experiments"]:
eT = input("Enter a valid entry type [items, experiments]: ")
# This allows for a shortcut: instead of prompting the type before and the elabid after I can just prompt both at the same time - e.g. e51 is exp. 51, i1108 is item 1108...
if eT[0] in ["e", "i"] and eT[-1].isnumeric():
try:
elabid = int(eT[1:])
eT = eT[0]
except Exception:
print("Usage: i|item|items|i[ELABID] for items, e|experiment|experiments|e[ELABID] for experiments.")
pass
match eT:
case "items" | "i" | "item":
entryType = "items"
case "experiments" | "e" | "exp" | "experiment":
entryType = "experiments"
case _:
pass
# This will probably be reworked in production
try:
elabid = elabid
except NameError:
elabid = input("Input elabid here [default = 1111]: ") or 1111
data = APIHandler(apikey).get_entry_from_elabid(elabid, entryType)
if entryType == "experiments":
layer = Layer(data)
result = layer.__dict__
result.pop("extra")
print(result)
elif entryType == "items":
if data.get("category_title") == "Sample":
item = Entrypoint(data)
elif data.get("category_title") in ["PLD Target", "Substrate"]:
item = Material(data)
print(item.get_compound_formula(apikey))
result = item.__dict__
result.pop("extra")
print(result)