Avoid too much lines in utils.py

Author: pwolnows

tools/accuracy_checker/accuracy_checker/annotation_converters/convert.py

@@ -16,9 +16,10 @@
 import warnings
 import platform
 import sys
-
+import os
 import copy
 import json
+import tempfile
 from pathlib import Path
 import pickle  # nosec B403  # disable import-pickle check
 from argparse import ArgumentParser
@@ -34,8 +35,7 @@
 )
 from ..data_readers import KaldiFrameIdentifier, KaldiMatrixIdentifier
 from ..utils import (
-    get_path, OrderedSet, cast_to_bool, is_relative_to, start_telemetry, send_telemetry_event, 
-    end_telemetry, AtomicWriteFileHandle
+    get_path, OrderedSet, cast_to_bool, is_relative_to, start_telemetry, send_telemetry_event, end_telemetry
 )
 from ..data_analyzer import BaseDataAnalyzer
 from .format_converter import BaseFormatConverter
@@ -410,3 +410,38 @@ def analyze_dataset(annotations, metadata):
     else:
         metadata = {'data_analysis': data_analysis}
     return metadata
+
+class AtomicWriteFileHandle:
+    """Ensure the file is written once in case of multi processes or threads."""
+
+    def __init__(self, file_path, open_mode):
+        self.target_path = file_path
+        self.mode = open_mode
+
+        self.temp_fd, self.temp_path = tempfile.mkstemp(dir=os.path.dirname(file_path))
+        self.temp_file = os.fdopen(self.temp_fd, open_mode)
+
+    def write(self, data):
+        self.temp_file.write(data)
+
+    def writelines(self, lines):
+        self.temp_file.writelines(lines)
+
+    def close(self):
+        if not self.temp_file.closed:
+            self.temp_file.close()
+            if not os.path.exists(self.target_path):
+                os.rename(self.temp_path, self.target_path)
+            else:
+                os.remove(self.temp_path)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
+
+    # Mimic other file object methods as needed
+    def __getattr__(self, item):
+        """Delegate attribute access to the underlying temporary file object."""
+        return getattr(self.temp_file, item)

tools/accuracy_checker/accuracy_checker/utils.py

@@ -24,7 +24,6 @@
 import sys
 import zlib
 import re
-import tempfile
 from enum import Enum
 
 from pathlib import Path
@@ -982,38 +981,3 @@ def ov_new_api_available():
         return True
     except ImportError:
         return False
-
-
-class AtomicWriteFileHandle:
-    """Ensure the file is written once in case of multi processes or threads."""
-
-    def __init__(self, file_path, open_mode):
-        self.target_path = file_path
-        self.mode = open_mode
-        self.temp_fd, self.temp_path = tempfile.mkstemp(dir=os.path.dirname(file_path))
-        self.temp_file = os.fdopen(self.temp_fd, open_mode)
-
-    def write(self, data):
-        self.temp_file.write(data)
-
-    def writelines(self, lines):
-        self.temp_file.writelines(lines)
-
-    def close(self):
-        if not self.temp_file.closed:
-            self.temp_file.close()
-            if not os.path.exists(self.target_path):
-                os.rename(self.temp_path, self.target_path)
-            else:
-                os.remove(self.temp_path)
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self.close()
-
-    # Mimic other file object methods as needed
-    def __getattr__(self, item):
-        """Delegate attribute access to the underlying temporary file object."""
-        return getattr(self.temp_file, item)

tools/accuracy_checker/tests/test_convert.py

@@ -0,0 +1,67 @@
+"""
+Copyright (c) 2018-2024 Intel Corporation
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import os
+import threading
+import warnings
+from accuracy_checker.annotation_converters.convert import AtomicWriteFileHandle
+
+def thread_access_file(file_path, data_dict, thread_id, write_lines):
+    if os.path.exists(file_path):
+        with open(file_path, 'r') as file:
+            read_lines = len(file.readlines())
+            # when new thread sees file it must have all lines already written
+            if read_lines != write_lines:
+                warn_message = f"Thread {thread_id}: Incorrect number of lines read from {file_path} ({read_lines} != {write_lines})"
+                warnings.warn(warn_message)
+                data_dict['assert'] = warn_message
+    else:
+        with AtomicWriteFileHandle(file_path, 'wt') as file:
+            for i in range(write_lines):
+                file.write(f"Thread {thread_id}:Line{i} {data_dict[thread_id]}\n")
+
+class TestAtomicWriteFileHandle:
+
+    def test_multithreaded_atomic_file_write(self):
+        target_file_path = "test_atomic_file.txt"
+        threads = []
+        num_threads = 10
+        write_lines = 10
+        data_chunks = [f"Data chunk {i}" for i in range(num_threads)]
+        threads_dict = {i: data_chunks[i] for i in range(len(data_chunks))}
+
+        if os.path.exists(target_file_path):
+            os.remove(target_file_path)
+
+        for i in range(num_threads):
+            thread = threading.Thread(target=thread_access_file, args=(target_file_path, threads_dict, i, write_lines))
+            threads.append(thread)
+
+        for thread in threads:
+            thread.start()
+
+        for i,thread in enumerate(threads):
+            thread.join()
+
+        with open(target_file_path, 'r') as file:
+            lines = file.readlines()
+
+        os.remove(target_file_path)
+
+        # check asserts passed from threads
+        assert 'assert' not in threads_dict.keys() , threads_dict['assert']
+
+        assert sum(1 for line in lines for data_chunk in data_chunks if data_chunk in line) == write_lines, f"data_chunks data not found in the {target_file_path} file"

tools/accuracy_checker/tests/test_utils.py

@@ -14,9 +14,7 @@
 limitations under the License.
 """
 
-import os
-import threading
-from accuracy_checker.utils import concat_lists, contains_all, contains_any, overrides, zipped_transform, AtomicWriteFileHandle
+from accuracy_checker.utils import concat_lists, contains_all, contains_any, overrides, zipped_transform
 
 
 def test_concat_lists():
@@ -127,37 +125,3 @@ class C:
 
         assert overrides(B, 'foo', A)
         assert not overrides(C, 'foo', A)
-
-
-def thread_write_to_file(file_path, data, thread_id):
-    with AtomicWriteFileHandle(file_path, 'wt') as file:
-        file.write(f"Thread {thread_id}: {data}\n")
-
-
-class TestAtomicWriteFileHandle:
-
-    def test_multithreaded_atomic_file_write(self):
-        target_file_path = "test_atomic_file.txt"
-        threads = []
-        num_threads = 8
-        data_chunks = [f"Data chunk {i}" for i in range(num_threads)]
-
-        if os.path.exists(target_file_path):
-            os.remove(target_file_path)
-
-        for i in range(num_threads):
-            thread = threading.Thread(target=thread_write_to_file, args=(target_file_path, data_chunks[i], i))
-            threads.append(thread)
-            
-        for thread in threads:
-            thread.start()            
-
-        for thread in threads:
-            thread.join()
-
-        with open(target_file_path, 'r') as file:
-            lines = file.readlines()
-
-        os.remove(target_file_path)
-
-        assert any(data_chunk in line for line in lines for data_chunk in data_chunks), f"data_chunks data not found in the file"