Move input path inspection + optional conversion to loading submodule

tracetools_analysis/test/tracetools_analysis/test_process_command.py

@@ -18,7 +18,7 @@ import shutil
 import tempfile
 import unittest
 
-from tracetools_analysis.process import inspect_input_path
+from tracetools_analysis.loading import inspect_input_path
 
 
 class TestProcessCommand(unittest.TestCase):

tracetools_analysis/tracetools_analysis/loading/__init__.py

@@ -16,17 +16,120 @@
 
 import os
 import pickle
+import sys
 from typing import Dict
 from typing import List
+from typing import Tuple
+
+from tracetools_read.trace import is_trace_directory
+
+from ..convert import convert
+from ..convert import DEFAULT_CONVERT_FILE_NAME
 
 
-def load_file(file_path: str) -> List[Dict]:
+def inspect_input_path(
+    input_path: str,
+    force_conversion: bool = False,
+) -> Tuple[str, bool]:
+    """
+    Check input path for a converted file or a trace directory.
+
+    If the input path is a file, it uses it as a converted file.
+    If the input path is a directory, it checks if there is a "converted" file directly inside it,
+    otherwise it tries to import the path as a trace directory.
+    If `force_conversion` is set to `True`, even if a converted file is found, it will ask to
+    re-create it.
+
+    :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
+    :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
+    """
+    input_path = os.path.expanduser(input_path)
+    converted_file_path = None
+    # Check if not a file
+    if not os.path.isfile(input_path):
+        input_directory = input_path
+        # Might be a (trace) directory
+        # Check if there is a converted file under the given directory
+        prospective_converted_file = os.path.join(input_directory, DEFAULT_CONVERT_FILE_NAME)
+        if os.path.isfile(prospective_converted_file):
+            # 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}')
+                return prospective_converted_file, True
+            else:
+                print(f'found converted file: {prospective_converted_file}')
+                return prospective_converted_file, False
+        else:
+            # Check if it is a trace directory
+            # Result could be unexpected because it will look for trace directories recursively
+            # (e.g. '/' is a valid trace directory if there is at least one trace anywhere)
+            if is_trace_directory(input_directory):
+                # Convert trace directory first to create converted file
+                return prospective_converted_file, True
+            else:
+                # We cannot do anything
+                print(
+                    f'cannot find either a trace directory or a converted file: {input_directory}',
+                    file=sys.stderr)
+                return None, None
+    else:
+        converted_file_path = input_path
+        if force_conversion:
+            # It's a file, but re-create it anyway
+            print(f'found converted file but will re-create it: {converted_file_path}')
+            return converted_file_path, True
+        else:
+            # Simplest use-case: given path is an existing converted file
+            # No need to print anything
+            return converted_file_path, False
+
+
+def convert_if_needed(
+    input_path: str,
+    force_conversion: bool = False,
+) -> str:
+    """
+    Inspect input path and convert trace 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
+    """
+    converted_file_path, create_converted_file = inspect_input_path(input_path, force_conversion)
+
+    if converted_file_path is None:
+        return None
+
+    # Convert trace directory to file if necessary
+    if create_converted_file:
+        input_directory = os.path.dirname(converted_file_path)
+        input_file_name = os.path.basename(converted_file_path)
+        convert(input_directory, input_file_name)
+
+    return converted_file_path
+
+
+def load_file(
+    input_path: str,
+    do_convert_if_needed: bool = False,
+    force_conversion: bool = False,
+) -> List[Dict]:
     """
     Load file containing converted trace events.
 
-    :param file_path: the path to the converted file to load
+    :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
     :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)
+    else:
+        file_path = input_path
+
     events = []
     with open(os.path.expanduser(file_path), 'rb') as f:
         p = pickle.Unpickler(f)

tracetools_analysis/tracetools_analysis/process.py

@@ -16,18 +16,12 @@
 """Entrypoint/script to process events from a converted file to build a ROS model."""
 
 import argparse
-import os
-import sys
 import time
 from typing import Optional
-from typing import Tuple
 
-from tracetools_analysis.convert import convert
-from tracetools_analysis.convert import DEFAULT_CONVERT_FILE_NAME
 from tracetools_analysis.loading import load_file
 from tracetools_analysis.processor import Processor
 from tracetools_analysis.processor.ros2 import Ros2Handler
-from tracetools_read.trace import is_trace_directory
 
 from . import time_diff_to_str
 
@@ -54,67 +48,6 @@ def parse_args():
     return parser.parse_args()
 
 
-def inspect_input_path(
-    input_path: str,
-    force_conversion: bool = False,
-) -> Tuple[str, bool]:
-    """
-    Check input path for a converted file or a trace directory.
-
-    If the input path is a file, it uses it as a converted file.
-    If the input path is a directory, it checks if there is a "converted" file directly inside it,
-    otherwise it tries to import the path as a trace directory.
-    If `force_conversion` is set to `True`, even if a converted file is found, it will ask to
-    re-create it.
-
-    :param input_path: the path to a converted file or trace directory
-    :param force_conversion: whether to re-creating converted file even if it is found
-    :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
-    """
-    input_path = os.path.expanduser(input_path)
-    converted_file_path = None
-    # Check if not a file
-    if not os.path.isfile(input_path):
-        input_directory = input_path
-        # Might be a (trace) directory
-        # Check if there is a converted file under the given directory
-        prospective_converted_file = os.path.join(input_directory, DEFAULT_CONVERT_FILE_NAME)
-        if os.path.isfile(prospective_converted_file):
-            # 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}')
-                return prospective_converted_file, True
-            else:
-                print(f'found converted file: {prospective_converted_file}')
-                return prospective_converted_file, False
-        else:
-            # Check if it is a trace directory
-            # Result could be unexpected because it will look for trace directories recursively
-            # (e.g. '/' is a valid trace directory if there is at least one trace anywhere)
-            if is_trace_directory(input_directory):
-                # Convert trace directory first to create converted file
-                return prospective_converted_file, True
-            else:
-                # We cannot do anything
-                print(
-                    f'cannot find either a trace directory or a converted file: {input_directory}',
-                    file=sys.stderr)
-                return None, None
-    else:
-        converted_file_path = input_path
-        if force_conversion:
-            # It's a file, but re-create it anyway
-            print(f'found converted file but will re-create it: {converted_file_path}')
-            return converted_file_path, True
-        else:
-            # Simplest use-case: given path is an existing converted file
-            print(f'found converted file: {converted_file_path}')
-            return converted_file_path, False
-
-
 def process(
     input_path: str,
     force_conversion: bool = False,
@@ -127,20 +60,9 @@ def process(
     :param force_conversion: whether to re-creating converted file even if it is found
     :param hide_results: whether to hide results and not print them
     """
-    converted_file_path, create_converted_file = inspect_input_path(input_path, force_conversion)
-
-    if converted_file_path is None:
-        return 1
-
-    # Convert trace directory to file if necessary
-    if create_converted_file:
-        input_directory = os.path.dirname(converted_file_path)
-        input_file_name = os.path.basename(converted_file_path)
-        convert(input_directory, input_file_name)
-
     start_time = time.time()
 
-    events = load_file(converted_file_path)
+    events = load_file(input_path, do_convert_if_needed=True, force_conversion=force_conversion)
     processor = Processor(Ros2Handler())
     processor.process(events)
 

tracetools_analysis/tracetools_analysis/scripts/memory_usage.py

@@ -25,11 +25,11 @@ from tracetools_analysis.utils.ros2 import Ros2DataModelUtil
 
 def main():
     if len(sys.argv) < 2:
-        print('Syntax: <converted tracefile>')
+        print('Syntax: [trace directory | converted tracefile]')
         sys.exit(1)
-    file_path = sys.argv[1]
+    input_path = sys.argv[1]
 
-    events = load_file(file_path)
+    events = load_file(input_path, do_convert_if_needed=True)
     ust_memory_handler = UserspaceMemoryUsageHandler()
     kernel_memory_handler = KernelMemoryUsageHandler()
     ros2_handler = Ros2Handler()