Extraer datos comprimidos zlib del archivo binario en python

Question

Mi empresa utiliza un formato de archivo heredado para datos de electromiografía, que ya no está en producción. Sin embargo, existe cierto interés en mantener la retrocompatibilidad, por lo que estoy estudiando la posibilidad de escribir un lector para ese formato de archivo.

Al analizar un antiguo código fuente muy intrincado escrito en Delphi, el lector/escritor de archivos utiliza ZLIB, y dentro de un HexEditor parece que hay un encabezado de archivo en ASCII binario (con campos como "Reproductor", "Analizador" fácilmente legible), seguido de una cadena comprimida que contiene datos brutos.

Mi duda es: ¿cómo debo proceder con el fin de identificar:

• Si se trata de una corriente comprimida;
• ¿Dónde se inicia el flujo comprimido y dónde termina?

De Wikipedia: datos comprimidos

zlib normalmente se escribe con una gzip o un zlib envoltura. El contenedor encapsula los datos sin procesar de DEFLATE agregando un encabezado y un avance. Esto proporciona identificación de flujo y error detección

¿Es esto relevante?

Estaré encantado de publicar más información, pero no sé lo que sería más relevante.

Gracias por cualquier pista.

EDITAR: Tengo la aplicación en funcionamiento, y puedo usarla para registrar datos reales de cualquier longitud de tiempo, obteniendo archivos incluso menores de 1kB si es necesario.

---

Algunos archivos de ejemplo:

Un recién creado uno, sin flujo de datos: https://dl.dropbox.com/u/4849855/Mio_File/HeltonEmpty.mio

Lo mismo que el anterior después de un tiempo muy corto (1 segundo?) Flujo de datos se ha guardado: https://dl.dropbox.com/u/4849855/Mio_File/HeltonFilled.mio

Una diferente, de un paciente llamado "manco" en lugar de "Helton", con un flujo aún más corto (ideal para la visualización Hex): https://dl.dropbox.com/u/4849855/Mio_File/manco_short.mio

Instrucciones: cada archivo debe ser el archivo de un paciente (una persona). Dentro de estos archivos, se guardan uno o más exámenes, cada examen consta de una o más series temporales. Los archivos provistos contienen solo un examen, con una serie de datos.

Answer 1

Para empezar, ¿por qué no explorar los archivos para todos los flujos postal válido (es lo suficientemente bueno para archivos pequeños y para averiguar el formato):

import zlib 
from glob import glob 

def zipstreams(filename): 
    """Return all zip streams and their positions in file.""" 
    with open(filename, 'rb') as fh: 
     data = fh.read() 
    i = 0 
    while i < len(data): 
     try: 
      zo = zlib.decompressobj() 
      yield i, zo.decompress(data[i:]) 
      i += len(data[i:]) - len(zo.unused_data) 
     except zlib.error: 
      i += 1 

for filename in glob('*.mio'): 
    print(filename) 
    for i, data in zipstreams(filename): 
     print (i, len(data)) 

Parece que los flujos de datos contienen doble precisión ascendente hacia la izquierda datos de coma flotante:

import numpy 
from matplotlib import pyplot 

for filename in glob('*.mio'): 
    for i, data in zipstreams(filename): 
     if data: 
      a = numpy.fromstring(data, '<f8') 
      pyplot.plot(a[1:]) 
      pyplot.title(filename + ' - %i' % i) 
      pyplot.show() 

Answer 2

zlib es una envoltura delgada alrededor datos comprimidos con el algoritmo DEFLATE y se define en RFC1950:

A zlib stream has the following structure: 

     0 1 
    +---+---+ 
    |CMF|FLG| (more-->) 
    +---+---+ 

    (if FLG.FDICT set) 

     0 1 2 3 
    +---+---+---+---+ 
    |  DICTID | (more-->) 
    +---+---+---+---+ 

    +=====================+---+---+---+---+ 
    |...compressed data...| ADLER32 | 
    +=====================+---+---+---+---+ 

lo que añade al menos dos, posiblemente seis bytes antes y 4 bytes con una ADLER32 suma de comprobación después de los datos comprimidos sin procesar DEFLATE.

El primer byte contiene el CMF (Método y banderas de compresión), que se divide en CM (Método de compresión) (los primeros 4 bits) y CINFO (info de compresión) (los últimos 4 bits) .

A partir de esto, es evidente que, lamentablemente, los primeros dos bytes de una secuencia zlib pueden variar mucho según el método de compresión y la configuración que se hayan utilizado.

Afortunadamente, me encontré con una publicación de Mark Adler, el autor del algoritmo ADLER32 , donde lists the most common and less common combinations of those two starting bytes.

Con eso fuera del camino, vamos a ver cómo podemos usar Python para examinar zlib:

>>> import zlib 
>>> msg = 'foo' 
>>> [hex(ord(b)) for b in zlib.compress(msg)] 
['0x78', '0x9c', '0x4b', '0xcb', '0xcf', '0x7', '0x0', '0x2', '0x82', '0x1', '0x45'] 

Así los datos zlib creados por el módulo de Python zlib (utilizando opciones por defecto) se inicia con 78 9c . Lo utilizaremos para crear un script que escriba un formato de archivo personalizado que contenga un preámbulo, algunos datos comprimidos zlib y un pie de página.

A continuación, escriba una segunda secuencia de comandos que escanea un archivo para que el patrón de dos bytes, comienza descomprimir todo lo que sigue como una corriente zlib y se da cuenta de donde termina el flujo y el pie de página aperturas.

create.py

import zlib 

msg = 'foo' 
filename = 'foo.compressed' 

compressed_msg = zlib.compress(msg) 
data = 'HEADER' + compressed_msg + 'FOOTER' 

with open(filename, 'wb') as outfile: 
    outfile.write(data) 

aquí tomamos msg, comprimirlo con zlib, y lo rodean con un encabezado y pie de página antes de escribirlo en un archivo.

Encabezado y pie de página son de longitud fija en este ejemplo, pero podrían, por supuesto, tener longitudes arbitrarias desconocidas.

Ahora para la secuencia de comandos que intenta encontrar una secuencia zlib en dicho archivo. Porque para este ejemplo sabemos exactamente qué marcador esperar, estoy usando solo uno, pero obviamente la lista ZLIB_MARKERS podría llenarse con todos los marcadores de la publicación mencionada anteriormente.

ident.py

import zlib 

ZLIB_MARKERS = ['\x78\x9c'] 
filename = 'foo.compressed' 

infile = open(filename, 'r') 
data = infile.read() 

pos = 0 
found = False 

while not found: 
    window = data[pos:pos+2] 
    for marker in ZLIB_MARKERS: 
     if window == marker: 
      found = True 
      start = pos 
      print "Start of zlib stream found at byte %s" % pos 
      break 
    if pos == len(data): 
     break 
    pos += 1 

if found: 
    header = data[:start] 

    rest_of_data = data[start:] 
    decomp_obj = zlib.decompressobj() 
    uncompressed_msg = decomp_obj.decompress(rest_of_data) 

    footer = decomp_obj.unused_data 

    print "Header: %s" % header 
    print "Message: %s" % uncompressed_msg 
    print "Footer: %s" % footer 

if not found: 
    print "Sorry, no zlib streams starting with any of the markers found." 

La idea es la siguiente:

• empezar por el principio del archivo y crear una ventana de dos bytes de búsqueda .

• Mueva la ventana de búsqueda hacia adelante en incrementos de un byte.

• Para cada ventana, compruebe si coincide con cualquiera de los dos marcadores de bytes que definimos.

• Si se encuentra una coincidencia, registre la posición de inicio, deje de buscar y intente descomprimir todo lo que sigue.

Ahora, encontrar el final de la secuencia no es tan trivial como la búsqueda de dos marcadores bytes. Las secuencias zlib no están terminadas por una secuencia de bytes fija ni su longitud se indica en ninguno de los campos de encabezado. En su lugar, finaliza con una suma de comprobación ADLER32 de cuatro bytes que debe coincidir con los datos hasta este punto.

La forma en que funciona es que la función interna C inflate() mantiene continuamente tratando de descomprimir el flujo a medida que lo lee, y si se encuentra con una suma de comprobación juego, señales que a su llamador, lo que indica que el resto de la datos ya no forman parte de la secuencia zlib.

En Python, este comportamiento se expone cuando se utilizan objetos de descompresión en lugar de simplemente llamando al . Llamar al decompress(string) en un objeto Decompress descomprimirá una secuencia zlib en string y devolverá los datos descomprimidos que formaban parte de la secuencia. Todo lo que sigue a la secuencia se almacenará en unused_data y puede ser recuperado posteriormente.

Esto debería producir la siguiente salida en un archivo creado con el primer guión :

Start of zlib stream found at byte 6 
Header: HEADER 
Message: foo 
Footer: FOOTER 

El ejemplo se puede modificar fácilmente para escribir el mensaje sin comprimir en un archivo en lugar de imprimirlo. Luego puede analizar aún más los datos comprimidos anteriormente zlib , e intentar identificar los campos conocidos en los metadatos en el encabezado y pie de página que ha separado.