Skip to content

continuous_timeseries#

Representation of continuous timeseries.

Modules:

Name Description
budget_compatible_pathways

Creation of emissions pathways compatible with a given budget
discrete_to_continuous

Conversion of timeseries from discrete to continuous
domain_helpers

Support for our domain handling
exceptions

Exceptions that are used throughout
formatting

Support for pretty formatting of our classes
pandas_accessors

API for pandas accessors.
plotting_helpers

Support for plotting
time_axis

Definition of TimeAxis
timeseries

Definition of a timeseries (Timeseries)
timeseries_continuous

Definition of a continuous timeseries (TimeseriesContinuous)
timeseries_discrete

Definition of a discrete timeseries (TimeseriesDiscrete)
typing

Helpful type hints
values_at_bounds

Definition of ValuesAtBounds
warnings

Warnings that are used throughout

Classes:

Name Description
InterpolationOption

Interpolation options
TimeAxis

Time axis representation
Timeseries

Timeseries representation
TimeseriesContinuous

Continuous time series representation
TimeseriesDiscrete

Discrete time series representation
ValuesAtBounds

Container for values to be used at the bounds of each time window in a timeseries

InterpolationOption #

Bases: IntEnum

Interpolation options

Attributes:

Name Type Description
Cubic

Cubic interpolation
Linear

Linear interpolation
PiecewiseConstantNextLeftClosed

Piecewise constant 'next' interpolation, each interval is closed on the left
PiecewiseConstantNextLeftOpen

Piecewise constant 'next' interpolation, each interval is open on the left
PiecewiseConstantPreviousLeftClosed

Piecewise constant 'previous' interpolation, each interval is closed on the left
PiecewiseConstantPreviousLeftOpen

Piecewise constant 'previous' interpolation, each interval is open on the left
Quadratic

Quadratic interpolation
Quartic

Quartic interpolation
Source code in src/continuous_timeseries/discrete_to_continuous/interpolation_option.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 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
 78
 79
 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
@unique
class InterpolationOption(IntEnum):
    """
    Interpolation options
    """

    Linear = 1
    """Linear interpolation"""

    Quadratic = 2
    """Quadratic interpolation"""

    Cubic = 3
    """Cubic interpolation"""

    Quartic = 4
    """Quartic interpolation"""

    PiecewiseConstantNextLeftClosed = 10
    """
    Piecewise constant 'next' interpolation, each interval is closed on the left

    In other words,
    between t(i) and t(i + 1), the value is equal to y(i + 1).
    At t(i), 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):                                    ixxxxxxxxxxxxxxxxxxxxxxxxxx
    y(3):                        ixxxxxxxxxxxo
    y(2):            ixxxxxxxxxxxo
    y(1): xxxxxxxxxxxo
          -----------|-----------|-----------|-----------|--------------
                  time(1)     time(2)     time(3)     time(4)
    ```
    """

    PiecewiseConstantNextLeftOpen = 11
    """
    Piecewise constant 'next' interpolation, each interval is open on the left

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

    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):                                    oxxxxxxxxxxxxxxxxxxxxxxxxxx
    y(3):                        oxxxxxxxxxxxi
    y(2):            oxxxxxxxxxxxi
    y(1): xxxxxxxxxxxi
          -----------|-----------|-----------|-----------|--------------
                  time(1)     time(2)     time(3)     time(4)
    ```
    """

    PiecewiseConstantPreviousLeftClosed = 12
    """
    Piecewise constant 'previous' interpolation, 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)
    ```
    """

    PiecewiseConstantPreviousLeftOpen = 13
    """
    Piecewise constant 'previous' interpolation, each interval is open 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).

    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):                                                oxxxxxxxxxxxxxx
    y(3):                                    oxxxxxxxxxxxi
    y(2):                        oxxxxxxxxxxxi
    y(1): xxxxxxxxxxxxxxxxxxxxxxxi
          -----------|-----------|-----------|-----------|--------------
                  time(1)     time(2)     time(3)     time(4)
    ```
    """

Cubic class-attribute instance-attribute #

Cubic = 3

Cubic interpolation

Linear class-attribute instance-attribute #

Linear = 1

Linear interpolation

PiecewiseConstantNextLeftClosed class-attribute instance-attribute #

PiecewiseConstantNextLeftClosed = 10

Piecewise constant 'next' interpolation, each interval is closed on the left

In other words, between t(i) and t(i + 1), the value is equal to y(i + 1). At t(i), 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):                                    ixxxxxxxxxxxxxxxxxxxxxxxxxx
y(3):                        ixxxxxxxxxxxo
y(2):            ixxxxxxxxxxxo
y(1): xxxxxxxxxxxo
      -----------|-----------|-----------|-----------|--------------
              time(1)     time(2)     time(3)     time(4)

PiecewiseConstantNextLeftOpen class-attribute instance-attribute #

PiecewiseConstantNextLeftOpen = 11

Piecewise constant 'next' interpolation, each interval is open on the left

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

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):                                    oxxxxxxxxxxxxxxxxxxxxxxxxxx
y(3):                        oxxxxxxxxxxxi
y(2):            oxxxxxxxxxxxi
y(1): xxxxxxxxxxxi
      -----------|-----------|-----------|-----------|--------------
              time(1)     time(2)     time(3)     time(4)

PiecewiseConstantPreviousLeftClosed class-attribute instance-attribute #

PiecewiseConstantPreviousLeftClosed = 12

Piecewise constant 'previous' interpolation, 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)

PiecewiseConstantPreviousLeftOpen class-attribute instance-attribute #

PiecewiseConstantPreviousLeftOpen = 13

Piecewise constant 'previous' interpolation, each interval is open 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).

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):                                                oxxxxxxxxxxxxxx
y(3):                                    oxxxxxxxxxxxi
y(2):                        oxxxxxxxxxxxi
y(1): xxxxxxxxxxxxxxxxxxxxxxxi
      -----------|-----------|-----------|-----------|--------------
              time(1)     time(2)     time(3)     time(4)

Quadratic class-attribute instance-attribute #

Quadratic = 2

Quadratic interpolation

Quartic class-attribute instance-attribute #

Quartic = 4

Quartic interpolation

TimeAxis #

Time axis representation

Methods:

Name Description
__str__

Get string representation of self
bounds_validator

Validate the received bounds

Attributes:

Name Type Description
bounds PINT_NUMPY_ARRAY

Bounds of each time step in the time axis.
bounds_2d PINT_NUMPY_ARRAY

Get the bounds of the time steps in two-dimensions
Source code in src/continuous_timeseries/time_axis.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 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
@define
class TimeAxis:
    """
    Time axis representation
    """

    bounds: PINT_NUMPY_ARRAY = field()
    """
    Bounds of each time step in the time axis.

    Must be one-dimensional and monotonically increasing.

    The first time step runs from `bounds[0]` to `bounds[1]`,
    the second from `bounds[1]` to `bounds[2]`,
    the third from `bounds[2]` to `bounds[3]` etc.
    (the nth step runs from `bounds[n-1]` to `bounds[n]`).

    As a result, if `bounds` has length n, then it defines n - 1 time steps.
    """

    @bounds.validator
    def bounds_validator(
        self,
        attribute: attr.Attribute[Any],
        value: PINT_NUMPY_ARRAY,
    ) -> None:
        """
        Validate the received bounds
        """
        try:
            shape = value.shape
        except AttributeError as exc:
            msg = (
                "`bounds` must be one-dimensional but "
                "an error was raised while trying to check its shape. "
                f"Received bounds={value}."
            )
            raise AssertionError(msg) from exc

        if len(shape) != 1:
            msg = (
                "`bounds` must be one-dimensional. "
                f"Received `bounds` with shape {shape}"
            )
            raise AssertionError(msg)

        deltas = value[1:] - value[:-1]
        if (deltas <= 0).any():
            msg = (
                "`bounds` must be strictly monotonically increasing. "
                f"Received bounds={value}"
            )
            raise ValueError(msg)

    # Let attrs take care of __repr__

    def __str__(self) -> str:
        """
        Get string representation of self
        """
        return continuous_timeseries.formatting.to_str(
            self,
            [a.name for a in self.__attrs_attrs__],
        )

    def _repr_pretty_(
        self,
        p: IPython.lib.pretty.RepresentationPrinter,
        cycle: bool,
        indent: int = 4,
    ) -> None:
        """
        Get IPython pretty representation of self

        Used by IPython notebooks and other tools
        """
        continuous_timeseries.formatting.to_pretty(
            self,
            [a.name for a in self.__attrs_attrs__],
            p=p,
            cycle=cycle,
        )

    def _repr_html_(self) -> str:
        """
        Get html representation of self

        Used by IPython notebooks and other tools
        """
        return continuous_timeseries.formatting.to_html(
            self, [a.name for a in self.__attrs_attrs__], prefix=f"{__name__}."
        )

    def _repr_html_internal_row_(self) -> str:
        """
        Get html representation of self to use as an internal row of another object

        Used to avoid our representations having more information than we'd like.
        """
        return continuous_timeseries.formatting.to_html(
            self,
            [a.name for a in self.__attrs_attrs__],
            include_header=False,
        )

    @property
    def bounds_2d(self) -> PINT_NUMPY_ARRAY:
        """
        Get the bounds of the time steps in two-dimensions

        This representation can be useful for some operations.

        Returns
        -------
        :
            Bounds of the time steps in two-dimensions
            (bounds is the second dimension i.e. has size 2).
        """
        starts = self.bounds[:-1]
        ends = self.bounds[1:]

        res: PINT_NUMPY_ARRAY = np.vstack([starts, ends]).T  # type: ignore # mypy confused by pint

        return res

bounds class-attribute instance-attribute #

bounds: PINT_NUMPY_ARRAY = field()

Bounds of each time step in the time axis.

Must be one-dimensional and monotonically increasing.

The first time step runs from bounds[0] to bounds[1], the second from bounds[1] to bounds[2], the third from bounds[2] to bounds[3] etc. (the nth step runs from bounds[n-1] to bounds[n]).

As a result, if bounds has length n, then it defines n - 1 time steps.

bounds_2d property #

bounds_2d: PINT_NUMPY_ARRAY

Get the bounds of the time steps in two-dimensions

This representation can be useful for some operations.

Returns:

Type Description
PINT_NUMPY_ARRAY

Bounds of the time steps in two-dimensions (bounds is the second dimension i.e. has size 2).

__str__ #

__str__() -> str

Get string representation of self

Source code in src/continuous_timeseries/time_axis.py 
115
116
117
118
119
120
121
122
def __str__(self) -> str:
    """
    Get string representation of self
    """
    return continuous_timeseries.formatting.to_str(
        self,
        [a.name for a in self.__attrs_attrs__],
    )

bounds_validator #

bounds_validator(
    attribute: Attribute[Any], value: PINT_NUMPY_ARRAY
) -> None

Validate the received bounds

Source code in src/continuous_timeseries/time_axis.py 
 79
 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
@bounds.validator
def bounds_validator(
    self,
    attribute: attr.Attribute[Any],
    value: PINT_NUMPY_ARRAY,
) -> None:
    """
    Validate the received bounds
    """
    try:
        shape = value.shape
    except AttributeError as exc:
        msg = (
            "`bounds` must be one-dimensional but "
            "an error was raised while trying to check its shape. "
            f"Received bounds={value}."
        )
        raise AssertionError(msg) from exc

    if len(shape) != 1:
        msg = (
            "`bounds` must be one-dimensional. "
            f"Received `bounds` with shape {shape}"
        )
        raise AssertionError(msg)

    deltas = value[1:] - value[:-1]
    if (deltas <= 0).any():
        msg = (
            "`bounds` must be strictly monotonically increasing. "
            f"Received bounds={value}"
        )
        raise ValueError(msg)

Timeseries #

Timeseries representation

Methods:

Name Description
__str__

Get string representation of self
antidifferentiate

Antidifferentiate the time series
differentiate

Differentiate the time series
from_arrays

Initialise from arrays
integrate

Integrate the time series
interpolate

Interpolate onto a new time axis
plot

Plot
update_interpolation

Update the interpolation
update_interpolation_integral_preserving

Update the interpolation while preserving the integral

Attributes:

Name Type Description
discrete TimeseriesDiscrete

Discrete view of the time series
name str

Name of the time series
time_axis TimeAxis

Time axis of the timeseries
timeseries_continuous TimeseriesContinuous

Continuous version of the timeseries
Source code in src/continuous_timeseries/timeseries.py 
 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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
@define
class Timeseries:
    """Timeseries representation"""

    time_axis: TimeAxis
    """
    Time axis of the timeseries

    Used for plotting and creating the discrete form of the time series.
    """

    timeseries_continuous: TimeseriesContinuous
    """Continuous version of the timeseries"""

    # Let attrs take care of __repr__

    def __str__(self) -> str:
        """
        Get string representation of self
        """
        return continuous_timeseries.formatting.to_str(
            self,
            [a.name for a in self.__attrs_attrs__],
        )

    def _repr_pretty_(
        self,
        p: IPython.lib.pretty.RepresentationPrinter,
        cycle: bool,
        indent: int = 4,
    ) -> None:
        """
        Get IPython pretty representation of self

        Used by IPython notebooks and other tools
        """
        continuous_timeseries.formatting.to_pretty(
            self,
            [a.name for a in self.__attrs_attrs__],
            p=p,
            cycle=cycle,
        )

    def _repr_html_(self) -> str:
        """
        Get html representation of self

        Used by IPython notebooks and other tools
        """
        return continuous_timeseries.formatting.to_html(
            self,
            [a.name for a in self.__attrs_attrs__],
            prefix="continuous_timeseries.",
        )

    def _repr_html_internal_row_(self) -> str:
        """
        Get html representation of self to use as an internal row of another object

        Used to avoid our representations having more information than we'd like.
        """
        return continuous_timeseries.formatting.to_html(
            self,
            [a.name for a in self.__attrs_attrs__],
            include_header=False,
        )

    @property
    def name(self) -> str:
        """
        Name of the time series
        """
        return self.timeseries_continuous.name

    @property
    def discrete(self) -> TimeseriesDiscrete:
        """
        Discrete view of the time series
        """
        values_at_bounds = ValuesAtBounds(
            self.timeseries_continuous.interpolate(self.time_axis)
        )

        return TimeseriesDiscrete(
            name=self.name,
            time_axis=self.time_axis,
            values_at_bounds=values_at_bounds,
        )

    @classmethod
    def from_arrays(
        cls,
        x: PINT_NUMPY_ARRAY,
        y: PINT_NUMPY_ARRAY,
        interpolation: InterpolationOption,
        name: str,
    ) -> Timeseries:
        """
        Initialise from arrays

        Parameters
        ----------
        x
            The x-values from which to initialise

        y
            The y-values from which to initialise

        interpolation
            Interpolation to apply when converting
            the discrete values to a continuous representation

        name
            The value to use to set the result's name attribute

        Returns
        -------
        :
            Initialised [`Timeseries`][(m)].
        """
        continuous = discrete_to_continuous(
            x=x,
            y=y,
            interpolation=interpolation,
            name=name,
        )

        return cls(
            time_axis=TimeAxis(x),
            timeseries_continuous=continuous,
        )

    def differentiate(
        self,
        name_res: str | None = None,
    ) -> Timeseries:
        """
        Differentiate the time series

        Parameters
        ----------
        name_res
            Name to apply to the result.

            If not supplied, we use `f"{self.name}_derivative`.

        Returns
        -------
        :
            Derivative of the time series
        """
        if name_res is None:
            name_res = f"{self.name}_derivative"

        derivative = self.timeseries_continuous.differentiate(
            name_res=name_res,
        )

        return type(self)(
            time_axis=self.time_axis,
            timeseries_continuous=derivative,
        )

    def integrate(
        self,
        integration_constant: PINT_SCALAR,
        name_res: str | None = None,
    ) -> Timeseries:
        """
        Integrate the time series

        Parameters
        ----------
        integration_constant
            The integration constant to use when performing the integration.

            This is required to ensure that the integral is a definite integral.

        name_res
            Name to apply to the result.

            If not supplied, we use `f"{self.name}_integral`.

        Returns
        -------
        :
            Integral of the time series
        """
        if name_res is None:
            name_res = f"{self.name}_integral"

        integral = self.timeseries_continuous.integrate(
            integration_constant=integration_constant,
            name_res=name_res,
        )

        return type(self)(
            time_axis=self.time_axis,
            timeseries_continuous=integral,
        )

    def antidifferentiate(
        self,
        name_res: str | None = None,
    ) -> Timeseries:
        """
        Antidifferentiate the time series

        Parameters
        ----------
        name_res
            Name to apply to the result.

            If not supplied, we use `f"{self.name}_antiderivative`.

        Returns
        -------
        :
            Indefinite integral of the time series
        """
        if name_res is None:
            name_res = f"{self.name}_antiderivative"

        antiderivative = self.timeseries_continuous.antidifferentiate(
            name_res=name_res,
        )

        return type(self)(
            time_axis=self.time_axis,
            timeseries_continuous=antiderivative,
        )

    def interpolate(
        self, time_axis: TimeAxis | PINT_NUMPY_ARRAY, allow_extrapolation: bool = False
    ) -> Timeseries:
        """
        Interpolate onto a new time axis

        Parameters
        ----------
        time_axis
            Time axis to update to

        allow_extrapolation
            Should extrapolation be allowed?

        Returns
        -------
        :
            `self`, interpolated onto `time_axis`.
        """
        if not isinstance(time_axis, TimeAxis):
            time_axis = TimeAxis(time_axis)

        if not allow_extrapolation:
            try:
                check_no_times_outside_domain(
                    time_axis.bounds,
                    domain=self.timeseries_continuous.domain,
                )
            except ValueError as exc:
                msg = f"Extrapolation is not allowed ({allow_extrapolation=})."
                raise ExtrapolationNotAllowedError(msg) from exc

        timeseries_continuous_new = evolve(
            self.timeseries_continuous,
            domain=(np.min(time_axis.bounds), np.max(time_axis.bounds)),
        )

        return type(self)(
            time_axis=time_axis,
            timeseries_continuous=timeseries_continuous_new,
        )

    def update_interpolation(
        self,
        interpolation: InterpolationOption,
        name_res: str | None = None,
        warn_if_values_at_bounds_change: bool = True,
        check_change_func: Callable[
            [PINT_NUMPY_ARRAY, PINT_NUMPY_ARRAY], None
        ] = pint.testing.assert_allclose,
    ) -> Timeseries:
        """
        Update the interpolation

        Note that this uses default interpolation choices.
        This might not always be what you want.

        Parameters
        ----------
        interpolation
            Interpolation to change to

        name_res
            Name of the result

            If not supplied, we use
            `f"{self.name}_{interpolation.name}"`.

        warn_if_values_at_bounds_change
            Should a warning be raised if the `interpolation`
            causes the values at the time bounds defined by `self.time_axis` to change?

        check_change_func
            Function to use to check if the values at the bounds have changed.

            If the values are different, this function should raise an `AssertionError`.

        Returns
        -------
        :
            `self` with its interpolation updated to `interpolation`.

        Warns
        -----
        InterpolationUpdateChangedValuesAtBoundsWarning
            If updating the interpolation could the values at the time bounds to change
            and `warn_if_values_at_bounds_change` is `True`.
        """
        if name_res is None:
            name_res = f"{self.name}_{interpolation.name}"

        continuous = discrete_to_continuous(
            x=self.time_axis.bounds,
            y=self.timeseries_continuous.interpolate(self.time_axis),
            name=self.name,
            interpolation=interpolation,
        )
        continuous.name = name_res

        res = type(self)(
            time_axis=self.time_axis,
            timeseries_continuous=continuous,
        )

        if warn_if_values_at_bounds_change:
            try:
                check_change_func(
                    self.discrete.values_at_bounds.values,
                    res.discrete.values_at_bounds.values,
                )

            except AssertionError:
                msg = (
                    f"Updating interpolation to {interpolation.name} "
                    "has caused the values "
                    "at the bounds defined by `self.time_axis` to change."
                )
                warnings.warn(msg, InterpolationUpdateChangedValuesAtBoundsWarning)

        return res

    def update_interpolation_integral_preserving(
        self,
        interpolation: InterpolationOption,
        name_res: str | None = None,
        warn_if_values_at_bounds_change: bool = True,
        check_change_func: Callable[
            [PINT_NUMPY_ARRAY, PINT_NUMPY_ARRAY], None
        ] = pint.testing.assert_allclose,
    ) -> Timeseries:
        """
        Update the interpolation while preserving the integral

        This is useful if the integral of your quantity needs to be preserved,
        e.g. you want to do integral-preserving interpolation of emissions
        so that mass is conserved.

        It is obviously not possible to do this at all time points.
        So it would be more precise to say
        that the integral is preserved at the points in `self.time_axis`.

        We recommend being a bit careful with this.
        In general, performing this operation with linear or higher-order interpolations
        may not lead to the most intuitive result because of the
        quadratic or higher-order fitting that is done in cumulative space
        (quadratic and higher-order fitting is a difficult problem in general,
        see, for example, the multiple boundary condition options in
        [scipy's cubic spline](https://docs.scipy.org/doc/scipy/reference/generated/scipy.interpolate.CubicSpline.html)
        ).

        Parameters
        ----------
        interpolation
            Interpolation to update to

        name_res
            Name of the result

            If not supplied, we use
            `f"{self.name}_integral-preserving-interpolation-{interpolation.name}"`.

        warn_if_values_at_bounds_change
            Passed to [`update_interpolation`][(c)].

        check_change_func
            Passed to [`update_interpolation`][(c)].

        Returns
        -------
        :
            `self` with interpolation updated to `interpolation`
            while preserving the integral at `self.time_axis`.
        """
        if interpolation in (
            InterpolationOption.PiecewiseConstantNextLeftOpen,
            InterpolationOption.PiecewiseConstantPreviousLeftClosed,
            InterpolationOption.PiecewiseConstantPreviousLeftOpen,
        ):
            raise UnreachableIntegralPreservingInterpolationTarget(interpolation)

        if name_res is None:
            name_res = (
                f"{self.name}_integral-preserving-interpolation-{interpolation.name}"
            )

        if interpolation in (
            InterpolationOption.PiecewiseConstantPreviousLeftClosed,
            InterpolationOption.PiecewiseConstantPreviousLeftOpen,
            InterpolationOption.PiecewiseConstantNextLeftClosed,
            InterpolationOption.PiecewiseConstantNextLeftOpen,
        ):
            interpolation_cumulative = InterpolationOption.Linear

        elif interpolation in (InterpolationOption.Linear,):
            interpolation_cumulative = InterpolationOption.Quadratic

        elif interpolation in (InterpolationOption.Quadratic,):
            interpolation_cumulative = InterpolationOption.Cubic

        elif interpolation in (InterpolationOption.Cubic,):
            interpolation_cumulative = InterpolationOption.Quartic

        else:  # pragma: no cover
            raise NotImplementedError(interpolation)

        # Value doesn't matter as the value will be lost when we differentiate.
        integration_constant = 0.0 * (
            self.timeseries_continuous.values_units
            * self.timeseries_continuous.time_units
        )

        res = (
            self.integrate(integration_constant)
            .update_interpolation(
                interpolation_cumulative,
                warn_if_values_at_bounds_change=warn_if_values_at_bounds_change,
                check_change_func=check_change_func,
            )
            .differentiate(name_res=name_res)
        )

        return res

    def plot(
        self,
        show_continuous: bool = True,
        continuous_plot_kwargs: dict[str, Any] | None = None,
        show_discrete: bool = False,
        discrete_plot_kwargs: dict[str, Any] | None = None,
        ax: matplotlib.axes.Axes | None = None,
    ) -> matplotlib.axes.Axes:
        """
        Plot

        Parameters
        ----------
        show_continuous
            Should we plot the continuous representation of `self`?

        continuous_plot_kwargs
            Passed to `self.timeseries_continuous.plot`

            For docs, see
            [`TimeseriesContinuous.plot`][(p)].

        show_discrete
            Should we plot the discrete representation of `self`?

        discrete_plot_kwargs
            Passed to `self.timeseries_discrete.plot`

            For docs, see
            [`TimeseriesDiscrete.plot`][(p)].

        ax
            Axes on which to plot.

            If not supplied, a set of axes will be created.

        Returns
        -------
        :
            Axes on which the data was plotted
        """
        if ax is None:
            try:
                import matplotlib.pyplot as plt
            except ImportError as exc:
                raise MissingOptionalDependencyError(
                    "TimeseriesContinuous.plot", requirement="matplotlib"
                ) from exc

            _, ax = plt.subplots()

        if continuous_plot_kwargs is None:
            continuous_plot_kwargs = {}

        if discrete_plot_kwargs is None:
            discrete_plot_kwargs = {}

        if show_continuous:
            self.timeseries_continuous.plot(
                time_axis=self.time_axis,
                ax=ax,
                **continuous_plot_kwargs,
            )

        if show_discrete:
            self.discrete.plot(
                ax=ax,
                **discrete_plot_kwargs,
            )

        return ax

discrete property #

discrete: TimeseriesDiscrete

Discrete view of the time series

name property #

name: str

Name of the time series

time_axis instance-attribute #

time_axis: TimeAxis

Time axis of the timeseries

Used for plotting and creating the discrete form of the time series.

timeseries_continuous instance-attribute #

timeseries_continuous: TimeseriesContinuous

Continuous version of the timeseries

__str__ #

__str__() -> str

Get string representation of self

Source code in src/continuous_timeseries/timeseries.py 
101
102
103
104
105
106
107
108
def __str__(self) -> str:
    """
    Get string representation of self
    """
    return continuous_timeseries.formatting.to_str(
        self,
        [a.name for a in self.__attrs_attrs__],
    )

antidifferentiate #

antidifferentiate(
    name_res: str | None = None,
) -> Timeseries

Antidifferentiate the time series

Parameters:

Name Type Description Default
name_res str | None

Name to apply to the result.

If not supplied, we use f"{self.name}_antiderivative.

 None

Returns:

Type Description
Timeseries

Indefinite integral of the time series
Source code in src/continuous_timeseries/timeseries.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
def antidifferentiate(
    self,
    name_res: str | None = None,
) -> Timeseries:
    """
    Antidifferentiate the time series

    Parameters
    ----------
    name_res
        Name to apply to the result.

        If not supplied, we use `f"{self.name}_antiderivative`.

    Returns
    -------
    :
        Indefinite integral of the time series
    """
    if name_res is None:
        name_res = f"{self.name}_antiderivative"

    antiderivative = self.timeseries_continuous.antidifferentiate(
        name_res=name_res,
    )

    return type(self)(
        time_axis=self.time_axis,
        timeseries_continuous=antiderivative,
    )

differentiate #

differentiate(name_res: str | None = None) -> Timeseries

Differentiate the time series

Parameters:

Name Type Description Default
name_res str | None

Name to apply to the result.

If not supplied, we use f"{self.name}_derivative.

 None

Returns:

Type Description
Timeseries

Derivative of the time series
Source code in src/continuous_timeseries/timeseries.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def differentiate(
    self,
    name_res: str | None = None,
) -> Timeseries:
    """
    Differentiate the time series

    Parameters
    ----------
    name_res
        Name to apply to the result.

        If not supplied, we use `f"{self.name}_derivative`.

    Returns
    -------
    :
        Derivative of the time series
    """
    if name_res is None:
        name_res = f"{self.name}_derivative"

    derivative = self.timeseries_continuous.differentiate(
        name_res=name_res,
    )

    return type(self)(
        time_axis=self.time_axis,
        timeseries_continuous=derivative,
    )

from_arrays classmethod #

from_arrays(
    x: PINT_NUMPY_ARRAY,
    y: PINT_NUMPY_ARRAY,
    interpolation: InterpolationOption,
    name: str,
) -> Timeseries

Initialise from arrays

Parameters:

Name Type Description Default
x PINT_NUMPY_ARRAY

The x-values from which to initialise

 required
y PINT_NUMPY_ARRAY

The y-values from which to initialise

 required
interpolation InterpolationOption

Interpolation to apply when converting the discrete values to a continuous representation

 required
name str

The value to use to set the result's name attribute

 required

Returns:

Type Description
Timeseries

Initialised Timeseries.
Source code in src/continuous_timeseries/timeseries.py 
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
210
211
212
213
214
215
@classmethod
def from_arrays(
    cls,
    x: PINT_NUMPY_ARRAY,
    y: PINT_NUMPY_ARRAY,
    interpolation: InterpolationOption,
    name: str,
) -> Timeseries:
    """
    Initialise from arrays

    Parameters
    ----------
    x
        The x-values from which to initialise

    y
        The y-values from which to initialise

    interpolation
        Interpolation to apply when converting
        the discrete values to a continuous representation

    name
        The value to use to set the result's name attribute

    Returns
    -------
    :
        Initialised [`Timeseries`][(m)].
    """
    continuous = discrete_to_continuous(
        x=x,
        y=y,
        interpolation=interpolation,
        name=name,
    )

    return cls(
        time_axis=TimeAxis(x),
        timeseries_continuous=continuous,
    )

integrate #

integrate(
    integration_constant: PINT_SCALAR,
    name_res: str | None = None,
) -> Timeseries

Integrate the time series

Parameters:

Name Type Description Default
integration_constant PINT_SCALAR

The integration constant to use when performing the integration.

This is required to ensure that the integral is a definite integral.

 required
name_res str | None

Name to apply to the result.

If not supplied, we use f"{self.name}_integral.

 None

Returns:

Type Description
Timeseries

Integral of the time series
Source code in src/continuous_timeseries/timeseries.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def integrate(
    self,
    integration_constant: PINT_SCALAR,
    name_res: str | None = None,
) -> Timeseries:
    """
    Integrate the time series

    Parameters
    ----------
    integration_constant
        The integration constant to use when performing the integration.

        This is required to ensure that the integral is a definite integral.

    name_res
        Name to apply to the result.

        If not supplied, we use `f"{self.name}_integral`.

    Returns
    -------
    :
        Integral of the time series
    """
    if name_res is None:
        name_res = f"{self.name}_integral"

    integral = self.timeseries_continuous.integrate(
        integration_constant=integration_constant,
        name_res=name_res,
    )

    return type(self)(
        time_axis=self.time_axis,
        timeseries_continuous=integral,
    )

interpolate #

interpolate(
    time_axis: TimeAxis | PINT_NUMPY_ARRAY,
    allow_extrapolation: bool = False,
) -> Timeseries

Interpolate onto a new time axis

Parameters:

Name Type Description Default
time_axis TimeAxis | PINT_NUMPY_ARRAY

Time axis to update to

 required
allow_extrapolation bool

Should extrapolation be allowed?

 False

Returns:

Type Description
Timeseries

self, interpolated onto time_axis.
Source code in src/continuous_timeseries/timeseries.py 
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
def interpolate(
    self, time_axis: TimeAxis | PINT_NUMPY_ARRAY, allow_extrapolation: bool = False
) -> Timeseries:
    """
    Interpolate onto a new time axis

    Parameters
    ----------
    time_axis
        Time axis to update to

    allow_extrapolation
        Should extrapolation be allowed?

    Returns
    -------
    :
        `self`, interpolated onto `time_axis`.
    """
    if not isinstance(time_axis, TimeAxis):
        time_axis = TimeAxis(time_axis)

    if not allow_extrapolation:
        try:
            check_no_times_outside_domain(
                time_axis.bounds,
                domain=self.timeseries_continuous.domain,
            )
        except ValueError as exc:
            msg = f"Extrapolation is not allowed ({allow_extrapolation=})."
            raise ExtrapolationNotAllowedError(msg) from exc

    timeseries_continuous_new = evolve(
        self.timeseries_continuous,
        domain=(np.min(time_axis.bounds), np.max(time_axis.bounds)),
    )

    return type(self)(
        time_axis=time_axis,
        timeseries_continuous=timeseries_continuous_new,
    )

plot #

plot(
    show_continuous: bool = True,
    continuous_plot_kwargs: dict[str, Any] | None = None,
    show_discrete: bool = False,
    discrete_plot_kwargs: dict[str, Any] | None = None,
    ax: Axes | None = None,
) -> Axes

Plot

Parameters:

Name Type Description Default
show_continuous bool

Should we plot the continuous representation of self?

 True
continuous_plot_kwargs dict[str, Any] | None

Passed to self.timeseries_continuous.plot

For docs, see TimeseriesContinuous.plot.

 None
show_discrete bool

Should we plot the discrete representation of self?

 False
discrete_plot_kwargs dict[str, Any] | None

Passed to self.timeseries_discrete.plot

For docs, see TimeseriesDiscrete.plot.

 None
ax Axes | None

Axes on which to plot.

If not supplied, a set of axes will be created.

 None

Returns:

Type Description
Axes

Axes on which the data was plotted
Source code in src/continuous_timeseries/timeseries.py 
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
def plot(
    self,
    show_continuous: bool = True,
    continuous_plot_kwargs: dict[str, Any] | None = None,
    show_discrete: bool = False,
    discrete_plot_kwargs: dict[str, Any] | None = None,
    ax: matplotlib.axes.Axes | None = None,
) -> matplotlib.axes.Axes:
    """
    Plot

    Parameters
    ----------
    show_continuous
        Should we plot the continuous representation of `self`?

    continuous_plot_kwargs
        Passed to `self.timeseries_continuous.plot`

        For docs, see
        [`TimeseriesContinuous.plot`][(p)].

    show_discrete
        Should we plot the discrete representation of `self`?

    discrete_plot_kwargs
        Passed to `self.timeseries_discrete.plot`

        For docs, see
        [`TimeseriesDiscrete.plot`][(p)].

    ax
        Axes on which to plot.

        If not supplied, a set of axes will be created.

    Returns
    -------
    :
        Axes on which the data was plotted
    """
    if ax is None:
        try:
            import matplotlib.pyplot as plt
        except ImportError as exc:
            raise MissingOptionalDependencyError(
                "TimeseriesContinuous.plot", requirement="matplotlib"
            ) from exc

        _, ax = plt.subplots()

    if continuous_plot_kwargs is None:
        continuous_plot_kwargs = {}

    if discrete_plot_kwargs is None:
        discrete_plot_kwargs = {}

    if show_continuous:
        self.timeseries_continuous.plot(
            time_axis=self.time_axis,
            ax=ax,
            **continuous_plot_kwargs,
        )

    if show_discrete:
        self.discrete.plot(
            ax=ax,
            **discrete_plot_kwargs,
        )

    return ax

update_interpolation #

update_interpolation(
    interpolation: InterpolationOption,
    name_res: str | None = None,
    warn_if_values_at_bounds_change: bool = True,
    check_change_func: Callable[
        [PINT_NUMPY_ARRAY, PINT_NUMPY_ARRAY], None
    ] = assert_allclose,
) -> Timeseries

Update the interpolation

Note that this uses default interpolation choices. This might not always be what you want.

Parameters:

Name Type Description Default
interpolation InterpolationOption

Interpolation to change to

 required
name_res str | None

Name of the result

If not supplied, we use f"{self.name}_{interpolation.name}".

 None
warn_if_values_at_bounds_change bool

Should a warning be raised if the interpolation causes the values at the time bounds defined by self.time_axis to change?

 True
check_change_func Callable[[PINT_NUMPY_ARRAY, PINT_NUMPY_ARRAY], None]

Function to use to check if the values at the bounds have changed.

If the values are different, this function should raise an AssertionError.

 assert_allclose

Returns:

Type Description
Timeseries

self with its interpolation updated to interpolation.

Warns:

Type Description
InterpolationUpdateChangedValuesAtBoundsWarning

If updating the interpolation could the values at the time bounds to change and warn_if_values_at_bounds_change is True.
Source code in src/continuous_timeseries/timeseries.py 
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
def update_interpolation(
    self,
    interpolation: InterpolationOption,
    name_res: str | None = None,
    warn_if_values_at_bounds_change: bool = True,
    check_change_func: Callable[
        [PINT_NUMPY_ARRAY, PINT_NUMPY_ARRAY], None
    ] = pint.testing.assert_allclose,
) -> Timeseries:
    """
    Update the interpolation

    Note that this uses default interpolation choices.
    This might not always be what you want.

    Parameters
    ----------
    interpolation
        Interpolation to change to

    name_res
        Name of the result

        If not supplied, we use
        `f"{self.name}_{interpolation.name}"`.

    warn_if_values_at_bounds_change
        Should a warning be raised if the `interpolation`
        causes the values at the time bounds defined by `self.time_axis` to change?

    check_change_func
        Function to use to check if the values at the bounds have changed.

        If the values are different, this function should raise an `AssertionError`.

    Returns
    -------
    :
        `self` with its interpolation updated to `interpolation`.

    Warns
    -----
    InterpolationUpdateChangedValuesAtBoundsWarning
        If updating the interpolation could the values at the time bounds to change
        and `warn_if_values_at_bounds_change` is `True`.
    """
    if name_res is None:
        name_res = f"{self.name}_{interpolation.name}"

    continuous = discrete_to_continuous(
        x=self.time_axis.bounds,
        y=self.timeseries_continuous.interpolate(self.time_axis),
        name=self.name,
        interpolation=interpolation,
    )
    continuous.name = name_res

    res = type(self)(
        time_axis=self.time_axis,
        timeseries_continuous=continuous,
    )

    if warn_if_values_at_bounds_change:
        try:
            check_change_func(
                self.discrete.values_at_bounds.values,
                res.discrete.values_at_bounds.values,
            )

        except AssertionError:
            msg = (
                f"Updating interpolation to {interpolation.name} "
                "has caused the values "
                "at the bounds defined by `self.time_axis` to change."
            )
            warnings.warn(msg, InterpolationUpdateChangedValuesAtBoundsWarning)

    return res

update_interpolation_integral_preserving #

update_interpolation_integral_preserving(
    interpolation: InterpolationOption,
    name_res: str | None = None,
    warn_if_values_at_bounds_change: bool = True,
    check_change_func: Callable[
        [PINT_NUMPY_ARRAY, PINT_NUMPY_ARRAY], None
    ] = assert_allclose,
) -> Timeseries

Update the interpolation while preserving the integral

This is useful if the integral of your quantity needs to be preserved, e.g. you want to do integral-preserving interpolation of emissions so that mass is conserved.

It is obviously not possible to do this at all time points. So it would be more precise to say that the integral is preserved at the points in self.time_axis.

We recommend being a bit careful with this. In general, performing this operation with linear or higher-order interpolations may not lead to the most intuitive result because of the quadratic or higher-order fitting that is done in cumulative space (quadratic and higher-order fitting is a difficult problem in general, see, for example, the multiple boundary condition options in scipy's cubic spline ).

Parameters:

Name Type Description Default
interpolation InterpolationOption

Interpolation to update to

 required
name_res str | None

Name of the result

If not supplied, we use f"{self.name}_integral-preserving-interpolation-{interpolation.name}".

 None
warn_if_values_at_bounds_change bool

Passed to update_interpolation.

 True
check_change_func Callable[[PINT_NUMPY_ARRAY, PINT_NUMPY_ARRAY], None]

Passed to update_interpolation.

 assert_allclose

Returns:

Type Description
Timeseries

self with interpolation updated to interpolation while preserving the integral at self.time_axis.
Source code in src/continuous_timeseries/timeseries.py 
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
def update_interpolation_integral_preserving(
    self,
    interpolation: InterpolationOption,
    name_res: str | None = None,
    warn_if_values_at_bounds_change: bool = True,
    check_change_func: Callable[
        [PINT_NUMPY_ARRAY, PINT_NUMPY_ARRAY], None
    ] = pint.testing.assert_allclose,
) -> Timeseries:
    """
    Update the interpolation while preserving the integral

    This is useful if the integral of your quantity needs to be preserved,
    e.g. you want to do integral-preserving interpolation of emissions
    so that mass is conserved.

    It is obviously not possible to do this at all time points.
    So it would be more precise to say
    that the integral is preserved at the points in `self.time_axis`.

    We recommend being a bit careful with this.
    In general, performing this operation with linear or higher-order interpolations
    may not lead to the most intuitive result because of the
    quadratic or higher-order fitting that is done in cumulative space
    (quadratic and higher-order fitting is a difficult problem in general,
    see, for example, the multiple boundary condition options in
    [scipy's cubic spline](https://docs.scipy.org/doc/scipy/reference/generated/scipy.interpolate.CubicSpline.html)
    ).

    Parameters
    ----------
    interpolation
        Interpolation to update to

    name_res
        Name of the result

        If not supplied, we use
        `f"{self.name}_integral-preserving-interpolation-{interpolation.name}"`.

    warn_if_values_at_bounds_change
        Passed to [`update_interpolation`][(c)].

    check_change_func
        Passed to [`update_interpolation`][(c)].

    Returns
    -------
    :
        `self` with interpolation updated to `interpolation`
        while preserving the integral at `self.time_axis`.
    """
    if interpolation in (
        InterpolationOption.PiecewiseConstantNextLeftOpen,
        InterpolationOption.PiecewiseConstantPreviousLeftClosed,
        InterpolationOption.PiecewiseConstantPreviousLeftOpen,
    ):
        raise UnreachableIntegralPreservingInterpolationTarget(interpolation)

    if name_res is None:
        name_res = (
            f"{self.name}_integral-preserving-interpolation-{interpolation.name}"
        )

    if interpolation in (
        InterpolationOption.PiecewiseConstantPreviousLeftClosed,
        InterpolationOption.PiecewiseConstantPreviousLeftOpen,
        InterpolationOption.PiecewiseConstantNextLeftClosed,
        InterpolationOption.PiecewiseConstantNextLeftOpen,
    ):
        interpolation_cumulative = InterpolationOption.Linear

    elif interpolation in (InterpolationOption.Linear,):
        interpolation_cumulative = InterpolationOption.Quadratic

    elif interpolation in (InterpolationOption.Quadratic,):
        interpolation_cumulative = InterpolationOption.Cubic

    elif interpolation in (InterpolationOption.Cubic,):
        interpolation_cumulative = InterpolationOption.Quartic

    else:  # pragma: no cover
        raise NotImplementedError(interpolation)

    # Value doesn't matter as the value will be lost when we differentiate.
    integration_constant = 0.0 * (
        self.timeseries_continuous.values_units
        * self.timeseries_continuous.time_units
    )

    res = (
        self.integrate(integration_constant)
        .update_interpolation(
            interpolation_cumulative,
            warn_if_values_at_bounds_change=warn_if_values_at_bounds_change,
            check_change_func=check_change_func,
        )
        .differentiate(name_res=name_res)
    )

    return res

TimeseriesContinuous #

Continuous time series representation

Methods:

Name Description
__str__

Get string representation of self
antidifferentiate

Antidifferentiate
differentiate

Differentiate
domain_validator

Validate the received values
integrate

Integrate
interpolate

Interpolate values on a given time axis
plot

Plot the function
to_discrete_timeseries

Convert to TimeseriesDiscrete

Attributes:

Name Type Description
domain tuple[PINT_SCALAR, PINT_SCALAR]

Domain over which the function can be evaluated
function ContinuousFunctionLike

The continuous function that represents this timeseries.
name str

Name of the timeseries
time_units PlainUnit

The units of the time axis
values_units PlainUnit

The units of the values
Source code in src/continuous_timeseries/timeseries_continuous.py 
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
@define
class TimeseriesContinuous:
    """
    Continuous time series representation
    """

    name: str
    """Name of the timeseries"""

    time_units: pint.facets.plain.PlainUnit
    """The units of the time axis"""

    values_units: pint.facets.plain.PlainUnit
    """The units of the values"""

    function: ContinuousFunctionLike
    """
    The continuous function that represents this timeseries.
    """

    domain: tuple[PINT_SCALAR, PINT_SCALAR] = field()
    """
    Domain over which the function can be evaluated
    """

    @domain.validator
    def domain_validator(
        self,
        attribute: attr.Attribute[Any],
        value: tuple[PINT_SCALAR, PINT_SCALAR],
    ) -> None:
        """
        Validate the received values
        """
        try:
            validate_domain(value)
        except AssertionError as exc:
            msg = "The value supplied for `domain` failed validation."
            raise ValueError(msg) from exc

    # Let attrs take care of __repr__

    def __str__(self) -> str:
        """
        Get string representation of self
        """
        return continuous_timeseries.formatting.to_str(
            self,
            [a.name for a in self.__attrs_attrs__],
        )

    def _repr_pretty_(
        self,
        p: IPython.lib.pretty.RepresentationPrinter,
        cycle: bool,
        indent: int = 4,
    ) -> None:
        """
        Get IPython pretty representation of self

        Used by IPython notebooks and other tools
        """
        continuous_timeseries.formatting.to_pretty(
            self,
            [a.name for a in self.__attrs_attrs__],
            p=p,
            cycle=cycle,
        )

    def _repr_html_(self) -> str:
        """
        Get html representation of self

        Used by IPython notebooks and other tools
        """
        return continuous_timeseries.formatting.to_html(
            self,
            [a.name for a in self.__attrs_attrs__],
            prefix="continuous_timeseries.",
        )

    def _repr_html_internal_row_(self) -> str:
        """
        Get html representation of self to use as an internal row of another object

        Used to avoid our representations having more information than we'd like.
        """
        return continuous_timeseries.formatting.to_html(
            self,
            [a.name for a in self.__attrs_attrs__],
            include_header=False,
        )

    def to_discrete_timeseries(
        self,
        time_axis: TimeAxis,
        allow_extrapolation: bool = False,
    ) -> TimeseriesDiscrete:
        """
        Convert to [`TimeseriesDiscrete`][(p)]

        Parameters
        ----------
        time_axis
            Time axis to use for the conversion

        allow_extrapolation
            Should extrapolation be allowed during the conversion?

        Returns
        -------
        :
            Discrete representation of `self`
        """
        # Late import to avoid circularity
        from continuous_timeseries.timeseries_discrete import TimeseriesDiscrete

        res = TimeseriesDiscrete(
            name=self.name,
            time_axis=time_axis,
            values_at_bounds=ValuesAtBounds(
                self.interpolate(time_axis, allow_extrapolation=allow_extrapolation)
            ),
        )

        return res

    def interpolate(
        self, time_axis: TimeAxis | PINT_NUMPY_ARRAY, allow_extrapolation: bool = False
    ) -> PINT_NUMPY_ARRAY:
        """
        Interpolate values on a given time axis

        Parameters
        ----------
        time_axis
            Time axis onto which to interpolate values

        allow_extrapolation
            Should extrapolation be allowed while interpolating?

        Returns
        -------
        :
            Interpolated values
        """
        if isinstance(time_axis, TimeAxis):
            time_axis = time_axis.bounds

        if not allow_extrapolation:
            try:
                check_no_times_outside_domain(
                    time_axis,
                    domain=self.domain,
                )
            except ValueError as exc:
                msg = f"Extrapolation is not allowed ({allow_extrapolation=})."
                raise ExtrapolationNotAllowedError(msg) from exc

        times_m = time_axis.to(self.time_units).m
        values_m = self.function(
            times_m,
            # We have already checked the domain above.
            # Hence, we want the function to extrapolate if needed.
            allow_extrapolation=True,
        )

        if np.isnan(values_m).any():  # pragma: no cover
            # This is an escape hatch.
            # In general, we expect `self.function` to handle NaNs
            # before we get to this point.
            msg = (
                "The result of calling `self.function` contains NaNs. "
                f"The result is {values_m!r}."
            )
            raise AssertionError(msg)

        res: PINT_NUMPY_ARRAY = values_m * self.values_units

        return res

    def integrate(
        self, integration_constant: PINT_SCALAR, name_res: str | None = None
    ) -> TimeseriesContinuous:
        """
        Integrate

        Parameters
        ----------
        integration_constant
            Integration constant to use when performing the integration

        name_res
            Name to use for the output.

            If not supplied, we use f"{self.name}_integral".

        Returns
        -------
        :
            Integral of `self`.
        """
        if name_res is None:
            name_res = f"{self.name}_integral"

        integral_values_units = self.values_units * self.time_units

        integral = self.function.integrate(
            integration_constant=integration_constant.to(integral_values_units).m,
            domain_start=self.domain[0].to(self.time_units).m,
        )

        return type(self)(
            name=name_res,
            time_units=self.time_units,
            values_units=integral_values_units,
            function=integral,
            domain=self.domain,
        )

    def antidifferentiate(self, name_res: str | None = None) -> TimeseriesContinuous:
        """
        Antidifferentiate

        Parameters
        ----------
        name_res
            Name to use for the output.

            If not supplied, we use f"{self.name}_antiderivative".

        Returns
        -------
        :
            Antiderivative of `self`.
        """
        if name_res is None:
            name_res = f"{self.name}_antiderivative"

        antiderivative_values_units = self.values_units * self.time_units

        antiderivative = self.function.antidifferentiate(
            domain_start=self.domain[0].to(self.time_units).m,
        )

        return type(self)(
            name=name_res,
            time_units=self.time_units,
            values_units=antiderivative_values_units,
            function=antiderivative,
            domain=self.domain,
        )

    def differentiate(self, name_res: str | None = None) -> TimeseriesContinuous:
        """
        Differentiate

        Parameters
        ----------
        name_res
            Name to use for the output.

            If not supplied, we use f"{self.name}_derivative".

        Returns
        -------
        :
            Integral of `self`.
        """
        if name_res is None:
            name_res = f"{self.name}_derivative"

        derivative_values_units = self.values_units / self.time_units

        derivative = self.function.differentiate()

        return type(self)(
            name=name_res,
            time_units=self.time_units,
            values_units=derivative_values_units,
            function=derivative,
            domain=self.domain,
        )

    def plot(
        self,
        time_axis: TimeAxis | PINT_NUMPY_ARRAY,
        res_increase: int = 500,
        label: str | None = None,
        ax: matplotlib.axes.Axes | None = None,
        warn_if_plotting_magnitudes: bool = True,
        **kwargs: Any,
    ) -> matplotlib.axes.Axes:
        """
        Plot the function

        We can't see an easy way to plot the continuous function exactly,
        so we approximate by interpolating very finely
        then just using a standard linear interpolation between the points.

        Parameters
        ----------
        time_axis
            Time axis to use for plotting.

            All points in `time_axis` will be included as plotting points.

        res_increase
            The amount by which to increase the resolution of the x-axis when plotting.

            If equal to 1, then only the points in `time_axis` will be plotted.
            If equal to 100, then there will be 100 times as many points
            plotted as the number of points in `time_axis`.
            If equal to n, then there will be n times as many points
            plotted as the number of points in `time_axis`.

        label
            Label to use when plotting the data.

            If not supplied, we use the `self.name`.

        ax
            Axes on which to plot.

            If not supplied, a set of axes will be created.

        warn_if_plotting_magnitudes
            Should a warning be raised if the units of the values
            are not considered while plotting?

        **kwargs
            Keyword arguments to pass to `ax.plot`.

        Returns
        -------
        :
            Axes on which the data was plotted
        """
        if isinstance(time_axis, TimeAxis):
            time_axis = time_axis.bounds

        if label is None:
            label = self.name

        if ax is None:
            try:
                import matplotlib.pyplot as plt
            except ImportError as exc:
                raise MissingOptionalDependencyError(
                    "TimeseriesContinuous.plot", requirement="matplotlib"
                ) from exc

            _, ax = plt.subplots()

        # Interpolate based on res_increase.
        # Then plot interpolated using linear joins
        # (as far as I can tell, this is the only general way to do this,
        # although it is slower than using e.g. step for piecewise constant stuff).)
        plot_points = get_plot_points(time_axis, res_increase=res_increase)
        plot_values = self.interpolate(plot_points)

        x_vals = get_plot_vals(
            plot_points,
            "time_axis",
            warn_if_magnitudes=warn_if_plotting_magnitudes,
        )
        y_vals = get_plot_vals(
            plot_values,
            "show_values",
            warn_if_magnitudes=warn_if_plotting_magnitudes,
        )

        ax.plot(x_vals, y_vals, label=label, **kwargs)

        return ax

domain class-attribute instance-attribute #

domain: tuple[PINT_SCALAR, PINT_SCALAR] = field()

Domain over which the function can be evaluated

function instance-attribute #

function: ContinuousFunctionLike

The continuous function that represents this timeseries.

name instance-attribute #

name: str

Name of the timeseries

time_units instance-attribute #

time_units: PlainUnit

The units of the time axis

values_units instance-attribute #

values_units: PlainUnit

The units of the values

__str__ #

__str__() -> str

Get string representation of self

Source code in src/continuous_timeseries/timeseries_continuous.py 
453
454
455
456
457
458
459
460
def __str__(self) -> str:
    """
    Get string representation of self
    """
    return continuous_timeseries.formatting.to_str(
        self,
        [a.name for a in self.__attrs_attrs__],
    )

antidifferentiate #

antidifferentiate(
    name_res: str | None = None,
) -> TimeseriesContinuous

Antidifferentiate

Parameters:

Name Type Description Default
name_res str | None

Name to use for the output.

If not supplied, we use f"{self.name}_antiderivative".

 None

Returns:

Type Description
TimeseriesContinuous

Antiderivative of self.
Source code in src/continuous_timeseries/timeseries_continuous.py 
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
def antidifferentiate(self, name_res: str | None = None) -> TimeseriesContinuous:
    """
    Antidifferentiate

    Parameters
    ----------
    name_res
        Name to use for the output.

        If not supplied, we use f"{self.name}_antiderivative".

    Returns
    -------
    :
        Antiderivative of `self`.
    """
    if name_res is None:
        name_res = f"{self.name}_antiderivative"

    antiderivative_values_units = self.values_units * self.time_units

    antiderivative = self.function.antidifferentiate(
        domain_start=self.domain[0].to(self.time_units).m,
    )

    return type(self)(
        name=name_res,
        time_units=self.time_units,
        values_units=antiderivative_values_units,
        function=antiderivative,
        domain=self.domain,
    )

differentiate #

differentiate(
    name_res: str | None = None,
) -> TimeseriesContinuous

Differentiate

Parameters:

Name Type Description Default
name_res str | None

Name to use for the output.

If not supplied, we use f"{self.name}_derivative".

 None

Returns:

Type Description
TimeseriesContinuous

Integral of self.
Source code in src/continuous_timeseries/timeseries_continuous.py 
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
def differentiate(self, name_res: str | None = None) -> TimeseriesContinuous:
    """
    Differentiate

    Parameters
    ----------
    name_res
        Name to use for the output.

        If not supplied, we use f"{self.name}_derivative".

    Returns
    -------
    :
        Integral of `self`.
    """
    if name_res is None:
        name_res = f"{self.name}_derivative"

    derivative_values_units = self.values_units / self.time_units

    derivative = self.function.differentiate()

    return type(self)(
        name=name_res,
        time_units=self.time_units,
        values_units=derivative_values_units,
        function=derivative,
        domain=self.domain,
    )

domain_validator #

domain_validator(
    attribute: Attribute[Any],
    value: tuple[PINT_SCALAR, PINT_SCALAR],
) -> None

Validate the received values

Source code in src/continuous_timeseries/timeseries_continuous.py 
436
437
438
439
440
441
442
443
444
445
446
447
448
449
@domain.validator
def domain_validator(
    self,
    attribute: attr.Attribute[Any],
    value: tuple[PINT_SCALAR, PINT_SCALAR],
) -> None:
    """
    Validate the received values
    """
    try:
        validate_domain(value)
    except AssertionError as exc:
        msg = "The value supplied for `domain` failed validation."
        raise ValueError(msg) from exc

integrate #

integrate(
    integration_constant: PINT_SCALAR,
    name_res: str | None = None,
) -> TimeseriesContinuous

Integrate

Parameters:

Name Type Description Default
integration_constant PINT_SCALAR

Integration constant to use when performing the integration

 required
name_res str | None

Name to use for the output.

If not supplied, we use f"{self.name}_integral".

 None

Returns:

Type Description
TimeseriesContinuous

Integral of self.
Source code in src/continuous_timeseries/timeseries_continuous.py 
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
def integrate(
    self, integration_constant: PINT_SCALAR, name_res: str | None = None
) -> TimeseriesContinuous:
    """
    Integrate

    Parameters
    ----------
    integration_constant
        Integration constant to use when performing the integration

    name_res
        Name to use for the output.

        If not supplied, we use f"{self.name}_integral".

    Returns
    -------
    :
        Integral of `self`.
    """
    if name_res is None:
        name_res = f"{self.name}_integral"

    integral_values_units = self.values_units * self.time_units

    integral = self.function.integrate(
        integration_constant=integration_constant.to(integral_values_units).m,
        domain_start=self.domain[0].to(self.time_units).m,
    )

    return type(self)(
        name=name_res,
        time_units=self.time_units,
        values_units=integral_values_units,
        function=integral,
        domain=self.domain,
    )

interpolate #

interpolate(
    time_axis: TimeAxis | PINT_NUMPY_ARRAY,
    allow_extrapolation: bool = False,
) -> PINT_NUMPY_ARRAY

Interpolate values on a given time axis

Parameters:

Name Type Description Default
time_axis TimeAxis | PINT_NUMPY_ARRAY

Time axis onto which to interpolate values

 required
allow_extrapolation bool

Should extrapolation be allowed while interpolating?

 False

Returns:

Type Description
PINT_NUMPY_ARRAY

Interpolated values
Source code in src/continuous_timeseries/timeseries_continuous.py 
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
def interpolate(
    self, time_axis: TimeAxis | PINT_NUMPY_ARRAY, allow_extrapolation: bool = False
) -> PINT_NUMPY_ARRAY:
    """
    Interpolate values on a given time axis

    Parameters
    ----------
    time_axis
        Time axis onto which to interpolate values

    allow_extrapolation
        Should extrapolation be allowed while interpolating?

    Returns
    -------
    :
        Interpolated values
    """
    if isinstance(time_axis, TimeAxis):
        time_axis = time_axis.bounds

    if not allow_extrapolation:
        try:
            check_no_times_outside_domain(
                time_axis,
                domain=self.domain,
            )
        except ValueError as exc:
            msg = f"Extrapolation is not allowed ({allow_extrapolation=})."
            raise ExtrapolationNotAllowedError(msg) from exc

    times_m = time_axis.to(self.time_units).m
    values_m = self.function(
        times_m,
        # We have already checked the domain above.
        # Hence, we want the function to extrapolate if needed.
        allow_extrapolation=True,
    )

    if np.isnan(values_m).any():  # pragma: no cover
        # This is an escape hatch.
        # In general, we expect `self.function` to handle NaNs
        # before we get to this point.
        msg = (
            "The result of calling `self.function` contains NaNs. "
            f"The result is {values_m!r}."
        )
        raise AssertionError(msg)

    res: PINT_NUMPY_ARRAY = values_m * self.values_units

    return res

plot #

plot(
    time_axis: TimeAxis | PINT_NUMPY_ARRAY,
    res_increase: int = 500,
    label: str | None = None,
    ax: Axes | None = None,
    warn_if_plotting_magnitudes: bool = True,
    **kwargs: Any,
) -> Axes

Plot the function

We can't see an easy way to plot the continuous function exactly, so we approximate by interpolating very finely then just using a standard linear interpolation between the points.

Parameters:

Name Type Description Default
time_axis TimeAxis | PINT_NUMPY_ARRAY

Time axis to use for plotting.

All points in time_axis will be included as plotting points.

 required
res_increase int

The amount by which to increase the resolution of the x-axis when plotting.

If equal to 1, then only the points in time_axis will be plotted. If equal to 100, then there will be 100 times as many points plotted as the number of points in time_axis. If equal to n, then there will be n times as many points plotted as the number of points in time_axis.

 500
label str | None

Label to use when plotting the data.

If not supplied, we use the self.name.

 None
ax Axes | None

Axes on which to plot.

If not supplied, a set of axes will be created.

 None
warn_if_plotting_magnitudes bool

Should a warning be raised if the units of the values are not considered while plotting?

 True
**kwargs Any

Keyword arguments to pass to ax.plot.

 {}

Returns:

Type Description
Axes

Axes on which the data was plotted
Source code in src/continuous_timeseries/timeseries_continuous.py 
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
def plot(
    self,
    time_axis: TimeAxis | PINT_NUMPY_ARRAY,
    res_increase: int = 500,
    label: str | None = None,
    ax: matplotlib.axes.Axes | None = None,
    warn_if_plotting_magnitudes: bool = True,
    **kwargs: Any,
) -> matplotlib.axes.Axes:
    """
    Plot the function

    We can't see an easy way to plot the continuous function exactly,
    so we approximate by interpolating very finely
    then just using a standard linear interpolation between the points.

    Parameters
    ----------
    time_axis
        Time axis to use for plotting.

        All points in `time_axis` will be included as plotting points.

    res_increase
        The amount by which to increase the resolution of the x-axis when plotting.

        If equal to 1, then only the points in `time_axis` will be plotted.
        If equal to 100, then there will be 100 times as many points
        plotted as the number of points in `time_axis`.
        If equal to n, then there will be n times as many points
        plotted as the number of points in `time_axis`.

    label
        Label to use when plotting the data.

        If not supplied, we use the `self.name`.

    ax
        Axes on which to plot.

        If not supplied, a set of axes will be created.

    warn_if_plotting_magnitudes
        Should a warning be raised if the units of the values
        are not considered while plotting?

    **kwargs
        Keyword arguments to pass to `ax.plot`.

    Returns
    -------
    :
        Axes on which the data was plotted
    """
    if isinstance(time_axis, TimeAxis):
        time_axis = time_axis.bounds

    if label is None:
        label = self.name

    if ax is None:
        try:
            import matplotlib.pyplot as plt
        except ImportError as exc:
            raise MissingOptionalDependencyError(
                "TimeseriesContinuous.plot", requirement="matplotlib"
            ) from exc

        _, ax = plt.subplots()

    # Interpolate based on res_increase.
    # Then plot interpolated using linear joins
    # (as far as I can tell, this is the only general way to do this,
    # although it is slower than using e.g. step for piecewise constant stuff).)
    plot_points = get_plot_points(time_axis, res_increase=res_increase)
    plot_values = self.interpolate(plot_points)

    x_vals = get_plot_vals(
        plot_points,
        "time_axis",
        warn_if_magnitudes=warn_if_plotting_magnitudes,
    )
    y_vals = get_plot_vals(
        plot_values,
        "show_values",
        warn_if_magnitudes=warn_if_plotting_magnitudes,
    )

    ax.plot(x_vals, y_vals, label=label, **kwargs)

    return ax

to_discrete_timeseries #

to_discrete_timeseries(
    time_axis: TimeAxis, allow_extrapolation: bool = False
) -> TimeseriesDiscrete

Convert to TimeseriesDiscrete

Parameters:

Name Type Description Default
time_axis TimeAxis

Time axis to use for the conversion

 required
allow_extrapolation bool

Should extrapolation be allowed during the conversion?

 False

Returns:

Type Description
TimeseriesDiscrete

Discrete representation of self
Source code in src/continuous_timeseries/timeseries_continuous.py 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
def to_discrete_timeseries(
    self,
    time_axis: TimeAxis,
    allow_extrapolation: bool = False,
) -> TimeseriesDiscrete:
    """
    Convert to [`TimeseriesDiscrete`][(p)]

    Parameters
    ----------
    time_axis
        Time axis to use for the conversion

    allow_extrapolation
        Should extrapolation be allowed during the conversion?

    Returns
    -------
    :
        Discrete representation of `self`
    """
    # Late import to avoid circularity
    from continuous_timeseries.timeseries_discrete import TimeseriesDiscrete

    res = TimeseriesDiscrete(
        name=self.name,
        time_axis=time_axis,
        values_at_bounds=ValuesAtBounds(
            self.interpolate(time_axis, allow_extrapolation=allow_extrapolation)
        ),
    )

    return res

TimeseriesDiscrete #

Discrete time series representation

Methods:

Name Description
__str__

Get string representation of self
plot

Plot the data
to_continuous_timeseries

Convert to TimeseriesContinuous
values_at_bounds_validator

Validate the received values

Attributes:

Name Type Description
name str

Name of the timeseries
time_axis TimeAxis

Time axis of the timeseries
values_at_bounds ValuesAtBounds

Values at the bounds defined by self.time_axis
Source code in src/continuous_timeseries/timeseries_discrete.py 
 45
 46
 47
 48
 49
 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
 78
 79
 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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
@define
class TimeseriesDiscrete:
    """
    Discrete time series representation
    """

    name: str
    """Name of the timeseries"""

    time_axis: TimeAxis
    """Time axis of the timeseries"""

    values_at_bounds: ValuesAtBounds = field()
    """
    Values at the bounds defined by `self.time_axis`

    Must hold values that are the same length as `self.time_axis`.
    """

    @values_at_bounds.validator
    def values_at_bounds_validator(
        self,
        attribute: attr.Attribute[Any],
        value: ValuesAtBounds,
    ) -> None:
        """
        Validate the received values
        """
        if value.values.shape != self.time_axis.bounds.shape:
            msg = (
                "`values_at_bounds` must have values "
                "that are the same shape as `self.time_axis.bounds`. "
                f"Received values_at_bounds.values.shape={value.values.shape} "
                f"while {self.time_axis.bounds.shape=}."
            )
            raise AssertionError(msg)

    # Let attrs take care of __repr__

    def __str__(self) -> str:
        """
        Get string representation of self
        """
        return continuous_timeseries.formatting.to_str(
            self,
            [a.name for a in self.__attrs_attrs__],
        )

    def _repr_pretty_(
        self,
        p: IPython.lib.pretty.RepresentationPrinter,
        cycle: bool,
        indent: int = 4,
    ) -> None:
        """
        Get IPython pretty representation of self

        Used by IPython notebooks and other tools
        """
        continuous_timeseries.formatting.to_pretty(
            self,
            [a.name for a in self.__attrs_attrs__],
            p=p,
            cycle=cycle,
        )

    def _repr_html_(self) -> str:
        """
        Get html representation of self

        Used by IPython notebooks and other tools
        """
        return continuous_timeseries.formatting.to_html(
            self,
            [a.name for a in self.__attrs_attrs__],
            prefix="continuous_timeseries.",
        )

    def _repr_html_internal_row_(self) -> str:
        """
        Get html representation of self to use as an internal row of another object

        Used to avoid our representations having more information than we'd like.
        """
        return continuous_timeseries.formatting.to_html(
            self,
            [a.name for a in self.__attrs_attrs__],
            include_header=False,
        )

    def to_continuous_timeseries(
        self,
        interpolation: InterpolationOption,
        warn_if_output_values_at_bounds_could_confuse: bool = True,
        check_change_func: Callable[
            [PINT_NUMPY_ARRAY, PINT_NUMPY_ARRAY], None
        ] = pint.testing.assert_allclose,
    ) -> TimeseriesContinuous:
        """
        Convert to [`TimeseriesContinuous`][(p)]

        Parameters
        ----------
        interpolation
            Interpolation to use for the conversion

        warn_if_output_values_at_bounds_could_confuse
            Should a warning be raised if the `interpolation` choice
            means that the value of the output timeseries at
            point `x(n)` is not equal to `y(n)`?

        check_change_func
            Function to use to check
            if the value of the output at `x(n)` is equal to `y(n)`.

            If the values are different, this function should raise an `AssertionError`.

        Returns
        -------
        :
            Continuous representation of `self` for the interpolation
            specified by `interpolation`
        """
        x = self.time_axis.bounds
        y = self.values_at_bounds.values
        res = discrete_to_continuous(
            x=x,
            y=y,
            name=self.name,
            interpolation=interpolation,
        )

        if warn_if_output_values_at_bounds_could_confuse:
            try:
                check_change_func(y, res.interpolate(x))

            except AssertionError:
                msg = (
                    f"Using the interpolation {interpolation.name} "
                    "means that the y-values do not line up exactly with the x-values. "
                    "In other words, in the output, "
                    "y(x(1)) is not equal to the input y(1), "
                    "y(x(2)) is not equal to the input y(2), "
                    "y(x(n)) is not equal to the input y(n) etc. "
                    "This may cause confusion. "
                    "Either ignore this warning, "
                    "suppress it "
                    "(by passing `warn_if_values_at_bounds_could_confuse=False` "
                    "or via Python's `warnings` module settings) "
                    "or choose a different interpolation option."
                )
                warnings.warn(msg, InterpolationUpdateChangedValuesAtBoundsWarning)

        return res

    def plot(
        self,
        label: str | None = None,
        ax: matplotlib.axes.Axes | None = None,
        warn_if_plotting_magnitudes: bool = True,
        **kwargs: Any,
    ) -> matplotlib.axes.Axes:
        """
        Plot the data

        Parameters
        ----------
        label
            Label to use when plotting the data.

            If not supplied, we use the `self.name`.

        ax
            Axes on which to plot.

            If not supplied, a set of axes will be created.

        warn_if_plotting_magnitudes
            Should a warning be raised if the units of the values
            are not considered while plotting?

        **kwargs
            Keyword arguments to pass to `ax.scatter`.

        Returns
        -------
        :
            Axes on which the data was plotted
        """
        if label is None:
            label = self.name

        if ax is None:
            try:
                import matplotlib.pyplot as plt
            except ImportError as exc:
                raise MissingOptionalDependencyError(
                    "TimeseriesDiscrete.plot", requirement="matplotlib"
                ) from exc

            _, ax = plt.subplots()

        x_vals = get_plot_vals(
            self.time_axis.bounds,
            "self.time_axis.bounds",
            warn_if_magnitudes=warn_if_plotting_magnitudes,
        )
        y_vals = get_plot_vals(
            self.values_at_bounds.values,
            "self.values_at_bounds.values",
            warn_if_magnitudes=warn_if_plotting_magnitudes,
        )

        ax.scatter(x_vals, y_vals, label=label, **kwargs)

        return ax

name instance-attribute #

name: str

Name of the timeseries

time_axis instance-attribute #

time_axis: TimeAxis

Time axis of the timeseries

values_at_bounds class-attribute instance-attribute #

values_at_bounds: ValuesAtBounds = field()

Values at the bounds defined by self.time_axis

Must hold values that are the same length as self.time_axis.

__str__ #

__str__() -> str

Get string representation of self

Source code in src/continuous_timeseries/timeseries_discrete.py 
84
85
86
87
88
89
90
91
def __str__(self) -> str:
    """
    Get string representation of self
    """
    return continuous_timeseries.formatting.to_str(
        self,
        [a.name for a in self.__attrs_attrs__],
    )

plot #

plot(
    label: str | None = None,
    ax: Axes | None = None,
    warn_if_plotting_magnitudes: bool = True,
    **kwargs: Any,
) -> Axes

Plot the data

Parameters:

Name Type Description Default
label str | None

Label to use when plotting the data.

If not supplied, we use the self.name.

 None
ax Axes | None

Axes on which to plot.

If not supplied, a set of axes will be created.

 None
warn_if_plotting_magnitudes bool

Should a warning be raised if the units of the values are not considered while plotting?

 True
**kwargs Any

Keyword arguments to pass to ax.scatter.

 {}

Returns:

Type Description
Axes

Axes on which the data was plotted
Source code in src/continuous_timeseries/timeseries_discrete.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def plot(
    self,
    label: str | None = None,
    ax: matplotlib.axes.Axes | None = None,
    warn_if_plotting_magnitudes: bool = True,
    **kwargs: Any,
) -> matplotlib.axes.Axes:
    """
    Plot the data

    Parameters
    ----------
    label
        Label to use when plotting the data.

        If not supplied, we use the `self.name`.

    ax
        Axes on which to plot.

        If not supplied, a set of axes will be created.

    warn_if_plotting_magnitudes
        Should a warning be raised if the units of the values
        are not considered while plotting?

    **kwargs
        Keyword arguments to pass to `ax.scatter`.

    Returns
    -------
    :
        Axes on which the data was plotted
    """
    if label is None:
        label = self.name

    if ax is None:
        try:
            import matplotlib.pyplot as plt
        except ImportError as exc:
            raise MissingOptionalDependencyError(
                "TimeseriesDiscrete.plot", requirement="matplotlib"
            ) from exc

        _, ax = plt.subplots()

    x_vals = get_plot_vals(
        self.time_axis.bounds,
        "self.time_axis.bounds",
        warn_if_magnitudes=warn_if_plotting_magnitudes,
    )
    y_vals = get_plot_vals(
        self.values_at_bounds.values,
        "self.values_at_bounds.values",
        warn_if_magnitudes=warn_if_plotting_magnitudes,
    )

    ax.scatter(x_vals, y_vals, label=label, **kwargs)

    return ax

to_continuous_timeseries #

to_continuous_timeseries(
    interpolation: InterpolationOption,
    warn_if_output_values_at_bounds_could_confuse: bool = True,
    check_change_func: Callable[
        [PINT_NUMPY_ARRAY, PINT_NUMPY_ARRAY], None
    ] = assert_allclose,
) -> TimeseriesContinuous

Convert to TimeseriesContinuous

Parameters:

Name Type Description Default
interpolation InterpolationOption

Interpolation to use for the conversion

 required
warn_if_output_values_at_bounds_could_confuse bool

Should a warning be raised if the interpolation choice means that the value of the output timeseries at point x(n) is not equal to y(n)?

 True
check_change_func Callable[[PINT_NUMPY_ARRAY, PINT_NUMPY_ARRAY], None]

Function to use to check if the value of the output at x(n) is equal to y(n).

If the values are different, this function should raise an AssertionError.

 assert_allclose

Returns:

Type Description
TimeseriesContinuous

Continuous representation of self for the interpolation specified by interpolation
Source code in src/continuous_timeseries/timeseries_discrete.py 
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
def to_continuous_timeseries(
    self,
    interpolation: InterpolationOption,
    warn_if_output_values_at_bounds_could_confuse: bool = True,
    check_change_func: Callable[
        [PINT_NUMPY_ARRAY, PINT_NUMPY_ARRAY], None
    ] = pint.testing.assert_allclose,
) -> TimeseriesContinuous:
    """
    Convert to [`TimeseriesContinuous`][(p)]

    Parameters
    ----------
    interpolation
        Interpolation to use for the conversion

    warn_if_output_values_at_bounds_could_confuse
        Should a warning be raised if the `interpolation` choice
        means that the value of the output timeseries at
        point `x(n)` is not equal to `y(n)`?

    check_change_func
        Function to use to check
        if the value of the output at `x(n)` is equal to `y(n)`.

        If the values are different, this function should raise an `AssertionError`.

    Returns
    -------
    :
        Continuous representation of `self` for the interpolation
        specified by `interpolation`
    """
    x = self.time_axis.bounds
    y = self.values_at_bounds.values
    res = discrete_to_continuous(
        x=x,
        y=y,
        name=self.name,
        interpolation=interpolation,
    )

    if warn_if_output_values_at_bounds_could_confuse:
        try:
            check_change_func(y, res.interpolate(x))

        except AssertionError:
            msg = (
                f"Using the interpolation {interpolation.name} "
                "means that the y-values do not line up exactly with the x-values. "
                "In other words, in the output, "
                "y(x(1)) is not equal to the input y(1), "
                "y(x(2)) is not equal to the input y(2), "
                "y(x(n)) is not equal to the input y(n) etc. "
                "This may cause confusion. "
                "Either ignore this warning, "
                "suppress it "
                "(by passing `warn_if_values_at_bounds_could_confuse=False` "
                "or via Python's `warnings` module settings) "
                "or choose a different interpolation option."
            )
            warnings.warn(msg, InterpolationUpdateChangedValuesAtBoundsWarning)

    return res

values_at_bounds_validator #

values_at_bounds_validator(
    attribute: Attribute[Any], value: ValuesAtBounds
) -> None

Validate the received values

Source code in src/continuous_timeseries/timeseries_discrete.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@values_at_bounds.validator
def values_at_bounds_validator(
    self,
    attribute: attr.Attribute[Any],
    value: ValuesAtBounds,
) -> None:
    """
    Validate the received values
    """
    if value.values.shape != self.time_axis.bounds.shape:
        msg = (
            "`values_at_bounds` must have values "
            "that are the same shape as `self.time_axis.bounds`. "
            f"Received values_at_bounds.values.shape={value.values.shape} "
            f"while {self.time_axis.bounds.shape=}."
        )
        raise AssertionError(msg)

ValuesAtBounds #

Container for values to be used at the bounds of each time window in a timeseries

This is a low-level container. It generally won't be used directly.

Methods:

Name Description
__str__

Get string representation of self
values_validator

Validate the received values

Attributes:

Name Type Description
values PINT_NUMPY_ARRAY

Values
Source code in src/continuous_timeseries/values_at_bounds.py 
 47
 48
 49
 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
 78
 79
 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
@define
class ValuesAtBounds:
    """
    Container for values to be used at the bounds of each time window in a timeseries

    This is a low-level container.
    It generally won't be used directly.
    """

    values: PINT_NUMPY_ARRAY = field()
    """
    Values

    Must be one-dimensional.
    """

    @values.validator
    def values_validator(
        self,
        attribute: attr.Attribute[Any],
        value: PINT_NUMPY_ARRAY,
    ) -> None:
        """
        Validate the received values
        """
        try:
            shape = value.shape
        except AttributeError as exc:
            msg = (
                "`values` must be one-dimensional but "
                "an error was raised while trying to check its shape. "
                f"Received values={value}."
            )
            raise AssertionError(msg) from exc

        if len(shape) != 1:
            msg = (
                "`values` must be one-dimensional. "
                f"Received `values` with shape {shape}"
            )
            raise AssertionError(msg)

    # Let attrs take care of __repr__

    def __str__(self) -> str:
        """
        Get string representation of self
        """
        return continuous_timeseries.formatting.to_str(
            self,
            [a.name for a in self.__attrs_attrs__],
        )

    def _repr_pretty_(
        self,
        p: IPython.lib.pretty.RepresentationPrinter,
        cycle: bool,
        indent: int = 4,
    ) -> None:
        """
        Get IPython pretty representation of self

        Used by IPython notebooks and other tools
        """
        continuous_timeseries.formatting.to_pretty(
            self,
            [a.name for a in self.__attrs_attrs__],
            p=p,
            cycle=cycle,
        )

    def _repr_html_(self) -> str:
        """
        Get html representation of self

        Used by IPython notebooks and other tools
        """
        return continuous_timeseries.formatting.to_html(
            self, [a.name for a in self.__attrs_attrs__], prefix=f"{__name__}."
        )

    def _repr_html_internal_row_(self) -> str:
        """
        Get html representation of self to use as an internal row of another object

        Used to avoid our representations having more information than we'd like.
        """
        return continuous_timeseries.formatting.to_html(
            self,
            [a.name for a in self.__attrs_attrs__],
            include_header=False,
        )

values class-attribute instance-attribute #

values: PINT_NUMPY_ARRAY = field()

Values

Must be one-dimensional.

__str__ #

__str__() -> str

Get string representation of self

Source code in src/continuous_timeseries/values_at_bounds.py 
91
92
93
94
95
96
97
98
def __str__(self) -> str:
    """
    Get string representation of self
    """
    return continuous_timeseries.formatting.to_str(
        self,
        [a.name for a in self.__attrs_attrs__],
    )

values_validator #

values_validator(
    attribute: Attribute[Any], value: PINT_NUMPY_ARRAY
) -> None

Validate the received values

Source code in src/continuous_timeseries/values_at_bounds.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@values.validator
def values_validator(
    self,
    attribute: attr.Attribute[Any],
    value: PINT_NUMPY_ARRAY,
) -> None:
    """
    Validate the received values
    """
    try:
        shape = value.shape
    except AttributeError as exc:
        msg = (
            "`values` must be one-dimensional but "
            "an error was raised while trying to check its shape. "
            f"Received values={value}."
        )
        raise AssertionError(msg) from exc

    if len(shape) != 1:
        msg = (
            "`values` must be one-dimensional. "
            f"Received `values` with shape {shape}"
        )
        raise AssertionError(msg)