Moved to opusFC from brukeropusreader

Author: Martin

converter_app/models.py

@@ -2,11 +2,11 @@
 import logging
 import os
 import pathlib
+import shutil
 import tarfile
 import tempfile
 import uuid
 from collections import defaultdict
-
 from pathlib import Path
 
 import magic
@@ -202,6 +202,7 @@ class File:
     def __init__(self, file):
         self._features = {}
         self.fp = file
+        self._temp_dir = None
 
         # read the file
         self.content = file.read()
@@ -218,6 +219,24 @@ def __init__(self, file):
         # decode file string
         self.string = self.content.decode(self.encoding, errors='ignore') if self.encoding != 'binary' else None
 
+    def __enter__(self):
+        # Create the temporary directory when the context is entered
+        self._temp_dir = tempfile.mkdtemp()
+        print(f"Temporary directory created: {self._temp_dir}")
+        return self._temp_dir
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        # Cleanup: Remove the temporary directory when the context is exited
+        if self._temp_dir and os.path.exists(self._temp_dir):
+            shutil.rmtree(self._temp_dir)
+
+    @property
+    def temp_dir(self):
+        """Returns temporary directory path for tar extraction"""
+        if not self._temp_dir or not os.path.exists(self._temp_dir):
+            self.__enter__()
+        return self._temp_dir
+
     @property
     def content_type(self):
         """
@@ -267,15 +286,21 @@ def is_tar_archive(self) -> bool:
         """
         return self.name.endswith(".gz") or self.name.endswith(".xz") or self.name.endswith(".tar")
 
+    def __del__(self):
+        self.__exit__(None, None, None)
+        print(f"Object {self.name} is being destroyed.")
+
+
 
-def extract_tar_archive(file: File, temp_dir: str) -> list[File]:
+def extract_tar_archive(file: File) -> list[File]:
     """
     If the file is a tar archive, this function extracts it and returns a list of all files
     :param file: Input file from the client
     :return: A list of all files extracted
     """
     if not file.is_tar_archive:
         return []
+    temp_dir = file.temp_dir
     file_list = []
     with tempfile.NamedTemporaryFile(delete=True) as temp_archive:
         try:

converter_app/readers/brucker_dotzero.py

@@ -1,6 +1,6 @@
 import logging
 
-from brukeropusreader.opus_parser import parse_meta, parse_data
+import opusFC
 from numpy import ndarray
 
 from converter_app.readers import Readers
@@ -18,7 +18,7 @@ class DotZeroReader(Reader):
 
     def __init__(self, file, *tar_files):
         super().__init__(file, *tar_files)
-        self._dotzero_file = []
+        self._dotzero_file = None
         self._dx_name = None
 
     def check(self):
@@ -29,15 +29,22 @@ def check(self):
         dotzero_extentions = ['.0']
 
         if self.is_tar_ball:
-            self._dotzero_file = [x for x in self.file_content if x.suffix.lower() in dotzero_extentions]
+            for x in self.file_content:
+                try:
+                    if opusFC.isOpusFile(x.fp.filename):
+                        self._dotzero_file = x.fp.filename
+                        return True
+                except (ValueError, TypeError, FileNotFoundError):
+                    pass
+
             dx_file = next((x for x in self.file_content if x.suffix.lower() == '.dx'), None)
             if dx_file is not None:
                 self._dx_name = dx_file.name[:-3]
 
-        if self.file.suffix.lower() in dotzero_extentions:
-            self._dotzero_file.append(self.file)
+        elif self.file.suffix.lower() in dotzero_extentions and opusFC.isOpusFile(self.file.fp.filename):
+            self._dotzero_file = self.file.fp.filename
 
-        return len(self._dotzero_file) > 0
+        return self._dotzero_file is not None
 
     def _add_to_meta(self, table, src, k=None):
         if k is None:
@@ -57,24 +64,22 @@ def _add_to_meta(self, table, src, k=None):
     def prepare_tables(self):
         tables = []
 
-        for dotzero_file in self._dotzero_file:
-            data = dotzero_file.content
-            meta_data = parse_meta(data)
-            opus_data = parse_data(data, meta_data)
-            if dotzero_file.name == self._dx_name:
-                table = self.append_table([])
-                tables.insert(0, table)
-            else:
-                table = self.append_table(tables)
-            table['metadata']['__FILE_NAME__'] = dotzero_file.name
-            self._add_to_meta(table, opus_data)
-            ab_x = opus_data.get_range("AB")
-            table['rows'] = [[val,  opus_data["AB"][i], opus_data["ScSm"][i], opus_data["ScRf"][i]] for i, val in enumerate(ab_x)]
-
-            table['columns'] += [{
-                'key': f'{idx}',
-                'name': value
-            } for idx, value in enumerate(["X", "AB", "ScSm", "ScRf"])]
+        dbs = opusFC.listContents(self._dotzero_file)  # List all data blocks in the file
+        for block in dbs:
+            table = self.append_table(tables)
+            data = opusFC.getOpusData( self._dotzero_file, block)  # Retrieve data from the specific block
+            for x in block:
+                table.add_metadata('__BLOCK__', str(x))
+
+            for key, value in data.parameters.items():
+                table.add_metadata(str(key), str(value))
+
+            table['rows'] = [[float(val), float(data.y[i])] for i, val in enumerate(data.x)]
+            table['columns'] = [{
+                        'key': f'{idx}',
+                        'name': name
+                    } for idx, name in enumerate(['X', 'Y'])]
+
         return tables
 
 

converter_app/readers/helper/reader.py

@@ -1,5 +1,4 @@
 import inspect
-import tempfile
 from collections import OrderedDict
 
 from converter_app.converters import logger
@@ -59,26 +58,24 @@ def match_reader(self, file: File):
         logger.debug('file_name=%s content_type=%s mime_type=%s encoding=%s',
                      file.name, file.content_type, file.mime_type, file.encoding)
 
-        with tempfile.TemporaryDirectory() as tmpdir:
+        archive_file_list = extract_tar_archive(file)
 
-            archive_file_list = extract_tar_archive(file, tmpdir)
+        for _identifier, reader in self.readers.items():
+            params = inspect.signature(reader).parameters
+            if len(params) > 1:
+                reader = reader(file, *archive_file_list)
+            else:
+                reader = reader(file)
 
-            for _identifier, reader in self.readers.items():
-                params = inspect.signature(reader).parameters
-                if len(params) > 1:
-                    reader = reader(file, *archive_file_list)
-                else:
-                    reader = reader(file)
+            result = reader.check()
 
-                result = reader.check()
+            logger.debug('For reader %s -> result=%s', reader.__class__.__name__, result)
 
-                logger.debug('For reader %s -> result=%s', reader.__class__.__name__, result)
-
-                # reset file pointer and return the reader it is the one
-                file.fp.seek(0)
-                for archive_file in archive_file_list:
-                    archive_file.fp.seek(0)
-                if result:
-                    return reader
+            # reset file pointer and return the reader it is the one
+            file.fp.seek(0)
+            for archive_file in archive_file_list:
+                archive_file.fp.seek(0)
+            if result:
+                return reader
 
         return None

requirements/common.txt

@@ -13,4 +13,4 @@ jcamp~=1.2.2
 PyMuPDF==1.23.7
 pylint==3.0.3
 str2bool~=1.1
-brukeropusreader~=1.3.4
+opusFC~=1.4.0

test_static/tests.py

@@ -67,9 +67,8 @@ def test_tar_unzip(client: FlaskClient):
         file = File(fs)
         assert file.is_tar_archive
         assert file.name == 'a.tar.gz'
-        with tempfile.TemporaryDirectory() as tmpdirname:
-            with open(os.path.join(os.path.dirname(__file__), 'a/a.txt.0'), 'r') as tf:
-                archive = extract_tar_archive(file, tmpdirname)
-                assert len(archive) == 1
-                assert archive[0].name == 'a.txt.0'
-                assert archive[0].string == tf.read()
+        with open(os.path.join(os.path.dirname(__file__), 'a/a.txt.0'), 'r') as tf:
+            archive = extract_tar_archive(file)
+            assert len(archive) == 1
+            assert archive[0].name == 'a.txt.0'
+            assert archive[0].string == tf.read()