erp5_wendelin_telecom_ingestion: unit test for nifty nii files

bt5/erp5_wendelin_telecom_ingestion/TestTemplateItem/portal_components/test.erp5.testDataIngestion.py

@@ -20,6 +20,8 @@ class TestDataIngestion(SecurityTestCase):
   TXT = "/txt"
   CSV = "/csv"
   TSV = "/tsv"
+  GZ = "/gz"
+  NII = "/nii"
   RANDOM = "/" + ''.join([random.choice(string.ascii_letters + string.digits) for _ in xrange(3)])
   CHUNK_SIZE_TXT = 50000
   CHUNK_SIZE_CSV = 25
@@ -184,6 +186,44 @@ class TestDataIngestion(SecurityTestCase):
     if os.path.exists(file_name):
       os.remove(file_name)
 
+  def perform_nii_test(self, gz):
+    import nibabel as nib
+    import numpy as np
+    reference = self.getRandomReference()
+    url = 'data_stream_module/mne_sample_for_test_nii'
+    if gz: url += "_gz"
+    sample_data_stream = self.context.restrictedTraverse(url)
+    data_chunk = sample_data_stream.getData()
+    ingestion_reference = self.ingest(data_chunk, reference, self.GZ if gz else self.NII)
+
+    file_name = "file_name.nii"
+    if gz: file_name += ".gz"
+    with open(file_name, 'w') as nii_file:
+      nii_file.write(data_chunk)
+    img = nib.load(file_name)
+    data = {}
+    for key in img.header:
+      try:
+        if type(img.header[key]) is np.ndarray:
+          content = img.header[key].tolist()
+          try:
+            if np.isnan(img.header[key]): content = "nan"
+          except:
+            pass
+        else:
+          content = img.header[key]
+        if content == 0: content = "0"
+        json.dumps(content)
+        if content != "":
+          data[key] = content
+      except Exception as e:
+        pass # ignore non serializable info
+    json_data = json.dumps(data)
+    if os.path.exists(file_name):
+      os.remove(file_name)
+
+    self.checkDataObjects(ingestion_reference, data_chunk, None, json_data)
+
   def test_full_data_ingestion(self):
     reference = self.getRandomReference()
     data_chunk, nparray, json_data = self.generateRawDataBytesAndArray()
@@ -217,6 +257,12 @@ class TestDataIngestion(SecurityTestCase):
 
     self.checkDataObjects(ingestion_reference, data_chunk, None, json_data)
 
+  def test_nii_data_ingestion(self):
+    self.perform_nii_test(gz=False)
+
+  def test_nii_gz_data_ingestion(self):
+    self.perform_nii_test(gz=True)
+
   def test_data_ingestion_splitted_file(self):
     reference = self.getRandomReference()
     ingestion_policy = self.getIngestionPolicy(reference, self.INGESTION_SCRIPT)

bt5/erp5_wendelin_telecom_ingestion/TestTemplateItem/portal_components/test.erp5.testDataIngestion.xml

@@ -46,9 +46,14 @@
             <key> <string>text_content_warning_message</string> </key>
             <value>
               <tuple>
-                <string>W:166, 34: Unused variable \'i\' (unused-variable)</string>
-                <string>W:166, 76: Unused variable \'j\' (unused-variable)</string>
-                <string>W:230,  4: Unused variable \'ingestion_id\' (unused-variable)</string>
+                <string>W:168, 34: Unused variable \'i\' (unused-variable)</string>
+                <string>W:168, 76: Unused variable \'j\' (unused-variable)</string>
+                <string>W:191,  4: Redefining name \'np\' from outer scope (line 9) (redefined-outer-name)</string>
+                <string>W:191,  4: Reimport \'numpy\' (imported line 9) (reimported)</string>
+                <string>W:207, 11: Using type() instead of isinstance() for a typecheck. (unidiomatic-typecheck)</string>
+                <string>W:211, 10: No exception type(s) specified (bare-except)</string>
+                <string>W:219, 26: Unused variable \'e\' (unused-variable)</string>
+                <string>W:276,  4: Unused variable \'ingestion_id\' (unused-variable)</string>
               </tuple>
             </value>
         </item>

bt5/erp5_wendelin_telecom_ingestion/bt/template_keep_last_workflow_history_only_path_list

@@ -11,4 +11,6 @@ portal_categories/function/**
 portal_categories/use/big_data/ingestion/batch_ingestion
 data_stream_module/embulk_download_script
 data_stream_module/embulk_download_script_win
-data_stream_module/mne_sample_for_test
\ No newline at end of file
+data_stream_module/mne_sample_for_test
+data_stream_module/mne_sample_for_test_nii
+data_stream_module/mne_sample_for_test_nii_gz
\ No newline at end of file

bt5/erp5_wendelin_telecom_ingestion/bt/template_keep_workflow_path_list

@@ -11,4 +11,6 @@ portal_categories/function/**
 portal_categories/use/big_data/ingestion/batch_ingestion
 data_stream_module/embulk_download_script
 data_stream_module/embulk_download_script_win
-data_stream_module/mne_sample_for_test
\ No newline at end of file
+data_stream_module/mne_sample_for_test
+data_stream_module/mne_sample_for_test_nii
+data_stream_module/mne_sample_for_test_nii_gz
\ No newline at end of file

bt5/erp5_wendelin_telecom_ingestion/bt/template_path_list

@@ -6,6 +6,8 @@ data_product_module/fif_descriptor
 data_stream_module/embulk_download_script
 data_stream_module/embulk_download_script_win
 data_stream_module/mne_sample_for_test
+data_stream_module/mne_sample_for_test_nii
+data_stream_module/mne_sample_for_test_nii_gz
 data_supply_module/**
 data_transformation_module/**
 person_module/**