Skip to content

continuous_timeseries.discrete_to_continuous.piecewise_constant_previous_left_closed#

Conversion of discrete to continuous using 'previous' piecewise constant steps

Each interval is closed on the left.

In other words, between t(i) and t(i + 1), the value is equal to y(i). At t(i + 1), the value is equal to y(i + 1).

If helpful, we have drawn a picture of how this works below. Symbols:

  • time: y-value selected for this time-value
  • i: closed (i.e. inclusive) boundary
  • o: open (i.e. exclusive) boundary
y(4):                                                ixxxxxxxxxxxxxx
y(3):                                    ixxxxxxxxxxxo
y(2):                        ixxxxxxxxxxxo
y(1): xxxxxxxxxxxxxxxxxxxxxxxo
      -----------|-----------|-----------|-----------|--------------
              time(1)     time(2)     time(3)     time(4)

Classes:

Name Description
PPolyPiecewiseConstantPreviousLeftClosed

Piecewise polynomial that implements our 'previous' constant left-closed logic

Functions:

Name Description
get_idxs

Get the indexes from self.y to return, given the times of interest and self.x

PPolyPiecewiseConstantPreviousLeftClosed #

Piecewise polynomial that implements our 'previous' constant left-closed logic

For full details of the logic, see the module's docstring.

We can't use scipy.interpolate.PPoly directly because it doesn't behave as we want at the first boundary. We could subclass scipy.interpolate.PPoly, but that is more trouble than its worth for such a simple implementation.

Methods:

Name Description
__call__

Evaluate the function at specific points
antidifferentiate

Antidifferentiate
differentiate

Differentiate
integrate

Integrate

Attributes:

Name Type Description
x NP_ARRAY_OF_FLOAT_OR_INT

Breakpoints between each piecewise constant interval
y NP_ARRAY_OF_FLOAT_OR_INT

The y-values which help define our spline.
Source code in src/continuous_timeseries/discrete_to_continuous/piecewise_constant_previous_left_closed.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
@define
class PPolyPiecewiseConstantPreviousLeftClosed:
    """
    Piecewise polynomial that implements our 'previous' constant left-closed logic

    For full details of the logic, see [the module's docstring][(m)].

    We can't use [`scipy.interpolate.PPoly`][scipy.interpolate.PPoly] directly
    because it doesn't behave as we want at the first boundary.
    We could subclass [`scipy.interpolate.PPoly`][scipy.interpolate.PPoly],
    but that is more trouble than its worth for such a simple implementation.
    """

    x: NP_ARRAY_OF_FLOAT_OR_INT
    """
    Breakpoints between each piecewise constant interval
    """

    y: NP_ARRAY_OF_FLOAT_OR_INT = field(validator=[piecewise_constant_y_validator])
    """
    The y-values which help define our spline.

    Note that these are not the same as the values at our boundaries,
    see [the module's docstring][(m)].
    """

    def __call__(
        self, x: NP_ARRAY_OF_FLOAT_OR_INT, allow_extrapolation: bool = False
    ) -> NP_ARRAY_OF_FLOAT_OR_INT:
        """
        Evaluate the function at specific points

        Parameters
        ----------
        x
            Points at which to evaluate the function

        allow_extrapolation
            Should extrapolation be allowed?

        Returns
        -------
        :
            The function, evaluated at `x`

        Raises
        ------
        ExtrapolationNotAllowedError
            The user attempted to extrapolate when it isn't allowed.
        """
        res = get_values(
            times=x,
            self_x=self.x,
            self_y=self.y,
            get_idxs=get_idxs,
            allow_extrapolation=allow_extrapolation,
        )

        return res

    def differentiate(self) -> ContinuousFunctionScipyPPoly:
        """
        Differentiate

        Returns
        -------
        :
            Derivative of the function
        """
        res = differentiate_piecewise_constant(self=self)

        return res

    def integrate(
        self, integration_constant: NP_FLOAT_OR_INT, domain_start: NP_FLOAT_OR_INT
    ) -> ContinuousFunctionScipyPPoly:
        """
        Integrate

        Parameters
        ----------
        integration_constant
            Integration constant

            This is required for the integral to be a definite integral.

        domain_start
            The start of the domain.

            This is required to ensure that we start at the right point
            when evaluating the definite integral.

        Returns
        -------
        :
            Integral of the function
        """
        res = integrate_piecewise_constant(
            self=self,
            integration_constant=integration_constant,
            domain_start=domain_start,
        )

        return res

    def antidifferentiate(
        self, domain_start: NP_FLOAT_OR_INT
    ) -> ContinuousFunctionScipyPPoly:
        """
        Antidifferentiate

        Parameters
        ----------
        domain_start
            The start of the domain.

            This is required to ensure that we start at the right point
            when evaluating the indefinite integral.

        Returns
        -------
        :
            Indefinite integral of the function
        """
        res = antidifferentiate_piecewise_constant(
            self=self,
            domain_start=domain_start,
        )

        return res

x instance-attribute #

x: NP_ARRAY_OF_FLOAT_OR_INT

Breakpoints between each piecewise constant interval

y class-attribute instance-attribute #

y: NP_ARRAY_OF_FLOAT_OR_INT = field(
    validator=[piecewise_constant_y_validator]
)

The y-values which help define our spline.

Note that these are not the same as the values at our boundaries, see the module's docstring.

__call__ #

__call__(
    x: NP_ARRAY_OF_FLOAT_OR_INT,
    allow_extrapolation: bool = False,
) -> NP_ARRAY_OF_FLOAT_OR_INT

Evaluate the function at specific points

Parameters:

Name Type Description Default
x NP_ARRAY_OF_FLOAT_OR_INT

Points at which to evaluate the function

 required
allow_extrapolation bool

Should extrapolation be allowed?

 False

Returns:

Type Description
NP_ARRAY_OF_FLOAT_OR_INT

The function, evaluated at x

Raises:

Type Description
ExtrapolationNotAllowedError

The user attempted to extrapolate when it isn't allowed.
Source code in src/continuous_timeseries/discrete_to_continuous/piecewise_constant_previous_left_closed.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def __call__(
    self, x: NP_ARRAY_OF_FLOAT_OR_INT, allow_extrapolation: bool = False
) -> NP_ARRAY_OF_FLOAT_OR_INT:
    """
    Evaluate the function at specific points

    Parameters
    ----------
    x
        Points at which to evaluate the function

    allow_extrapolation
        Should extrapolation be allowed?

    Returns
    -------
    :
        The function, evaluated at `x`

    Raises
    ------
    ExtrapolationNotAllowedError
        The user attempted to extrapolate when it isn't allowed.
    """
    res = get_values(
        times=x,
        self_x=self.x,
        self_y=self.y,
        get_idxs=get_idxs,
        allow_extrapolation=allow_extrapolation,
    )

    return res

antidifferentiate #

antidifferentiate(
    domain_start: NP_FLOAT_OR_INT,
) -> ContinuousFunctionScipyPPoly

Antidifferentiate

Parameters:

Name Type Description Default
domain_start NP_FLOAT_OR_INT

The start of the domain.

This is required to ensure that we start at the right point when evaluating the indefinite integral.

 required

Returns:

Type Description
ContinuousFunctionScipyPPoly

Indefinite integral of the function
Source code in src/continuous_timeseries/discrete_to_continuous/piecewise_constant_previous_left_closed.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def antidifferentiate(
    self, domain_start: NP_FLOAT_OR_INT
) -> ContinuousFunctionScipyPPoly:
    """
    Antidifferentiate

    Parameters
    ----------
    domain_start
        The start of the domain.

        This is required to ensure that we start at the right point
        when evaluating the indefinite integral.

    Returns
    -------
    :
        Indefinite integral of the function
    """
    res = antidifferentiate_piecewise_constant(
        self=self,
        domain_start=domain_start,
    )

    return res

differentiate #

differentiate() -> ContinuousFunctionScipyPPoly

Differentiate

Returns:

Type Description
ContinuousFunctionScipyPPoly

Derivative of the function
Source code in src/continuous_timeseries/discrete_to_continuous/piecewise_constant_previous_left_closed.py 
140
141
142
143
144
145
146
147
148
149
150
151
def differentiate(self) -> ContinuousFunctionScipyPPoly:
    """
    Differentiate

    Returns
    -------
    :
        Derivative of the function
    """
    res = differentiate_piecewise_constant(self=self)

    return res

integrate #

integrate(
    integration_constant: NP_FLOAT_OR_INT,
    domain_start: NP_FLOAT_OR_INT,
) -> ContinuousFunctionScipyPPoly

Integrate

Parameters:

Name Type Description Default
integration_constant NP_FLOAT_OR_INT

Integration constant

This is required for the integral to be a definite integral.

 required
domain_start NP_FLOAT_OR_INT

The start of the domain.

This is required to ensure that we start at the right point when evaluating the definite integral.

 required

Returns:

Type Description
ContinuousFunctionScipyPPoly

Integral of the function
Source code in src/continuous_timeseries/discrete_to_continuous/piecewise_constant_previous_left_closed.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def integrate(
    self, integration_constant: NP_FLOAT_OR_INT, domain_start: NP_FLOAT_OR_INT
) -> ContinuousFunctionScipyPPoly:
    """
    Integrate

    Parameters
    ----------
    integration_constant
        Integration constant

        This is required for the integral to be a definite integral.

    domain_start
        The start of the domain.

        This is required to ensure that we start at the right point
        when evaluating the definite integral.

    Returns
    -------
    :
        Integral of the function
    """
    res = integrate_piecewise_constant(
        self=self,
        integration_constant=integration_constant,
        domain_start=domain_start,
    )

    return res

get_idxs #

get_idxs(
    times: NP_ARRAY_OF_FLOAT_OR_INT,
    self_x: NP_ARRAY_OF_FLOAT_OR_INT,
) -> NDArray[integer]

Get the indexes from self.y to return, given the times of interest and self.x

This function defines the key logic of the interpolation implementation.

Parameters:

Name Type Description Default
times NP_ARRAY_OF_FLOAT_OR_INT

Times for which to retrieve the values

 required
self_x NP_ARRAY_OF_FLOAT_OR_INT

The points which define the piecewise constant intervals

 required

Returns:

Type Description
NDArray[integer]

The indexes from self.y to return
Source code in src/continuous_timeseries/discrete_to_continuous/piecewise_constant_previous_left_closed.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def get_idxs(
    times: NP_ARRAY_OF_FLOAT_OR_INT, self_x: NP_ARRAY_OF_FLOAT_OR_INT
) -> npt.NDArray[np.integer]:
    """
    Get the indexes from `self.y` to return, given the times of interest and `self.x`

    This function defines the key logic of the interpolation implementation.

    Parameters
    ----------
    times
        Times for which to retrieve the values

    self_x
        The points which define the piecewise constant intervals

    Returns
    -------
    :
        The indexes from `self.y` to return
    """
    res_idxs: npt.NDArray[np.integer] = (
        np.searchsorted(a=self_x, v=np.atleast_1d(times), side="right") - 1
    )
    # Fix up any underrun
    res_idxs[res_idxs == -1] = 0

    return res_idxs