Coverage for rivapy / tools / _converter.py: 9%

1# -*- coding: utf-8 -*- 

2 

3 

4from sys import version_info as _version_info 

5from inspect import \ 

6 getmembers as _getmembers, \ 

7 isfunction as _isfunction, \ 

8 ismethod as _ismethod 

9from typing import \ 

10 List as _List, \ 

11 Union as _Union 

12from datetime import datetime as _datetime, date as _date 

13from numpy import empty as _empty, array as _array, ndarray as _ndarray 

14from rivapy import _pyvacon_available 

15 

16if _pyvacon_available: 

17 from pyvacon.pyvacon_swig import ptime as _ptime, CouponDescription as _CouponDescription, vectorCouponDescription as _vectorCouponDescription, BaseObject as _BaseObject, \ 

18 vectorVectorDouble as _vectorVectorDouble, vectorDouble as _vectorDouble 

19 def create_ptime(date: _Union[_date, _datetime, _ptime]) -> _ptime: 

20 """ 

21 Converts dates from given datetime or date into ptime format. Leaves dates given in ptime format unchanged. 

22 

23 Args: 

24 date (date, datetime, ptime): The input datetime/ptime which will be converted to ptime. 

25 

26 Returns: 

27 _ptime: (converted) date 

28 """ 

29 if isinstance(date, _ptime): 

30 return date 

31 elif isinstance(date, _datetime): 

32 return _ptime(date.year, date.month, date.day, date.hour, date.minute, date.second) 

33 else: 

34 return _ptime(date.year, date.month, date.day, 0, 0, 0) 

35 

36 

37 def _convert(x: _Union[_Union[_date, _datetime, _ptime], _List[_Union[_date, _datetime, _ptime]], 

38 _List[_CouponDescription]]) -> _Union[_ptime, _List[_ptime], _vectorCouponDescription]: 

39 """ 

40 Converts variables (mostly from python) to c++ types: 

41 - date, datetime, ptime -> ptime 

42 - List[date, datetime, ptime] -> List[ptime] 

43 - List[CouponDescription] -> vectorCouponDescription 

44 

45 Args: 

46 x: Variable to be converted. 

47 

48 Returns: 

49 __Union[_ptime, _List[_ptime], _vectorCouponDescription]: Converted variable. 

50 """ 

51 if isinstance(x, (_date, _datetime, _ptime)): 

52 return create_ptime(x) 

53 if isinstance(x, list) and len(x) > 0: # Warum hier mit Länge > 0, wohingegen ... 

54 if isinstance(x[0], (_date, _datetime, _ptime)): 

55 return [create_ptime(y) for y in x] 

56 if isinstance(x[0], _CouponDescription): # ... hier keine Mindestlänge verlangt wird? 

57 coupons = _vectorCouponDescription() 

58 for coupon in x: 

59 coupons.append(coupon) 

60 return coupons 

61 return x 

62 

63 

64 def converter(f): 

65 def wrapper(*args, **kwargs): 

66 new_args = [_convert(x) for x in args] 

67 result = f(*new_args, **kwargs) 

68 return result 

69 return wrapper 

70 

71 

72 def _add_converter(cls): 

73 if _version_info >= (3, 0,): 

74 members = _getmembers(cls, _isfunction) 

75 else: 

76 members = _getmembers(cls, _ismethod) 

77 

78 for attr, item in members: 

79 setattr(cls, attr, converter(item)) 

80 for name, method in _getmembers(cls, lambda o: isinstance(o, property)): 

81 setattr(cls, name, property(converter(method.fget), converter(method.fset))) 

82 setattr(cls, '__str__', _get_string_rep) 

83 setattr(cls, 'get_dictionary', _get_dict_repr) 

84 return cls 

85 

86 

87 def _get_dict_repr(obj): 

88 import json 

89 

90 def cleanup_dict(dictionary): 

91 if not isinstance(dictionary, dict): 

92 return dictionary 

93 if len(dictionary) == 1: 

94 for v in dictionary.values(): 

95 return v 

96 new_dict = {} 

97 for item, value in dictionary.items(): 

98 if item != 'cereal_class_version' and item != 'polymorphic_id' and item != 'UID_': 

99 if isinstance(value, dict): 

100 if len(value) == 1: 

101 for v in value.values(): 

102 new_dict[item] = v 

103 else: 

104 new_dict[item] = cleanup_dict(value) 

105 elif isinstance(value, list): 

106 new_dict[item] = [cleanup_dict(vv) for vv in value] 

107 else: 

108 new_dict[item] = value 

109 return new_dict 

110 

111 represent = str(_BaseObject.getString(obj)) + '}' 

112 d = json.loads(represent) 

113 return cleanup_dict(d['value0']) 

114 

115 

116 def _get_string_rep(obj): 

117 dictionary = _get_dict_repr(obj) 

118 return str(dictionary) 

119 

120 

121 def create_datetime(date: _ptime) -> _datetime: 

122 """[summary] 

123 

124 Args: 

125 date (ptime): [description] 

126 

127 Returns: 

128 datetime: [description] 

129 """ 

130 return _datetime(date.year(), date.month(), date.day(), date.hours(), date.minutes(), date.seconds()) 

131 

132 

133 def to_np_matrix(mat: _vectorVectorDouble) -> _ndarray: 

134 """[summary] 

135 

136 Args: 

137 mat (vectorVectorDouble): [description] 

138 

139 Returns: 

140 ndarray: [description] 

141 """ 

142 if len(mat) == 0: 

143 return _empty([0, 0]) 

144 result = _empty([len(mat), len(mat[0])]) 

145 for i in range(len(mat)): 

146 for j in range(len(mat[i])): 

147 result[i][j] = mat[i][j] 

148 return result 

149 

150 

151 def from_np_matrix(mat: _array) -> _vectorVectorDouble: 

152 rows, cols = mat.shape 

153 result = _vectorVectorDouble(rows) 

154 for i in range(rows): 

155 tmp = _vectorDouble(cols) 

156 for j in range(cols): 

157 tmp[j] = mat[i][j] 

158 result[i] = tmp 

159 return result 

160 

161else: 

162 def _add_converter(cls): 

163 raise Exception("pyvacon is not available. Please install pyvacon to use this function.")