Skip to content

Commit

Permalink
Skip empty files by default when searching for a p1log file. (#321)
Browse files Browse the repository at this point in the history
# Changes
- Skip empty files by default in Python `find_log_file()` function
  • Loading branch information
adamshapiro0 authored Jun 11, 2024
2 parents a086a52 + aff602e commit e83c6ed
Showing 1 changed file with 25 additions and 6 deletions.
31 changes: 25 additions & 6 deletions python/fusion_engine_client/utils/log.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@
DEFAULT_LOG_BASE_DIR = os.path.expanduser("~/point_one/logs")


def find_log_by_pattern(pattern, log_base_dir=DEFAULT_LOG_BASE_DIR, allow_multiple=False,
def find_log_by_pattern(pattern, log_base_dir=DEFAULT_LOG_BASE_DIR, allow_multiple=False, skip_empty_files=True,
log_test_filenames=(MANIFEST_FILE_NAME,), return_test_file=False):
"""!
@brief Perform a pattern match to locate a log directory containing the specified files.
Expand Down Expand Up @@ -71,6 +71,7 @@ def find_log_by_pattern(pattern, log_base_dir=DEFAULT_LOG_BASE_DIR, allow_multip
@param log_base_dir The base directory to be searched.
@param allow_multiple If `True`, return multiple matching logs if present, and return an empty list if no logs match
the pattern. Otherwise, raise an exception if either multiple or zero logs are found.
@param skip_empty_files If `True`, ignore files that exist but are 0 bytes.
@param log_test_filenames A list of input files to locate within the log directory. If _one_ of the listed files is
found, the directory is considered a valid log. If `None` or an empty list, skip the test file requirement.
@param return_test_file If `True`, return the path to the located test file.
Expand Down Expand Up @@ -118,9 +119,17 @@ def find_log_by_pattern(pattern, log_base_dir=DEFAULT_LOG_BASE_DIR, allow_multip
# - `path/to/foo_/abcd1234` <-- NO MATCH
if fnmatch.fnmatch(path, match_pattern) and fnmatch.fnmatch(dirname, last_element):
test_file = None
for f in log_test_filenames:
for i, f in enumerate(log_test_filenames):
test_file_path = os.path.join(path, f)
if os.path.exists(test_file_path):
# If the file exists but is empty, see if we should skip it. Otherwise, print a warning.
if os.path.getsize(test_file_path) == 0:
if skip_empty_files:
_logger.warning(f"File '{test_file_path}' is 0 bytes. Skipping.")
continue
else:
_logger.error(f"Selected file '{test_file_path}' is 0 bytes.")

test_file = test_file_path
break

Expand Down Expand Up @@ -154,7 +163,8 @@ def find_log_by_pattern(pattern, log_base_dir=DEFAULT_LOG_BASE_DIR, allow_multip


def find_log_file(input_path, candidate_files=None, return_output_dir=False, return_log_id=False,
log_base_dir=DEFAULT_LOG_BASE_DIR, check_exact_match=True, check_pattern_match=True):
log_base_dir=DEFAULT_LOG_BASE_DIR, check_exact_match=True, check_pattern_match=True,
skip_empty_files=True):
"""!
@brief Locate a log directory containing the specified file(s).
Expand Down Expand Up @@ -190,15 +200,16 @@ def find_log_file(input_path, candidate_files=None, return_output_dir=False, ret
only perform a pattern search.
@param check_pattern_match If `True` and `input_path` does not refer to a log file or directory, perform a pattern
match using `input_path` as the pattern.
@param skip_empty_files If `True`, ignore files that exist but are 0 bytes.
@return The path to the located file or a tuple containing:
- The path to the located file.
- The path to the located output directory. Only provided if `return_output_dir` is `True`.
- The log ID string, or `None` if the requested file is not part of a FusionEngine log. Only provided if
`return_log_id` is `True`.
"""
def _get_log_id(file_path):
parent_dir = os.path.dirname(os.path.abspath(input_path))
def _get_log_id(path):
parent_dir = os.path.dirname(os.path.abspath(path))
return os.path.basename(parent_dir)

# Check if the input path is a file. If so, return it and set the output directory to its parent directory.
Expand All @@ -223,12 +234,20 @@ def _get_log_id(file_path):
log_id = None

def _search_directory(dir_path):
for f in candidate_files:
for i, f in enumerate(candidate_files):
if f is None:
continue

test_path = os.path.join(dir_path, f)
if os.path.exists(test_path):
# If the file exists but is empty, see if we should skip it. Otherwise, print a warning.
if os.path.getsize(test_path) == 0:
if skip_empty_files:
_logger.warning(f"File '{test_path}' is 0 bytes. Skipping.")
continue
else:
_logger.error(f"Selected file '{test_path}' is 0 bytes.")

return test_path, dir_path, _get_log_id(test_path)
return None, None, None

Expand Down

0 comments on commit e83c6ed

Please sign in to comment.