Added block walker implementations + tests

.gitignore

@@ -5,3 +5,5 @@ __pycache__
 /.cache
 /.pytest_cache
 /.coverage
+
+/tests/blendfiles/cache_ocean/

blender_asset_tracer/blendfile/iterators.py

@@ -1,9 +1,10 @@
 from . import BlendFileBlock
+from .dna import FieldPath
 
 
-def listbase(block: BlendFileBlock) -> BlendFileBlock:
+def listbase(block: BlendFileBlock, next_path: FieldPath=b'next') -> BlendFileBlock:
     """Generator, yields all blocks in the ListBase linked list."""
     while block:
         yield block
-        next_ptr = block[b'next']
+        next_ptr = block[next_path]
         block = block.bfile.find_block_from_address(next_ptr)

blender_asset_tracer/tracer/__init__.py

@@ -2,8 +2,8 @@ import logging
 import pathlib
 import typing
 
-from blender_asset_tracer import bpathlib, blendfile
-from . import result, blocks
+from blender_asset_tracer import blendfile
+from . import result, block_walkers
 
 log = logging.getLogger(__name__)
 
@@ -18,7 +18,7 @@ def deps(bfilepath: pathlib.Path) -> typing.Iterator[result.BlockUsage]:
 
     with blendfile.BlendFile(bfilepath) as bfile:
         for block in asset_holding_blocks(bfile):
-            yield from blocks.from_block(block)
+            yield from block_walkers.from_block(block)
 
         # TODO: handle library blocks for recursion.
 

blender_asset_tracer/tracer/block_walkers.py

@@ -0,0 +1,174 @@
+"""Block walkers.
+
+From a BlendFileBlock, the block walker functions yield BlockUsage objects.
+The top-level block walkers are implemented as _from_block_XX() function,
+where XX is the DNA code of the block.
+"""
+
+import functools
+import logging
+import sys
+import typing
+
+from blender_asset_tracer import blendfile, bpathlib
+from blender_asset_tracer.blendfile import iterators
+from . import result, cdefs
+
+log = logging.getLogger(__name__)
+
+_warned_about_types = set()
+
+
+def from_block(block: blendfile.BlendFileBlock) -> typing.Iterator[result.BlockUsage]:
+    assert block.code != b'DATA'
+
+    module = sys.modules[__name__]
+    funcname = '_from_block_' + block.code.decode().lower()
+    try:
+        block_reader = getattr(module, funcname)
+    except AttributeError:
+        if block.code not in _warned_about_types:
+            log.warning('No reader implemented for block type %r', block.code.decode())
+            _warned_about_types.add(block.code)
+        return
+
+    log.debug('Tracing block %r', block)
+    yield from block_reader(block)
+
+
+def skip_packed(wrapped):
+    """Decorator, skip blocks where 'packedfile' is set to true."""
+
+    @functools.wraps(wrapped)
+    def wrapper(block: blendfile.BlendFileBlock, *args, **kwargs):
+        if block.get(b'packedfile', default=False):
+            return
+
+        yield from wrapped(block, *args, **kwargs)
+
+    return wrapper
+
+
+def _from_block_cf(block: blendfile.BlendFileBlock) -> typing.Iterator[result.BlockUsage]:
+    """Cache file data blocks."""
+    path, field = block.get(b'filepath', return_field=True)
+    yield result.BlockUsage(block, path, path_full_field=field)
+
+
+@skip_packed
+def _from_block_im(block: blendfile.BlendFileBlock) -> typing.Iterator[result.BlockUsage]:
+    """Image data blocks."""
+    # old files miss this
+    image_source = block.get(b'source', default=cdefs.IMA_SRC_FILE)
+    if image_source not in {cdefs.IMA_SRC_FILE, cdefs.IMA_SRC_SEQUENCE, cdefs.IMA_SRC_MOVIE}:
+        return
+
+    pathname, field = block.get(b'name', return_field=True)
+    is_sequence = image_source == cdefs.IMA_SRC_SEQUENCE
+
+    yield result.BlockUsage(block, pathname, is_sequence, path_full_field=field)
+
+
+def _from_block_me(block: blendfile.BlendFileBlock) -> typing.Iterator[result.BlockUsage]:
+    """Mesh data blocks."""
+    block_external = block.get_pointer((b'ldata', b'external'), None)
+    if block_external is None:
+        block_external = block.get_pointer((b'fdata', b'external'), None)
+    if block_external is None:
+        return
+
+    path, field = block_external.get(b'filename', return_field=True)
+    yield result.BlockUsage(block, path, path_full_field=field)
+
+
+def _from_block_mc(block: blendfile.BlendFileBlock) -> typing.Iterator[result.BlockUsage]:
+    """MovieClip data blocks."""
+    path, field = block.get(b'name', return_field=True)
+    # TODO: The assumption that this is not a sequence may not be true for all modifiers.
+    yield result.BlockUsage(block, path, is_sequence=False, path_full_field=field)
+
+
+def _from_block_ob(block: blendfile.BlendFileBlock) -> typing.Iterator[result.BlockUsage]:
+    """Object data blocks."""
+    # 'ob->modifiers[...].filepath'
+    ob_idname = block[b'id', b'name']
+    mods = block.get_pointer((b'modifiers', b'first'))
+    for mod_idx, block_mod in enumerate(iterators.listbase(mods, next_path=(b'modifier', b'next'))):
+        block_name = b'%s.modifiers[%d]' % (ob_idname, mod_idx)
+        log.debug('Tracing modifier %s', block_name.decode())
+
+        mod_type = block_mod[b'modifier', b'type']
+        if mod_type == cdefs.eModifierType_Ocean:
+            if block_mod[b'cached']:
+                path, field = block_mod.get(b'cachepath', return_field=True)
+                # The path indicates the directory containing the cached files.
+                yield result.BlockUsage(block_mod, path, is_sequence=True, path_full_field=field,
+                                        block_name=block_name)
+
+        elif mod_type == cdefs.eModifierType_MeshCache:
+            path, field = block_mod.get(b'filepath', return_field=True)
+            yield result.BlockUsage(block_mod, path, is_sequence=False, path_full_field=field,
+                                    block_name=block_name)
+
+
+def _from_block_sc(block: blendfile.BlendFileBlock) -> typing.Iterator[result.BlockUsage]:
+    """Scene data blocks."""
+    # Sequence editor is the only interesting bit.
+    block_ed = block.get_pointer(b'ed')
+    if block_ed is None:
+        return
+
+    single_asset_types = {cdefs.SEQ_TYPE_MOVIE, cdefs.SEQ_TYPE_SOUND_RAM, cdefs.SEQ_TYPE_SOUND_HD}
+    asset_types = single_asset_types.union({cdefs.SEQ_TYPE_IMAGE})
+
+    def iter_seqbase(seqbase) -> typing.Iterator[result.BlockUsage]:
+        """Generate results from a ListBase of sequencer strips."""
+
+        for seq in iterators.listbase(seqbase):
+            seq.refine_type(b'Sequence')
+            seq_type = seq[b'type']
+
+            if seq_type == cdefs.SEQ_TYPE_META:
+                # Recurse into this meta-sequence.
+                subseq = seq.get_pointer((b'seqbase', b'first'))
+                yield from iter_seqbase(subseq)
+                continue
+
+            if seq_type not in asset_types:
+                continue
+
+            seq_strip = seq.get_pointer(b'strip')
+            if seq_strip is None:
+                continue
+            seq_stripdata = seq_strip.get_pointer(b'stripdata')
+            if seq_stripdata is None:
+                continue
+
+            dirname, dn_field = seq_strip.get(b'dir', return_field=True)
+            basename, bn_field = seq_stripdata.get(b'name', return_field=True)
+            asset_path = bpathlib.BlendPath(dirname) / basename
+
+            is_sequence = seq_type not in single_asset_types
+            yield result.BlockUsage(seq, asset_path,
+                                    is_sequence=is_sequence,
+                                    path_dir_field=dn_field,
+                                    path_base_field=bn_field)
+
+    sbase = block_ed.get_pointer((b'seqbase', b'first'))
+    yield from iter_seqbase(sbase)
+
+
+@skip_packed
+def _from_block_so(block: blendfile.BlendFileBlock) -> typing.Iterator[result.BlockUsage]:
+    """Sound data blocks."""
+    path, field = block.get(b'name', return_field=True)
+    yield result.BlockUsage(block, path, path_full_field=field)
+
+
+@skip_packed
+def _from_block_vf(block: blendfile.BlendFileBlock) -> typing.Iterator[result.BlockUsage]:
+    """Vector Font data blocks."""
+    path, field = block.get(b'name', return_field=True)
+    if path == b'<builtin>':  # builtin font
+        return
+    yield result.BlockUsage(block, path, path_full_field=field)

blender_asset_tracer/tracer/blocks.py

@@ -1,49 +0,0 @@
-import logging
-import sys
-import typing
-
-from blender_asset_tracer import blendfile
-from . import result, cdefs
-
-log = logging.getLogger(__name__)
-
-_warned_about_types = set()
-
-
-class NoReaderImplemented(NotImplementedError):
-    """There is no reader implementation for a specific block code."""
-
-    def __init__(self, message: str, code: bytes):
-        super().__init__(message)
-        self.code = code
-
-
-def from_block(block: blendfile.BlendFileBlock) -> typing.Iterator[result.BlockUsage]:
-    assert block.code != b'DATA'
-
-    module = sys.modules[__name__]
-    funcname = '_from_block_' + block.code.decode().lower()
-    try:
-        block_reader = getattr(module, funcname)
-    except AttributeError:
-        if block.code not in _warned_about_types:
-            log.warning('No reader implemented for block type %r', block.code.decode())
-            _warned_about_types.add(block.code)
-        return
-
-    yield from block_reader(block)
-
-
-def _from_block_im(block: blendfile.BlendFileBlock) -> typing.Iterator[result.BlockUsage]:
-    # old files miss this
-    image_source = block.get(b'source', default=cdefs.IMA_SRC_FILE)
-    if image_source not in {cdefs.IMA_SRC_FILE, cdefs.IMA_SRC_SEQUENCE, cdefs.IMA_SRC_MOVIE}:
-        return
-    if block[b'packedfile']:
-        return
-
-    pathname, field = block.get(b'name', return_field=True)
-
-    # TODO: the receiver should inspect the 'source' property too, and if set
-    # to cdefs.IMA_SRC_SEQUENCE yield the entire sequence of files.
-    yield result.BlockUsage(block, field, pathname)

blender_asset_tracer/tracer/cdefs.py

@@ -34,6 +34,7 @@ IMA_SRC_SEQUENCE = 2
 IMA_SRC_MOVIE = 3
 
 # DNA_modifier_types.h
+eModifierType_Ocean = 39
 eModifierType_MeshCache = 46
 
 # DNA_particle_types.h

blender_asset_tracer/tracer/result.py

@@ -1,35 +1,73 @@
+from blender_asset_tracer import blendfile, bpathlib
+from blender_asset_tracer.blendfile import dna
 
 
 class BlockUsage:
     """Represents the use of an asset by a data block.
 
+    :ivar block_name: an identifying name for this block. Defaults to the ID
+        name of the block.
     :ivar block:
-    :type block: blendfile.BlendFileBlock
-    :ivar field:
-    :type field: dna.Field
     :ivar asset_path:
-    :type asset_path: bpathlib.BlendPath
+    :ivar is_sequence: Indicates whether this file is alone (False), the
+        first of a sequence (True, and the path points to a file), or a
+        directory containing a sequence (True, and path points to a directory).
+        In certain cases such files should be reported once (f.e. when
+        rewriting the source field to another path), and in other cases the
+        sequence should be expanded (f.e. when copying all assets to a BAT
+        Pack).
+    :ivar path_full_field: field containing the full path of this asset.
+    :ivar path_dir_field: field containing the parent path (i.e. the
+        directory) of this asset.
+    :ivar path_base_field: field containing the basename of this asset.
     """
 
-    def __init__(self, block, field, asset_path):
-        self.block_idname = block[b'id', b'name']
-
-        from blender_asset_tracer import blendfile, bpathlib
-        from blender_asset_tracer.blendfile import dna
+    def __init__(self,
+                 block: blendfile.BlendFileBlock,
+                 asset_path: bpathlib.BlendPath,
+                 is_sequence: bool = False,
+                 path_full_field: dna.Field = None,
+                 path_dir_field: dna.Field = None,
+                 path_base_field: dna.Field = None,
+                 block_name: bytes = '',
+                 ):
+        if block_name:
+            self.block_name = block_name
+        else:
+            try:
+                self.block_name = block[b'id', b'name']
+            except KeyError:
+                try:
+                    self.block_name = block[b'name']
+                except KeyError:
+                    self.block_name = b'-unnamed-'
 
         assert isinstance(block, blendfile.BlendFileBlock)
-        assert isinstance(field, dna.Field), 'field should be dna.Field, not %r' % type(field)
         assert isinstance(asset_path, (bytes, bpathlib.BlendPath)), \
             'asset_path should be BlendPath, not %r' % type(asset_path)
 
+        if path_full_field is None:
+            assert isinstance(path_dir_field, dna.Field), \
+                'path_dir_field should be dna.Field, not %r' % type(path_dir_field)
+            assert isinstance(path_base_field, dna.Field), \
+                'path_base_field should be dna.Field, not %r' % type(path_base_field)
+        else:
+            assert isinstance(path_full_field, dna.Field), \
+                'path_full_field should be dna.Field, not %r' % type(path_full_field)
+
         if isinstance(asset_path, bytes):
             asset_path = bpathlib.BlendPath(asset_path)
 
         self.block = block
-        self.field = field
         self.asset_path = asset_path
+        self.is_sequence = bool(is_sequence)
+        self.path_full_field = path_full_field
+        self.path_dir_field = path_dir_field
+        self.path_base_field = path_base_field
 
     def __repr__(self):
-        return '<BlockUsage name=%r type=%r field=%r asset=%r>' % (
-            self.block_idname, self.block.dna_type_name,
-            self.field.name.name_full.decode(), self.asset_path)
+        return '<BlockUsage name=%r type=%r field=%r asset=%r%s>' % (
+            self.block_name, self.block.dna_type_name,
+            self.path_full_field.name.name_full.decode(), self.asset_path,
+            ' sequence' if self.is_sequence else ''
+        )

tests/abstract_test.py

@@ -1,3 +1,4 @@
+import logging
 import pathlib
 import unittest
 
@@ -6,6 +7,7 @@ class AbstractBlendFileTest(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.blendfiles = pathlib.Path(__file__).with_name('blendfiles')
+        logging.basicConfig(level=logging.INFO)
 
     def setUp(self):
         self.bf = None

tests/blendfiles/imgseq/LICENSE.txt

@@ -0,0 +1,3 @@
+License: CC-BY
+Created by: Blender Animation Studio
+            https://cloud.blender.org/p/dailydweebs/

tests/test_tracer.py

@@ -1,15 +1,19 @@
+import collections
 import logging
 
+from blender_asset_tracer import tracer, blendfile
 from abstract_test import AbstractBlendFileTest
 
-from blender_asset_tracer import tracer, blendfile
+# Mimicks a BlockUsage, but without having to set the block to an expected value.
+Expect = collections.namedtuple(
+    'Expect',
+    'type full_field dirname_field basename_field asset_path is_sequence')
 
 
 class AbstractTracerTest(AbstractBlendFileTest):
     @classmethod
     def setUpClass(cls):
         super().setUpClass()
-        logging.basicConfig(level=logging.INFO)
         logging.getLogger('blender_asset_tracer.tracer').setLevel(logging.DEBUG)
 
 
@@ -56,11 +60,97 @@ class AssetHoldingBlocksTest(AbstractTracerTest):
         self.assertEqual(37, blocks_seen)
 
 
-class DepsTest(AbstractBlendFileTest):
+class DepsTest(AbstractTracerTest):
+
+    def assert_deps(self, blend_fname, expects):
+        for dep in tracer.deps(self.blendfiles / blend_fname):
+            exp = expects[dep.block_name]
+            self.assertEqual(exp.type, dep.block.dna_type.dna_type_id.decode())
+            self.assertEqual(exp.asset_path, dep.asset_path)
+            self.assertEqual(exp.is_sequence, dep.is_sequence)
+
+            if exp.full_field is not None:
+                self.assertEqual(exp.full_field, dep.path_full_field.name.name_full.decode())
+            if exp.dirname_field is not None:
+                self.assertEqual(exp.dirname_field, dep.path_dir_field.name.name_full.decode())
+            if exp.basename_field is not None:
+                self.assertEqual(exp.basename_field, dep.path_base_field.name.name_full.decode())
+
+            del expects[dep.block_name]  # should be seen only once
+        # All expected uses should have been seen.
+        self.assertEqual({}, expects)
+
     def test_no_deps(self):
-        for dep in tracer.deps(self.blendfiles / 'basic_file.blend'):
-            self.fail(dep)
+        self.assert_deps('basic_file.blend', {})
 
     def test_ob_mat_texture(self):
-        for dep in tracer.deps(self.blendfiles / 'material_textures.blend'):
-            self.fail(repr(dep))
+        expects = {
+            b'IMbrick_dotted_04-bump': Expect(
+                'Image', 'name[1024]', None, None,
+                b'//textures/Bricks/brick_dotted_04-bump.jpg', False),
+            b'IMbrick_dotted_04-color': Expect(
+                'Image', 'name[1024]', None, None,
+                b'//textures/Bricks/brick_dotted_04-color.jpg', False),
+            # This data block is in there, but the image is packed, so it
+            # shouldn't be in the results.
+            # b'IMbrick_dotted_04-specular': Expect(
+            #     'Image', 'name[1024]', None, None,
+            #     b'//textures/Bricks/brick_dotted_04-specular.jpg', False),
+            b'IMbuildings_roof_04-color': Expect(
+                'Image', 'name[1024]', None, None,
+                b'//textures/Textures/Buildings/buildings_roof_04-color.png', False),
+        }
+        self.assert_deps('material_textures.blend', expects)
+
+    def test_seq_image_sequence(self):
+        expects = {
+            b'SQ000210.png': Expect(
+                'Sequence', None, 'dir[768]', 'name[256]', b'//imgseq/000210.png', True),
+            b'SQvideo-tiny.mkv': Expect(
+                'Sequence', None, 'dir[768]', 'name[256]',
+                b'//../../../../cloud/pillar/testfiles/video-tiny.mkv', False),
+
+            # The sound will be referenced twice, from the sequence strip and an SO data block.
+            b'SQvideo-tiny.001': Expect(
+                'Sequence', None, 'dir[768]', 'name[256]',
+                b'//../../../../cloud/pillar/testfiles/video-tiny.mkv', False),
+            b'SOvideo-tiny.mkv': Expect(
+                'bSound', 'name[1024]', None, None,
+                b'//../../../../cloud/pillar/testfiles/video-tiny.mkv', False),
+        }
+        self.assert_deps('image_sequencer.blend', expects)
+
+    def test_block_cf(self):
+        self.assert_deps('alembic-user.blend', {
+            b'CFclothsim.abc': Expect('CacheFile', 'filepath[1024]', None, None,
+                                      b'//clothsim.abc', False),
+        })
+
+    def test_block_mc(self):
+        self.assert_deps('movieclip.blend', {
+            b'MCvideo.mov': Expect('MovieClip', 'name[1024]', None, None,
+                                   b'//../../../../cloud/pillar/testfiles/video.mov', False),
+        })
+
+    def test_block_me(self):
+        self.assert_deps('multires_external.blend', {
+            b'MECube': Expect('Mesh', 'filename[1024]', None, None, b'//Cube.btx', False),
+        })
+
+    def test_ocean(self):
+        self.assert_deps('ocean_modifier.blend', {
+            b'OBPlane.modifiers[0]': Expect('OceanModifierData', 'cachepath[1024]', None, None,
+                                            b'//cache_ocean', True),
+        })
+
+    def test_mesh_cache(self):
+        self.assert_deps('meshcache-user.blend', {
+            b'OBPlane.modifiers[0]': Expect('MeshCacheModifierData', 'filepath[1024]', None, None,
+                                            b'//meshcache.mdd', False),
+        })
+
+    def test_block_vf(self):
+        self.assert_deps('with_font.blend', {
+            b'VFHack-Bold': Expect('VFont', 'name[1024]', None, None,
+                                   b'/usr/share/fonts/truetype/hack/Hack-Bold.ttf', False),
+        })