Coverage for foxplot / hot_series.py: 94%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# 

4# SPDX-License-Identifier: Apache-2.0 

5 

6"""Series data unpacked from input dictionaries.""" 

7 

8from typing import Any, List 

9 

10import numpy as np 

11 

12from .labeled_series import LabeledSeries 

13from .series import Series 

14 

15 

16class HotSeries(LabeledSeries): 

17 """Indexed time-series in which we can still insert values. 

18 

19 Values are appended in order. When a record is missing this key, the 

20 previous value carries over (the gap is filled lazily on the next update 

21 or at freeze time). 

22 """ 

23 

24 __values: List[Any] 

25 __last_value: Any 

26 __next_index: int 

27 

28 def __init__(self, label: str): 

29 """Initialize a new indexed series. 

30 

31 Args: 

32 label: Label of the series in the input data. 

33 """ 

34 super().__init__(label) 

35 self.__values = [] 

36 self.__last_value = None 

37 self.__next_index = 0 

38 

39 def __len__(self): 

40 """Length of the indexed series.""" 

41 return len(self.__values) 

42 

43 def __repr__(self): 

44 """String representation of the series.""" 

45 return f"Time series with values: {self.__values}" 

46 

47 def _update(self, index: int, value: Any) -> None: 

48 """Update the value at a given time index. 

49 

50 Args: 

51 index: Time index. 

52 value: New value. 

53 """ 

54 next_index = self.__next_index 

55 if index > next_index: 

56 self.__values.extend([self.__last_value] * (index - next_index)) 

57 self.__values.append(value) 

58 self.__last_value = value 

59 self.__next_index = index + 1 

60 

61 def _freeze(self, max_index: int) -> Series: 

62 """Get indexed series as a NumPy array. 

63 

64 Args: 

65 max_index: The output array will range from 0 (first time from the 

66 input) to this maximum index (excluded). 

67 

68 Returns: 

69 Indexed series as a frozen :class:`Series`. 

70 """ 

71 next_index = self.__next_index 

72 if next_index < max_index: 

73 self.__values.extend( 

74 [self.__last_value] * (max_index - next_index) 

75 ) 

76 last_value = self.__last_value 

77 array = ( 

78 np.array(self.__values, dtype=np.float64) 

79 if isinstance(last_value, (int, float)) 

80 else np.array(self.__values) 

81 ) 

82 return Series(label=self._label, values=array, times=None)