1 # Copyright (c) 2018 Cisco and/or its affiliates.
2 # Licensed under the Apache License, Version 2.0 (the "License");
3 # you may not use this file except in compliance with the License.
4 # You may obtain a copy of the License at:
6 # http://www.apache.org/licenses/LICENSE-2.0
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS,
10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11 # See the License for the specific language governing permissions and
12 # limitations under the License.
14 """General purpose utilities.
17 import multiprocessing
24 from os import walk, makedirs, environ
25 from os.path import join, isdir
26 from shutil import move, Error
28 from datetime import datetime
30 from errors import PresentationError
31 from jumpavg.BitCountingClassifier import BitCountingClassifier
35 """Calculate mean value from the items.
37 :param items: Mean value is calculated from these items.
43 return float(sum(items)) / len(items)
47 """Calculate stdev from the items.
49 :param items: Stdev is calculated from these items.
56 variance = [(x - avg) ** 2 for x in items]
57 stddev = sqrt(mean(variance))
61 def relative_change(nr1, nr2):
62 """Compute relative change of two values.
64 :param nr1: The first number.
65 :param nr2: The second number.
68 :returns: Relative change of nr1.
72 return float(((nr2 - nr1) / nr1) * 100)
75 def get_files(path, extension=None, full_path=True):
76 """Generates the list of files to process.
78 :param path: Path to files.
79 :param extension: Extension of files to process. If it is the empty string,
80 all files will be processed.
81 :param full_path: If True, the files with full path are generated.
85 :returns: List of files to process.
90 for root, _, files in walk(path):
91 for filename in files:
93 if filename.endswith(extension):
95 file_list.append(join(root, filename))
97 file_list.append(filename)
99 file_list.append(join(root, filename))
104 def get_rst_title_char(level):
105 """Return character used for the given title level in rst files.
107 :param level: Level of the title.
109 :returns: Character used for the given title level in rst files.
112 chars = ('=', '-', '`', "'", '.', '~', '*', '+', '^')
113 if level < len(chars):
119 def execute_command(cmd):
120 """Execute the command in a subprocess and log the stdout and stderr.
122 :param cmd: Command to execute.
124 :returns: Return code of the executed command, stdout and stderr.
125 :rtype: tuple(int, str, str)
129 proc = subprocess.Popen(
131 stdout=subprocess.PIPE,
132 stderr=subprocess.PIPE,
136 stdout, stderr = proc.communicate()
143 if proc.returncode != 0:
144 logging.error(" Command execution failed.")
145 return proc.returncode, stdout, stderr
148 def get_last_successful_build_number(jenkins_url, job_name):
149 """Get the number of the last successful build of the given job.
151 :param jenkins_url: Jenkins URL.
152 :param job_name: Job name.
153 :type jenkins_url: str
155 :returns: The build number as a string.
159 url = "{}/{}/lastSuccessfulBuild/buildNumber".format(jenkins_url, job_name)
160 cmd = "wget -qO- {url}".format(url=url)
162 return execute_command(cmd)
165 def get_last_completed_build_number(jenkins_url, job_name):
166 """Get the number of the last completed build of the given job.
168 :param jenkins_url: Jenkins URL.
169 :param job_name: Job name.
170 :type jenkins_url: str
172 :returns: The build number as a string.
176 url = "{}/{}/lastCompletedBuild/buildNumber".format(jenkins_url, job_name)
177 cmd = "wget -qO- {url}".format(url=url)
179 return execute_command(cmd)
182 def get_build_timestamp(jenkins_url, job_name, build_nr):
183 """Get the timestamp of the build of the given job.
185 :param jenkins_url: Jenkins URL.
186 :param job_name: Job name.
187 :param build_nr: Build number.
188 :type jenkins_url: str
191 :returns: The timestamp.
192 :rtype: datetime.datetime
195 url = "{jenkins_url}/{job_name}/{build_nr}".format(jenkins_url=jenkins_url,
198 cmd = "wget -qO- {url}".format(url=url)
200 timestamp = execute_command(cmd)
202 return datetime.fromtimestamp(timestamp/1000)
205 def archive_input_data(spec):
206 """Archive the report.
208 :param spec: Specification read from the specification file.
209 :type spec: Specification
210 :raises PresentationError: If it is not possible to archive the input data.
213 logging.info(" Archiving the input data files ...")
215 extension = spec.input["file-format"]
216 data_files = get_files(spec.environment["paths"]["DIR[WORKING,DATA]"],
218 dst = spec.environment["paths"]["DIR[STATIC,ARCH]"]
219 logging.info(" Destination: {0}".format(dst))
225 for data_file in data_files:
226 logging.info(" Moving the file: {0} ...".format(data_file))
229 except (Error, OSError) as err:
230 raise PresentationError("Not possible to archive the input data.",
233 logging.info(" Done.")
236 def classify_anomalies(data):
237 """Process the data and return anomalies and trending values.
239 Gather data into groups with average as trend value.
240 Decorate values within groups to be normal,
241 the first value of changed average as a regression, or a progression.
243 :param data: Full data set with unavailable samples replaced by nan.
244 :type data: OrderedDict
245 :returns: Classification and trend values
246 :rtype: 2-tuple, list of strings and list of floats
248 # Nan mean something went wrong.
249 # Use 0.0 to cause that being reported as a severe regression.
250 bare_data = [0.0 if np.isnan(sample.avg) else sample
251 for _, sample in data.iteritems()]
252 # TODO: Put analogous iterator into jumpavg library.
253 groups = BitCountingClassifier().classify(bare_data)
254 groups.reverse() # Just to use .pop() for FIFO.
260 for _, sample in data.iteritems():
261 if np.isnan(sample.avg):
262 classification.append("outlier")
263 avgs.append(sample.avg)
265 if values_left < 1 or active_group is None:
267 while values_left < 1: # Ignore empty groups (should not happen).
268 active_group = groups.pop()
269 values_left = len(active_group.values)
270 avg = active_group.metadata.avg
271 classification.append(active_group.metadata.classification)
275 classification.append("normal")
278 return classification, avgs
281 def convert_csv_to_pretty_txt(csv_file, txt_file):
282 """Convert the given csv table to pretty text table.
284 :param csv_file: The path to the input csv file.
285 :param txt_file: The path to the output pretty text file.
291 with open(csv_file, 'rb') as csv_file:
292 csv_content = csv.reader(csv_file, delimiter=',', quotechar='"')
293 for row in csv_content:
294 if txt_table is None:
295 txt_table = prettytable.PrettyTable(row)
297 txt_table.add_row(row)
298 txt_table.align["Test case"] = "l"
300 with open(txt_file, "w") as txt_file:
301 txt_file.write(str(txt_table))
304 class Worker(multiprocessing.Process):
305 """Worker class used to process tasks in separate parallel processes.
308 def __init__(self, work_queue, data_queue, func):
311 :param work_queue: Queue with items to process.
312 :param data_queue: Shared memory between processes. Queue which keeps
313 the result data. This data is then read by the main process and used
314 in further processing.
315 :param func: Function which is executed by the worker.
316 :type work_queue: multiprocessing.JoinableQueue
317 :type data_queue: multiprocessing.Manager().Queue()
318 :type func: Callable object
320 super(Worker, self).__init__()
321 self._work_queue = work_queue
322 self._data_queue = data_queue
326 """Method representing the process's activity.
331 self.process(self._work_queue.get())
333 self._work_queue.task_done()
335 def process(self, item_to_process):
336 """Method executed by the runner.
338 :param item_to_process: Data to be processed by the function.
339 :type item_to_process: tuple
341 self._func(self.pid, self._data_queue, *item_to_process)