RF64 implementation

wavinfo/rf64_parser.py

@@ -0,0 +1,40 @@
+import struct
+from collections import namedtuple
+from . import riff_parser 
+
+RF64Context = namedtuple('RF64Context','sample_count bigchunk_table')
+
+
+def parse_rf64(stream):
+    #print("starting parse_rf64")
+    start = stream.tell()
+    assert( stream.read(4) == b'WAVE' ) 
+
+    ds64_chunk = riff_parser.parse_chunk(stream)
+
+    ds64_field_spec = "<QQQI"
+    ds64_fields_size = struct.calcsize(ds64_field_spec)
+    assert(ds64_chunk.ident == b'ds64')
+    
+    ds64_data = ds64_chunk.read_data(stream)
+    assert(len(ds64_data) >= ds64_fields_size )
+    
+    #print("Read ds64 chunk: len()",len(ds64_data))
+    riff_size, data_size, sample_count, length_lookup_table = struct.unpack( ds64_field_spec , ds64_data[0:ds64_fields_size] )
+    
+    bigchunk_table = {}
+    chunksize64format = "<4sL"
+    chunksize64size = struct.calcsize(chunksize64format)
+    #print("Found chunks64s:", length_lookup_table)
+
+    for n in range(length_lookup_table):
+        bigname, bigsize = struct.unpack_from( chunksize64format , ds64data, offset= ds64_fields_size ) 
+        bigchunk_table[bigname] = bigsize
+
+    bigchunk_table[b'data'] = data_size
+    bigchunk_table[b'RF64'] = riff_size
+ 
+    stream.seek(start, 0)
+    #print("returning from parse_rf64, context: ",RF64Context( sample_count=sample_count, bigchunk_table=bigchunk_table ) )
+    return RF64Context( sample_count=sample_count, bigchunk_table=bigchunk_table )
+    

wavinfo/riff_parser.py

@@ -2,6 +2,7 @@
 import struct
 import pdb
 from collections import namedtuple
+from .rf64_parser import parse_rf64
 
 class ListChunkDescriptor(namedtuple('ListChunkDescriptor' , 'signature children')):
 
@@ -18,19 +19,20 @@ class ListChunkDescriptor(namedtuple('ListChunkDescriptor' , 'signature children
                             return chunk
 
 
-class ChunkDescriptor(namedtuple('ChunkDescriptor', 'ident start length') ):
+class ChunkDescriptor(namedtuple('ChunkDescriptor', 'ident start length rf64_context') ):
     def read_data(self, from_stream):
         from_stream.seek(self.start)
         return from_stream.read(self.length)
 
-def parse_list_chunk(stream, length):
+def parse_list_chunk(stream, length, rf64_context =None):
     start = stream.tell()
 
     signature = stream.read(4)
 
+    #print("Parsing list chunk with siganture: ", signature)
     children = []
     while (stream.tell() - start) < length:
-        child_chunk = parse_chunk(stream)
+        child_chunk = parse_chunk(stream, rf64_context= rf64_context)
         if child_chunk:
              children.append(child_chunk)
         else: 
@@ -38,27 +40,31 @@ def parse_list_chunk(stream, length):
 
     return ListChunkDescriptor(signature=signature, children=children)
 
-def parse_chunk(stream):
-    #breakpoint()
+def parse_chunk(stream, rf64_context=None):
     ident = stream.read(4)
     if len(ident) != 4:
         return
 
     sizeb = stream.read(4)
     size  = struct.unpack('<I',sizeb)[0]
-
+    
+    if size == 0xFFFFFFFF:
+        if rf64_context is None and ident == b'RF64':
+            rf64_context = parse_rf64(stream=stream)
+        
+        size = rf64_context.bigchunk_table[ident] 
+            
     displacement = size
     if displacement % 2 is not 0:
         displacement = displacement + 1
 
-    if ident in [b'RIFF',b'LIST']:
-        return parse_list_chunk(stream=stream, length=size)
+    if ident in [b'RIFF',b'LIST', b'RF64']:
+        #print("Parsing list chunk with ident: ", ident)
+        return parse_list_chunk(stream=stream, length=size, rf64_context=rf64_context)
     else:
         start = stream.tell()
         stream.seek(displacement,1)
-        return ChunkDescriptor(ident=ident, start=start, length=size)
-
-
-
+        #print("Parsing chunk with start=%i, ident=%s" % (start, ident))
+        return ChunkDescriptor(ident=ident, start=start, length=size, rf64_context=rf64_context)