DEV Community

AndreiKeino
AndreiKeino

Posted on

Constant State Detection in Time Series with Python.

Constant State Detection in Time Series with Python.

See this code on Github

In this article we will detect constant state section in time series with Python.

You can take a look at the preprint with the introduction to the algorithm here.

We will use the package that requires Python 3.10+

Installation:

Recommended:

1. Create a virtual environment.

For Linux: execute the console command

python3 -m venv your_path_to_new_virtual_environment

for other operating system there could be different command(s)

2. Activate it.

For Linux: execute the console command

source your_path_to_new_virtual_environment/bin/activate

for other operating system there could be different command(s)

3. Install the package.

execute the console command

pip install csd-algo

Quick check if it is working

now execute the console command

python -m csd_algo.usage_example

If all have been done correctly, you should see the console output and a couple of plots like this:

Example figure:

Example figure

Now create the github_example.py file and paste the following code into it:


import requests
import matplotlib.pyplot as plt
from csd_algo.client_axiliary import report_results, __version__
from csd_algo.models import (MAX_CSD_POINTS_FREE, CSD_Data,  
                    to_jsonable_dict)
from csd_algo.generate_timeseries_numpy import generate_timseries

plt.close('all')

url_csd_algo = 'https://dazzling-faithful-banteng.anvil.app/constant-state-detection'

ts: list[float] = generate_timseries(
    0.5, 
    MAX_CSD_POINTS_FREE,
    ).tolist()

# process the timeseries with the default settings:

csd_data = CSD_Data(csd_algo_input=ts, 
                    csd_algo_min_window_width=16, 
                    csd_algo_max_window_height=0.08)

# send the data to the server and receive the reply
response = requests.post(url_csd_algo, 
                         json=to_jsonable_dict(csd_data))

# validate received data
out_data = CSD_Data.model_validate(response.json())

# print and plot the processing results
report_results(out_data, 
               f'CSD plot, Algorithm type: postprocessing')

print(f'Client version is {__version__}')

Save the file and execute the console command

python github_example.py

Now you should see some console output and just one plot like the picture above.

After completing and verifying these steps, you can adjust the input data and algorithm parameters to test the algorithm implementation according to your requirements.

It might be helpful to take a look at the code for the class used as the webservice input data:


class CSD_Data(BaseModel):
    """
    class to hold the Constant State Detection algorithm data
    #
    values:
    # 
    csd_algo_input: 
        [in] input timeseries 
    # csd_algo_min_window_width: 
        [in] minimal window width
    # csd_algo_max_window_length: 
        [in] maximal window height
    # Algo_Type: 
        [in] AlgorithmType value
    # Window_Height_Type: 
        [in] WindowHeightType value
    # start: 
        [out] list of start constant section series indexes inclusive
    # finish: 
        [out] list of finish constant section series indexes inclusive
    # start_section_series: 
        [out] list of start continuous constant 
        state section indexes inclusive
    # finish_section_series: 
        [out] list of finish continuous constant 
        state section indexes inclusive
    # api_key: 
        [in] reserved for future use
    # csd_algo_info: 
        [out] list of infromation strings
    # csd_algo_warnings: 
        [out] list of warning strings
    # csd_algo_errors: 
        [out] list of error strings
    # 
    see also: 
    https://figshare.com/articles/preprint/The_effective_algorithm_for_constant_state_detection_in_time_series_V_2_3_/28375532?file=56562641

    """

    csd_algo_input: list[float]
    csd_algo_min_window_width: int
    csd_algo_max_window_height: float
    Algo_Type: AlgorithmType = AlgorithmType.POSTPROCESSING_V1
    Window_Height_Type: WindowHeightType = WindowHeightType.Relative
    start: list[int] = []
    finish: list[int] = []
    start_section_series: list[int] = []
    finish_section_series: list[int] = []
    api_key: str = ''
    csd_algo_info: list[str] = []
    csd_algo_warnings: list[str] = []
    csd_algo_errors: list[str] = []

The main parameters here are:

  • csd_algo_input: list[float] - this is the timeseries input data for the algorithm
  • csd_algo_min_window_width: int - this is the minimal window width input for the algorithm
  • csd_algo_max_window_height: float - this is the maximal window height input for the algorithm

It is better to leave the parameters Algo_Type and Window_Height_Type values to have their default values for the beginning. Algo_Type.POSTPROCESSING_V1 value means advanced algorithm which chooses the 'best' constant state section with the minimal height from sections that have the maximal length from one continuous constant state section series.

It might also be helpful to take a look at the code to create the numpy array used for testing:


def generate_timseries(b: float, points_cnt: int=256, 
                       periods: int=1, 
                       phi: float = 0) -> np.array:
        """
        function to generate the test timeseries for the 
        constant state detection algorithm demo. returns the 
        float numpy arrray with the sin(x) + b * sin(2 * x + phi)
        sum of the first and the second harmonic inside.

        input parameters:
        * b - float, the value of the second harmonic
        * points_cnt - int, the count of the points in the output array
        * periods - int, the period count of the first harmonic
        * phi - float, the phase of the second harmonic in the 
         PI * radians, 
        """

        points = int(periods * points_cnt)
        phase = phi * np.pi
        x = np.arange(points)
        y = np.sin(2.0 * np.pi * periods * x / points)
        sin2 = b * np.sin(4.0 * np.pi * periods * x / points + phase)
        y += sin2
        return y

Using these definitions, one can easily test the algorithm with one's own input data; for example load them from a file, just don't forget about the MAX_CSD_POINTS_FREE input point count restriction.

Important note: Unfortunately, I can't guarantee at the moment that the webservice will work with the good uptime and response time, as this is just a free test version of the webservice. There might be times when the webservice will be inaccessible for hours, maybe even for days.

Top comments (0)