Module defs.OFR
+The docstring for this submodule
+Classes
+-
+
+class Distribution +(variable_array, function_array, form='log10') +
+-
++
Object representing a distribution.
+Description
+Bla Bla Bla
+Attributes
+-
+
var:np.ndarray
+- A 1D array containing the (independent) variable, for instance the diameter $D$. +
fct:np.ndarray
+- rA 1D array containing the dependent functional value, for instance count density $\frac{d\log_{10}N}{d\log_{10}D}$ +
+++Expand source code +
+
+class Distribution(): + + r"""Object representing a distribution. + + Description + ----------- + Bla Bla Bla + + Attributes + ---------- + var: np.ndarray + A 1D array containing the (independent) variable, for instance the diameter $D$. + fct: np.ndarray + rA 1D array containing the dependent functional value, for instance count density $\frac{d\log_{10}N}{d\log_{10}D}$ + """ + + def __init__(self, variable_array, function_array, form = 'log10'): + + self.var = variable_array + self.fct = function_array + + self.__form = form + + def transform(self, to = 'linear'): + + """A method to transform distributions between different representations + + Description + ----------- + Bla bla bla + + Parameters + ---------- + to: str + Indicating into which representation the distribution should be transformed. Possible values are: `linear`, `ln` or `log10` + """ + + if self.__form == to: + warnings.warn("Already in desired format. Transformation has no effect.") + return + + # TODO: check the transformations! + # TODO: check the order of the transformation (first variable, then function, or vice versa) to avoid having a mess! + + match to: + case 'linear': + if self.__form == 'ln': + self.var = np.exp(self.var) + self.fct = np.divide(self.fct, self.var) + self.__form = 'linear' + elif self.__form == 'log10': + self.var = np.power(10, self.var) + self.fct = np.divide(self.fct, self.var * np.log(10)) + self.__form = 'linear' + case 'ln': + if self.__form == 'linear': + self.var = np.log(self.var) + self.fct = np.multiply(self.var, self.fct) + self.__form = 'ln' + elif self.__form == 'log10': + self.var = np.log(np.power(10, self.var)) + self.fct = np.divide(self.fct, np.log(10)) + self.__form = 'ln' + case 'log10': + if self.__form == 'linear': + self.fct = np.multiply(self.var, self.fct * np.log(10)) + self.var = np.log10(self.var) + self.__form = 'log10' + elif self.__form == 'ln': + self.var = np.log10(np.exp(self.var)) + self.fct = np.multiply(self.fct, np.log(10)) + self.__form = 'log10' + + def update_units(self, new_var_units, new_fct_units): + + """A method to update the units of the distribution + + Description + ----------- + Bla Bla Bla + + Parameters + ---------- + new_var_units: float + The SI factor used to transform the independent variable `self.var` + new_fct_units: float + The SI factor used to transform the dependent variable `self.fct` + """ + + old_from = self.__form + self.transform(to='linear') + + self.var = self.var * new_var_units + self.fct = self.fct * new_fct_units + + self.transform(to=old_from) + + def integrate(self) -> np.float64: + + """A method to integrate over the spectrum $dN/d f(D)$ + + Description + ----------- + Bla Bla Bla + """ + + return np.trapezoid(y = self.fct, x = self.var) + + def median(): + # TODO: implement that method + pass + + def mode(): + # TODO: implement that method + pass + + def mean(): + # TODO: implement that method + pass + + def variance(): + # TODO: implement that method + passMethods
+-
+
+def integrate(self) ‑> numpy.float64 +
+-
++
A method to integrate over the spectrum $dN/d f(D)$
+Description
+Bla Bla Bla
+ +def mean() +
+- + + +
+def median() +
+- + + +
+def mode() +
+- + + +
+def transform(self, to='linear') +
+-
++
A method to transform distributions between different representations
+Description
+Bla bla bla
+Parameters
+-
+
to:str
+- Indicating into which representation the distribution should be transformed. Possible values are:
linear,lnorlog10
+
+ +def update_units(self, new_var_units, new_fct_units) +
+-
++
A method to update the units of the distribution
+Description
+Bla Bla Bla
+Parameters
+-
+
new_var_units:float
+- The SI factor used to transform the independent variable
self.var
+ new_fct_units:float
+- The SI factor used to transform the dependent variable
self.fct
+
+ +def variance() +
+- + + +
+ +class OFRAnalyser +(data_source_file, flows_source_file, sample_number_list, data_separator=',') +
+-
++
An object representing an OFR experiment
+Description
+Bla Bla Bla
+Attributes
+-
+
data_source_file:str
+- A string pointing to the data source file +
flows_source_file:str
+- A string pointing to the flow data source file +
sample_number_list:[int]
+- A list of integers indicating which samples to consider +
max_samples:int
+- The highest sample available in the data +
data:pandas.DataFrame
+- A dataframe containing all the data +
+++Expand source code +
+
+class OFRAnalyser: + + """An object representing an OFR experiment + + Description + ----------- + Bla Bla Bla + + Attributes + ---------- + data_source_file: str + A string pointing to the data source file + flows_source_file: str + A string pointing to the flow data source file + sample_number_list: [int] + A list of integers indicating which samples to consider + max_samples: int + The highest sample available in the data + data: pandas.DataFrame + A dataframe containing all the data + """ + + def __init__(self, data_source_file, flows_source_file, sample_number_list, data_separator = ',') -> None: + + self.data_source_file = data_source_file + self.flows_source_file = flows_source_file + self.sample_number_list = sample_number_list + self.__data_separator = data_separator + + with open(self.data_source_file, 'r', errors = 'ignore') as file: + for i, line in enumerate(file.readlines()): + if i == 17: + items = line.split(',') + max_sample_number = int(len(items) / 2) + else: + continue + + self.max_samples = max_sample_number + self.data = None + + def __sample_number_list_check(self, sample_number_list) -> None: + + # TODO: implement a similar check for the flow data file! + + last_sample = sample_number_list[-1] + + if last_sample > self.max_samples: + raise IndexError("The last sample(s) are out of range of the number of recorded samples") + + def GetData(self) -> None: + + """Method to get the data + + Description + ----------- + Bla Bla Bla + """ + + # TODO make this work also for a specific sample! + # TODO convert everything is SI units, making live a lot easier... + + self.__sample_number_list_check(self.sample_number_list) + + numbers_of_sample = [] # (-) + number_concentrations = [] # (#/m^3) + length_concentrations = [] # (m/m^3) + surface_concentrations = [] # (m^2/m^3) + volume_concentrations = [] # (m^3/m^3) + aerosol_densities = [] # (kg/m^3) + mass_concentrations = [] # (kg/m^3) + starting_times = [] # (s) + stopping_times = [] # (s) + sampling_times = [] # (s) + dry_air_flows = [] # (m^3/s) + species_flows = [] # (m^3/s) + ozone_flows = [] # (m^3/s) + total_flows = None # (m^3/s) + cpc_flows = [] # (m^3/s) + mass_production_rate = None # (kg/s) + mass_produced = None # (kg) + cumulative_mass_produced = None # (kg) + + + for sample_number in self.sample_number_list: + sample = OFRSample(source_file=self.data_source_file, sample_number = sample_number, data_separator = self.__data_separator) + + numbers_of_sample.append(sample_number) + number_concentrations.append(sample.Count.integrate()) + length_concentrations.append(sample.Length.integrate()) + surface_concentrations.append(sample.Surface.integrate()) + volume_concentrations.append(sample.Volume.integrate()) + aerosol_densities.append(sample.SOA_density) + mass_concentrations.append(sample.Mass.integrate()) + + starting_times.append((sample_number - 1) * sample.sampling_time) + stopping_times.append(sample_number * sample.sampling_time) + + sampling_times.append(sample.sampling_time) + + out_flag = False + + with open(self.flows_source_file, 'r', errors='ignore') as file: + + j = 1 + + for line_counter, line in enumerate(file.readlines()): + + if out_flag == True: + continue + + if line_counter >= 6: + + if line.startswith('-'): + out_flag = True + continue + + i = 0 + items = line.split() + + while i < int(items[-1]): + + if j in self.sample_number_list: + dry_air_flows.append(float(items[0]) * 6E-4) + species_flows.append(float(items[1]) * 6E-4) + ozone_flows.append(float(items[2]) * 6E-4) + cpc_flows.append(float(items[3]) * 6E-4) + + i += 1 + j += 1 + + else: + continue + + total_flows = [(flow_1 + flow_2 + flow_3) for flow_1, flow_2, flow_3 in zip(dry_air_flows, species_flows, ozone_flows)] + mass_production_rate = [(mass * flow) for mass, flow in zip(mass_concentrations, total_flows)] + mass_produced = [(mpr * duration) for mpr, duration in zip(mass_production_rate, sampling_times)] + cumulative_mass_produced = np.cumsum(np.array(mass_produced)).tolist() + + self.data = pd.DataFrame({'Sample (-)': numbers_of_sample, + 'Number (#/m^3)': number_concentrations, + 'Length (m/m^3)': length_concentrations, + 'Surface (m^2/m^3)': surface_concentrations, + 'Volume (m^3/m^3)': volume_concentrations, + 'Density (kg/m^3)': aerosol_densities, + 'Mass (kg/m^3)': mass_concentrations, + 'Start (s)': starting_times, + 'Stop (s)': stopping_times, + 'Duration (s)': sampling_times, + 'Dry Air Flow (m^3/s)': dry_air_flows, + 'Species Flow (m^3/s)': species_flows, + 'Ozone Flow (m^3/s)': ozone_flows, + 'Total Flow (m^3/s)': total_flows, + 'CPC Flow (m^3/s)': cpc_flows, + 'Mass Production Rate (kg/s)': mass_production_rate, + 'Mass Produced (kg)': mass_produced, + 'Cumulative Mass Produced (kg)': cumulative_mass_produced})Methods
+-
+
+def GetData(self) ‑> None +
+-
++
Method to get the data
+Description
+Bla Bla Bla
+
+ +class OFRSample +(source_file, sample_number, data_separator=',') +
+-
++
Object to represent a single sample from the OFR
+Description
+Bla bla bla
+Attributes
+-
+
number:int
+- The sample number +
sourcefile:str
+- The path to the data file +
headlines:int
+- The number of lines in the data file until raw data is reached +
lower_bound:float
+- The lower bound diameter (nm) of the SMPS +
upper_bound:float
+- The upper bound diameter (nm) of the SMPS +
scan_up:float
+- The time (s) to ramp up voltage in SMPS +
scan_down:float
+- The time (s) to ramp voltage back down in SMPS +
delta_time:float
+- The time interval (s) between two measured data points of the raw count density +
SOA_density:float
+- The density (kg/m3) of the SOA produced in the OFR +
+++Expand source code +
+
+class OFRSample: + + """Object to represent a single sample from the OFR + + Description + ----------- + Bla bla bla + + Attributes + ---------- + + number: int + The sample number + sourcefile: str + The path to the data file + headlines: int + The number of lines in the data file until raw data is reached + lower_bound: float + The lower bound diameter (nm) of the SMPS + upper_bound: float + The upper bound diameter (nm) of the SMPS + scan_up: float + The time (s) to ramp up voltage in SMPS + scan_down: float + The time (s) to ramp voltage back down in SMPS + delta_time: float + The time interval (s) between two measured data points of the raw count density + SOA_density: float + The density (kg/m3) of the SOA produced in the OFR + """ + + def __get_data(self) -> np.float64: + + footer_flag = False + footer_start_index = 0 + data_flag = False + data_lines = [] + + midpoints = [] + counts = [] + + with open(self.source_file, 'r', errors = 'ignore') as file: + for i, line in enumerate(file.readlines()): + if i == 20: + data_flag = True + continue + if data_flag == True: + data_lines.append(line) + if line.split(self.__data_separator)[0] == 'Scan Up Time(s)': + data_flag = False + footer_flag = True + footer_start_index = i + self.scan_up_time = float(line.split(self.__data_separator)[1]) + continue + if footer_flag == False: + continue + else: + if i == footer_start_index + 1: + self.scan_down_time = float(line.split(self.__data_separator)[1]) + continue + elif i == footer_start_index + 6: + self.aerosol_flow = float(line.split(self.__data_separator)[1]) + continue + elif i == footer_start_index + 11: + self.lower_diameter_bound = float(line.split(self.__data_separator)[1]) + continue + elif i == footer_start_index + 12: + self.upper_diameter_bound = float(line.split(self.__data_separator)[1]) + continue + elif i == footer_start_index + 13: + self.SOA_density = float(line.split(self.__data_separator)[1]) + continue + + self.sampling_time = self.scan_up_time + self.scan_down_time + data_lines.pop(-1) + + for line in data_lines: + items = line.split(self.__data_separator) + local_diameter_midpoint = float(items[0]) + if self.lower_diameter_bound <= local_diameter_midpoint and local_diameter_midpoint <= self.upper_diameter_bound: + midpoints.append(local_diameter_midpoint) + items.pop(0) + counts.append(float(items[(self.sample_number - 1) * 2])) + else: + continue + + self.aerosol_flow = self.aerosol_flow * 6E-4 + self.SOA_density = self.SOA_density * 1E3 + self.lower_diameter_bound = self.lower_diameter_bound * 1E-9 + self.upper_diameter_bound = self.upper_diameter_bound * 1E-9 + + + # Input data from file is in (nm) + return [np.array(midpoints), np.array(counts)] + + def __init__(self, source_file, sample_number, data_separator = ','): + + ## VALUES READ IN FROM DISK + #-------------------------- + + # Sample information + self.source_file = source_file # where the data is stored, i.e. "C:/Users/tschan_d/..." + self.sample_number = sample_number # sample number, i.e. 33 + self.__data_separator = data_separator + + self.scan_up_time = None # Time to scan up the voltage range in DMA, in (s) + self.scan_down_time = None # Time to scan down the voltage range in DMA, in(s) + self.sampling_time = None # Total time to scan voltage up and back down in DMA, in (s) + + self.lower_diameter_bound = None # Lower Diameter bound of CPC, in (nm) + self.upper_diameter_bound = None # Upper Diameter bound of CPC, in (nm) + self.aerosol_flow = None # Aerosol flow, in (L/min) + self.SOA_density = None # Expected density of SOA, in (g/cm3) + + # Distributions of the relevant quantities (all relative to nanometer) + data = self.__get_data() + + # Getting Distribution objects + self.Count = Distribution(variable_array = np.log10(data[0]), function_array = data[1], form='log10') + self.Count.update_units(new_fct_units=1E15, new_var_units=1E-9) + + self.Length = Distribution(variable_array = np.log10(data[0]), function_array = data[0]*data[1], form = 'log10') + self.Length.update_units(new_fct_units=1E6, new_var_units=1E-9) + + self.Surface = Distribution(variable_array = np.log(data[0]), function_array = ((math.pi)* data[0] ** 2) * data[1], form = 'log10') + self.Surface.update_units(new_fct_units=1E-3, new_var_units=1E-9) + + self.Volume = Distribution(variable_array = np.log10(data[0]),function_array = ((math.pi / 6) * data[0] ** 3) * data[1], form = 'log10') + self.Volume.update_units(new_fct_units=1E-12, new_var_units=1E-9) + + self.Mass = Distribution(variable_array = self.Volume.var, function_array = self.Volume.fct * self.SOA_density, form='log10')
+