Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/nfstream/utils.py: 46%

1""" 

2------------------------------------------------------------------------------------------------------------------------ 

3utils.py 

4Copyright (C) 2019-22 - NFStream Developers 

5This file is part of NFStream, a Flexible Network Data Analysis Framework (https://www.nfstream.org/). 

6NFStream is free software: you can redistribute it and/or modify it under the terms of the GNU Lesser General Public 

7License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later 

8version. 

9NFStream is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty 

10of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details. 

11You should have received a copy of the GNU Lesser General Public License along with NFStream. 

12If not, see <http://www.gnu.org/licenses/>. 

13------------------------------------------------------------------------------------------------------------------------ 

14""" 

15 

16import json 

17import platform 

18import psutil 

19from threading import Timer 

20from collections import namedtuple 

21from enum import Enum, IntEnum 

22 

23 

24class NFEvent(Enum): 

25 FLOW = -1 

26 ERROR = -2 

27 SOCKET_CREATE = -3 

28 SOCKET_REMOVE = -4 

29 ALL_AFFINITY_SET = -5 

30 

31 

32class NFMode(IntEnum): 

33 SINGLE_FILE = 0 

34 INTERFACE = 1 

35 MULTIPLE_FILES = 2 

36 

37 

38InternalError = namedtuple("InternalError", ["id", "message"]) 

39 

40InternalState = namedtuple("InternalState", ["id"]) 

41 

42 

43def validate_flows_per_file(n): 

44 """Simple parameter validator""" 

45 if not isinstance(n, int) or isinstance(n, int) and n < 0: 

46 raise ValueError("Please specify a valid flows_per_file parameter (>= 0).") 

47 

48 

49def validate_rotate_files(n): 

50 """Simple parameter validator""" 

51 if not isinstance(n, int) or isinstance(n, int) and n < 0: 

52 raise ValueError("Please specify a valid rotate_files parameter (>= 0).") 

53 

54 

55def create_csv_file_path(path, source): 

56 """File path creator""" 

57 if path is None: 

58 if type(source) == list: 

59 return str(source[0]) + ".csv" 

60 return str(source) + ".csv" 

61 return path 

62 

63 

64def csv_converter(values): 

65 """Convert non numeric values to string using their __str__ method""" 

66 for idx, value in enumerate(values): 

67 if not isinstance(value, float) and not isinstance(value, int): 

68 if value is None: 

69 values[idx] = "" 

70 else: 

71 values[idx] = str(values[idx]) 

72 

73 

74def open_file(path, chunked, chunk_idx, rotate_files): 

75 """File opener taking chunk mode into consideration""" 

76 if not chunked: 

77 return open(path, "w", newline='', encoding="utf-8") 

78 else: 

79 if rotate_files: 

80 return open( 

81 path.replace("csv", "{}.csv".format(chunk_idx % rotate_files)), "w", newline='', encoding="utf-8" 

82 ) 

83 return open(path.replace("csv", "{}.csv".format(chunk_idx)), "w", newline='', encoding="utf-8") 

84 

85 

86def update_performances(performances, is_linux, flows_count): 

87 """Update performance report and check platform for consistency""" 

88 drops = 0 

89 processed = 0 

90 ignored = 0 

91 load = [] 

92 for meter in performances: 

93 if is_linux: 

94 drops += meter[0].value 

95 ignored += meter[2].value 

96 else: 

97 drops = max(meter[0].value, drops) 

98 ignored = max(meter[2].value, ignored) 

99 processed += meter[1].value 

100 load.append(meter[1].value) 

101 print( 

102 json.dumps( 

103 { 

104 "flows_expired": flows_count.value, 

105 "packets_processed": processed, 

106 "packets_ignored": ignored, 

107 "packets_dropped_filtered_by_kernel": drops, 

108 "meters_packets_processing_balance": load, 

109 } 

110 ) 

111 ) 

112 

113 

114class RepeatedTimer(object): 

115 """Repeated timer thread""" 

116 

117 def __init__(self, interval, function, *args, **kwargs): 

118 self._timer = None 

119 self.interval = interval 

120 self.function = function 

121 self.args = args 

122 self.kwargs = kwargs 

123 self.is_running = False 

124 self.start() 

125 

126 def _run(self): 

127 self.is_running = False 

128 self.start() 

129 self.function(*self.args, **self.kwargs) 

130 

131 def start(self): 

132 if not self.is_running: 

133 self._timer = Timer(self.interval, self._run) 

134 self._timer.start() 

135 self.is_running = True 

136 

137 def stop(self): 

138 self._timer.cancel() 

139 self.is_running = False 

140 

141 

142def chunks_of_list(lst, n): 

143 """create list of chunks of size n from a list""" 

144 for i in range(0, len(lst), n): 

145 yield lst[i : i + n] 

146 

147 

148def set_affinity(idx): 

149 """CPU affinity setter""" 

150 if platform.system() == "Linux": 

151 c_cpus = psutil.Process().cpu_affinity() 

152 temp = list(chunks_of_list(c_cpus, 2)) 

153 x = len(temp) 

154 try: 

155 psutil.Process().cpu_affinity(list(temp[idx % x])) 

156 except OSError as err: 

157 print("WARNING: failed to set CPU affinity ({err})".format(err)) 

158 

159 

160def available_cpus_count(): 

161 if platform.system() == "Linux": 

162 return len(psutil.Process().cpu_affinity()) 

163 return psutil.cpu_count(logical=True)