pymdea.core¶

def analyze_with_stripes(
    self: Self,
    fit_start: int,
    fit_stop: int,
    fit_method: Literal["siegel", "theilsen", "ls"] = "siegel",
    stripes: int = 20,
) -> Self:
    """Run a modified diffusion entropy analysis.

    Parameters
    ----------
    fit_start : float
        Fraction of maximum window length at which to start linear fit.
    fit_stop : float
        Fraction of maximum window length at which to stop linear fit.
    fit_method : str {"siegel", "theilsen", "ls"}, optional
        Linear fit method to use. By default "siegel"
    stripes : int, optional, default: 20
        Number of stripes to apply to input time-series during
        analysis.

    Returns
    -------
    Self @ Engine
        Object containing the results and inputs of the diffusion
        entropy analysis.

    Raises
    ------
    ValueError
        If n_stripes < 2. At least two stripes must be applied for
        DEA to provide a meaningful result.

    Notes
    -----
    Prefer the siegel or theilsen methods. Least squares linear
    fits can introduce bias when done over log-scale data, see
    Clauset, A., Shalizi, C.R. and Newman, M.E., 2009. Power-law
    distributions in empirical data. SIAM review, 51(4), pp.661-703.
    https://doi.org/10.1137/070710111.
    https://arxiv.org/pdf/0706.1062.pdf.

    """
    if stripes < 2:  # noqa: PLR2004
        msg = "n_stripes must be greater than 1"
        raise ValueError(msg)
    self.number_of_stripes = stripes
    self.fit_start = fit_start
    self.fit_stop = fit_stop
    self.fit_method = fit_method
    self._apply_stripes()
    self._get_events()
    self._make_trajectory()
    self._calculate_entropy()
    self._calculate_scaling()
    self._calculate_mu()
    self.print_result()

def analyze_without_stripes(
    self: Self,
    fit_start: int,
    fit_stop: int,
    fit_method: Literal["siegel", "theilsen", "ls"] = "siegel",
) -> Self:
    """Run a regular diffusion entropy analysis.

    Parameters
    ----------
    fit_start : float
        Fraction of maximum window length at which to start linear fit.
    fit_stop : float
        Fraction of maximum window length at which to stop linear fit.
    fit_method : str {"siegel", "theilsen", "ls"}, optional
        Linear fit method to use. By default "siegel"

    Returns
    -------
    Self @ Engine
        Object containing the results and inputs of the diffusion
        entropy analysis.

    Notes
    -----
    Prefer the siegel or theilsen methods. Least squares linear
    fits can introduce bias when done over log-scale data, see
    Clauset, A., Shalizi, C.R. and Newman, M.E., 2009. Power-law
    distributions in empirical data. SIAM review, 51(4), pp.661-703.
    https://doi.org/10.1137/070710111.
    https://arxiv.org/pdf/0706.1062.pdf.

    """
    self.trajectory = self.data
    self.fit_start = fit_start
    self.fit_stop = fit_stop
    self.fit_method = fit_method
    self._calculate_entropy()
    self._calculate_scaling()
    self._calculate_mu()
    self.print_result()

def print_result(self: Self) -> str:
    """Print out result of analysis."""
    self.result = Table(box=box.SIMPLE)
    self.result.add_column("δ")
    self.result.add_column("μ (rule 1)")
    self.result.add_column("μ (rule 2)")
    self.result.add_column("μ (rule 3)")
    self.result.add_row(
        f"{self.delta:.5f}",
        f"{self.mu1:.5f}",
        f"{self.mu2:.5f}",
        f"{self.mu3:.5f}",
    )
    console = Console()
    console.print(self.result)
    return self

def make_diffusion_process(
    self: Self,
    kind: Literal["cn", "gn", "fgn", "fbm"] = "gn",
    length: int = 10000,
    a: float = 0,
) -> Self:
    """Generate diffusion process data.

    Parameters
    ----------
    kind : str {"cn", "gn", "fgn", "fbm"}, optional, default "cn"
        Type of diffusion noise to generate. If "cn", generate a
        colored noise with spectral power `a`. If "gn", generate a
        Gaussian noise. If "fgn", generate a fractional Gaussian
        noise with Hurst index H = `a`. If "fbm", generate a
        fractional Brownian motion with Hurst index H=`a`.
    length : int, optional, default 10000
        Length of time-series to generate.
    a : float, optional, default 0
        Only used if `kind` is "fgn", "fbm", or "cn". If `kind` is
        "fgn" or "fbm", this sets the Hurst index of the process.
        If `kind` is "cn" this sets the index of the power law
        spectrum for the noise, 1/(f^`a`).

    Returns
    -------
    Self @ Loader
        An instance of the Loader object.

    """
    if kind == "gn":
        process = stochastic.processes.noise.GaussianNoise()
        self.data = process.sample(length)
    if kind == "cn":
        process = stochastic.processes.noise.ColoredNoise(beta=a)
        self.data = process.sample(length)
    if kind == "fgn":
        process = stochastic.processes.noise.FractionalGaussianNoise(hurst=a)
        self.data = process.sample(length)
    if kind == "fbm":
        process = stochastic.processes.continuous.FractionalBrownianMotion(hurst=a)
        self.data = process.sample(length)
    return self

def make_sample_data(self: Self, length: int = 100000, seed: int = 1) -> np.ndarray:
    """Generate an array of sample data.

    Parameters
    ----------
    length : int, optional, default: 100000
        Number of time-steps to produce in the sample data.
    seed : int, optional, default: 1
        Seed for random number generation.

    Returns
    -------
    Self @ Loader
        An instance of the Loader object.

    """
    rng = np.random.default_rng(seed=seed)
    random_walk = rng.choice([-1, 1], size=length).cumsum()
    random_walk[0] = 0  # always start from 0
    self.seed = seed
    self.data = random_walk
    return self

def read_data_file(self: Self, filepath: str, column_name: str) -> pl.DataFrame:
    """Read input data from file.

    Parameters
    ----------
    filepath : str
        System path to a file containing data. Must include the
        full file name, including the extension. Example:
        "/example/path/to/file.csv"
    column_name : str
        Name of the column in the data file which contains the time
        series data values.

    Returns
    -------
    Self @ Loader
        An instance of the Loader object.

    Raises
    ------
    ValueError
        If filepath points to a file of type other than
        CSV. Support for more types of files is a work in
        progress.

    """
    filepath = Path(filepath)
    supported_types = [".csv"]
    if filepath.suffix not in supported_types:
        msg = f"filetype must be one of: {supported_types}."
        raise ValueError(msg)
    if filepath.suffix == ".csv":
        self.data = pl.scan_csv(filepath).select(column_name).to_numpy()
    return self