NF - incorporate Ians code with tests; fix 4 byte boundaries

Author: matthew-brett

nibabel/dicom/csareader.py

@@ -8,32 +8,75 @@ class CSAReadError(Exception):
     pass
 
 
-def read(fileobj):
-    ''' Read CSA header from fileobj
+def read(csa_str):
+    ''' Read CSA header from string `csa_str`
 
     Parameters
     ----------
-    fileobj : file-like
-       file-like object implementing ``read, seek, tell`.  The data is
-       taken to start at position 0 and end at the end of the file. 
+    csa_str : str
+       byte string containing CSA header information
 
     Returns
     -------
     header : dict
        header information
-    tags : sequence
-       sequence of tags from header
     '''
-    fileobj.seek(0)
-    hdr_id = fileobj.read(4)
-    hdr = {}
-    tags = []
+    hdr_id = csa_str[:4]
+    csa_len = len(csa_str)
+    csa_dict = {'tags': {}}
     if hdr_id != 'SV10':
-        hdr['type'] = 1
-        hdr['n_tags'], = unpack('<i', hdr_id)
+        csa_dict['type'] = 1
+        csa_dict['n_tags'], = unpack('<i', hdr_id)
+        ptr = 4
     else:
-        hdr['type'] = 2
-        hdr['unused1'] = fileobj.read(4)
-        hdr['n_tags'], = unpack('<I', fileobj.read(4))
-    hdr['unused3'], = unpack('<I', fileobj.read(4))
-    return hdr, tags
+        csa_dict['type'] = 2
+        csa_dict['unused0'] = csa_str[4:8]
+        csa_dict['n_tags'], = unpack('<I', csa_str[8:12])
+        ptr = 12
+    csa_dict['check'], = unpack('<I', csa_str[ptr:ptr+4])
+    ptr +=4
+    for tag_no in range(csa_dict['n_tags']):
+        name, vm, vr, syngodt, n_items, xx = \
+            unpack('<64sI4sIII',csa_str[ptr:ptr+84])
+        name = nt_str(name)
+        tag = {'pointer': ptr,
+               'n_items': n_items,
+               'last3': xx}
+        ptr+=84
+        assert n_items < 100
+        for item_no in range(n_items):
+            tag['pointer'] = ptr
+            x0,x1,x2,x3 = unpack('<4i',csa_str[ptr:ptr+16])
+            tag['xx'] = [x0,x1,x2,x3]
+            ptr+=16
+            item_len = x1
+            if (ptr + item_len) > csa_len:
+                raise CSAReadError('Item is too long, aborting read')
+            format = '<%ds' % item_len
+            tag['value'] = unpack(format,csa_str[ptr:ptr+item_len])
+            plus4 = item_len % 4
+            if plus4 != 0:
+                ptr+=item_len + (4-plus4)
+            else:
+                ptr+=item_len
+            tag['end_pointer'] = ptr
+        csa_dict['tags'][name] = tag
+    return csa_dict
+
+
+def nt_str(s):
+    ''' Strip string to first null
+
+    Parameters
+    ----------
+    s : str
+
+    Returns
+    -------
+    sdash : str
+       s stripped to first occurence of null (0)
+    '''
+    zero_pos = s.find(chr(0))
+    if zero_pos == -1:
+        return s
+    return s[:zero_pos]

nibabel/dicom/tests/test_csareader.py

@@ -17,15 +17,15 @@
 
 data_path = pjoin(os.path.dirname(__file__), 'data')
 
-CSA2_B0 = pjoin(data_path, 'csa2_b0.bin')
-CSA2_B1000 = pjoin(data_path, 'csa2_b1000.bin')
+CSA2_B0 = open(pjoin(data_path, 'csa2_b0.bin')).read()
+CSA2_B1000 = open(pjoin(data_path, 'csa2_b1000.bin')).read()
 
 
 @parametric
 def test_csa():
-    csa_f = open(CSA2_B0, 'rb')
-    hdr, tags = csa.read(csa_f)
-    yield assert_equal(hdr['type'], 2)
-    yield assert_equal(hdr['n_tags'], 83)
+    csa_info = csa.read(CSA2_B0)
+    yield assert_equal(csa_info['type'], 2)
+    yield assert_equal(csa_info['n_tags'], 83)
+    tags = csa_info['tags']
     yield assert_equal(len(tags), 83)
-    
+    print csa_info