Module defs.SMPS
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, flow_source_file, smpl_number_list, on_time_interval, 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
flow_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
on_time_interval:[pd.Timestamp]- A list with two elements containing the pd.Timestamp objects when the VOC flow was switched on and off, respectively.
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 flow_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 on_time_interval: [pd.Timestamp] A list with two elements containing the pd.Timestamp objects when the VOC flow was switched on and off, respectively. 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, flow_source_file, smpl_number_list, on_time_interval, data_separator = ',') -> None: self.data_source_file = data_source_file self.flow_source_file = flow_source_file self.smpl_number_list = smpl_number_list self.on_time_interval = on_time_interval 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 __smpl_number_list_check(self, smpl_number_list) -> None: # TODO: implement a similar check for the flow data file! last_sample = smpl_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.__smpl_number_list_check(self.smpl_number_list) # Quantities of interest: numbers_of_sample = [] # (-) starting_times = [] # (timestamp) sampling_times = [] # (s) 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) filter_flows = [] # (m^3/s) mass_production_rate = None # (kg/s) mass_produced = None # (kg) cumulative_mass_produced = None # (kg) for sample_number in self.smpl_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) starting_times.append(sample.start_time) 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()) sampling_times.append(sample.sampling_time) out_flag = False with open(self.flow_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.smpl_number_list: filter_flows.append(float(items[0]) * (1E-3) / 60) i += 1 j += 1 else: continue mass_production_rate = [(mass * flow) for mass, flow in zip(mass_concentrations, filter_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, 'Starting Time (-)': starting_times, 'Sampling Time (s)': sampling_times, '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, 'Filter Flow (m^3/s)': filter_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
start_time:pd.datetime- The starting time of the sample recording
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 start_time: pd.datetime The starting time of the sample recording 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 = [] date_line = None time_line = None with open(self.source_file, 'r', errors = 'ignore') as file: for i, line in enumerate(file.readlines()): if i == 18: date_line = line continue if i == 19: time_line = line continue 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) # obtaining the starting time raw_date_items = date_line.split(self.__data_separator) raw_time_items = time_line.split(self.__data_separator) del raw_date_items[0] del raw_time_items[0] raw_date = raw_date_items[(self.sample_number - 1) * 2].splitlines()[0] raw_time = raw_time_items[(self.sample_number - 1) * 2].splitlines()[0] self.start_time = pd.to_datetime(raw_date + ' ' + raw_time, format=r'%m/%d/%y %H:%M:%S') 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.start_time = None 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')