Compare commits

...

5 Commits

5 changed files with 76 additions and 13 deletions

View File

@@ -9,7 +9,7 @@ class APIHandler:
'''Init method, apikey suggested but not required (empty by default).''' '''Init method, apikey suggested but not required (empty by default).'''
self.auth = {"Authorization" : apikey} self.auth = {"Authorization" : apikey}
self.content = {"Content-Type" : "application/json"} self.content = {"Content-Type" : "application/json"}
self.dump = {**self.auth, **self.content} self.header = {**self.auth, **self.content}
self.elaburl = ELABFTW_API_URL self.elaburl = ELABFTW_API_URL
def get_entry_from_elabid(self, elabid, entryType="items"): def get_entry_from_elabid(self, elabid, entryType="items"):
''' '''
@@ -18,14 +18,22 @@ class APIHandler:
Entry type can be either "experiments" or "items". Entry type can be either "experiments" or "items".
''' '''
# TO-DO: validation and error handling on entryType value. # TO-DO: validation and error handling on entryType value.
header = self.dump header = self.header
response = requests.get( response = requests.get(
headers = header, headers = header,
url = f"{self.elaburl}/{entryType}/{elabid}", url = f"{self.elaburl}/{entryType}/{elabid}",
verify=True verify=True
) )
if response.status_code // 100 in [2,3]: if response.status_code // 100 in [1,2,3]:
entry_data = response.json() entry_data = response.json()
return entry_data return entry_data
elif response.status_code // 100 == 4:
match response.status_code:
case 401|403:
raise ConnectionError(f"Invalid API key or authentication method.")
case 404:
raise ConnectionError(f"404: Not Found. This means there's no resource with this elabid (wrong elabid?) on your eLabFTW (wrong endpoint?).")
case _:
raise ConnectionError(f"HTTP request failed with status code: {response.status_code} (NOTE: 4xx means user's fault).")
else: else:
raise ConnectionError(f"HTTP request failed with status code: {response.status_code}.") raise ConnectionError(f"There's a problem on the server. Status code: {response.status_code}.")

View File

@@ -14,6 +14,7 @@ class Layer:
def __init__(self, layer_data): def __init__(self, layer_data):
try: try:
self.extra = layer_data["metadata_decoded"]["extra_fields"] self.extra = layer_data["metadata_decoded"]["extra_fields"]
self.layer_number = self.extra["Layer Progressive Number"]["value"] # integer
self.target_elabid = self.extra["Target"]["value"] # elabid self.target_elabid = self.extra["Target"]["value"] # elabid
self.rheed_system_elabid = self.extra["RHEED System"]["value"] # elabid self.rheed_system_elabid = self.extra["RHEED System"]["value"] # elabid
self.laser_system_elabid = self.extra["Laser System"]["value"] # elabid self.laser_system_elabid = self.extra["Laser System"]["value"] # elabid
@@ -83,10 +84,10 @@ class Entrypoint:
def __init__(self, sample_data): def __init__(self, sample_data):
try: try:
self.extra = sample_data["metadata_decoded"]["extra_fields"] self.extra = sample_data["metadata_decoded"]["extra_fields"]
self.linked_items = sample_data["items_links"] self.linked_items = sample_data["items_links"] # dict
self.batch_elabid = self.extra["Substrate batch"]["value"] self.batch_elabid = self.extra["Substrate batch"]["value"] # elabid
self.linked_experiments = sample_data["related_experiments_links"] self.linked_experiments = sample_data["related_experiments_links"] # dict
self.linked_experiments_elabid = [ i["entityid"] for i in self.linked_experiments ] self.linked_experiments_elabid = [ i["entityid"] for i in self.linked_experiments ] # list of elabid
except KeyError as k: except KeyError as k:
# Some keys are not required and can be called through the .get() method - which is permissive and allows null values; # 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: # Other keys are required so if they can't be called (invalid or null) raise error and stop execution of the program:
@@ -133,5 +134,5 @@ class Material:
if __name__=="__main__": if __name__=="__main__":
head = Header("MyApiKey-123456789abcdef") head = Header("MyApiKey-123456789abcdef")
print(f"Example header:\n\t{head.dump}\n") print(f"Example header:\n\t{head.header}\n")
print("Warning: you're not supposed to be running this as the main program.") print("Warning: you're not supposed to be running this as the main program.")

View File

@@ -1,3 +1,6 @@
"""
Currently unused!
"""
import json, requests import json, requests
from APIHandler import APIHandler from APIHandler import APIHandler

View File

@@ -4,6 +4,55 @@ from APIHandler import APIHandler
from classes import * from classes import *
def call_entrypoint_from_elabid(elabid):
try:
sample_data = APIHandler(apikey).get_entry_from_elabid(elabid, entryType="items")
if not sample_data.get("category_title") == "Sample":
raise ValueError("The resource you selected is not a sample, therefore it can't be used as an entrypoint.")
sample_object = Entrypoint(sample_data)
except ConnectionError as e:
raise ConnectionError(e)
return sample_object # Entrypoint-class object
def call_material_from_elabid(elabid):
try:
material_data = APIHandler(apikey).get_entry_from_elabid(elabid, entryType="items")
# TO-DO: correct this typo on elabftw: Subtrate → Substrate.
if not sample_data.get("category_title") in ["PLD Target", "Substrate batch", "Subtrate batch"]:
raise ValueError(f"The referenced resource (elabid = {elabid}) is not a material.")
material_object = Material(material_data)
except ConnectionError as e:
raise ConnectionError(e)
return material_object # Material-class object
def call_layers_from_list(elabid_list):
list_of_layers = []
for elabid in elabid_list:
try:
layer_data = APIHandler(apikey).get_entry_from_elabid(elabid, entryType="experiments")
if not layer_data.get("category_title") == "PLD Deposition":
continue
layer_object = Layer(layer_data)
list_of_layers.append(layer_object)
except ConnectionError as e:
nums = [ layer.layer_number for layer in list_of_layers ]
nums.sort()
print(f"LISTA DEI LAYER PROCESSATI FINORA (in ordine sparso):\n" + str(nums))
raise ConnectionError(f"An error occurred while fetching the experiment with elabid = {elabid}:\n" +
str(e) + f"\nPlease solve the problem before retrying." + "\n\n" +
f"Last resource attempted to call: {ELABFTW_API_URL}/experiments/{elabid}"
)
return list_of_layers # list of Layer-class objects
def from_entrypoint_to_material(sample_object):
material_elabid = sample_object.batch_elabid
material_object = call_material_from_elabid(material_elabid)
return sample_object, material_object
sample_object = call_entrypoint_from_elabid(elabid)
from_entrypoint_to_material(sample_object)
if __name__=="__main__": if __name__=="__main__":
print(f"=======================\n===== DEBUG MODE! =====\n=======================\n") print(f"=======================\n===== DEBUG MODE! =====\n=======================\n")
ELABFTW_API_URL = "https://elabftw.fisica.unina.it/api/v2" ELABFTW_API_URL = "https://elabftw.fisica.unina.it/api/v2"
@@ -19,14 +68,14 @@ if __name__=="__main__":
eT = eT[0] eT = eT[0]
except Exception: except Exception:
print("Usage: i|item|items|i[ELABID] for items, e|experiment|experiments|e[ELABID] for experiments.") print("Usage: i|item|items|i[ELABID] for items, e|experiment|experiments|e[ELABID] for experiments.")
pass continue
match eT: match eT:
case "items" | "i" | "item": case "items" | "i" | "item":
entryType = "items" entryType = "items"
case "experiments" | "e" | "exp" | "experiment": case "experiments" | "e" | "exp" | "experiment":
entryType = "experiments" entryType = "experiments"
case _: case _:
pass continue
# This will probably be reworked in production # This will probably be reworked in production
try: try:
elabid = elabid elabid = elabid
@@ -44,6 +93,8 @@ if __name__=="__main__":
elif data.get("category_title") in ["PLD Target", "Substrate"]: elif data.get("category_title") in ["PLD Target", "Substrate"]:
item = Material(data) item = Material(data)
print(item.get_compound_formula(apikey)) print(item.get_compound_formula(apikey))
else:
raise Exception("The selected item or experiment is not in one of the following categories: [Sample, PLD Target, Substrate, PLD Deposition].")
result = item.__dict__ result = item.__dict__
result.pop("extra") result.pop("extra")
print(result) print(result)

View File

@@ -9,13 +9,13 @@ class Header:
'''Init method, apikey suggested but not required (empty by default).''' '''Init method, apikey suggested but not required (empty by default).'''
self.auth = {"Authorization" : apikey} self.auth = {"Authorization" : apikey}
self.content = {"Content-Type" : "application/json"} self.content = {"Content-Type" : "application/json"}
self.dump = {**self.auth, **self.content} self.header = {**self.auth, **self.content}
def get_entry_from_elabid(elabid, entryType="items"): def get_entry_from_elabid(elabid, entryType="items"):
''' '''
Function which returns entrypoint data (as dictionary) from its elabid. Function which returns entrypoint data (as dictionary) from its elabid.
''' '''
header = Header(apikey).dump header = Header(apikey).header
response = requests.get( response = requests.get(
headers = header, headers = header,
url = f"{ELABFTW_API_URL}/{entryType}/{elabid}", url = f"{ELABFTW_API_URL}/{entryType}/{elabid}",