niklas/tracetools_analysis
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch '34-add-quiet-option-to-not-display-processing-conversion-output' into 'master'

Browse source 
Add 'quiet' option to not display processing/conversion output

Closes #34

See merge request micro-ROS/ros_tracing/tracetools_analysis!62
This commit is contained in:
Christophe Bedard 2020-05-23 14:23:31 +00:00
commit 9a01789769
4 changed files with 74 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

31
tracetools_analysis/test/tracetools_analysis/test_process_command.py → tracetools_analysis/test/tracetools_analysis/test_loading.py
View file

@ -1,5 +1,6 @@
#!/usr/bin/env python3
# Copyright 2019 Apex.AI, Inc.
# Copyright 2020 Christophe Bedard
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
@ -13,6 +14,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import contextlib
from io import StringIO
import os
import shutil
import tempfile
@ -21,7 +24,7 @@ import unittest
from tracetools_analysis.loading import inspect_input_path
class TestProcessCommand(unittest.TestCase):
class TestLoading(unittest.TestCase):
def __init__(self, *args) -> None:
super().__init__(
@ -60,31 +63,43 @@ class TestProcessCommand(unittest.TestCase):
def tearDown(self):
shutil.rmtree(self.test_dir_path)
def test_inspect_input_path(self) -> None:
def test_inspect_input_path(
self,
quiet: bool = False,
) -> None:
# Should find converted file under directory
file_path, create_file = inspect_input_path(self.with_converted_file_dir, False)
file_path, create_file = inspect_input_path(self.with_converted_file_dir, False, quiet)
self.assertEqual(self.converted_file_path, file_path)
self.assertFalse(create_file)
# Should find it but set it to be re-created
file_path, create_file = inspect_input_path(self.with_converted_file_dir, True)
file_path, create_file = inspect_input_path(self.with_converted_file_dir, True, quiet)
self.assertEqual(self.converted_file_path, file_path)
self.assertTrue(create_file)
# Should fail to find converted file under directory
file_path, create_file = inspect_input_path(self.without_converted_file_dir, False)
file_path, create_file = inspect_input_path(self.without_converted_file_dir, False, quiet)
self.assertIsNone(file_path)
self.assertFalse(create_file)
file_path, create_file = inspect_input_path(self.without_converted_file_dir, True)
file_path, create_file = inspect_input_path(self.without_converted_file_dir, True, quiet)
self.assertIsNone(file_path)
self.assertFalse(create_file)
# Should accept any file path if it exists
file_path, create_file = inspect_input_path(self.random_file_path, False)
file_path, create_file = inspect_input_path(self.random_file_path, False, quiet)
self.assertEqual(self.random_file_path, file_path)
self.assertFalse(create_file)
# Should set it to be re-created
file_path, create_file = inspect_input_path(self.random_file_path, True)
file_path, create_file = inspect_input_path(self.random_file_path, True, quiet)
self.assertEqual(self.random_file_path, file_path)
self.assertTrue(create_file)
# TODO try with a trace directory
def test_inspect_input_path_quiet(self) -> None:
temp_stdout = StringIO()
with contextlib.redirect_stdout(temp_stdout):
# Just run the other test again
self.test_inspect_input_path(quiet=True)
# Shouldn't be any output
output = temp_stdout.getvalue()
self.assertEqual(0, len(output), f'was not quiet: "{output}"')

17
tracetools_analysis/test/tracetools_analysis/test_processor.py
View file

@ -12,6 +12,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import contextlib
from io import StringIO
from typing import Dict
from typing import Set
import unittest
@ -172,6 +174,21 @@ class TestProcessor(unittest.TestCase):
result = processor.get_handler_by_type(StubHandler1)
self.assertTrue(result is handler1)
def test_processor_quiet(self) -> None:
handler1 = StubHandler1()
mock_event = {
'_name': 'myeventname',
'_timestamp': 0,
'cpu_id': 0,
}
temp_stdout = StringIO()
with contextlib.redirect_stdout(temp_stdout):
processor = Processor(handler1, quiet=True)
processor.process([mock_event])
# Shouldn't be any output
output = temp_stdout.getvalue()
self.assertEqual(0, len(output), f'Processor was not quiet: "{output}"')
if __name__ == '__main__':
unittest.main()

21
tracetools_analysis/tracetools_analysis/loading/__init__.py
View file

@ -31,6 +31,7 @@ from ..convert import DEFAULT_CONVERT_FILE_NAME
def inspect_input_path(
input_path: str,
force_conversion: bool = False,
quiet: bool = False,
) -> Tuple[Optional[str], bool]:
"""
Check input path for a converted file or a trace directory.
@ -43,6 +44,7 @@ def inspect_input_path(
:param input_path: the path to a converted file or trace directory
:param force_conversion: whether to re-create converted file even if it is found
:param quiet: whether to not print any normal output
:return:
the path to a converted file (or `None` if could not find),
`True` if the given converted file should be (re-)created, `False` otherwise
@ -59,9 +61,13 @@ def inspect_input_path(
# Use that as the converted input file
converted_file_path = prospective_converted_file
if force_conversion:
print(f'found converted file but will re-create it: {prospective_converted_file}')
if not quiet:
print(
f'found converted file but will re-create it: {prospective_converted_file}'
)
return prospective_converted_file, True
else:
if not quiet:
print(f'found converted file: {prospective_converted_file}')
return prospective_converted_file, False
else:
@ -81,6 +87,7 @@ def inspect_input_path(
converted_file_path = input_path
if force_conversion:
# It's a file, but re-create it anyway
if not quiet:
print(f'found converted file but will re-create it: {converted_file_path}')
return converted_file_path, True
else:
@ -92,15 +99,21 @@ def inspect_input_path(
def convert_if_needed(
input_path: str,
force_conversion: bool = False,
quiet: bool = False,
) -> Optional[str]:
"""
Inspect input path and convert trace directory to file if necessary.
:param input_path: the path to a converted file or trace directory
:param force_conversion: whether to re-create converted file even if it is found
:param quiet: whether to not print any output
:return: the path to the converted file, or `None` if it failed
"""
converted_file_path, create_converted_file = inspect_input_path(input_path, force_conversion)
converted_file_path, create_converted_file = inspect_input_path(
input_path,
force_conversion,
quiet,
)
if converted_file_path is None:
return None
@ -118,6 +131,7 @@ def load_file(
input_path: str,
do_convert_if_needed: bool = True,
force_conversion: bool = False,
quiet: bool = False,
) -> List[Dict]:
"""
Load file containing converted trace events.
@ -125,10 +139,11 @@ def load_file(
:param input_path: the path to a converted file or trace directory
:param do_convert_if_needed: whether to create the converted file if needed (else, let it fail)
:param force_conversion: whether to re-create converted file even if it is found
:param quiet: whether to not print any output
:return: the list of events read from the file
"""
if do_convert_if_needed or force_conversion:
file_path = convert_if_needed(input_path, force_conversion)
file_path = convert_if_needed(input_path, force_conversion, quiet)
else:
file_path = input_path

12
tracetools_analysis/tracetools_analysis/processor/__init__.py
View file

@ -289,12 +289,14 @@ class Processor():
def __init__(
self,
*handlers: EventHandler,
quiet: bool = False,
**kwargs,
) -> None:
"""
Create a Processor.
:param handlers: the `EventHandler`s to use for processing
:param quiet: whether to not print any output, like progress information
:param kwargs: the parameters to pass on to new handlers
"""
self._initial_handlers = list(handlers)
@ -303,9 +305,10 @@ class Processor():
self._expanded_handlers = self._expand_dependencies(*handlers, **kwargs)
self._handler_multimap = self._get_handler_maps(self._expanded_handlers)
self._register_with_handlers(self._expanded_handlers)
self._quiet = quiet
self._progress_display = ProcessingProgressDisplay(
[type(handler).__name__ for handler in self._expanded_handlers],
)
) if not self._quiet else None
self._processing_done = False
@staticmethod
@ -401,12 +404,17 @@ class Processor():
self._check_required_events(events)
if not self._processing_done:
# Split into two versions so that performance is optimal
if self._progress_display is None:
for event in events:
self._process_event(event)
else:
self._progress_display.set_work_total(len(events))
for event in events:
self._process_event(event)
self._progress_display.did_work()
self._processing_done = True
self._progress_display.done(erase=erase_progress)
self._processing_done = True
def _process_event(self, event: DictEvent) -> None:
"""Process a single event."""