first attempt to refactor the test code into something more elegant

also changes default elabid to newer resource

src/APIHandler.py

@@ -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}.")

src/classes.py

@@ -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=""):
+    def __init__(self, layer_data):
-        '''Init method, apikey suggested but not required (empty by default).'''
+        try:
-        self.auth = {"Authorization" : apikey}
+            self.extra = layer_data["metadata_decoded"]["extra_fields"]
-        self.content = {"Content-Type" : "application/json"}
+            self.target_elabid = self.extra["Target"]["value"] # elabid
-        self.dump = {**self.auth, **self.content}
+            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.")

src/functions.py

@@ -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.")

src/main.py

@@ -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
+if __name__=="__main__":
-    - Category (should check it's "PLD Deposition")
+    print(f"=======================\n===== DEBUG MODE! =====\n=======================\n")
-    - Layer number - if present (PLD depositions)
+    ELABFTW_API_URL = "https://elabftw.fisica.unina.it/api/v2"
-    - Deposition time - returns error if not present
+    apikey = getpass("Paste API key here: ")
-    - Repetition rate - returns error if not present
+    # TEST. In production the entryType will probably just be "items" since the entrypoint is an item (sample).
-    '''
+    entryType = None
-    # header = {
+    while entryType not in ["items", "experiments"]:
-    #     "Authorization": apikey,
+        eT = input("Enter a valid entry type [items, experiments]: ")
-    #     "Content-Type": "application/json"
+        # 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():
-    sample_data = requests.get(
+            try:
-        headers = header,
+                elabid = int(eT[1:])
-        url = f"https://elabftw.fisica.unina.it/api/v2/items/{elabid}",
+                eT = eT[0]
-        verify=True
+            except Exception:
-    ).json()
+                print("Usage: i|item|items|i[ELABID] for items, e|experiment|experiments|e[ELABID] for experiments.")
-    related_experiments = sample_data["related_experiments_links"]
+                pass
-    result = []
+        match eT:
-    for exp in related_experiments:
+            case "items" | "i" | "item":
-        experiment_data = requests.get(
+                entryType = "items"
-            headers = header,
+            case "experiments" | "e" | "exp" | "experiment":
-            url = f"https://elabftw.fisica.unina.it/api/v2/experiments/{exp.get("entityid")}",
+                entryType = "experiments"
-            verify=True
+            case _:
-        ).json()
+                pass
-        extra = experiment_data["metadata_decoded"]["extra_fields"]
+    # This will probably be reworked in production
-        result.append(
+    try:
-            {"title": exp.get("title"),
+        elabid = elabid
-             "layer_number": extra.get("Layer Progressive Number").get("value"),
+    except NameError:
-             "category": exp.get("category_title"),
+        elabid = input("Input elabid here [default = 1111]: ") or 1111
-             "deposition_time": extra.get("Duration").get("value"),
+    data = APIHandler(apikey).get_entry_from_elabid(elabid, entryType)
-             "repetition_rate": extra.get("Repetition rate").get("value")}
+    if entryType == "experiments":
-        )
+        layer = Layer(data)
-    return result
+        result = layer.__dict__
-
+        result.pop("extra")
-apikey = getpass("Paste API key here: ") # consider replacing with .env file
+        print(result)
-header = Header(apikey).dump
+    elif entryType == "items":
-result = get_sample_layers_data(1108) # edit id at will in case of deletion of remote source
+        if data.get("category_title") == "Sample":
-print(json.dumps(result))
+            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)

tests/chained_requests.py

@@ -1,6 +1,15 @@
 import os, json, requests
 from getpass import getpass
-from classes import Header
+
+class Header:
+    '''
+    Class to standardize the format of the headers of our http requests.
+    '''
+    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 get_entry_from_elabid(elabid, entryType="items"):
     '''
@@ -24,6 +33,8 @@ class Layer:
     '''
     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.
@@ -44,14 +55,12 @@ class Layer:
             except ValueError:
                 # Since number_of_pulses is required, if it can't be calculated raise error:
                 raise ValueError("""
-    Warning: either Duration or Repetition Rate are empty or invalid.
+    Fatal: either Duration or Repetition Rate are empty or invalid.
-    If you think this is an error, please edit your eLabFTW entry and retry.
+    This has to be an error, since these fields are required by the NeXus standard.
-    Setting Number of Pulses to NoneType.
+    Please edit your eLabFTW entry and retry.
                 """)
-            # TO-DO: remove trailing space on eLabFTW's template for deposition layers
+            self.temperature = self.extra["Heater temperature"]["value"] # Note: this field used to have a trailing space in its name
-            self.temperature = self.extra["Heater temperature "]["value"] # TYPO: trailing space, must fix on elabftw
+            self.process_pressure = self.extra["Process pressure"]["value"] # Note: this field used to have a trailing space in its name
-            self.process_pressure = self.extra["Process pressure "]["value"] # TYPO: trailing space, must fix on elabftw
-            # </todo>
             self.heating_method = self.extra["Heating Method"]["value"]
             self.layer_thickness = self.extra["Thickness"]["value"]
             self.buffer_gas = self.extra["Buffer gas"]["value"]
@@ -94,6 +103,8 @@ 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.
     '''
@@ -115,6 +126,8 @@ 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;
@@ -128,11 +141,21 @@ class Material:
             # 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(self):
+    def get_compound_data(self):
-        compound_data = get_entry_from_elabid(self.compound_elabid, entryType="items")
+        raw_compound_data = get_entry_from_elabid(self.compound_elabid, entryType="items")
-        formula = compound_data["metadata_decoded"]["extra_fields"].get("Chemical formula")
+        name = raw_compound_data["title"]
-        formula_value = formula.get("value")
+        extra = raw_compound_data["metadata_decoded"]["extra_fields"]
-        return formula_value
+        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):
+        formula = self.get_compound_data().get("chemical_formula")
+        return formula
 
 
 
@@ -163,7 +186,7 @@ if __name__=="__main__":
     try:
         elabid = elabid
     except NameError:
-        elabid = input("Input elabid here [default = 1108]: ") or 1108
+        elabid = input("Input elabid here [default = 1111]: ") or 1111
     data = get_entry_from_elabid(elabid, entryType)
     if entryType == "experiments":
         layer = Layer(data)
@@ -175,7 +198,7 @@ if __name__=="__main__":
             item = Entrypoint(data)
         elif data.get("category_title") in ["PLD Target", "Substrate"]:
             item = Material(data)
-            print(item.get_compound())
+            print(item.get_compound_formula())
         result = item.__dict__
         result.pop("extra")
         print(result)