Module pyangstrom.wrappers.caching
Expand source code
from pathlib import Path
from dataclasses import dataclass
from typing import get_type_hints
import multiprocessing
import pandas as pd
import numpy as np
from pyangstromHT.high_T_angstrom_method import (
parallel_temperature_average_batch_experimental_results,
)
from pyangstromRT.blmcmc import calculate_theoretical_results
from pyangstrom.hu.RT import TheorConfig
from pyangstrom.wrappers.data_extraction import McmcConfig, FrameConfig
from pyangstrom.wrappers.data_extraction import mcmc_analysis, get_first_frame
from pyangstrom.wrappers.helpers import ht_df_to_rt_df
class FigCacheConfig(McmcConfig, TheorConfig, FrameConfig):
pass
class CacheConfig(FigCacheConfig):
pass
class BatchData:
@dataclass
class Record:
df_temp: pd.DataFrame
df_ht_amp_phase: pd.DataFrame
def __init__(
self,
working_directory: Path | str,
parameters_file: str,
*,
__df_temp=None,
__df_ht_amp_phase=None
) -> None:
self._p_wd = Path(working_directory)
self._fn_config = parameters_file
self.__lst_df_temp = __df_temp
self.__lst_df_ht_amp_phase = __df_ht_amp_phase
def iterdata(self):
if self.__lst_df_temp is None or self.__lst_df_ht_amp_phase is None:
self.__lst_df_temp, self.__lst_df_ht_amp_phase = parallel_temperature_average_batch_experimental_results(
self._fn_config,
code_directory=f'{self._p_wd}/',
data_directory=f'{self._p_wd / "temperature data"}/',
num_cores=multiprocessing.cpu_count(),
)
return (
BatchData.Record(*v) for v
in zip(self.__lst_df_temp, self.__lst_df_ht_amp_phase)
)
class FigCache:
def __init__(
self,
working_directory: Path | str,
dict_config: FigCacheConfig,
df_rt_amp_phase: pd.DataFrame,
*,
__df_mcmc_results=None,
__arr_theor_amp=None,
__arr_theor_phase=None,
__arr_first_frame=None,
__alpha_fitting=None,
__h_fitting=None,
):
self._p_wd = Path(working_directory)
self._dict_config = dict_config
self._df_rt_amp_phase = df_rt_amp_phase
self.__df_mcmc_results = __df_mcmc_results
self.__arr_theor_amp = __arr_theor_amp
self.__arr_theor_phase = __arr_theor_phase
self.__arr_first_frame = __arr_first_frame
self.__alpha_fitting = __alpha_fitting
self.__h_fitting = __h_fitting
def get_df_mcmc_results(self):
if self.__df_mcmc_results is None:
self.__df_mcmc_results = mcmc_analysis(
self._p_wd,
self._dict_config,
self._df_rt_amp_phase,
)
return self.__df_mcmc_results
def get_arr_theor_amp(self):
if self.__arr_theor_amp is None:
self.__set_arr_theor()
return self.__arr_theor_amp
def get_arr_theor_phase(self):
if self.__arr_theor_phase is None:
self.__set_arr_theor()
return self.__arr_theor_phase
def get_arr_first_frame(self):
if self.__arr_first_frame is None:
self.__arr_first_frame = get_first_frame(
self._p_wd,
self._dict_config,
)
return self.__arr_first_frame
def __set_arr_theor(self):
self.__arr_theor_amp, self.__arr_theor_phase = calculate_theoretical_results(
self._dict_config,
self._dict_config,
self._df_rt_amp_phase,
[
self.__get_alpha_fitting(),
self.__get_h_fitting(),
]
)
def __get_alpha_fitting(self):
if not self.__alpha_fitting:
df_mcmc_results = self.get_df_mcmc_results()
self.__alpha_fitting = np.mean(df_mcmc_results['alpha'])
return self.__alpha_fitting
def __get_h_fitting(self):
if not self.__h_fitting:
df_mcmc_results = self.get_df_mcmc_results()
self.__h_fitting = np.mean(df_mcmc_results['h'])
return self.__h_fitting
class Cache:
def __init__(
self,
working_directory: Path | str,
parameters_file: str,
px=25e-6, # Length of pixel in meters
*,
__datatable: BatchData=None,
__lst_figcache: list=None
) -> None:
self._p_wd = Path(working_directory)
self._fn_config = parameters_file
self._px = px
self.__datatable = __datatable
self.__lst_figcache = __lst_figcache
def iterdata(self):
return zip(self.__get_datatable().iterdata(), self.__get_lst_fig_cache())
def __get_datatable(self):
if not self.__datatable:
self.__datatable = BatchData(self._p_wd, self._fn_config)
return self.__datatable
def __get_lst_fig_cache(self):
if not self.__lst_figcache:
df_config = pd.read_csv(
self._p_wd / 'batch process information' / self._fn_config,
usecols=get_type_hints(CacheConfig).keys(),
dtype=get_type_hints(CacheConfig),
)
datatable = self.__get_datatable()
self.__lst_figcache = [
FigCache(
self._p_wd,
config._asdict(),
ht_df_to_rt_df(batchdata.df_ht_amp_phase, self._px),
)
for config, batchdata
in zip(df_config.itertuples(), datatable.iterdata())
]
return self.__lst_figcache
Classes
class BatchData (working_directory: pathlib.Path | str, parameters_file: str)
-
Expand source code
class BatchData: @dataclass class Record: df_temp: pd.DataFrame df_ht_amp_phase: pd.DataFrame def __init__( self, working_directory: Path | str, parameters_file: str, *, __df_temp=None, __df_ht_amp_phase=None ) -> None: self._p_wd = Path(working_directory) self._fn_config = parameters_file self.__lst_df_temp = __df_temp self.__lst_df_ht_amp_phase = __df_ht_amp_phase def iterdata(self): if self.__lst_df_temp is None or self.__lst_df_ht_amp_phase is None: self.__lst_df_temp, self.__lst_df_ht_amp_phase = parallel_temperature_average_batch_experimental_results( self._fn_config, code_directory=f'{self._p_wd}/', data_directory=f'{self._p_wd / "temperature data"}/', num_cores=multiprocessing.cpu_count(), ) return ( BatchData.Record(*v) for v in zip(self.__lst_df_temp, self.__lst_df_ht_amp_phase) )
Class variables
var Record
-
Record(df_temp: pandas.core.frame.DataFrame, df_ht_amp_phase: pandas.core.frame.DataFrame)
Methods
def iterdata(self)
-
Expand source code
def iterdata(self): if self.__lst_df_temp is None or self.__lst_df_ht_amp_phase is None: self.__lst_df_temp, self.__lst_df_ht_amp_phase = parallel_temperature_average_batch_experimental_results( self._fn_config, code_directory=f'{self._p_wd}/', data_directory=f'{self._p_wd / "temperature data"}/', num_cores=multiprocessing.cpu_count(), ) return ( BatchData.Record(*v) for v in zip(self.__lst_df_temp, self.__lst_df_ht_amp_phase) )
class Cache (working_directory: pathlib.Path | str, parameters_file: str, px=2.5e-05)
-
Expand source code
class Cache: def __init__( self, working_directory: Path | str, parameters_file: str, px=25e-6, # Length of pixel in meters *, __datatable: BatchData=None, __lst_figcache: list=None ) -> None: self._p_wd = Path(working_directory) self._fn_config = parameters_file self._px = px self.__datatable = __datatable self.__lst_figcache = __lst_figcache def iterdata(self): return zip(self.__get_datatable().iterdata(), self.__get_lst_fig_cache()) def __get_datatable(self): if not self.__datatable: self.__datatable = BatchData(self._p_wd, self._fn_config) return self.__datatable def __get_lst_fig_cache(self): if not self.__lst_figcache: df_config = pd.read_csv( self._p_wd / 'batch process information' / self._fn_config, usecols=get_type_hints(CacheConfig).keys(), dtype=get_type_hints(CacheConfig), ) datatable = self.__get_datatable() self.__lst_figcache = [ FigCache( self._p_wd, config._asdict(), ht_df_to_rt_df(batchdata.df_ht_amp_phase, self._px), ) for config, batchdata in zip(df_config.itertuples(), datatable.iterdata()) ] return self.__lst_figcache
Methods
def iterdata(self)
-
Expand source code
def iterdata(self): return zip(self.__get_datatable().iterdata(), self.__get_lst_fig_cache())
class CacheConfig (*args, **kwargs)
-
dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)
Expand source code
class CacheConfig(FigCacheConfig): pass
Ancestors
- builtins.dict
Class variables
var L : float
var N_chains : int
var N_sample : int
var R0_pixels : int
var R_analysis_pixels : int
var anguler_range : str
var cp : float
var exp_amp_phase_extraction_method : str
var f_heating : float
var gap_pixels : int
var r : float
var rec_name : str
var rho : float
var x0_pixels : int
var y0_pixels : int
class FigCache (working_directory: pathlib.Path | str, dict_config: FigCacheConfig, df_rt_amp_phase: pandas.core.frame.DataFrame)
-
Expand source code
class FigCache: def __init__( self, working_directory: Path | str, dict_config: FigCacheConfig, df_rt_amp_phase: pd.DataFrame, *, __df_mcmc_results=None, __arr_theor_amp=None, __arr_theor_phase=None, __arr_first_frame=None, __alpha_fitting=None, __h_fitting=None, ): self._p_wd = Path(working_directory) self._dict_config = dict_config self._df_rt_amp_phase = df_rt_amp_phase self.__df_mcmc_results = __df_mcmc_results self.__arr_theor_amp = __arr_theor_amp self.__arr_theor_phase = __arr_theor_phase self.__arr_first_frame = __arr_first_frame self.__alpha_fitting = __alpha_fitting self.__h_fitting = __h_fitting def get_df_mcmc_results(self): if self.__df_mcmc_results is None: self.__df_mcmc_results = mcmc_analysis( self._p_wd, self._dict_config, self._df_rt_amp_phase, ) return self.__df_mcmc_results def get_arr_theor_amp(self): if self.__arr_theor_amp is None: self.__set_arr_theor() return self.__arr_theor_amp def get_arr_theor_phase(self): if self.__arr_theor_phase is None: self.__set_arr_theor() return self.__arr_theor_phase def get_arr_first_frame(self): if self.__arr_first_frame is None: self.__arr_first_frame = get_first_frame( self._p_wd, self._dict_config, ) return self.__arr_first_frame def __set_arr_theor(self): self.__arr_theor_amp, self.__arr_theor_phase = calculate_theoretical_results( self._dict_config, self._dict_config, self._df_rt_amp_phase, [ self.__get_alpha_fitting(), self.__get_h_fitting(), ] ) def __get_alpha_fitting(self): if not self.__alpha_fitting: df_mcmc_results = self.get_df_mcmc_results() self.__alpha_fitting = np.mean(df_mcmc_results['alpha']) return self.__alpha_fitting def __get_h_fitting(self): if not self.__h_fitting: df_mcmc_results = self.get_df_mcmc_results() self.__h_fitting = np.mean(df_mcmc_results['h']) return self.__h_fitting
Methods
def get_arr_first_frame(self)
-
Expand source code
def get_arr_first_frame(self): if self.__arr_first_frame is None: self.__arr_first_frame = get_first_frame( self._p_wd, self._dict_config, ) return self.__arr_first_frame
def get_arr_theor_amp(self)
-
Expand source code
def get_arr_theor_amp(self): if self.__arr_theor_amp is None: self.__set_arr_theor() return self.__arr_theor_amp
def get_arr_theor_phase(self)
-
Expand source code
def get_arr_theor_phase(self): if self.__arr_theor_phase is None: self.__set_arr_theor() return self.__arr_theor_phase
def get_df_mcmc_results(self)
-
Expand source code
def get_df_mcmc_results(self): if self.__df_mcmc_results is None: self.__df_mcmc_results = mcmc_analysis( self._p_wd, self._dict_config, self._df_rt_amp_phase, ) return self.__df_mcmc_results
class FigCacheConfig (*args, **kwargs)
-
dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)
Expand source code
class FigCacheConfig(McmcConfig, TheorConfig, FrameConfig): pass
Ancestors
- builtins.dict
Class variables
var L : float
var N_chains : int
var N_sample : int
var R0_pixels : int
var R_analysis_pixels : int
var anguler_range : str
var cp : float
var exp_amp_phase_extraction_method : str
var f_heating : float
var gap_pixels : int
var r : float
var rec_name : str
var rho : float
var x0_pixels : int
var y0_pixels : int