epysurv

https://epysurv.readthedocs.io/en/latest/
Admin
epysurv is a Python wrapper around the R surveillance package for predicting disease outbreaks using...
Tokens:46,689
Snippets:290
Trust Score:9.6
Update:5 months ago
Show doc for...
Tokens:
Raw
### Install epysurv using Conda

Source: https://epysurv.readthedocs.io/en/latest/1_quickstart.html

Follow these steps to create a new conda environment and install epysurv.

```bash
conda create -n epysurv
conda activate epysurv
conda install -c conda-forge epysurv

```

--------------------------------

### Generate Outbreak Start Indices

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/simulation/naive_poisson.html

Determines random start indices for outbreaks, ensuring they do not overlap and are within valid bounds. It iteratively selects start points and removes surrounding weeks from consideration to prevent overlaps.

```python
import random
from typing import Set

import numpy as np
import pandas as pd
from scipy import stats



[docs]
def get_outbreak_begins(n: int, outbreak_length: int, n_outbreaks: int) -> Set[int]:
    possible_outbreaks_starts = set(range(outbreak_length, n - outbreak_length - 1))
    outbreaks_starts = set()

    for _ in range(n_outbreaks):
        start = random.choice(tuple(possible_outbreaks_starts))
        outbreaks_starts.add(start)
        for i in range(start - outbreak_length, start + outbreak_length * 2):
            try:
                possible_outbreaks_starts.remove(i)
            except KeyError:
                continue
    return outbreaks_starts
```

--------------------------------

### Input Validation for Expanding Windows

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/data/filter_combination.html

Validates input parameters for the expanding_windows method. Checks if the start date is after the first case, and if the start date plus the minimum length offset is before the middle date.

```python
    def _validate_input(self, min_len_in_weeks, split_years):
        # TODO: while this makes sense, it is turned off for now, because most of the data, does not
        # have a case in 2019 yet.
        # if self.data.ReportingDate.max() < split_years.end:
        #     raise ValueError(f'The end date must be before the last case, but is {split_years.end}')
        if split_years.start < self.data.ReportingDate.min():
            raise ValueError(
                f"The start date must be after the first case, but is {split_years.start}"
            )
        if split_years.middle < split_years.start + timedelta_weeks(min_len_in_weeks):
            raise ValueError(
                f"The start date plus the offset must be before the middle date, "
                f"but is {split_years.start + timedelta_weeks(min_len_in_weeks)}"
            )
```

--------------------------------

### Simulate Outbreaks with Poisson Distribution

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/simulation/naive_poisson.html

Simulates weekly case counts using Poisson distributions for baseline and outbreak periods. It generates baseline cases and then adds outbreak cases at determined start times, ensuring at least one case during an outbreak. Returns a pandas DataFrame with detailed case counts.

```python
def simulate_outbreaks(
    n: int = 104,
    outbreak_length: int = 5,
    n_outbreaks: int = 3,
    mu: float = 1,
    outbreak_mu: float = 10,
) -> pd.DataFrame:
    """Simulate outbreaks based on Poisson distribution.

    Parameters
    ----------
    n
        Number of weeks.
    outbreak_length
        Number of weeks each outbreak is long.
    n_outbreaks
        Number of outbreaks.
    mu
        Mean for the baseline.
    outbreak_mu
        Mean for the outbreaks.

    Returns
    -------
        Simulated case counts per week, separated into baseline and outbreak cases.
    """
    baseline = stats.poisson.rvs(mu=mu, size=n)
    n_outbreak_cases = np.zeros_like(baseline)
    n_cases = baseline.copy()

    outbreaks_starts = get_outbreak_begins(n, outbreak_length, n_outbreaks)

    for start in outbreaks_starts:
        outbreak_cases = stats.poisson.rvs(mu=outbreak_mu, size=outbreak_length)
        outbreak_cases += (
            outbreak_cases == 0
        )  # Ensure that there is a least on case during the outbreak.
        n_cases[start : start + outbreak_length] += outbreak_cases
        n_outbreak_cases[start : start + outbreak_length] = outbreak_cases

    data = pd.DataFrame(
        {
            "n_cases": n_cases,
            "n_outbreak_cases": n_outbreak_cases,
            "outbreak": n_outbreak_cases > 0,
            "baseline": baseline,
        },
        index=pd.date_range(start="2020", periods=baseline.size, freq="W-MON"),
    )

    data.index.name = "date"
    return data
```

--------------------------------

### Configure GLR Surveillance Algorithm

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/models/timepoint/glr.html

Sets up control parameters for the `surveillance.glrpois` function. This includes defining the detection range, ARL threshold, window size (M), change type, direction of testing, and the type of upperbound statistic to return.

```python
control = r.list(
            **{
                "range": detection_range,
                "c.ARL": self.glr_test_threshold,
                "m0": robjects.NULL,
                # Mtilde is set to 1, since that is the only valid value for "epi" and "intercept"
                "Mtilde": 1,
                "M": self.m,
                "change": self.change,
                # Role of theta: If NULL then the GLR scheme is used. If not NULL the prespecified value for κ or λ is used in a recursive LR scheme, which is faster."
                "theta": robjects.NULL,
                "dir": r.c(*self.direction),
                "ret": self.upperbound_statistic,
            }
        )
```

--------------------------------

### RKI Algorithm Implementation

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/models/timepoint/rki.html

Defines the RKI class, inheriting from STSBasedAlgorithm. It configures parameters for the RKI surveillance algorithm, including historical data inclusion and window sizes.

```python
from dataclasses import dataclass

from rpy2.robjects import r
from rpy2.robjects.packages import importr

from ._base import STSBasedAlgorithm

surveillance = importr("surveillance")



@dataclass
class RKI(STSBasedAlgorithm):
    """The old algorithm from the Robert Koch Institute.

    Attributes
    ----------
    years_back
        How many years back in time to include when forming the base counts.
    window_half_width
        Number of weeks to include before and after the current week in each year.
    include_recent_year
        Is a boolean to decide if the year of timePoint also contributes w reference values.
    """

    years_back: int = 0
    window_half_width: int = 6
    include_recent_year: bool = True

    def _call_surveillance_algo(self, sts, detection_range):
        control = r.list(
            range=detection_range,
            b=self.years_back,
            w=self.window_half_width,
            actY=self.include_recent_year,
        )

        surv = surveillance.rki(sts, control=control)
        return surv

```

--------------------------------

### SplitYears Class Definition

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/data/filter_combination.html

Defines a data structure to hold start, middle, and end timestamps for splitting time series data into training and testing sets. Ensures that the timestamps are consecutive.

```python
from collections import namedtuple
from dataclasses import dataclass, field

import pandas as pd

from .utils import timedelta_weeks



[docs]class SplitYears:
    """Data structure that holds the years data should be split into training and test set.

    start to middle is the training data. middle to end is the test data.
    """

    def __init__(self, start: pd.Timestamp, middle: pd.Timestamp, end: pd.Timestamp):
        if not (start < middle < end):
            raise ValueError("start, middle and end must be consecutive.")
        self.start = start
        self.middle = middle
        self.end = end


[docs]    @classmethod
    def from_ts_input(cls, start, middle, end):
        """Create instance from inputs that are passed through ``pd.Timestamp``."""
        start = pd.Timestamp(start)
        middle = pd.Timestamp(middle)
        end = pd.Timestamp(end)
        return cls(start, middle, end)
```

--------------------------------

### Cusum Model Implementation

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/models/timepoint/cusum.html

This snippet defines the Cusum class, inheriting from STSBasedAlgorithm. It includes attributes for configuring the CUSUM algorithm, such as reference value, decision boundary, and methods for calculating expected numbers and data transformation. The `_call_surveillance_algo` method demonstrates how to set up and call the R 'surveillance' package's CUSUM function with the specified parameters.

```python
from dataclasses import dataclass
from typing import *  # NOQA

from rpy2 import robjects
from rpy2.robjects import r
from rpy2.robjects.packages import importr

from ._base import STSBasedAlgorithm

surveillance = importr("surveillance")




@dataclass
class Cusum(STSBasedAlgorithm):
    r"""The Cusum model.

    Attributes
    ----------
    reference_value
    decision_boundary
    expected_numbers_method
        How to determine the expected number of cases – the following arguments are possible: {"glm", "mean"}.

        ``mean``
            Use the mean of all data points passed to ``fit``.
        ``glm``
            Fit a glm to the data ponts passed to ``fit``.
    transform
        One of the following transformations (warning: Anscombe and NegBin transformations are experimental)
        - standard standardized variables z1 (based on asymptotic normality) - This is the default.
        - rossi standardized variables z3 as proposed by Rossi
        - anscombe anscombe residuals – experimental
        - anscombe2nd anscombe residuals as in Pierce and Schafer (1986) based on 2nd order approximation of E(X) – experimental
        - pearsonNegBin compute Pearson residuals for NegBin – experimental
        - anscombeNegBin anscombe residuals for NegBin – experimental
        - ``"none"`` no transformation
    negbin_alpha
        Parameter of the negative binomial distribution, such that the variance is :math:`m + α \cdot m2`.

    References
    ----------
    .. [1] G. Rossi, L. Lampugnani and M. Marchi (1999), An approximate CUSUM procedure for surveillance of health events, Statistics in Medicine, 18, 2111–2122
    .. [2] D. A. Pierce and D. W. Schafer (1986), Residuals in Generalized Linear Models, Journal of the
        American Statistical Association, 81, 977–986
    """

    reference_value: float = 1.04
    decision_boundary: float = 2.26
    expected_numbers_method: str = "mean"
    transform: str = "standard"
    negbin_alpha: float = 0.1

    def _call_surveillance_algo(self, sts, detection_range):
        control = r.list(
            range=detection_range,
            k=self.reference_value,
            h=self.decision_boundary,
            m=robjects.NULL
            if self.expected_numbers_method == "mean"
            else self.expected_numbers_method,
            trans=self.transform,
            alpha=self.negbin_alpha,
        )
        surv = surveillance.cusum(sts, control=control)
        return surv

```

--------------------------------

### Fit EarsC1 Model and Predict

Source: https://epysurv.readthedocs.io/en/latest/demo.html

Initializes an EarsC1 model, fits it to the training data, and generates predictions on the test data. Displays the first few rows of the predictions.

```python
model = EarsC1()
model.fit(train)
pred = model.predict(test)
pred.head()
```

--------------------------------

### Fit FarringtonFlexible Model and Predict

Source: https://epysurv.readthedocs.io/en/latest/demo.html

Initializes a FarringtonFlexible model, fits it to the training data, generates predictions on the test data, and plots these predictions.

```python
model = FarringtonFlexible()
model.fit(train)
pred = model.predict(test)
plot_prediction(train, test, pred)
```

--------------------------------

### Import Epysurv Models and Visualization Tools

Source: https://epysurv.readthedocs.io/en/latest/demo.html

Imports specific epidemiological models (EarsC1, FarringtonFlexible) and visualization functions (plot_prediction, plot_confusion_matrix) from the epysurv library, along with sklearn's confusion_matrix.

```python
from epysurv.models.timepoint import EarsC1, FarringtonFlexible
from epysurv.visualization.model_diagnostics import plot_prediction, plot_confusion_matrix
from sklearn.metrics import confusion_matrix
```

--------------------------------

### RKI Algorithm Class

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/models/timepoint/rki.html

The RKI class represents the old algorithm from the Robert Koch Institute. It inherits from STSBasedAlgorithm and includes attributes for configuring the historical data window and recent year inclusion.

```APIDOC
## Class: RKI

### Description
The old algorithm from the Robert Koch Institute.

### Attributes
- **years_back** (int) - How many years back in time to include when forming the base counts.
- **window_half_width** (int) - Number of weeks to include before and after the current week in each year.
- **include_recent_year** (bool) - Is a boolean to decide if the year of timePoint also contributes w reference values.

### Internal Method
#### _call_surveillance_algo(sts, detection_range)

This method calls the underlying R `rki` function from the `surveillance` package with the specified control parameters derived from the RKI instance attributes.
```

--------------------------------

### Expanding Windows Transformation

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/data/filter_combination.html

Transforms case records into expanding time series for training and testing. Requires a minimum length in weeks and a SplitYears object to define the train/test split points. Fills missing values with 0 and assigns an 'outbreak' boolean based on the number of outbreak cases.

```python
[docs]    def expanding_windows(
        self,
        min_len_in_weeks: int,
        split_years: SplitYears
    ) -> TimeseriesClassificationData:
        """
        Transform case records into expanding time series.

        Parameters
        ----------
        min_len_in_weeks
            The minimum length of each time series.
        split_years
            The years at which to split the data into train and test data.

        Returns
        -------
        Compound object of train and test data as generators and dataframes.
        """
        self._validate_input(min_len_in_weeks, split_years)

        train_data = self.data.query(
            "@split_years.start <= ReportingDate < @split_years.middle"
        )
        test_data = self.data.query(
            "@split_years.start <= ReportingDate < @split_years.end"
        )

        offset = split_years.start + timedelta_weeks(min_len_in_weeks)

        true_train = (
            pd.DataFrame(
                index=pd.date_range(
                    offset, split_years.middle, freq=FREQ, closed="left"
                )
            )
            .join(_to_recent_timeseries(train_data))
            .fillna(0)
            .assign(outbreak=lambda df: df.n_outbreak_cases > 0)
        )
        true_test = (
            pd.DataFrame(
                index=pd.date_range(
                    split_years.middle, split_years.end, freq=FREQ, closed="left"
                )
            )
            .join(
                _to_recent_timeseries(
                    self.data.query(
                        "@split_years.middle <= ReportingDate < @split_years.end"
                    )
                )
            )
            .fillna(0)
            .assign(outbreak=lambda df: df.n_outbreak_cases > 0)
        )

        train_gen = self._expanding_frame(
            train_data,
            true_train,
            offset=offset,
            start=split_years.start,
            end=split_years.middle,
        )
        test_gen = self._expanding_frame(
            test_data,
            true_test,
            offset=split_years.middle,
            start=split_years.start,
            end=split_years.end,
        )

        return TimeseriesClassificationData(true_train, true_test, train_gen, test_gen)
```

--------------------------------

### PointSource.simulate

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/simulation/point_source.html

Simulates outbreaks based on the PointSource model parameters. It can either generate a random state chain or use a provided state chain.

```APIDOC
## simulate(length: int, state_weight: float = 0, state: Optional[Sequence[int]] = None) -> pd.DataFrame

### Description
Simulate outbreaks.

### Parameters
#### Parameters
- **length** (int) - Number of weeks to model. ``length`` is ignored if ``state`` is given. In this case, the length of ``state`` is used.
- **state** (Optional[Sequence[int]]) - Use a state chain to define the status at this time point (outbreak or not). If not given, a Markov chain is generated automatically.
- **state_weight** (float) - Additional weight for an outbreak which influences the distribution parameter mu.

### Returns
- **DataFrame** - A ``DataFrame`` of simulated case counts per week, separated into baseline and outbreak cases.
```

--------------------------------

### simulate

Source: https://epysurv.readthedocs.io/en/latest/api_doc/epysurv.simulation.html

Simulates outbreak data for a specified length or using a given state chain. It can optionally incorporate a state weight to influence the distribution parameter.

```APIDOC
## simulate

### Description
Simulate outbreaks.

### Method
`simulate(length: int, state_weight: Optional[float] = None, state: Optional[Sequence[int]] = None) -> pandas.core.frame.DataFrame`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
* **length** (int) - Number of weeks to model. `length` is ignored if `state` is given. In this case the length of `state` is used.
* **state** (Optional[Sequence[int]]) - Use a state chain to define the status at this time point (outbreak or not). If not given, a Markov chain is generated automatically.
* **state_weight** (Optional[float]) - Additional weight for an outbreak which influences the distribution parameter μμ.

### Response
#### Success Response (200)
* A `DataFrame` of an endemic time series where each row contains the case counts of this week.
* It also contains the mean case count value based on the underlying sinus model.

#### Response Example
None provided in source.
```

--------------------------------

### salmonella()

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/data/salmonella_data.html

Loads the training and testing datasets for Salmonella newport in Germany.

```APIDOC
## salmonella()

### Description
Loads the training and testing datasets for Salmonella newport in Germany.

### Returns
tuple: A tuple containing two pandas DataFrames: (train, test).
```

--------------------------------

### epysurv.simulation.naive_poisson.simulate_outbreaks

Source: https://epysurv.readthedocs.io/en/latest/api_doc/epysurv.simulation.html

Simulates outbreaks based on Poisson distribution, returning a DataFrame of simulated case counts per week.

```APIDOC
## simulate_outbreaks

### Description
Simulate outbreaks based on Poisson distribution.

### Parameters
- **n** (int) - Optional - Number of weeks.
- **outbreak_length** (int) - Optional - Number of weeks each outbreak is long.
- **n_outbreaks** (int) - Optional - Number of outbreaks.
- **mu** (float) - Optional - Mean for the baseline.
- **outbreak_mu** (float) - Optional - Mean for the outbreaks.

### Returns
- **pandas.core.frame.DataFrame** - Simulated case counts per week, separated into baseline and outbreak cases.
```

--------------------------------

### Timeseries Interface for Timepoint Algorithms

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/models/timeseries/convert_interface.html

These classes create a timeseries interface for existing timepoint algorithms by inheriting from NonLearningTimeseriesClassificationMixin and the respective algorithm class. This ensures a consistent API for timepoint-based classification tasks.

```python
"""Put a timeseries interface in front of all timepoint algorithms."""
from ..timepoint import (
    CDC,
    HMM,
    RKI,
    Bayes,
    Boda,
    Cusum,
    EarsC1,
    EarsC2,
    Farrington,
    FarringtonFlexible,
    GLRNegativeBinomial,
    GLRPoisson,
    OutbreakP,
)
from ._base import NonLearningTimeseriesClassificationMixin




[docs]
class Bayes(NonLearningTimeseriesClassificationMixin, Bayes):
    pass




[docs]
class Boda(NonLearningTimeseriesClassificationMixin, Boda):
    pass




[docs]
class CDC(NonLearningTimeseriesClassificationMixin, CDC):
    pass




[docs]
class Cusum(NonLearningTimeseriesClassificationMixin, Cusum):
    pass




[docs]
class EarsC1(NonLearningTimeseriesClassificationMixin, EarsC1):
    pass




[docs]
class EarsC2(NonLearningTimeseriesClassificationMixin, EarsC2):
    pass




[docs]
class Farrington(NonLearningTimeseriesClassificationMixin, Farrington):
    pass




[docs]
class FarringtonFlexible(NonLearningTimeseriesClassificationMixin, FarringtonFlexible):
    pass




[docs]
class GLRNegativeBinomial(
    NonLearningTimeseriesClassificationMixin, GLRNegativeBinomial
):
    pass




[docs]
class GLRPoisson(NonLearningTimeseriesClassificationMixin, GLRPoisson):
    pass




[docs]
class HMM(NonLearningTimeseriesClassificationMixin, HMM):
    pass




[docs]
class OutbreakP(NonLearningTimeseriesClassificationMixin, OutbreakP):
    pass




[docs]
class RKI(NonLearningTimeseriesClassificationMixin, RKI):
    pass

```

--------------------------------

### RKI Model

Source: https://epysurv.readthedocs.io/en/latest/api_doc/epysurv.models.timepoint.html

The old algorithm from the Robert Koch Institute for epidemiological surveillance.

```APIDOC
## epysurv.models.timepoint.rki.RKI

### Description
The old algorithm from the Robert Koch Institute for epidemiological surveillance.

### Parameters
- **years_back** (int) - Optional - How many years back in time to include when forming the base counts. Default is 0.
- **window_half_width** (int) - Optional - Number of weeks to include before and after the current week in each year. Default is 6.
- **include_recent_year** (bool) - Optional - Is a boolean to decide if the year of timePoint also contributes w reference values. Default is True.
```

--------------------------------

### Calling the Surveillance Algorithm

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/models/timepoint/farrington.html

This method constructs the control list and calls the `surveillance.farringtonFlexible` function with the provided time series data and control parameters. It is used internally for running the outbreak detection algorithm.

```python
def _call_surveillance_algo(self, sts, detection_range):
        control = r.list(
            range=detection_range,
            b=self.years_back,
            w=self.window_half_width,
            reweight=self.reweight,
            weightsThreshold=self.weights_threshold,
            alpha=self.alpha,
            trend=self.trend,
            trend_threshold=self.trend_threshold,
            limit54=r.c(self.min_cases_in_past_periods, self.past_period_cutoff),
            powertrans=self.power_transform,
            pastWeeksNotIncluded=self.past_weeks_not_included,
            thresholdMethod=self.threshold_method,
        )

        surv = surveillance.farringtonFlexible(sts, control=control)
        return surv
```

--------------------------------

### Import Data and Plotting Libraries

Source: https://epysurv.readthedocs.io/en/latest/demo.html

Imports necessary libraries for data manipulation (pandas), plotting (matplotlib.pyplot), and epidemiological data handling (epysurv.data).

```python
import matplotlib.pyplot as plt
import pandas as pd
from epysurv import data as epidata
```

--------------------------------

### Append to System Path

Source: https://epysurv.readthedocs.io/en/latest/demo.html

Appends the parent directory to the system path to allow importing modules from a specific location.

```python
import sys
sys.path.append(".." )
```

--------------------------------

### Seasonal Noise Simulation with Negative Binomial Distribution

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/simulation/seasonal_noise.html

Models a time series using a negative binomial distribution, incorporating trend and seasonality via Fourier terms. Use this for count data with overdispersion.

```python
from dataclasses import dataclass
from typing import Optional, Sequence

import numpy as np
import pandas as pd
import rpy2.robjects.packages as rpackages
from epysurv.simulation.base import BaseSimulation
from epysurv.simulation.utils import add_date_time_index_to_frame, r_list_to_frame
from rpy2 import robjects
from scipy.stats import nbinom, poisson

surveillance = rpackages.importr("surveillance")


@dataclass
class SeasonalNoiseNegativeBinomial(BaseSimulation):
    r"""A time series simulation that generates case counts based on a negative binomial model.

    The model is described by a mean :math:`\mu`, variance :math:`\phi \cdot \mu`, and a linear predictor including
    trend and seasonality determined by Fourier terms. :math:`\mu` of the model depends on the current week and
    is defined as follows:

        :math:`\mu(t) = \exp \left\{ \theta + \beta t + \sum_{j=1}^{m} \left\{ \gamma_{1} \cos (\frac{2\pi j t}{52}) 
        + \gamma_{2} \sin (\frac{2\pi j t}{52}) \right\} \right\}`

    where :math:`t` is the current week, :math:`m` the seasonality length, :math:`\beta` equals to the trend parameter,
    :math:`\gamma` is a seasonality parameter, and :math:`\theta` is the baseline frequency of the cases.

    The simulation is then run using
    :math:`\mu` and the dispersion parameter :math:`\phi` to specify the
    negative binomial model we draw case counts from.

    Parameters
    ----------
    baseline_frequency
        Baseline frequency of cases.
    dispersion
        Regulates the overdispersion compared to the Poisson distribution (:math:`\phi \cdot \mu`).
    seasonality_cos
        Seasonality parameter to model :math:`\cos` of the Fourier term.
    seasonality_sin
        seasonality parameter to model :math:`\sin` of the Fourier term.
    seasonality_length
    """

    baseline_frequency: float = 1.0
    dispersion: float = 1.0
    seasonality_cos: Sequence[float] = () 
    seasonality_sin: Sequence[float] = ()
    seasonality_length: int = 1

    def simulate(
        self,
        length: int,
        state_weight: Optional[float] = None,
        state: Optional[Sequence[int]] = None,
    ) -> pd.DataFrame:
        r"""
        Simulate outbreaks.

        Parameters
        ----------
        length
            Number of weeks to model. ``length`` is ignored if ``state`` is given. In this case the length of ``state``
            is used.
        state
            Use a state chain to define the status at this time point (outbreak or not). If not given, a Markov chain is
            generated automatically.
        state_weight
            Additional weight for an outbreak which influences the distribution parameter :math:`\mu`.

        Returns
        -------
            A ``DataFrame`` of an endemic time series where each row contains the case counts of this week.
            It also contains the mean case count value based on the underlying sinus model.
        """
        if self.seed:
            base = robjects.packages.importr("base")
            base.set_seed(self.seed)
        simulated = surveillance.sim_seasonalNoiseNB(
            alpha=self.baseline_frequency,
            phi=self.dispersion,
            beta=self.trend,
            frequency=self.seasonality_length,
            length=length,
            seasonal_cos=robjects.FloatVector(self.seasonality_cos),
            seasonal_sin=robjects.FloatVector(self.seasonality_sin),
            state=robjects.NULL if state is None else robjects.IntVector(state),
            K=robjects.NULL if state_weight is None else state_weight,
        )

        simulated = r_list_to_frame(simulated, ["mu", "seasonalBackground"])
        simulated = simulated.pipe(add_date_time_index_to_frame).rename(
            columns={"mu": "mean", "seasonalBackground": "n_cases"}
        )
        return simulated
```

--------------------------------

### Plot Confusion Matrix for FarringtonFlexible

Source: https://epysurv.readthedocs.io/en/latest/demo.html

Generates and displays a confusion matrix for the FarringtonFlexible model, comparing predicted alarms against actual outbreaks.

```python
plot_confusion_matrix(confusion_matrix(test["outbreak"], pred["alarm"]),
                    class_names=["no outbreak", "outbreak"])
```

--------------------------------

### Call the surveillance R package for Boda model

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/models/timepoint/boda.html

Implements the _call_surveillanc_algo method to interface with the R 'surveillance' package's 'boda' function. It checks for the 'INLA' package dependency and constructs the control list for the R function.

```python
def _call_surveillance_algo(self, sts, detection_range):
        try:
            importr("INLA")
        except RRuntimeError:
            raise ImportError(
                "For the Boda algortihm to run you need the INLA package (http://www.r-inla.org/). "
                'Install it by running install.packages("INLA", repos = c(getOption("repos"), INLA = "https://inla.r-inla-download.org/R/stable"), dep = TRUE) ' 
                "in the R console."
            )
        control = r.list(
            **{
                "range": detection_range,
                "X": robjects.NULL,
                "trend": self.trend,
                "season": self.season,
                "prior": self.prior,
                "alpha": self.alpha,
                "mc.munu": self.mc_munu,
                "mc.y": self.mc_y,
                "samplingMethod": self.sampling_method,
                "quantileMethod": self.quantile_method,
            }
        )
        surv = surveillance.boda(sts, control=control)
        return surv
```

--------------------------------

### EarsC1 Model Implementation

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/models/timepoint/ears.html

Implements the EARS C1 detection method. This model is suitable for data with limited historical values, using recent past counts to establish a baseline and detect anomalies.

```python
class EarsC1(_EarsBase):
    """Computes a threshold for the number of counts based on values from the recent past.

    This is then compared to the observed number of counts. If the observation is above
    a specific quantile of the prediction interval, then an alarm is raised.
    This method is especially useful for data without many
    historic values, since it only needs counts from the recent past.

    Attributes
    ----------
    alpha
        An approximate (two-sided)(1 − α) prediction interval is calculated.
    baseline
        How many time points to use for calculating the baseline.
    min_sigma
        If minSigma is higher than 0, the quantity zAlpha * minSigma is then the alerting threshold if the baseline is zero.

    References
    ----------
    .. [1] Fricker, R.D., Hegler, B.L, and Dunfee, D.A. (2008). Comparing syndromic surveillance detection
        methods: EARS versus a CUSUM-based methodology, 27:3407-3429, Statistics in medicine.
    .. [2] Salmon, M., Schumacher, D. and Höhle, M. (2016): Monitoring count time series in R: Aberration
        detection in public health surveillance. Journal of Statistical Software, 70 (10), 1-35. doi: 10.18637/jss.v070.i10
    """

    method = "C1"

```

--------------------------------

### simulate

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/simulation/seasonal_noise.html

Simulates outbreaks based on seasonal noise parameters. It generates a DataFrame of endemic time series data.

```APIDOC
## simulate

### Description
Simulate outbreaks.

### Parameter
- **length** (int) - Number of weeks to model.

### Returns
- A ``DataFrame`` of an endemic time series where each row contains the case counts ot this week.

### Method Signature
def simulate(self, length: int) -> pd.DataFrame:
    r"""Simulate outbreaks.

    Parameter
    ---------
    length
        Number of weeks to model.

    Returns
    -------
        A ``DataFrame`` of an endemic time series where each row contains the case counts ot this week.
    """
    if self.seed:
        np.random.seed(self.seed)
    mu_s = [
        np.exp(
            self.baseline_frequency + self.trend * week + self._seasonality(week)
        )
        for week in range(length)
    ]
    if self.dispersion == 1:
        cases = [poisson.rvs(mu, size=1)[0] for mu in mu_s]
    else:
        cases = []
        for mu in mu_s:
            r = np.float(mu / (self.dispersion - 1))
            p = r / (r + mu)
            cases.append(nbinom.rvs(r, p, size=1)[0])
    return (
        pd.DataFrame({"n_cases": cases})
        .pipe(add_date_time_index_to_frame)
        .assign(timestep=list(range(1, length + 1)))
    )
```

--------------------------------

### Bayes Model Implementation

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/models/timepoint/bayes.html

Defines the Bayes timepoint model, inheriting from STSBasedAlgorithm. It includes parameters for historical data inclusion, window size, recent year inclusion, and the alpha threshold for upper bound calculation. The `_call_surveillance_algo` method configures and calls the R 'bayes' function.

```python
from dataclasses import dataclass

from rpy2.robjects import r
from rpy2.robjects.packages import importr

from ._base import STSBasedAlgorithm

surveillance = importr("surveillance")



@dataclass
class Bayes(STSBasedAlgorithm):
    """
    Evaluation of timepoints with the Bayes subsystem.

    Attributes
    ----------
    years_back
        How many years back in time to include when forming the base counts.
    window_half_width
        Number of weeks to include before and after the current week in each year.
    include_recent_year
        is a boolean to decide if the year of timePoint also contributes w reference values.
    alpha
        The parameter alpha is the (1 − α)-quantile to use in order to calculate the upper threshold. As default b, w, actY are set for the Bayes 1 system with alpha=0.05.

    References
    ----------
    .. [1] Riebler, A. (2004), Empirischer Vergleich von statistischen Methoden zur Ausbruchserkennung bei
        Surveillance Daten, Bachelor’s thesis
    .. [2] Höhle, M., & Riebler, A. (2005). Höhle, Riebler: The R-Package “surveillance.” Sonderforschungsbereich (Vol. 386). Retrieved from https://epub.ub.uni-muenchen.de/1791/1/paper_422.pdf
    """

    years_back: int = 0
    window_half_width: int = 6
    include_recent_year: bool = True
    alpha: float = 0.05

    def _call_surveillance_algo(self, sts, detection_range):
        control = r.list(
            range=detection_range,
            b=self.years_back,
            w=self.window_half_width,
            actY=self.include_recent_year,
            alpha=self.alpha,
        )

        surv = surveillance.bayes(sts, control=control)
        return surv

```

--------------------------------

### HMM

Source: https://epysurv.readthedocs.io/en/latest/api_doc/epysurv.models.timepoint.html

Hidden Markov Model for outbreak detection. This model can incorporate trend and seasonality.

```APIDOC
## HMM

### Description
Hidden Markov model for outbreak detection.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Constructor
`HMM(n_observations: int = -1, n_hidden_states: int = 2, trend: bool = True, n_harmonics: int = 1, equal_covariate_effects: bool = False)`

### Attributes
- **n_observations** (int): Number of observations.
- **n_hidden_states** (int): Number of hidden states.
- **trend** (bool): Whether to include a trend component.
- **n_harmonics** (int): Number of harmonics for seasonality.
- **equal_covariate_effects** (bool): Whether to assume equal covariate effects across states.
```

--------------------------------

### Simulate Seasonal Noise with Poisson Distribution

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/simulation/seasonal_noise.html

Simulates an endemic time series using a Poisson distribution. The mean is modeled with a sinusoidal function, trend, and optional state weights. Use this for simulating count data with seasonal patterns and a trend.

```python
from dataclasses import dataclass
from typing import Optional, Sequence

import numpy as np
import pandas as pd
import rpy2.robjects.packages as rpackages
from epysurv.simulation.base import BaseSimulation
from epysurv.simulation.utils import add_date_time_index_to_frame, r_list_to_frame
from rpy2 import robjects
from scipy.stats import nbinom, poisson

surveillance = rpackages.importr("surveillance")


@dataclass
class SeasonalNoisePoisson(BaseSimulation):
    """Simulation of an endemic time series based on a Poisson distribution.

    The mean of the Poisson distribution is modelled as:

        :math:`\mu(t) = \exp{(A\sin{(frequency \cdot \omega \cdot (t + \phi))}
        + \alpha + \beta \cdot t + K \cdot state)}`

    with :math:`\omega = \pi / 52`, :math:`A` being the amplitude, :math:`\beta` the trend parameter, :math:`t`
    the current week, and :math:`\theta` the seasonal move.

    Parameters
    ----------
    amplitude
        Amplitude of the sine. Determines the range of simulated cases.
    alpha
        Parameter to move simulation along the y-axis (negative values are not allowed) with `alpha` >= `amplitude`.
    frequency
        Factor in oscillation term. Is multiplied with the annual term :math:`\omega` and the current time point.
    seasonal_move
        A term added to each time point :math:`t` to move the curve along the x-axis.
    seed
        Seed for the random number generation.
    trend
        Controls the influence of the current week on :math:`\mu`.

    References
    ----------
        http://surveillance.r-forge.r-project.org/
    """

    alpha: float = 1.0
    amplitude: float = 1.0
    frequency: int = 1
    seasonal_move: int = 0
    seed: Optional[int] = None
    trend: float = 0.0


    def simulate(
        self,
        length: int,
        state_weight: Optional[float] = None,
        state: Optional[Sequence[int]] = None,
    ) -> pd.DataFrame:
        r"""
        Simulate outbreaks.

        Parameters
        ----------
        length
            Number of weeks to model. ``length`` is ignored if ``state`` is given. In this case the length of ``state``
            is used.
        state
            Use a state chain to define the status at this time point (outbreak or not). If not given, a Markov chain is
            generated automatically.
        state_weight
            Additional weight for an outbreak which influences the distribution parameter :math:`\mu`.

        Returns
        -------
            A ``DataFrame`` of an endemic time series where each row contains the case counts of this week.
            It also contains the mean case count value based on the underlying sinus model.
        """
        if self.seed:
            base = robjects.packages.importr("base")
            base.set_seed(self.seed)
        simulated = surveillance.sim_seasonalNoise(
            A=self.amplitude,
            alpha=self.alpha,
            beta=self.trend,
            phi=self.seasonal_move,
            length=length,
            frequency=self.frequency,
            state=robjects.NULL if state is None else robjects.IntVector(state),
            K=robjects.NULL if state_weight is None else state_weight,
        )

        simulated = r_list_to_frame(simulated, ["mu", "seasonalBackground"])
        simulated = simulated.pipe(add_date_time_index_to_frame).rename(
            columns={"mu": "mean", "seasonalBackground": "n_cases"}
        )
        return simulated
```

--------------------------------

### RKI Timeseries Model

Source: https://epysurv.readthedocs.io/en/latest/api_doc/epysurv.models.timeseries.html

The RKI model provides a time-series interface to the RKI timepoint algorithm. It includes parameters for the lookback period, window half-width, and whether to include the most recent year's data.

```APIDOC
## epysurv.models.timeseries.convert_interface.RKI

### Description
Provides a timeseries interface to the RKI timepoint algorithm.

### Parameters
- **years_back** (int) - Optional - Number of past years to consider.
- **window_half_width** (int) - Optional - Half-width of the sliding window.
- **include_recent_year** (bool) - Optional - Whether to include the most recent year's data.
```

--------------------------------

### EarsC3 Model Implementation

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/models/timepoint/ears.html

Implements the EARS C3 detection method. This model also relies on recent past counts for thresholding and anomaly detection, suitable for time series with limited historical data.

```python
@dataclass
class EarsC3(_EarsBase):
    """The EarsC3 model.

    Computes a threshold for the number of counts based on values from the recent past. This is then
    compared to the observed number of counts. If the observation is above a specific quantile of the
    prediction interval, then an alarm is raised. This method is especially useful for data without many
    historic values, since it only needs counts from the recent past.

    Attributes
    ----------
    alpha
        An approximate (two-sided)(1 − α) prediction interval is calculated.
    baseline
        How many time points to use for calculating the baseline.


    References
    ----------
    .. [1] Fricker, R.D., Hegler, B.L, and Dunfee, D.A. (2008). Comparing syndromic surveillance detection
        methods: EARS versus a CUSUM-based methodology, 27:3407-3429, Statistics in medicine.
    .. [2] Salmon, M., Schumacher, D. and Höhle, M. (2016): Monitoring count time series in R: Aberration
        detection in public health surveillance. Journal of Statistical Software, 70 (10), 1-35. doi: 10.18637/jss.v070.i10
    """

    alpha: float = 0.001
    baseline: int = 7

    def _call_surveillance_algo(self, sts, detection_range):
        control = r.list(
            range=detection_range,
            method="C3",
            baseline=self.baseline,
            minSigma=self.min_sigma,
            alpha=self.alpha,
        )
        surv = surveillance.earsC(sts, control=control)
        return surv

```

--------------------------------

### Simulate Outbreaks with PointSource

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/simulation/point_source.html

Simulates epidemic outbreaks using the PointSource model. This method generates a time series of case counts, separated into baseline and outbreak cases, based on the configured parameters. It can optionally use a predefined state chain.

```python
from dataclasses import dataclass
from typing import Optional, Sequence

import pandas as pd
import rpy2.robjects.packages as rpackages
from rpy2 import robjects

from epysurv.simulation.base import BaseSimulation
from epysurv.simulation.utils import add_date_time_index_to_frame, r_list_to_frame

surveillance = rpackages.importr("surveillance")
base = rpackages.importr("base")



@dataclass
class PointSource(BaseSimulation):
    r"""Simulation of epidemics which were introduced by point sources.

    The basis of this programme is a combination of a Hidden Markov Model
    (to get random time points for outbreaks) and a simple model
    (compare :class:`epysurv.simulation.SeasonalNoise`) to simulate the baseline.

    Parameters
    ----------
    amplitude
        Amplitude of the sine. Determines the possible range of simulated seasonal cases.
    alpha
        Parameter to move along the y-axis (negative values are not allowed) with `alpha` >= `amplitude`.
    frequency
        Factor in oscillation term. Is multiplied with the annual term :math:`\omega` and the current time point.
    p
        Probability to get a new outbreak at time :math:`t` if there was one at time :math:`t-1`.
    r
        Probability to get no new outbreak at time :math:`t` if there was none at time :math:`t-1`.
    seasonal_move
        A term added to time point :math:`t` to move the curve along the x-axis.
    seed
        Seed for the random number generation.
    trend
        Controls the influence of the current week on :math:`\mu`.

    References
    ----------
        http://surveillance.r-forge.r-project.org/
    """

    alpha: float = 1.0
    amplitude: float = 1.0
    frequency: int = 1
    p: float = 0.99
    r: float = 0.01
    seasonal_move: int = 0
    seed: Optional[int] = None
    trend: float = 0.0


    def simulate(
        self,
        length: int,
        state_weight: float = 0,
        state: Optional[Sequence[int]] = None,
    ) -> pd.DataFrame:
        """Simulate outbreaks.

        Parameters
        ----------
        length
            Number of weeks to model. ``length`` is ignored if ``state`` is given. In this case, the length of
            ``state`` is used.
        state
            Use a state chain to define the status at this time point (outbreak or not). If not given, a Markov chain is
            generated automatically.
        state_weight
            Additional weight for an outbreak which influences the distribution parameter mu.

        Returns
        -------
            A ``DataFrame`` of simulated case counts per week, separated into baseline and outbreak cases.
        """
        if self.seed:
            base.set_seed(self.seed)
        simulated = surveillance.sim_pointSource(
            p=self.p,
            r=self.r,
            length=length,
            A=self.amplitude,
            alpha=self.alpha,
            beta=self.trend,
            phi=self.seasonal_move,
            frequency=self.frequency,
            state=robjects.NULL if state is None else robjects.IntVector(state),
            K=state_weight,
        )

        simulated_as_frame = r_list_to_frame(simulated, ["observed", "state"])
        return simulated_as_frame.pipe(add_date_time_index_to_frame).rename(
            columns={"observed": "n_cases", "state": "is_outbreak"}
        )

```

--------------------------------

### Farrington Flexible Surveillance Control Parameters

Source: https://epysurv.readthedocs.io/en/latest/_modules/epysurv/models/timepoint/farrington.html

Defines the control parameters for the `farringtonFlexible` surveillance algorithm. These parameters configure aspects like the lookback period, window size, and thresholding methods for outbreak detection.

```python
years_back: int = 3
window_half_width: int = 3
reweight: bool = True
weights_threshold: float = 2.58
alpha: float = 0.01
trend: bool = True
trend_threshold: float = 0.05
past_period_cutoff: int = 4
min_cases_in_past_periods: int = 5
power_transform: str = "2/3"
past_weeks_not_included: int = 26
threshold_method: str = "delta"
```
Tokens:
Raw