add relevant functions

FAIRFlowChemistry/core/calibration.py

@@ -1,4 +1,5 @@
 import sdRDM
+import numpy as np
 
 from typing import List, Optional
 from uuid import uuid4
@@ -66,3 +67,25 @@ class Calibration(
     _commit: Optional[str] = PrivateAttr(
         default="ff495cb3c7e3baec101ecf174569b19e722565cc"
     )
+
+    def calibrate(self):
+        """
+        Calibrate the regression model on seen data
+        """
+
+        self.regression_coefficients = np.polynomial.polynomial.polyfit(
+            self.peak_areas.values, self.concentrations.values, self.degree
+        ).tolist()
+
+    def predict(self, x: list) -> np.ndarray:
+        """
+        Predict with regression model
+
+        Args:
+            x (1D list): New locations for which predictions should be made
+
+        Returns:
+           (1D numpy array): Predicted data at new locations
+        """
+
+        return np.polynomial.Polynomial(self.regression_coefficients)(np.array(x))

FAIRFlowChemistry/core/experiment.py

@@ -1,5 +1,8 @@
 import sdRDM
 
+import yaml
+import pandas as pd
+
 from typing import List, Optional
 from uuid import uuid4
 from pydantic import PrivateAttr
@@ -14,9 +17,10 @@
 from .metadata import Metadata
 from .measurement import Measurement
 from .data import Data
+from .datatype import DataType
 from .speciesdata import SpeciesData
 from .measurementtype import MeasurementType
-
+from .quantity import Quantity
 
 @forge_signature
 class Experiment(
@@ -146,3 +150,93 @@ def add_to_species_data(
         self.species_data.append(SpeciesData(**params))
 
         return self.species_data[-1]
+
+    def initialize_species_from_yaml(self, yaml_file: str):
+        """
+        Function that initializes species from a yaml file
+
+        Args:
+            yaml_file (str): Path to the yaml file
+        """
+
+        with open(yaml_file) as f:
+            species_data = yaml.safe_load(f)
+
+        for species, item in species_data.items():
+
+            if not "calibration" in item.keys():
+                raise KeyError(f"No calibration data provided for species: {species}!")
+            else:
+                if not "peak_areas" in item["calibration"].keys():
+                    raise KeyError(
+                        f"No peak areas provided for calibration of species: {species}!"
+                    )
+                if not "concentrations" in item["calibration"].keys():
+                    raise KeyError(
+                        "No concentrations provided for calibration of species:"
+                        f" {species}!"
+                    )
+
+            if not "chemical_formula" in item.keys():
+                raise KeyError(f"No chemical formula provided for species: {species}!")
+
+            if not "correction_factor" in item.keys():
+                raise KeyError(f"No correction factor provided for species: {species}!")
+
+            if not "electron_transfer" in item.keys():
+                raise KeyError(f"No electron transfer provided for species: {species}!")
+
+            # Create Calibration object and fit it to the given data
+            calibration = Calibration(
+                peak_areas=Data(
+                    quantity="Peak area", values=item["calibration"]["peak_areas"]
+                ),
+                concentrations=Data(
+                    quantity=Quantity.CONCENTRATION.value,
+                    values=item["calibration"]["concentrations"],
+                ),
+            )
+            calibration.calibrate()
+
+            # Add species
+            self.add_to_species_data(
+                species=species,
+                chemical_formula=item["chemical_formula"],
+                calibration=calibration,
+                correction_factor=item["correction_factor"],
+                electron_transfer=item["electron_transfer"],
+            )
+
+    @property
+    def volumetric_flow_time_course(self) -> list:
+        """This property extracts the volumetric flow time as well as the flow it self from the experiment class
+
+        Returns:
+            list: Datetime list and flow value list
+        """
+        volumetric_flow_datetime_list = []
+        volumetric_flow_values_list = []
+
+        mfm_measurements = self.get(
+            "measurements", "measurement_type", "MFM measurement"
+        )[0]
+        for mfm_measurement in mfm_measurements:
+            volumetric_flow_datetime_list.extend(
+                mfm_measurement.get(
+                    "experimental_data", "quantity", Quantity.DATETIME.value
+                )[0][0].values
+            )
+            volumetric_flow_values_list.extend(
+                mfm_measurement.get(
+                    "experimental_data", "quantity", Quantity.VOLUMETRICFLOWRATE.value
+                )[0][0].values
+            )
+
+        # If data is directly read in from the experiment, it is the correct format, if read from json dataset, it is a string and needs to be converted
+        if not type(volumetric_flow_datetime_list[0]) == DataType.DATETIME.value:
+            volumetric_flow_datetime_list = [
+                pd.to_datetime(timestamp, format="%Y-%m-%dT%H:%M:%S").to_pydatetime()
+                for timestamp in volumetric_flow_datetime_list
+            ]
+
+        return [volumetric_flow_datetime_list, volumetric_flow_values_list]