Compare commits

..

2 Commits

Author SHA256 Message Date
865f5cab6b untested: adds methods to Layer class to fetch attachments list
one method fetches all
one filters textual uploads
one filters png and bmp images
2026-05-08 23:40:14 +02:00
0102bb282e improves documentation, tabbing and error handling in APIHandler class
Claude Code helped with autocompletion, the rest is my work
2026-05-08 23:31:36 +02:00
2 changed files with 97 additions and 26 deletions

View File

@@ -4,7 +4,18 @@ import elabapi_python as elabapi
class APIHandler: class APIHandler:
""" """
Class to standardize the format of the headers of our http requests. Class which handles all interactions with the eLabFTW API.
It provides methods to retrieve data from the API and download attachments.
It relies minimally on the elabapi-python library, which is used only for downloading attachments
(since the API doesn't support downloading attachments AFAIK).
Args:
api_key: A valid API key for the eLabFTW instance where the data is stored, with permissions to access the relevant entries.
eLabFTW's API keys are well documented here: https://doc.elabftw.net/docs/usage/api/.
If you don't have an API key and are uncapable of creating one, contact your eLabFTW administrator.
Or RTFM and create one yourself, it's not that hard.
ELABFTW_API_URL: Complete URL of the eLabFTW instance's root for the API endpoints.
In full caps because it won't (shouldn't) be changed much.
""" """
# TO-DO: remove static url. # TO-DO: remove static url.
@@ -20,40 +31,54 @@ class APIHandler:
def get_entry_from_elabid(self, elabid, entryType="items"): 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. Returns raw data (as dictionary) from its elabid and entry type.
Entry type can be either "experiments" or "items". args:
elabid: elabftw internal id of the selected resource.
entryType: Resource type. Anything other than "experiments" or "items" WILL raise an error.
""" """
# TO-DO: validation and error handling on entryType value. if entryType not in ["experiments", "items"]:
raise Exception(
"You can only download attachments from experiments or items."
)
header = self.header header = self.header
response = requests.get( response = requests.get(
headers=header, url=f"{self.elaburl}/{entryType}/{elabid}", verify=True headers=header, url=f"{self.elaburl}/{entryType}/{elabid}", verify=True
) )
if response.status_code // 100 in [1, 2, 3]:
entry_data = response.json() # Response is 5xx = server error:
return entry_data if response.status_code // 100 == 5:
elif response.status_code // 100 == 4: raise ConnectionError(
f"There's a problem on the server. Status code: {response.status_code}."
)
# Response is 4xx = client error:
if response.status_code // 100 == 4:
match response.status_code: match response.status_code:
case 401 | 403: case 401 | 403:
# Forbidden or unauthorized:
raise ConnectionError( raise ConnectionError(
f"Invalid API key, authentication method or elabid. Check if an item with ID = {elabid} actually exists." f"Invalid API key, authentication method or elabid. Check if an item with ID = {elabid} actually exists."
) )
case 404: case 404:
# Lapalissian:
raise ConnectionError( raise ConnectionError(
f"404: Not Found. This means there's no resource with this elabid (wrong elabid?) on your eLabFTW (wrong endpoint?)." f"404: Not Found. This means there's no resource with this elabid (wrong elabid?) on your eLabFTW (wrong endpoint?)."
) )
case 400: case 400:
# I genuinely have no idea:
raise ConnectionError( raise ConnectionError(
f"400: Bad Request. This means the API endpoint you tried to reach is invalid. Did you tamper with the source code? If not, contact the developer." f"400: Bad Request. This means the API endpoint you tried to reach is invalid. Did you tamper with the source code? If not, contact the developer."
) )
case _: case _:
# For some fucking reason, this is the only error I actually get from the API...
raise ConnectionError( raise ConnectionError(
f"HTTP request failed with status code: {response.status_code} (NOTE: 4xx means user's fault)." f"HTTP request failed with status code: {response.status_code} (NOTE: 4xx means user's fault)."
) )
else:
raise ConnectionError( entry_data = response.json()
f"There's a problem on the server. Status code: {response.status_code}." return entry_data
)
def download_attachments_data(self, elabid, entryType="experiments"): def download_attachments_data(self, elabid, entryType="experiments"):
""" """
@@ -123,4 +148,3 @@ class APIHandler:
with open(os.path.join(dump_dir, f"exp{elabid}-{file}"), "wb") as f: with open(os.path.join(dump_dir, f"exp{elabid}-{file}"), "wb") as f:
f.write(raw_data) f.write(raw_data)
return return

View File

@@ -15,8 +15,10 @@ class Layer:
def __init__(self, layer_data): def __init__(self, layer_data):
try: try:
self.elabid = layer_data["id"]
self.operator = layer_data["fullname"] self.operator = layer_data["fullname"]
self.extra = layer_data["metadata_decoded"]["extra_fields"] self.extra = layer_data["metadata_decoded"]["extra_fields"]
self.uploads = layer_data["uploads"] # dict
self.layer_number = self.extra["Layer Progressive Number"][ self.layer_number = self.extra["Layer Progressive Number"][
"value" "value"
] # integer ] # integer
@@ -127,14 +129,14 @@ class Layer:
self.start_time = layer_data.get("created_at") or None self.start_time = layer_data.get("created_at") or None
self.description = layer_data.get("body") or None self.description = layer_data.get("body") or None
def get_instruments(self, apikey): def get_instruments(self, api_key):
raw_lasersys_data = APIHandler(apikey).get_entry_from_elabid( raw_lasersys_data = APIHandler(api_key).get_entry_from_elabid(
self.laser_system_elabid, entryType="items" self.laser_system_elabid, entryType="items"
) )
raw_chamber_data = APIHandler(apikey).get_entry_from_elabid( raw_chamber_data = APIHandler(api_key).get_entry_from_elabid(
self.chamber_elabid, entryType="items" self.chamber_elabid, entryType="items"
) )
raw_rheedsys_data = APIHandler(apikey).get_entry_from_elabid( raw_rheedsys_data = APIHandler(api_key).get_entry_from_elabid(
self.rheed_system_elabid, entryType="items" self.rheed_system_elabid, entryType="items"
) )
instruments_used = { instruments_used = {
@@ -144,15 +146,61 @@ class Layer:
} }
return instruments_used return instruments_used
# Three possible approaches for the next two methods: either return the raw data as retrieved from eLabFTW, def list_attachments(self):
# or process it and return only the relevant information, or even simply return the list of filenames. """
# Returns a dictionary of all the attachments linked to the layer, where:
def fetch_textual_uploads(self, api_key): * Each key is the attachment's elabid;
""" """ * Each value is a dictionary containing the attachment's filename, hashname and related experiment elabid (= self.elabid).
return
def fetch_rheed_images(self, api_key): Data is already in layer_data, so the API key is unrequired. Same goes for:
return * fetch_textual_uploads() - no arguments;
* fetch_images() - no arguments.
"""
# Remember: Layers are experiments, so we only need to look for attachments in the experiment endpoint.
attachments = {
attachment["id"]: {
"filename": attachment["real_name"],
"hashname": attachment["long_name"],
"related_experiment": attachment["item_id"],
}
for attachment in self.uploads
}
return attachments
def fetch_textual_uploads(self):
"""
Starting from the list of attachments, filters out and returns a list of the textual uploads linked to the layer, which can be either plain text, csv, tsv etc.
Returns only their names, so that the user may select which one to import into the NeXus file as a dataset.
It only looks for .txt, .csv and .tsv files, although it could be easily modified to include other formats.
It is also file extension-sensitive, so anything not ending with .txt, .csv or .tsv won't be retrieved.
That's because the API (v5.3.11) doesn't provide MIME Type or similar metadata on the attachments, so the only way to know if an attachment is an image or not is through its filename.
"""
attachments = self.list_attachments()
textual_uploads = {
attachment: attachments[attachment]
for attachment in attachments
if attachments[attachments]["filename"][-4:] in (".txt", ".csv", ".tsv")
}
return textual_uploads
def fetch_images(self):
"""
Starting from the list of attachments, filters out and returns a Starting from the list of attachments, filters out and returns a list of all the (PNG or BMP) images attached to the layer.
Hopefully one of them is a RHEED pattern.
Returns only their names, so that the user may select which one to import into the NeXus file as a RHEED acquisition.
It only looks for .png and .bmp files, although it could be easily modified to include other formats.
It is also file extension-sensitive, so anything not ending with .png or .bmp won't be retrieved, even if it's an actual image.
That's because the API (v5.3.11) doesn't provide MIME Type or similar metadata on the attachments, so the only way to know if an attachment is an image or not is through its filename.
"""
attachments = self.list_attachments()
pictures = {
attachment: attachments[attachment]
for attachment in attachments
if attachments[attachments]["filename"][-4:] in (".png", ".bmp")
}
return pictures
class Entrypoint: class Entrypoint:
@@ -280,4 +328,3 @@ if __name__ == "__main__":
head = APIHandler("MyApiKey-123456789abcdef") head = APIHandler("MyApiKey-123456789abcdef")
print(f"Example header:\n\t{head.header}\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.")