Estrai dati compressi zlib dal file binario in python

Question

La mia azienda utilizza un formato di file legacy per dati Electromiography, che non è più in produzione. Tuttavia, c'è un certo interesse nel mantenere la retro-compatibilità, quindi sto studiando la possibilità di scrivere un lettore per quel formato di file.

Analizzando un codice sorgente molto complicato scritto in Delphi, il lettore/scrittore di file utilizza ZLIB, e all'interno di un HexEditor sembra che ci sia un'intestazione di file in ASCII binario (con campi come "Player", "Analyzer" facilmente leggibile), seguito da una stringa compressa contenente dati non elaborati.

Il mio dubbio è: come devo procedere al fine di identificare:

• Se si tratta di un flusso compresso;
• Da dove viene avviato il flusso compresso e dove finisce?

Da Wikipedia: dati compressi

zlib è tipicamente scritto con una gzip o uno zlib involucro. Il wrapper incapsula i dati DEFLATE grezzi aggiungendo un'intestazione e un trailer . Questo fornisce l'identificazione flusso e l'errore di rilevamento

È questo rilevante?

Sarò lieto di pubblicare ulteriori informazioni, ma non so quale sarebbe più rilevante.

Grazie per qualsiasi suggerimento.

EDIT: Ho l'applicazione funzionante e posso usarlo per registrare dati reali di qualsiasi durata, ottenendo file anche inferiori a 1kB se necessario.

---

Alcuni file di esempio:

A appena creato uno, senza flusso di dati: https://dl.dropbox.com/u/4849855/Mio_File/HeltonEmpty.mio

La stessa di cui sopra, dopo un brevissimo (1 secondo?) Datastream è stata salvata: https://dl.dropbox.com/u/4849855/Mio_File/HeltonFilled.mio

Uno diverso, da un paziente denominato "manco" anziché "Helton", con un flusso ancora più corto (ideale per la visualizzazione Hex): https://dl.dropbox.com/u/4849855/Mio_File/manco_short.mio

Istruzioni: ogni file dovrebbe essere il file di un paziente (una persona). All'interno di questi file, vengono salvati uno o più esami, ciascun esame consistente in una o più serie temporali. I file forniti contengono solo un esame, con una serie di dati.

Answer 1

Per iniziare, perché non eseguire la scansione dei file per tutti i flussi valida zip (va bene anche per file di piccole dimensioni e di capire il 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)) 

Sembra che i flussi di dati contengono little-endian doppia precisione dati in virgola mobile:

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 è un involucro sottile intorno dati compressi con l'algoritmo DEFLATE ed è definito nella 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 | 
    +=====================+---+---+---+---+ 

Quindi aggiunge almeno due, forse sei byte prima e 4 byte con un ADLER32 checksum dopo i dati compressi grezzi di DEFLATE.

Il primo byte contiene il CMF (Metodo di compressione e bandiere), che è diviso in CM (Metodo di compressione) (primi 4 bit) e CINFO (Compression informazioni) (ultimi 4 bit) .

Da questo è abbastanza chiaro che purtroppo già i primi due byte di un flusso zlib può variare molto a seconda di cosa compressione metodo e impostazioni sono stati utilizzati.

Per fortuna, sono incappato in un post di Mark Adler, l'autore dell'algoritmo ADLER32 , dove ha lists the most common and less common combinations of those two starting bytes.

Con quella di mezzo, diamo un'occhiata a come possiamo usare Python per esaminare 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'] 

Quindi i dati zlib creati dal modulo di Python zlib (utilizzando le opzioni di default) inizia con 78 9c . Lo useremo per creare uno script che scriva un formato di file personalizzato che contenga un preambolo, alcuni dati compressi con zlib e un piè di pagina.

Abbiamo poi scrivere un secondo script che esegue la scansione di un file per quel modello di due byte, inizia la decompressione tutto ciò che segue come un flusso di zlib e capisce dove finisce stream e inizia il piè di pagina.

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) 

Qui prendiamo msg, comprimerlo con zlib, e circondano con un colpo di testa e piè di pagina prima che lo scriviamo in un file.

L'intestazione e il piè di pagina sono di lunghezza fissa in questo esempio, ma potrebbero naturalmente avere lunghezze arbitrarie e sconosciute.

Ora per lo script che tenta di trovare uno stream zlib in tale file. Perché per questo esempio sappiamo esattamente quale indicatore attendersi sto usando solo uno, ma ovviamente l'elenco ZLIB_MARKERS potrebbe essere riempito con tutti i marcatori dal post menzionati sopra.

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." 

L'idea è questa:

• cominciare all'inizio del file e creare una finestra di due byte di ricerca .

• Sposta la finestra di ricerca in avanti con incrementi di un byte.

• Per ogni finestra controllare se corrisponde a uno dei due indicatori di byte definiti .

• Se viene trovata una corrispondenza, registrare la posizione di partenza, interrompere la ricerca e provare a decomprimere tutto ciò che segue.

Ora, trovando la fine del flusso non è così banale come alla ricerca di due marcatore byte. i flussi zlib non sono né terminati da una sequenza di byte fissi, né è la lunghezza indicata in nessuno dei campi dell'intestazione. Invece è terminato da un checksum ADLER32 a quattro byte che deve corrispondere ai dati fino a questo punto.

Il modo in cui funziona è che la funzione interna C inflate() mantiene continuamente cercando di decomprimere il flusso come lo legge, e se viene rilevato un checksum corrispondenza, segnali che al chiamante, indicando che il resto della I dati non fanno più parte del flusso zlib.

In Python questo comportamento viene esposto quando si utilizzano oggetti di decompressione anziché semplicemente chiamando zlib.decompress(). Chiamando decompress(string) su un Decompress l'oggetto decomprimerà un flusso zlib in string e restituirà i dati decompressi che facevano parte del flusso. Tutto ciò che segue lo stream verrà archiviato in unused_data e potrà essere recuperato in seguito dallo .

Questo dovrebbe produrre il seguente output su un file creato con il primo script:

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

L'esempio può essere facilmente modificato per scrivere il messaggio non compresso in un file invece di stamparlo. Quindi è possibile analizzare ulteriormente i dati compressi precedentemente zlib e provare a identificare i campi noti nei metadati nell'intestazione e nel piè di pagina separati.