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