btrfs incremental snapshots: find UUID in sent data

starting from the code from btrfs-snapshots-diff.py [github.com] is was able to make a script tailored to my needs. i can use it this way to get the uuids:

with BtrfsStream('snapshot_0-1.data') as btrfs_stream:
    print(btrfs_stream.get_send_command())
# ('BTRFS_SEND_C_SNAPSHOT', 
#      (UUID('01234567-89ab-cdef-0123-456789abcdef'), 
#       UUID('fedcba98-7654-3210-fedc-ba9876543210')))

with the class BtrfsStream as below.

i made some modifications to the original code:

• python 3 (instead of python 2)
• iterating over the file instead of reading all of it into the memory
• added contextmanager features in order to use it in a with satement

the code is used is then:

from struct import unpack
import io
from uuid import UUID

class BtrfsStream:

    # From btrfs/send.h
    send_cmds = (
        'BTRFS_SEND_C_UNSPEC BTRFS_SEND_C_SUBVOL BTRFS_SEND_C_SNAPSHOT '
        'BTRFS_SEND_C_MKFILE BTRFS_SEND_C_MKDIR BTRFS_SEND_C_MKNOD '
        'BTRFS_SEND_C_MKFIFO BTRFS_SEND_C_MKSOCK BTRFS_SEND_C_SYMLINK '
        'BTRFS_SEND_C_RENAME BTRFS_SEND_C_LINK BTRFS_SEND_C_UNLINK '
        'BTRFS_SEND_C_RMDIR BTRFS_SEND_C_SET_XATTR BTRFS_SEND_C_REMOVE_XATTR '
        'BTRFS_SEND_C_WRITE BTRFS_SEND_C_CLONE BTRFS_SEND_C_TRUNCATE '
        'BTRFS_SEND_C_CHMOD BTRFS_SEND_C_CHOWN BTRFS_SEND_C_UTIMES '
        'BTRFS_SEND_C_END BTRFS_SEND_C_UPDATE_EXTENT').split()

    send_attrs = (
        'BTRFS_SEND_A_UNSPEC BTRFS_SEND_A_UUID BTRFS_SEND_A_CTRANSID '
        'BTRFS_SEND_A_INO BTRFS_SEND_A_SIZE BTRFS_SEND_A_MODE '
        'BTRFS_SEND_A_UID BTRFS_SEND_A_GID BTRFS_SEND_A_RDEV '
        'BTRFS_SEND_A_CTIME BTRFS_SEND_A_MTIME BTRFS_SEND_A_ATIME '
        'BTRFS_SEND_A_OTIME BTRFS_SEND_A_XATTR_NAME '
        'BTRFS_SEND_A_XATTR_DATA BTRFS_SEND_A_PATH BTRFS_SEND_A_PATH_TO '
        'BTRFS_SEND_A_PATH_LINK BTRFS_SEND_A_FILE_OFFSET BTRFS_SEND_A_DATA '
        'BTRFS_SEND_A_CLONE_UUID BTRFS_SEND_A_CLONE_CTRANSID '
        'BTRFS_SEND_A_CLONE_PATH BTRFS_SEND_A_CLONE_OFFSET '
        'BTRFS_SEND_A_CLONE_LEN').split()

    # From btrfs/ioctl.h:#define BTRFS_UUID_SIZE 16
    BTRFS_UUID_SIZE = 16
    HEADER_SIZE = 17

    # Headers length
    l_head = 10
    l_tlv = 4

    def __init__(self, path):
        '''
        '''

        self.path = path
        self._stream = None

    def __enter__(self):
        '''
        enter for context manager
        '''

        self._stream = open(self.path, 'rb')
        self._read_header()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        '''
        exit for context manager
        '''

        self._stream.close()

    def _read_header(self):
        '''
        read the header
        '''

        header = self.read(BtrfsStream.HEADER_SIZE, assert_lengh=False)

        if len(header) < BtrfsStream.HEADER_SIZE:
            raise IOError('Invalid stream length\n')

        magic, null, self.version = unpack('<12scI', header)
        if magic != b'btrfs-stream':
            raise IOError('Not a Btrfs stream!')

    def seek(self, offset, whence=io.SEEK_SET):
        '''
        seek to a given point
        '''

        self._stream.seek(offset)

    def tell(self):
        '''
        tell where we are
        '''

        return self._stream.tell()

    def read(self, n_bytes, assert_lengh=True):
        '''
        try to read n_bytes
        '''

        tell_before = self.tell()
        btes = self._stream.read(n_bytes)

        if assert_lengh is True and len(btes) != n_bytes:
            msg = ('could only read {} instead of {} at offset {}'
                   ).format(len(btes), n_bytes, tell_before)
            raise IOError(msg)
        return btes

    def tlv_get(self, attr_type):
        attr, l_attr = unpack('<HH', self.read(BtrfsStream.l_tlv))
        if self.send_attrs[attr] != attr_type:
            raise ValueError('Unexpected attribute %s' % self.send_attrs[attr])
        ret, = unpack('<H', self.read(2))
        return ret

    def _tlv_get_string(self, attr_type):
        attr, l_attr = unpack('<HH', self.read(BtrfsStream.l_tlv))
        if self.send_attrs[attr] != attr_type:
            raise ValueError('Unexpected attribute %s' % self.send_attrs[attr])
        ret, = unpack('<%ds' % l_attr, self.read(l_attr))
        return ret

    def _tlv_get_u64(self, attr_type):
        attr, l_attr = unpack('<HH', self.read(BtrfsStream.l_tlv))
        if self.send_attrs[attr] != attr_type:
            raise ValueError('Unexpected attribute %s' % self.send_attrs[attr])
        ret, = unpack('<Q', self.read(l_attr))
        return ret

    def _tlv_get_uuid(self, attr_type):
        attr, l_attr = unpack('<HH', self.read(BtrfsStream.l_tlv))
        if self.send_attrs[attr] != attr_type:
            raise ValueError('Unexpected attribute %s' % self.send_attrs[attr])
        return UUID(bytes=self.read(l_attr))

    def get_send_command(self):
        '''
        search uuids only.
        '''
        # start at the right point in the file
        self.seek(BtrfsStream.HEADER_SIZE)

        while True:

            l_cmd, cmd, crc = unpack('<IHI', self.read(BtrfsStream.l_head))
            tell_before_cmd = self.tell()

            try:
                command = self.send_cmds[cmd]
            except:
                raise ValueError('Unkown command %d' % cmd)

            if command == 'BTRFS_SEND_C_SNAPSHOT':
                self._tlv_get_string('BTRFS_SEND_A_PATH')
                uuid = self._tlv_get_uuid('BTRFS_SEND_A_UUID')
                self._tlv_get_u64('BTRFS_SEND_A_CTRANSID')
                clone_uuid = self._tlv_get_uuid('BTRFS_SEND_A_CLONE_UUID')
                return 'BTRFS_SEND_C_SNAPSHOT', (uuid, clone_uuid)

            elif command == 'BTRFS_SEND_C_SUBVOL':
                self._tlv_get_string('BTRFS_SEND_A_PATH')
                uuid = self._tlv_get_uuid('BTRFS_SEND_A_UUID')
                return 'BTRFS_SEND_C_SUBVOL', (uuid, )

            elif command == 'BTRFS_SEND_C_CLONE':
                self._tlv_get_string('BTRFS_SEND_A_PATH')
                self._tlv_get_u64('BTRFS_SEND_A_FILE_OFFSET')
                self._tlv_get_u64('BTRFS_SEND_A_CLONE_LEN')
                clone_uuid = self._tlv_get_uuid('BTRFS_SEND_A_CLONE_UUID')
                return 'BTRFS_SEND_C_CLONE', (clone_uuid, )

            elif command == 'BTRFS_SEND_C_END':
                return

            self.seek(tell_before_cmd + l_cmd)