Trích xuất dữ liệu nén zlib từ tệp nhị phân trong python

Question

Công ty của tôi sử dụng định dạng tệp kế thừa cho dữ liệu Điện tâm đồ, không còn trong quá trình sản xuất nữa. Tuy nhiên, có một số quan tâm trong việc duy trì tính tương thích retro, vì vậy tôi đang nghiên cứu khả năng để viết một người đọc cho định dạng tập tin đó.

Bằng cách phân tích mã nguồn trước đây rất phức tạp được viết bằng Delphi, trình đọc/ghi tệp sử dụng ZLIB và bên trong HexEditor có vẻ như có tiêu đề tệp nhị phân ASCII (với các trường như "Trình phát", "Máy phân tích" dễ đọc), theo sau là một chuỗi nén chứa dữ liệu thô.

nghi ngờ của tôi là: làm thế nào tôi nên tiến hành để xác định:

• Nếu nó là một dòng nén;
• Bắt đầu luồng nén ở đâu và kết thúc ở đâu;

Từ Wikipedia:

zlib dữ liệu nén thường được viết với một gzip hoặc một zlib wrapper. Trình bao bọc đóng gói dữ liệu DEFLATE thô bằng cách thêm tiêu đề và đoạn giới thiệu . Điều này cung cấp xác định luồng và lỗi phát hiện

Điều này có liên quan không?

Tôi rất vui khi đăng thêm thông tin, nhưng tôi không biết điều gì sẽ phù hợp nhất.

Cảm ơn mọi gợi ý.

EDIT: Tôi có ứng dụng đang hoạt động và có thể sử dụng nó để ghi lại dữ liệu thực tế của bất kỳ thời lượng nào, nhận tệp thậm chí nhỏ hơn 1kB nếu cần.

---

Một số file mẫu:

Một vừa tạo một, mà không datastream: https://dl.dropbox.com/u/4849855/Mio_File/HeltonEmpty.mio

Giống như trên sau khi một rất ngắn (1 giây?) Datastream đã được lưu: https://dl.dropbox.com/u/4849855/Mio_File/HeltonFilled.mio

Một loại khác, từ một bệnh nhân có tên "manco" thay vì "Helton", với một dòng ngắn hơn (lý tưởng cho xem Hex): https://dl.dropbox.com/u/4849855/Mio_File/manco_short.mio

Hướng dẫn: mỗi tệp phải là tệp của bệnh nhân (một người). Bên trong các tệp này, một hoặc nhiều bài kiểm tra được lưu, mỗi bài kiểm tra bao gồm một hoặc nhiều chuỗi thời gian. Các tệp được cung cấp chỉ chứa một bài kiểm tra, với một chuỗi dữ liệu.

Answer 1

Để bắt đầu, tại sao không quét các tập tin cho mọi dòng zip hợp lệ (đó là đủ tốt cho các tập tin nhỏ và để tìm ra định dạng):

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)) 

Hình như các dòng dữ liệu chứa chính xác đôi chút về cuối nhỏ dữ liệu dấu chấm động:

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 là một wrapper mỏng xung quanh dữ liệu nén bằng thuật toán DEFLATE và được định nghĩa trong 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 | 
    +=====================+---+---+---+---+ 

Vì vậy, nó bổ sung thêm ít nhất hai, có thể là sáu byte trước và 4 byte với một ADLER32 tổng kiểm tra sau khi dữ liệu được nén DEFLATE thô.

Byte đầu tiên chứa CMF (Compression Method và cờ), được chia vào CM (phương pháp nén) (lần đầu tiên 4 bit) và CINFO (Compression thông tin) (cuối cùng 4 bit) .

Từ đây rõ ràng là không may hai byte đầu tiên của luồng zlib có thể khác nhau tùy thuộc vào phương pháp nén và cài đặt đã được sử dụng.

May mắn thay, tôi tình cờ gặp một bài đăng của Mark Adler, tác giả của thuật toán ADLER32 , trong đó ông lists the most common and less common combinations of those two starting bytes.

Với điều đó ra khỏi con đường, chúng ta hãy xem làm thế nào chúng ta có thể sử dụng Python để kiểm tra 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'] 

Vì vậy, các dữ liệu zlib tạo ra bởi mô-đun zlib Python (sử dụng tùy chọn mặc định) bắt đầu với 78 9c . Chúng tôi sẽ sử dụng điều đó để tạo tập lệnh viết định dạng tệp tùy chỉnh kết hợp phần mở đầu, một số dữ liệu nén zlib và chân trang.

Sau đó, chúng tôi viết tập lệnh thứ hai quét một tệp cho mẫu hai byte đó, bắt đầu giải nén mọi thứ theo sau dòng zlib và tìm ra nơi luồng kết thúc và chân trang bắt đầu.

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) 

Ở đây chúng ta hãy msg, nén nó với zlib, và bao quanh nó với một header và chân trước khi chúng tôi viết nó ra vào một tập tin.

Đầu trang và chân trang có chiều dài cố định trong ví dụ này, nhưng tất nhiên chúng có thể là có độ dài tùy ý, không xác định.

Bây giờ cho tập lệnh tìm cách tìm luồng zlib trong một tệp như vậy. Bởi vì đối với ví dụ này, chúng tôi biết chính xác điểm đánh dấu nào mong đợi tôi chỉ sử dụng một điểm, nhưng rõ ràng là danh sách ZLIB_MARKERS có thể được lấp đầy với tất cả các điểm đánh dấu từ bài đăng được đề cập ở trên.

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

Ý tưởng là thế này:

• Bắt đầu vào lúc bắt đầu của tập tin và tạo ra một hai byte cửa sổ tìm kiếm .

• Di chuyển cửa sổ tìm kiếm tiến lên theo từng bước một byte.

• Đối với mọi kiểm tra cửa sổ nếu nó khớp với bất kỳ dấu hai byte nào, chúng tôi được xác định.

• Nếu tìm thấy kết quả phù hợp, hãy ghi lại vị trí bắt đầu, ngừng tìm kiếm và cố gắng giải nén mọi thứ theo sau.

Bây giờ, việc tìm kết thúc luồng không phải là tầm thường khi tìm hai dấu mốc byte. các luồng zlib không bị kết thúc bởi một chuỗi byte cố định và cũng không phải là độ dài của chúng được chỉ ra trong bất kỳ trường tiêu đề nào. Thay vào đó, nó bị chấm dứt bởi tổng kiểm tra ADLER32 bốn byte phải khớp với dữ liệu đến thời điểm này.

Cách nó hoạt động là chức năng C nội inflate() tục giữ cố gắng để giải nén các dòng như nó đọc nó, và nếu nó đi qua một checksum phù hợp, tín hiệu đó để gọi của nó, chỉ ra rằng phần còn lại của dữ liệu không còn là một phần của luồng zlib nữa.

Trong Python, hành vi này được hiển thị khi sử dụng các đối tượng giải nén thay vì chỉ đơn giản là gọi zlib.decompress(). Gọi số decompress(string) trên đối tượng Decompress sẽ giải nén luồng zlib theo số string và trả về dữ liệu giải nén là một phần của luồng. Mọi thứ theo sau luồng sẽ được lưu trữ trong unused_data và có thể được truy xuất sau đó.

này nên cho kết quả sau trên một tập tin được tạo ra với kịch bản đầu tiên:

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

Ví dụ có thể dễ dàng được sửa đổi để viết tin nhắn không nén vào một tập tin thay vì in nó. Sau đó, bạn có thể phân tích thêm dữ liệu nén zlib trước đây và cố gắng xác định các trường đã biết trong siêu dữ liệu trong đầu trang và chân trang mà bạn đã tách ra.