Archived
initial commit
This commit is contained in:
Pullusb
@@ -0,0 +1,773 @@
|
||||
# ***** BEGIN GPL LICENSE BLOCK *****
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or
|
||||
# modify it under the terms of the GNU General Public License
|
||||
# as published by the Free Software Foundation; either version 2
|
||||
# of the License, or (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software Foundation,
|
||||
# Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
|
||||
#
|
||||
# ***** END GPL LICENCE BLOCK *****
|
||||
#
|
||||
# (c) 2009, At Mind B.V. - Jeroen Bakker
|
||||
# (c) 2014, Blender Foundation - Campbell Barton
|
||||
# (c) 2018, Blender Foundation - Sybren A. Stüvel
|
||||
|
||||
import atexit
|
||||
import collections
|
||||
import functools
|
||||
import gzip
|
||||
import logging
|
||||
import os
|
||||
import struct
|
||||
import pathlib
|
||||
import shutil
|
||||
import tempfile
|
||||
import typing
|
||||
|
||||
from . import exceptions, dna_io, dna, header
|
||||
from blender_asset_tracer import bpathlib
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
FILE_BUFFER_SIZE = 1024 * 1024
|
||||
BLENDFILE_MAGIC = b'BLENDER'
|
||||
GZIP_MAGIC = b'\x1f\x8b'
|
||||
BFBList = typing.List['BlendFileBlock']
|
||||
|
||||
_cached_bfiles = {} # type: typing.Dict[pathlib.Path, BlendFile]
|
||||
|
||||
|
||||
def open_cached(path: pathlib.Path, mode='rb',
|
||||
assert_cached: typing.Optional[bool] = None) -> 'BlendFile':
|
||||
"""Open a blend file, ensuring it is only opened once."""
|
||||
my_log = log.getChild('open_cached')
|
||||
bfile_path = bpathlib.make_absolute(path)
|
||||
|
||||
if assert_cached is not None:
|
||||
is_cached = bfile_path in _cached_bfiles
|
||||
if assert_cached and not is_cached:
|
||||
raise AssertionError('File %s was not cached' % bfile_path)
|
||||
elif not assert_cached and is_cached:
|
||||
raise AssertionError('File %s was cached' % bfile_path)
|
||||
|
||||
try:
|
||||
bfile = _cached_bfiles[bfile_path]
|
||||
except KeyError:
|
||||
my_log.debug('Opening non-cached %s', path)
|
||||
bfile = BlendFile(path, mode=mode)
|
||||
_cached_bfiles[bfile_path] = bfile
|
||||
else:
|
||||
my_log.debug('Returning cached %s', path)
|
||||
|
||||
return bfile
|
||||
|
||||
|
||||
@atexit.register
|
||||
def close_all_cached() -> None:
|
||||
if not _cached_bfiles:
|
||||
# Don't even log anything when there is nothing to close
|
||||
return
|
||||
|
||||
log.debug('Closing %d cached blend files', len(_cached_bfiles))
|
||||
for bfile in list(_cached_bfiles.values()):
|
||||
bfile.close()
|
||||
_cached_bfiles.clear()
|
||||
|
||||
|
||||
def _cache(path: pathlib.Path, bfile: 'BlendFile'):
|
||||
"""Add a BlendFile to the cache."""
|
||||
bfile_path = bpathlib.make_absolute(path)
|
||||
_cached_bfiles[bfile_path] = bfile
|
||||
|
||||
|
||||
def _uncache(path: pathlib.Path):
|
||||
"""Remove a BlendFile object from the cache."""
|
||||
bfile_path = bpathlib.make_absolute(path)
|
||||
_cached_bfiles.pop(bfile_path, None)
|
||||
|
||||
|
||||
class BlendFile:
|
||||
"""Representation of a blend file.
|
||||
|
||||
:ivar filepath: which file this object represents.
|
||||
:ivar raw_filepath: which file is accessed; same as filepath for
|
||||
uncompressed files, but a temporary file for compressed files.
|
||||
:ivar fileobj: the file object that's being accessed.
|
||||
"""
|
||||
log = log.getChild('BlendFile')
|
||||
|
||||
def __init__(self, path: pathlib.Path, mode='rb') -> None:
|
||||
"""Create a BlendFile instance for the blend file at the path.
|
||||
|
||||
Opens the file for reading or writing pending on the access. Compressed
|
||||
blend files are uncompressed to a temporary location before opening.
|
||||
|
||||
:param path: the file to open
|
||||
:param mode: see mode description of pathlib.Path.open()
|
||||
"""
|
||||
self.filepath = path
|
||||
self.raw_filepath = path
|
||||
self._is_modified = False
|
||||
self.fileobj = self._open_file(path, mode)
|
||||
|
||||
self.blocks = [] # type: BFBList
|
||||
"""BlendFileBlocks of this file, in disk order."""
|
||||
|
||||
self.code_index = collections.defaultdict(list) # type: typing.Dict[bytes, BFBList]
|
||||
self.structs = [] # type: typing.List[dna.Struct]
|
||||
self.sdna_index_from_id = {} # type: typing.Dict[bytes, int]
|
||||
self.block_from_addr = {} # type: typing.Dict[int, BlendFileBlock]
|
||||
|
||||
self.header = header.BlendFileHeader(self.fileobj, self.raw_filepath)
|
||||
self.block_header_struct = self.header.create_block_header_struct()
|
||||
self._load_blocks()
|
||||
|
||||
def _open_file(self, path: pathlib.Path, mode: str) -> typing.IO[bytes]:
|
||||
"""Open a blend file, decompressing if necessary.
|
||||
|
||||
This does not parse the blend file yet, just makes sure that
|
||||
self.fileobj is opened and that self.filepath and self.raw_filepath
|
||||
are set.
|
||||
|
||||
:raises exceptions.BlendFileError: when the blend file doesn't have the
|
||||
correct magic bytes.
|
||||
"""
|
||||
|
||||
if 'b' not in mode:
|
||||
raise ValueError('Only binary modes are supported, not %r' % mode)
|
||||
|
||||
self.filepath = path
|
||||
|
||||
fileobj = path.open(mode, buffering=FILE_BUFFER_SIZE) # typing.IO[bytes]
|
||||
fileobj.seek(0, os.SEEK_SET)
|
||||
|
||||
magic = fileobj.read(len(BLENDFILE_MAGIC))
|
||||
if magic == BLENDFILE_MAGIC:
|
||||
self.is_compressed = False
|
||||
self.raw_filepath = path
|
||||
return fileobj
|
||||
|
||||
if magic[:2] == GZIP_MAGIC:
|
||||
self.is_compressed = True
|
||||
|
||||
log.debug("compressed blendfile detected: %s", path)
|
||||
# Decompress to a temporary file.
|
||||
tmpfile = tempfile.NamedTemporaryFile()
|
||||
fileobj.seek(0, os.SEEK_SET)
|
||||
with gzip.GzipFile(fileobj=fileobj, mode=mode) as gzfile:
|
||||
magic = gzfile.read(len(BLENDFILE_MAGIC))
|
||||
if magic != BLENDFILE_MAGIC:
|
||||
raise exceptions.BlendFileError("Compressed file is not a blend file", path)
|
||||
|
||||
data = magic
|
||||
while data:
|
||||
tmpfile.write(data)
|
||||
data = gzfile.read(FILE_BUFFER_SIZE)
|
||||
|
||||
# Further interaction should be done with the uncompressed file.
|
||||
self.raw_filepath = pathlib.Path(tmpfile.name)
|
||||
fileobj.close()
|
||||
return tmpfile
|
||||
|
||||
fileobj.close()
|
||||
raise exceptions.BlendFileError("File is not a blend file", path)
|
||||
|
||||
def _load_blocks(self) -> None:
|
||||
"""Read the blend file to load its DNA structure to memory."""
|
||||
|
||||
self.structs.clear()
|
||||
self.sdna_index_from_id.clear()
|
||||
while True:
|
||||
block = BlendFileBlock(self)
|
||||
if block.code == b'ENDB':
|
||||
break
|
||||
|
||||
if block.code == b'DNA1':
|
||||
self.decode_structs(block)
|
||||
else:
|
||||
self.fileobj.seek(block.size, os.SEEK_CUR)
|
||||
|
||||
self.blocks.append(block)
|
||||
self.code_index[block.code].append(block)
|
||||
self.block_from_addr[block.addr_old] = block
|
||||
|
||||
if not self.structs:
|
||||
raise exceptions.NoDNA1Block("No DNA1 block in file, not a valid .blend file",
|
||||
self.filepath)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
clsname = self.__class__.__qualname__
|
||||
if self.filepath == self.raw_filepath:
|
||||
return '<%s %r>' % (clsname, self.filepath)
|
||||
return '<%s %r reading from %r>' % (clsname, self.filepath, self.raw_filepath)
|
||||
|
||||
def __enter__(self) -> 'BlendFile':
|
||||
return self
|
||||
|
||||
def __exit__(self, exctype, excvalue, traceback) -> None:
|
||||
self.close()
|
||||
|
||||
def copy_and_rebind(self, path: pathlib.Path, mode='rb') -> None:
|
||||
"""Change which file is bound to this BlendFile.
|
||||
|
||||
This allows cloning a previously opened file, and rebinding it to reuse
|
||||
the already-loaded DNA structs and data blocks.
|
||||
"""
|
||||
log.debug('Rebinding %r to %s', self, path)
|
||||
|
||||
self.close()
|
||||
_uncache(self.filepath)
|
||||
|
||||
self.log.debug('Copying %s to %s', self.filepath, path)
|
||||
# TODO(Sybren): remove str() calls when targeting Python 3.6+
|
||||
shutil.copy(str(self.filepath), str(path))
|
||||
|
||||
self.fileobj = self._open_file(path, mode=mode)
|
||||
_cache(path, self)
|
||||
|
||||
@property
|
||||
def is_modified(self) -> bool:
|
||||
return self._is_modified
|
||||
|
||||
def mark_modified(self) -> None:
|
||||
"""Recompess the file when it is closed."""
|
||||
self.log.debug('Marking %s as modified', self.raw_filepath)
|
||||
self._is_modified = True
|
||||
|
||||
def find_blocks_from_code(self, code: bytes) -> typing.List['BlendFileBlock']:
|
||||
assert isinstance(code, bytes)
|
||||
return self.code_index[code]
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close the blend file.
|
||||
|
||||
Recompresses the blend file if it was compressed and changed.
|
||||
"""
|
||||
if not self.fileobj:
|
||||
return
|
||||
|
||||
if self._is_modified:
|
||||
log.debug('closing blend file %s after it was modified', self.raw_filepath)
|
||||
|
||||
if self._is_modified and self.is_compressed:
|
||||
log.debug("recompressing modified blend file %s", self.raw_filepath)
|
||||
self.fileobj.seek(os.SEEK_SET, 0)
|
||||
|
||||
with gzip.open(str(self.filepath), 'wb') as gzfile:
|
||||
while True:
|
||||
data = self.fileobj.read(FILE_BUFFER_SIZE)
|
||||
if not data:
|
||||
break
|
||||
gzfile.write(data)
|
||||
log.debug("compressing to %s finished", self.filepath)
|
||||
|
||||
# Close the file object after recompressing, as it may be a temporary
|
||||
# file that'll disappear as soon as we close it.
|
||||
self.fileobj.close()
|
||||
self._is_modified = False
|
||||
|
||||
try:
|
||||
del _cached_bfiles[self.filepath]
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
def ensure_subtype_smaller(self, sdna_index_curr, sdna_index_next) -> None:
|
||||
# never refine to a smaller type
|
||||
curr_struct = self.structs[sdna_index_curr]
|
||||
next_struct = self.structs[sdna_index_next]
|
||||
if curr_struct.size > next_struct.size:
|
||||
raise RuntimeError("Can't refine to smaller type (%s -> %s)" %
|
||||
(curr_struct.dna_type_id.decode('utf-8'),
|
||||
next_struct.dna_type_id.decode('utf-8')))
|
||||
|
||||
def decode_structs(self, block: 'BlendFileBlock'):
|
||||
"""
|
||||
DNACatalog is a catalog of all information in the DNA1 file-block
|
||||
"""
|
||||
self.log.debug("building DNA catalog")
|
||||
|
||||
# Get some names in the local scope for faster access.
|
||||
structs = self.structs
|
||||
sdna_index_from_id = self.sdna_index_from_id
|
||||
endian = self.header.endian
|
||||
shortstruct = endian.USHORT
|
||||
shortstruct2 = endian.USHORT2
|
||||
intstruct = endian.UINT
|
||||
assert intstruct.size == 4
|
||||
|
||||
def pad_up_4(off: int) -> int:
|
||||
return (off + 3) & ~3
|
||||
|
||||
data = self.fileobj.read(block.size)
|
||||
types = []
|
||||
typenames = []
|
||||
|
||||
offset = 8
|
||||
names_len = intstruct.unpack_from(data, offset)[0]
|
||||
offset += 4
|
||||
|
||||
self.log.debug("building #%d names" % names_len)
|
||||
for _ in range(names_len):
|
||||
typename = endian.read_data0_offset(data, offset)
|
||||
offset = offset + len(typename) + 1
|
||||
typenames.append(dna.Name(typename))
|
||||
|
||||
offset = pad_up_4(offset)
|
||||
offset += 4
|
||||
types_len = intstruct.unpack_from(data, offset)[0]
|
||||
offset += 4
|
||||
self.log.debug("building #%d types" % types_len)
|
||||
for _ in range(types_len):
|
||||
dna_type_id = endian.read_data0_offset(data, offset)
|
||||
types.append(dna.Struct(dna_type_id))
|
||||
offset += len(dna_type_id) + 1
|
||||
|
||||
offset = pad_up_4(offset)
|
||||
offset += 4
|
||||
self.log.debug("building #%d type-lengths" % types_len)
|
||||
for i in range(types_len):
|
||||
typelen = shortstruct.unpack_from(data, offset)[0]
|
||||
offset = offset + 2
|
||||
types[i].size = typelen
|
||||
|
||||
offset = pad_up_4(offset)
|
||||
offset += 4
|
||||
|
||||
structs_len = intstruct.unpack_from(data, offset)[0]
|
||||
offset += 4
|
||||
log.debug("building #%d structures" % structs_len)
|
||||
pointer_size = self.header.pointer_size
|
||||
for sdna_index in range(structs_len):
|
||||
struct_type_index, fields_len = shortstruct2.unpack_from(data, offset)
|
||||
offset += 4
|
||||
|
||||
dna_struct = types[struct_type_index]
|
||||
sdna_index_from_id[dna_struct.dna_type_id] = sdna_index
|
||||
structs.append(dna_struct)
|
||||
|
||||
dna_offset = 0
|
||||
|
||||
for field_index in range(fields_len):
|
||||
field_type_index, field_name_index = shortstruct2.unpack_from(data, offset)
|
||||
offset += 4
|
||||
|
||||
dna_type = types[field_type_index]
|
||||
dna_name = typenames[field_name_index]
|
||||
|
||||
if dna_name.is_pointer or dna_name.is_method_pointer:
|
||||
dna_size = pointer_size * dna_name.array_size
|
||||
else:
|
||||
dna_size = dna_type.size * dna_name.array_size
|
||||
|
||||
field = dna.Field(dna_type, dna_name, dna_size, dna_offset)
|
||||
dna_struct.append_field(field)
|
||||
dna_offset += dna_size
|
||||
|
||||
def abspath(self, relpath: bpathlib.BlendPath) -> bpathlib.BlendPath:
|
||||
"""Construct an absolute path from a blendfile-relative path."""
|
||||
|
||||
if relpath.is_absolute():
|
||||
return relpath
|
||||
|
||||
bfile_dir = self.filepath.absolute().parent
|
||||
root = bpathlib.BlendPath(bfile_dir)
|
||||
abspath = relpath.absolute(root)
|
||||
|
||||
my_log = self.log.getChild('abspath')
|
||||
my_log.debug('Resolved %s relative to %s to %s', relpath, self.filepath, abspath)
|
||||
|
||||
return abspath
|
||||
|
||||
def dereference_pointer(self, address: int) -> 'BlendFileBlock':
|
||||
"""Return the pointed-to block, or raise SegmentationFault."""
|
||||
|
||||
try:
|
||||
return self.block_from_addr[address]
|
||||
except KeyError:
|
||||
raise exceptions.SegmentationFault('address does not exist', address) from None
|
||||
|
||||
def struct(self, name: bytes) -> dna.Struct:
|
||||
index = self.sdna_index_from_id[name]
|
||||
return self.structs[index]
|
||||
|
||||
|
||||
@functools.total_ordering
|
||||
class BlendFileBlock:
|
||||
"""
|
||||
Instance of a struct.
|
||||
"""
|
||||
|
||||
# Due to the huge number of BlendFileBlock objects created for packing a
|
||||
# production-size blend file, using slots here actually makes the
|
||||
# dependency tracer significantly (p<0.001) faster. In my test case the
|
||||
# speed improvement was 16% for a 'bam list' command.
|
||||
__slots__ = (
|
||||
'bfile', 'code', 'size', 'addr_old', 'sdna_index',
|
||||
'count', 'file_offset', 'endian', '_id_name',
|
||||
)
|
||||
|
||||
log = log.getChild('BlendFileBlock')
|
||||
old_structure = struct.Struct(b'4sI')
|
||||
"""old blend files ENDB block structure"""
|
||||
|
||||
def __init__(self, bfile: BlendFile) -> None:
|
||||
self.bfile = bfile
|
||||
|
||||
# Defaults; actual values are set by interpreting the block header.
|
||||
self.code = b''
|
||||
self.size = 0
|
||||
self.addr_old = 0
|
||||
self.sdna_index = 0
|
||||
self.count = 0
|
||||
self.file_offset = 0
|
||||
"""Offset in bytes from start of file to beginning of the data block.
|
||||
|
||||
Points to the data after the block header.
|
||||
"""
|
||||
self.endian = bfile.header.endian
|
||||
self._id_name = ... # type: typing.Union[None, ellipsis, bytes]
|
||||
|
||||
header_struct = bfile.block_header_struct
|
||||
data = bfile.fileobj.read(header_struct.size)
|
||||
if len(data) != header_struct.size:
|
||||
self.log.warning("Blend file %s seems to be truncated, "
|
||||
"expected %d bytes but could read only %d",
|
||||
bfile.filepath, header_struct.size, len(data))
|
||||
self.code = b'ENDB'
|
||||
return
|
||||
|
||||
# header size can be 8, 20, or 24 bytes long
|
||||
# 8: old blend files ENDB block (exception)
|
||||
# 20: normal headers 32 bit platform
|
||||
# 24: normal headers 64 bit platform
|
||||
if len(data) <= 15:
|
||||
self.log.debug('interpreting block as old-style ENB block')
|
||||
blockheader = self.old_structure.unpack(data)
|
||||
self.code = self.endian.read_data0(blockheader[0])
|
||||
return
|
||||
|
||||
blockheader = header_struct.unpack(data)
|
||||
self.code = self.endian.read_data0(blockheader[0])
|
||||
if self.code != b'ENDB':
|
||||
self.size = blockheader[1]
|
||||
self.addr_old = blockheader[2]
|
||||
self.sdna_index = blockheader[3]
|
||||
self.count = blockheader[4]
|
||||
self.file_offset = bfile.fileobj.tell()
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return "<%s.%s (%s), size=%d at %s>" % (
|
||||
self.__class__.__name__,
|
||||
self.dna_type_name,
|
||||
self.code.decode(),
|
||||
self.size,
|
||||
hex(self.addr_old),
|
||||
)
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash((self.code, self.addr_old, self.bfile.filepath))
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if not isinstance(other, BlendFileBlock):
|
||||
return False
|
||||
return (self.code == other.code and
|
||||
self.addr_old == other.addr_old and
|
||||
self.bfile.filepath == other.bfile.filepath)
|
||||
|
||||
def __lt__(self, other: 'BlendFileBlock') -> bool:
|
||||
"""Order blocks by file path and offset within that file."""
|
||||
if not isinstance(other, BlendFileBlock):
|
||||
raise NotImplemented()
|
||||
my_key = self.bfile.filepath, self.file_offset
|
||||
other_key = other.bfile.filepath, other.file_offset
|
||||
return my_key < other_key
|
||||
|
||||
def __bool__(self) -> bool:
|
||||
"""Data blocks are always True."""
|
||||
return True
|
||||
|
||||
@property
|
||||
def dna_type(self) -> dna.Struct:
|
||||
return self.bfile.structs[self.sdna_index]
|
||||
|
||||
@property
|
||||
def dna_type_id(self) -> bytes:
|
||||
return self.dna_type.dna_type_id
|
||||
|
||||
@property
|
||||
def dna_type_name(self) -> str:
|
||||
return self.dna_type_id.decode('ascii')
|
||||
|
||||
@property
|
||||
def id_name(self) -> typing.Optional[bytes]:
|
||||
"""Same as block[b'id', b'name']; None if there is no such field.
|
||||
|
||||
Evaluated only once, so safe to call multiple times without producing
|
||||
excessive disk I/O.
|
||||
"""
|
||||
if self._id_name is ...:
|
||||
try:
|
||||
self._id_name = self[b'id', b'name']
|
||||
except KeyError:
|
||||
self._id_name = None
|
||||
|
||||
# TODO(Sybren): figure out how to let mypy know self._id_name cannot
|
||||
# be ellipsis at this point.
|
||||
return self._id_name # type: ignore
|
||||
|
||||
def refine_type_from_index(self, sdna_index: int):
|
||||
"""Change the DNA Struct associated with this block.
|
||||
|
||||
Use to make a block type more specific, for example when you have a
|
||||
modifier but need to access it as SubSurfModifier.
|
||||
|
||||
:param sdna_index: the SDNA index of the DNA type.
|
||||
"""
|
||||
assert type(sdna_index) is int
|
||||
sdna_index_curr = self.sdna_index
|
||||
self.bfile.ensure_subtype_smaller(sdna_index_curr, sdna_index)
|
||||
self.sdna_index = sdna_index
|
||||
|
||||
def refine_type(self, dna_type_id: bytes):
|
||||
"""Change the DNA Struct associated with this block.
|
||||
|
||||
Use to make a block type more specific, for example when you have a
|
||||
modifier but need to access it as SubSurfModifier.
|
||||
|
||||
:param dna_type_id: the name of the DNA type.
|
||||
"""
|
||||
assert isinstance(dna_type_id, bytes)
|
||||
sdna_index = self.bfile.sdna_index_from_id[dna_type_id]
|
||||
self.refine_type_from_index(sdna_index)
|
||||
|
||||
def abs_offset(self, path: dna.FieldPath) -> typing.Tuple[int, int]:
|
||||
"""Compute the absolute file offset of the field.
|
||||
|
||||
:returns: tuple (offset in bytes, length of array in items)
|
||||
"""
|
||||
field, field_offset = self.dna_type.field_from_path(self.bfile.header.pointer_size, path)
|
||||
return self.file_offset + field_offset, field.name.array_size
|
||||
|
||||
def get(self,
|
||||
path: dna.FieldPath,
|
||||
default=...,
|
||||
null_terminated=True,
|
||||
as_str=False,
|
||||
return_field=False
|
||||
) -> typing.Any:
|
||||
"""Read a property and return the value.
|
||||
|
||||
:param path: name of the property (like `b'loc'`), tuple of names
|
||||
to read a sub-property (like `(b'id', b'name')`), or tuple of
|
||||
name and index to read one item from an array (like
|
||||
`(b'loc', 2)`)
|
||||
:param default: The value to return when the field does not exist.
|
||||
Use Ellipsis (the default value) to raise a KeyError instead.
|
||||
:param null_terminated: Only used when reading bytes or strings. When
|
||||
True, stops reading at the first zero byte; be careful with this
|
||||
when reading binary data.
|
||||
:param as_str: When True, automatically decode bytes to string
|
||||
(assumes UTF-8 encoding).
|
||||
:param return_field: When True, returns tuple (dna.Field, value).
|
||||
Otherwise just returns the value.
|
||||
"""
|
||||
self.bfile.fileobj.seek(self.file_offset, os.SEEK_SET)
|
||||
|
||||
dna_struct = self.bfile.structs[self.sdna_index]
|
||||
field, value = dna_struct.field_get(
|
||||
self.bfile.header, self.bfile.fileobj, path,
|
||||
default=default,
|
||||
null_terminated=null_terminated, as_str=as_str,
|
||||
)
|
||||
if return_field:
|
||||
return value, field
|
||||
return value
|
||||
|
||||
def get_recursive_iter(self,
|
||||
path: dna.FieldPath,
|
||||
path_root: dna.FieldPath = b'',
|
||||
default=...,
|
||||
null_terminated=True,
|
||||
as_str=True,
|
||||
) -> typing.Iterator[typing.Tuple[dna.FieldPath, typing.Any]]:
|
||||
"""Generator, yields (path, property value) tuples.
|
||||
|
||||
If a property cannot be decoded, a string representing its DNA type
|
||||
name is used as its value instead, between pointy brackets.
|
||||
"""
|
||||
path_full = path # type: dna.FieldPath
|
||||
if path_root:
|
||||
if isinstance(path_root, bytes):
|
||||
path_root = (path_root,)
|
||||
if isinstance(path, bytes):
|
||||
path = (path,)
|
||||
path_full = tuple(path_root) + tuple(path)
|
||||
|
||||
try:
|
||||
# Try accessing as simple property
|
||||
yield (path_full,
|
||||
self.get(path_full, default, null_terminated, as_str))
|
||||
except exceptions.NoReaderImplemented as ex:
|
||||
# This was not a simple property, so recurse into its DNA Struct.
|
||||
dna_type = ex.dna_type
|
||||
struct_index = self.bfile.sdna_index_from_id.get(dna_type.dna_type_id)
|
||||
if struct_index is None:
|
||||
yield (path_full, "<%s>" % dna_type.dna_type_id.decode('ascii'))
|
||||
return
|
||||
|
||||
# Recurse through the fields.
|
||||
for f in dna_type.fields:
|
||||
yield from self.get_recursive_iter(f.name.name_only, path_full, default=default,
|
||||
null_terminated=null_terminated, as_str=as_str)
|
||||
|
||||
def hash(self) -> int:
|
||||
"""Generate a pointer-independent hash for the block.
|
||||
|
||||
Generates a 'hash' that can be used instead of addr_old as block id,
|
||||
which should be 'stable' across .blend file load & save (i.e. it does
|
||||
not changes due to pointer addresses variations).
|
||||
"""
|
||||
# TODO This implementation is most likely far from optimal... and CRC32
|
||||
# is not kown as the best hashing algo either. But for now does the job!
|
||||
import zlib
|
||||
|
||||
dna_type = self.dna_type
|
||||
pointer_size = self.bfile.header.pointer_size
|
||||
|
||||
hsh = 1
|
||||
for path, value in self.items_recursive():
|
||||
field, _ = dna_type.field_from_path(pointer_size, path)
|
||||
if field.name.is_pointer:
|
||||
continue
|
||||
hsh = zlib.adler32(str(value).encode(), hsh)
|
||||
return hsh
|
||||
|
||||
def set(self, path: bytes, value):
|
||||
dna_struct = self.bfile.structs[self.sdna_index]
|
||||
self.bfile.mark_modified()
|
||||
self.bfile.fileobj.seek(self.file_offset, os.SEEK_SET)
|
||||
return dna_struct.field_set(self.bfile.header, self.bfile.fileobj, path, value)
|
||||
|
||||
def get_pointer(
|
||||
self, path: dna.FieldPath,
|
||||
default=...,
|
||||
) -> typing.Union[None, 'BlendFileBlock']:
|
||||
"""Same as get() but dereferences a pointer.
|
||||
|
||||
:raises exceptions.SegmentationFault: when there is no datablock with
|
||||
the pointed-to address.
|
||||
"""
|
||||
result = self.get(path, default=default)
|
||||
|
||||
# If it's not an integer, we have no pointer to follow and this may
|
||||
# actually be a non-pointer property.
|
||||
if type(result) is not int:
|
||||
return result
|
||||
|
||||
if result == 0:
|
||||
return None
|
||||
|
||||
try:
|
||||
return self.bfile.dereference_pointer(result)
|
||||
except exceptions.SegmentationFault as ex:
|
||||
ex.field_path = path
|
||||
raise
|
||||
|
||||
def iter_array_of_pointers(self, path: dna.FieldPath, array_size: int) \
|
||||
-> typing.Iterator['BlendFileBlock']:
|
||||
"""Dereference pointers from an array-of-pointers field.
|
||||
|
||||
Use this function when you have a field like Mesh materials:
|
||||
`Mat **mat`
|
||||
|
||||
:param path: The array-of-pointers field.
|
||||
:param array_size: Number of items in the array. If None, the
|
||||
on-disk size of the DNA field is divided by the pointer size to
|
||||
obtain the array size.
|
||||
"""
|
||||
if array_size == 0:
|
||||
return
|
||||
|
||||
array = self.get_pointer(path)
|
||||
assert array is not None
|
||||
assert array.code == b'DATA', \
|
||||
'Array data block should have code DATA, is %r' % array.code.decode()
|
||||
file_offset = array.file_offset
|
||||
|
||||
endian = self.bfile.header.endian
|
||||
ps = self.bfile.header.pointer_size
|
||||
|
||||
for i in range(array_size):
|
||||
fileobj = self.bfile.fileobj
|
||||
fileobj.seek(file_offset + ps * i, os.SEEK_SET)
|
||||
address = endian.read_pointer(fileobj, ps)
|
||||
if address == 0:
|
||||
continue
|
||||
yield self.bfile.dereference_pointer(address)
|
||||
|
||||
def iter_fixed_array_of_pointers(self, path: dna.FieldPath) \
|
||||
-> typing.Iterator['BlendFileBlock']:
|
||||
"""Yield blocks from a fixed-size array field.
|
||||
|
||||
Use this function when you have a field like lamp textures:
|
||||
`MTex *mtex[18]`
|
||||
|
||||
The size of the array is determined automatically by the size in bytes
|
||||
of the field divided by the pointer size of the blend file.
|
||||
|
||||
:param path: The array field.
|
||||
:raises KeyError: if the path does not exist.
|
||||
"""
|
||||
|
||||
dna_struct = self.dna_type
|
||||
ps = self.bfile.header.pointer_size
|
||||
endian = self.bfile.header.endian
|
||||
fileobj = self.bfile.fileobj
|
||||
|
||||
field, offset_in_struct = dna_struct.field_from_path(ps, path)
|
||||
array_size = field.size // ps
|
||||
|
||||
for i in range(array_size):
|
||||
fileobj.seek(self.file_offset + offset_in_struct + ps * i, os.SEEK_SET)
|
||||
address = endian.read_pointer(fileobj, ps)
|
||||
if not address:
|
||||
# Fixed-size arrays contain 0-pointers.
|
||||
continue
|
||||
yield self.bfile.dereference_pointer(address)
|
||||
|
||||
def __getitem__(self, path: dna.FieldPath):
|
||||
return self.get(path)
|
||||
|
||||
def __setitem__(self, item: bytes, value) -> None:
|
||||
self.set(item, value)
|
||||
|
||||
def keys(self) -> typing.Iterator[bytes]:
|
||||
"""Generator, yields all field names of this block."""
|
||||
return (f.name.name_only for f in self.dna_type.fields)
|
||||
|
||||
def values(self) -> typing.Iterable[typing.Any]:
|
||||
for k in self.keys():
|
||||
try:
|
||||
yield self[k]
|
||||
except exceptions.NoReaderImplemented as ex:
|
||||
yield '<%s>' % ex.dna_type.dna_type_id.decode('ascii')
|
||||
|
||||
def items(self) -> typing.Iterable[typing.Tuple[bytes, typing.Any]]:
|
||||
for k in self.keys():
|
||||
try:
|
||||
yield (k, self[k])
|
||||
except exceptions.NoReaderImplemented as ex:
|
||||
yield (k, '<%s>' % ex.dna_type.dna_type_id.decode('ascii'))
|
||||
|
||||
def items_recursive(self) -> typing.Iterator[typing.Tuple[dna.FieldPath, typing.Any]]:
|
||||
"""Generator, yields (property path, property value) recursively for all properties."""
|
||||
for k in self.keys():
|
||||
yield from self.get_recursive_iter(k, as_str=False)
|
||||
@@ -0,0 +1,334 @@
|
||||
# ***** BEGIN GPL LICENSE BLOCK *****
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or
|
||||
# modify it under the terms of the GNU General Public License
|
||||
# as published by the Free Software Foundation; either version 2
|
||||
# of the License, or (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software Foundation,
|
||||
# Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
|
||||
#
|
||||
# ***** END GPL LICENCE BLOCK *****
|
||||
#
|
||||
# (c) 2009, At Mind B.V. - Jeroen Bakker
|
||||
# (c) 2014, Blender Foundation - Campbell Barton
|
||||
# (c) 2018, Blender Foundation - Sybren A. Stüvel
|
||||
import logging
|
||||
import os
|
||||
import typing
|
||||
|
||||
from . import header, exceptions
|
||||
|
||||
# Either a simple path b'propname', or a tuple (b'parentprop', b'actualprop', arrayindex)
|
||||
FieldPath = typing.Union[bytes, typing.Iterable[typing.Union[bytes, int]]]
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Name:
|
||||
"""dna.Name is a C-type name stored in the DNA as bytes."""
|
||||
|
||||
def __init__(self, name_full: bytes) -> None:
|
||||
self.name_full = name_full
|
||||
self.name_only = self.calc_name_only()
|
||||
self.is_pointer = self.calc_is_pointer()
|
||||
self.is_method_pointer = self.calc_is_method_pointer()
|
||||
self.array_size = self.calc_array_size()
|
||||
|
||||
def __repr__(self):
|
||||
return '%s(%r)' % (type(self).__qualname__, self.name_full)
|
||||
|
||||
def as_reference(self, parent) -> bytes:
|
||||
if not parent:
|
||||
return self.name_only
|
||||
return parent + b'.' + self.name_only
|
||||
|
||||
def calc_name_only(self) -> bytes:
|
||||
result = self.name_full.strip(b'*()')
|
||||
index = result.find(b'[')
|
||||
if index == -1:
|
||||
return result
|
||||
return result[:index]
|
||||
|
||||
def calc_is_pointer(self) -> bool:
|
||||
return b'*' in self.name_full
|
||||
|
||||
def calc_is_method_pointer(self):
|
||||
return b'(*' in self.name_full
|
||||
|
||||
def calc_array_size(self):
|
||||
result = 1
|
||||
partial_name = self.name_full
|
||||
|
||||
while True:
|
||||
idx_start = partial_name.find(b'[')
|
||||
if idx_start < 0:
|
||||
break
|
||||
|
||||
idx_stop = partial_name.find(b']')
|
||||
result *= int(partial_name[idx_start + 1:idx_stop])
|
||||
partial_name = partial_name[idx_stop + 1:]
|
||||
|
||||
return result
|
||||
|
||||
|
||||
class Field:
|
||||
"""dna.Field is a coupled dna.Struct and dna.Name.
|
||||
|
||||
It also contains the file offset in bytes.
|
||||
|
||||
:ivar name: the name of the field.
|
||||
:ivar dna_type: the type of the field.
|
||||
:ivar size: size of the field on disk, in bytes.
|
||||
:ivar offset: cached offset of the field, in bytes.
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
dna_type: 'Struct',
|
||||
name: Name,
|
||||
size: int,
|
||||
offset: int) -> None:
|
||||
self.dna_type = dna_type
|
||||
self.name = name
|
||||
self.size = size
|
||||
self.offset = offset
|
||||
|
||||
def __repr__(self):
|
||||
return '<%r %r (%s)>' % (type(self).__qualname__, self.name, self.dna_type)
|
||||
|
||||
|
||||
class Struct:
|
||||
"""dna.Struct is a C-type structure stored in the DNA."""
|
||||
|
||||
log = log.getChild('Struct')
|
||||
|
||||
def __init__(self, dna_type_id: bytes, size: int = None) -> None:
|
||||
"""
|
||||
:param dna_type_id: name of the struct in C, like b'AlembicObjectPath'.
|
||||
:param size: only for unit tests; typically set after construction by
|
||||
BlendFile.decode_structs(). If not set, it is calculated on the fly
|
||||
when struct.size is evaluated, based on the available fields.
|
||||
"""
|
||||
self.dna_type_id = dna_type_id
|
||||
self._size = size
|
||||
self._fields = [] # type: typing.List[Field]
|
||||
self._fields_by_name = {} # type: typing.Dict[bytes, Field]
|
||||
|
||||
def __repr__(self):
|
||||
return '%s(%r)' % (type(self).__qualname__, self.dna_type_id)
|
||||
|
||||
@property
|
||||
def size(self) -> int:
|
||||
if self._size is None:
|
||||
if not self._fields:
|
||||
raise ValueError('Unable to determine size of fieldless %r' % self)
|
||||
last_field = max(self._fields, key=lambda f: f.offset)
|
||||
self._size = last_field.offset + last_field.size
|
||||
return self._size
|
||||
|
||||
@size.setter
|
||||
def size(self, new_size: int):
|
||||
self._size = new_size
|
||||
|
||||
def append_field(self, field: Field):
|
||||
self._fields.append(field)
|
||||
self._fields_by_name[field.name.name_only] = field
|
||||
|
||||
@property
|
||||
def fields(self) -> typing.List[Field]:
|
||||
"""Return the fields of this Struct.
|
||||
|
||||
Do not modify the returned list; use append_field() instead.
|
||||
"""
|
||||
return self._fields
|
||||
|
||||
def has_field(self, field_name: bytes) -> bool:
|
||||
return field_name in self._fields_by_name
|
||||
|
||||
def field_from_path(self,
|
||||
pointer_size: int,
|
||||
path: FieldPath) \
|
||||
-> typing.Tuple[Field, int]:
|
||||
"""
|
||||
Support lookups as bytes or a tuple of bytes and optional index.
|
||||
|
||||
C style 'id.name' --> (b'id', b'name')
|
||||
C style 'array[4]' --> (b'array', 4)
|
||||
|
||||
:returns: the field itself, and its offset taking into account the
|
||||
optional index. The offset is relative to the start of the struct,
|
||||
i.e. relative to the BlendFileBlock containing the data.
|
||||
:raises KeyError: if the field does not exist.
|
||||
"""
|
||||
if isinstance(path, tuple):
|
||||
name = path[0]
|
||||
if len(path) >= 2 and not isinstance(path[1], bytes):
|
||||
name_tail = path[2:]
|
||||
index = path[1]
|
||||
assert isinstance(index, int)
|
||||
else:
|
||||
name_tail = path[1:]
|
||||
index = 0
|
||||
else:
|
||||
name = path
|
||||
name_tail = ()
|
||||
index = 0
|
||||
|
||||
if not isinstance(name, bytes):
|
||||
raise TypeError('name should be bytes, but is %r' % type(name))
|
||||
|
||||
field = self._fields_by_name.get(name)
|
||||
if not field:
|
||||
raise KeyError('%r has no field %r, only %r' %
|
||||
(self, name, sorted(self._fields_by_name.keys())))
|
||||
|
||||
offset = field.offset
|
||||
if index:
|
||||
if field.name.is_pointer:
|
||||
index_offset = pointer_size * index
|
||||
else:
|
||||
index_offset = field.dna_type.size * index
|
||||
if index_offset >= field.size:
|
||||
raise OverflowError('path %r is out of bounds of its DNA type %s' %
|
||||
(path, field.dna_type))
|
||||
offset += index_offset
|
||||
|
||||
if name_tail:
|
||||
subval, suboff = field.dna_type.field_from_path(pointer_size, name_tail)
|
||||
return subval, suboff + offset
|
||||
|
||||
return field, offset
|
||||
|
||||
def field_get(self,
|
||||
file_header: header.BlendFileHeader,
|
||||
fileobj: typing.IO[bytes],
|
||||
path: FieldPath,
|
||||
default=...,
|
||||
null_terminated=True,
|
||||
as_str=True,
|
||||
) -> typing.Tuple[typing.Optional[Field], typing.Any]:
|
||||
"""Read the value of the field from the blend file.
|
||||
|
||||
Assumes the file pointer of `fileobj` is seek()ed to the start of the
|
||||
struct on disk (e.g. the start of the BlendFileBlock containing the
|
||||
data).
|
||||
|
||||
:param file_header:
|
||||
:param fileobj:
|
||||
:param path:
|
||||
:param default: The value to return when the field does not exist.
|
||||
Use Ellipsis (the default value) to raise a KeyError instead.
|
||||
:param null_terminated: Only used when reading bytes or strings. When
|
||||
True, stops reading at the first zero byte. Be careful with this
|
||||
default when reading binary data.
|
||||
:param as_str: When True, automatically decode bytes to string
|
||||
(assumes UTF-8 encoding).
|
||||
:returns: The field instance and the value. If a default value was passed
|
||||
and the field was not found, (None, default) is returned.
|
||||
"""
|
||||
try:
|
||||
field, offset = self.field_from_path(file_header.pointer_size, path)
|
||||
except KeyError:
|
||||
if default is ...:
|
||||
raise
|
||||
return None, default
|
||||
|
||||
fileobj.seek(offset, os.SEEK_CUR)
|
||||
|
||||
dna_type = field.dna_type
|
||||
dna_name = field.name
|
||||
endian = file_header.endian
|
||||
|
||||
# Some special cases (pointers, strings/bytes)
|
||||
if dna_name.is_pointer:
|
||||
return field, endian.read_pointer(fileobj, file_header.pointer_size)
|
||||
if dna_type.dna_type_id == b'char':
|
||||
return field, self._field_get_char(file_header, fileobj, field, null_terminated, as_str)
|
||||
|
||||
simple_readers = {
|
||||
b'int': endian.read_int,
|
||||
b'short': endian.read_short,
|
||||
b'uint64_t': endian.read_ulong,
|
||||
b'float': endian.read_float,
|
||||
}
|
||||
try:
|
||||
simple_reader = simple_readers[dna_type.dna_type_id]
|
||||
except KeyError:
|
||||
raise exceptions.NoReaderImplemented(
|
||||
"%r exists but not simple type (%r), can't resolve field %r" %
|
||||
(path, dna_type.dna_type_id.decode(), dna_name.name_only),
|
||||
dna_name, dna_type) from None
|
||||
|
||||
if isinstance(path, tuple) and len(path) > 1 and isinstance(path[-1], int):
|
||||
# The caller wants to get a single item from an array. The offset we seeked to already
|
||||
# points to this item. In this case we do not want to look at dna_name.array_size,
|
||||
# because we want a single item from that array.
|
||||
return field, simple_reader(fileobj)
|
||||
|
||||
if dna_name.array_size > 1:
|
||||
return field, [simple_reader(fileobj) for _ in range(dna_name.array_size)]
|
||||
return field, simple_reader(fileobj)
|
||||
|
||||
def _field_get_char(self,
|
||||
file_header: header.BlendFileHeader,
|
||||
fileobj: typing.IO[bytes],
|
||||
field: 'Field',
|
||||
null_terminated: typing.Optional[bool],
|
||||
as_str: bool) -> typing.Any:
|
||||
dna_name = field.name
|
||||
endian = file_header.endian
|
||||
|
||||
if field.size == 1:
|
||||
# Single char, assume it's bitflag or int value, and not a string/bytes data...
|
||||
return endian.read_char(fileobj)
|
||||
|
||||
if null_terminated or (null_terminated is None and as_str):
|
||||
data = endian.read_bytes0(fileobj, dna_name.array_size)
|
||||
else:
|
||||
data = fileobj.read(dna_name.array_size)
|
||||
|
||||
if as_str:
|
||||
return data.decode('utf8')
|
||||
return data
|
||||
|
||||
def field_set(self,
|
||||
file_header: header.BlendFileHeader,
|
||||
fileobj: typing.IO[bytes],
|
||||
path: bytes,
|
||||
value: typing.Any):
|
||||
"""Write a value to the blend file.
|
||||
|
||||
Assumes the file pointer of `fileobj` is seek()ed to the start of the
|
||||
struct on disk (e.g. the start of the BlendFileBlock containing the
|
||||
data).
|
||||
"""
|
||||
assert isinstance(path, bytes), 'path should be bytes, but is %s' % type(path)
|
||||
|
||||
field, offset = self.field_from_path(file_header.pointer_size, path)
|
||||
|
||||
dna_type = field.dna_type
|
||||
dna_name = field.name
|
||||
endian = file_header.endian
|
||||
|
||||
if dna_type.dna_type_id != b'char':
|
||||
msg = "Setting type %r is not supported for %s.%s" % (
|
||||
dna_type, self.dna_type_id.decode(), dna_name.name_full.decode())
|
||||
raise exceptions.NoWriterImplemented(msg, dna_name, dna_type)
|
||||
|
||||
fileobj.seek(offset, os.SEEK_CUR)
|
||||
|
||||
if self.log.isEnabledFor(logging.DEBUG):
|
||||
filepos = fileobj.tell()
|
||||
thing = 'string' if isinstance(value, str) else 'bytes'
|
||||
self.log.debug('writing %s %r at file offset %d / %x', thing, value, filepos, filepos)
|
||||
|
||||
if isinstance(value, str):
|
||||
return endian.write_string(fileobj, value, dna_name.array_size)
|
||||
else:
|
||||
return endian.write_bytes(fileobj, value, dna_name.array_size)
|
||||
@@ -0,0 +1,163 @@
|
||||
# ***** BEGIN GPL LICENSE BLOCK *****
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or
|
||||
# modify it under the terms of the GNU General Public License
|
||||
# as published by the Free Software Foundation; either version 2
|
||||
# of the License, or (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software Foundation,
|
||||
# Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
|
||||
#
|
||||
# ***** END GPL LICENCE BLOCK *****
|
||||
#
|
||||
# (c) 2009, At Mind B.V. - Jeroen Bakker
|
||||
# (c) 2014, Blender Foundation - Campbell Barton
|
||||
# (c) 2018, Blender Foundation - Sybren A. Stüvel
|
||||
"""Read-write utility functions."""
|
||||
|
||||
import struct
|
||||
import typing
|
||||
|
||||
|
||||
class EndianIO:
|
||||
# TODO(Sybren): note as UCHAR: struct.Struct = None and move actual structs to LittleEndianTypes
|
||||
UCHAR = struct.Struct(b'<B')
|
||||
USHORT = struct.Struct(b'<H')
|
||||
USHORT2 = struct.Struct(b'<HH') # two shorts in a row
|
||||
SSHORT = struct.Struct(b'<h')
|
||||
UINT = struct.Struct(b'<I')
|
||||
SINT = struct.Struct(b'<i')
|
||||
FLOAT = struct.Struct(b'<f')
|
||||
ULONG = struct.Struct(b'<Q')
|
||||
|
||||
@classmethod
|
||||
def _read(cls, fileobj: typing.IO[bytes], typestruct: struct.Struct):
|
||||
data = fileobj.read(typestruct.size)
|
||||
try:
|
||||
return typestruct.unpack(data)[0]
|
||||
except struct.error as ex:
|
||||
raise struct.error('%s (read %d bytes)' % (ex, len(data))) from None
|
||||
|
||||
@classmethod
|
||||
def read_char(cls, fileobj: typing.IO[bytes]):
|
||||
return cls._read(fileobj, cls.UCHAR)
|
||||
|
||||
@classmethod
|
||||
def read_ushort(cls, fileobj: typing.IO[bytes]):
|
||||
return cls._read(fileobj, cls.USHORT)
|
||||
|
||||
@classmethod
|
||||
def read_short(cls, fileobj: typing.IO[bytes]):
|
||||
return cls._read(fileobj, cls.SSHORT)
|
||||
|
||||
@classmethod
|
||||
def read_uint(cls, fileobj: typing.IO[bytes]):
|
||||
return cls._read(fileobj, cls.UINT)
|
||||
|
||||
@classmethod
|
||||
def read_int(cls, fileobj: typing.IO[bytes]):
|
||||
return cls._read(fileobj, cls.SINT)
|
||||
|
||||
@classmethod
|
||||
def read_float(cls, fileobj: typing.IO[bytes]):
|
||||
return cls._read(fileobj, cls.FLOAT)
|
||||
|
||||
@classmethod
|
||||
def read_ulong(cls, fileobj: typing.IO[bytes]):
|
||||
return cls._read(fileobj, cls.ULONG)
|
||||
|
||||
@classmethod
|
||||
def read_pointer(cls, fileobj: typing.IO[bytes], pointer_size: int):
|
||||
"""Read a pointer from a file."""
|
||||
|
||||
if pointer_size == 4:
|
||||
return cls.read_uint(fileobj)
|
||||
if pointer_size == 8:
|
||||
return cls.read_ulong(fileobj)
|
||||
raise ValueError('unsupported pointer size %d' % pointer_size)
|
||||
|
||||
@classmethod
|
||||
def write_string(cls, fileobj: typing.IO[bytes], astring: str, fieldlen: int) -> int:
|
||||
"""Write a (truncated) string as UTF-8.
|
||||
|
||||
The string will always be written 0-terminated.
|
||||
|
||||
:param fileobj: the file to write to.
|
||||
:param astring: the string to write.
|
||||
:param fieldlen: the field length in bytes.
|
||||
:returns: the number of bytes written.
|
||||
"""
|
||||
assert isinstance(astring, str)
|
||||
encoded = astring.encode('utf-8')
|
||||
|
||||
# Take into account we also need space for a trailing 0-byte.
|
||||
maxlen = fieldlen - 1
|
||||
|
||||
if len(encoded) >= maxlen:
|
||||
encoded = encoded[:maxlen]
|
||||
|
||||
# Keep stripping off the last byte until the string
|
||||
# is valid UTF-8 again.
|
||||
while True:
|
||||
try:
|
||||
encoded.decode('utf8')
|
||||
except UnicodeDecodeError:
|
||||
encoded = encoded[:-1]
|
||||
else:
|
||||
break
|
||||
|
||||
return fileobj.write(encoded + b'\0')
|
||||
|
||||
@classmethod
|
||||
def write_bytes(cls, fileobj: typing.IO[bytes], data: bytes, fieldlen: int) -> int:
|
||||
"""Write (truncated) bytes.
|
||||
|
||||
When len(data) < fieldlen, a terminating b'\0' will be appended.
|
||||
|
||||
:returns: the number of bytes written.
|
||||
"""
|
||||
assert isinstance(data, (bytes, bytearray))
|
||||
if len(data) >= fieldlen:
|
||||
to_write = data[0:fieldlen]
|
||||
else:
|
||||
to_write = data + b'\0'
|
||||
|
||||
return fileobj.write(to_write)
|
||||
|
||||
@classmethod
|
||||
def read_bytes0(cls, fileobj, length):
|
||||
data = fileobj.read(length)
|
||||
return cls.read_data0(data)
|
||||
|
||||
@classmethod
|
||||
def read_data0_offset(cls, data, offset):
|
||||
add = data.find(b'\0', offset) - offset
|
||||
return data[offset:offset + add]
|
||||
|
||||
@classmethod
|
||||
def read_data0(cls, data):
|
||||
add = data.find(b'\0')
|
||||
if add < 0:
|
||||
return data
|
||||
return data[:add]
|
||||
|
||||
|
||||
class LittleEndianTypes(EndianIO):
|
||||
pass
|
||||
|
||||
|
||||
class BigEndianTypes(LittleEndianTypes):
|
||||
UCHAR = struct.Struct(b'>B')
|
||||
USHORT = struct.Struct(b'>H')
|
||||
USHORT2 = struct.Struct(b'>HH') # two shorts in a row
|
||||
SSHORT = struct.Struct(b'>h')
|
||||
UINT = struct.Struct(b'>I')
|
||||
SINT = struct.Struct(b'>i')
|
||||
FLOAT = struct.Struct(b'>f')
|
||||
ULONG = struct.Struct(b'>Q')
|
||||
@@ -0,0 +1,76 @@
|
||||
# ***** BEGIN GPL LICENSE BLOCK *****
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or
|
||||
# modify it under the terms of the GNU General Public License
|
||||
# as published by the Free Software Foundation; either version 2
|
||||
# of the License, or (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software Foundation,
|
||||
# Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
|
||||
#
|
||||
# ***** END GPL LICENCE BLOCK *****
|
||||
#
|
||||
# (c) 2009, At Mind B.V. - Jeroen Bakker
|
||||
# (c) 2014, Blender Foundation - Campbell Barton
|
||||
# (c) 2018, Blender Foundation - Sybren A. Stüvel
|
||||
|
||||
|
||||
import pathlib
|
||||
|
||||
|
||||
class BlendFileError(Exception):
|
||||
"""Raised when there was an error reading/parsing a blend file."""
|
||||
|
||||
def __init__(self, message: str, filepath: pathlib.Path) -> None:
|
||||
super().__init__(message)
|
||||
self.filepath = filepath
|
||||
|
||||
def __str__(self):
|
||||
return '%s: %s' % (super().__str__(), self.filepath)
|
||||
|
||||
|
||||
class NoDNA1Block(BlendFileError):
|
||||
"""Raised when the blend file contains no DNA1 block."""
|
||||
|
||||
|
||||
class NoReaderImplemented(NotImplementedError):
|
||||
"""Raised when reading a property of a non-implemented type.
|
||||
|
||||
This indicates that the property should be read using some dna.Struct.
|
||||
|
||||
:type dna_name: blender_asset_tracer.blendfile.dna.Name
|
||||
:type dna_type: blender_asset_tracer.blendfile.dna.Struct
|
||||
"""
|
||||
|
||||
def __init__(self, message: str, dna_name, dna_type) -> None:
|
||||
super().__init__(message)
|
||||
self.dna_name = dna_name
|
||||
self.dna_type = dna_type
|
||||
|
||||
|
||||
class NoWriterImplemented(NotImplementedError):
|
||||
"""Raised when writing a property of a non-implemented type.
|
||||
|
||||
:type dna_name: blender_asset_tracer.blendfile.dna.Name
|
||||
:type dna_type: blender_asset_tracer.blendfile.dna.Struct
|
||||
"""
|
||||
|
||||
def __init__(self, message: str, dna_name, dna_type) -> None:
|
||||
super().__init__(message)
|
||||
self.dna_name = dna_name
|
||||
self.dna_type = dna_type
|
||||
|
||||
|
||||
class SegmentationFault(Exception):
|
||||
"""Raised when a pointer to a non-existant datablock was dereferenced."""
|
||||
|
||||
def __init__(self, message: str, address: int, field_path=None) -> None:
|
||||
super().__init__(message)
|
||||
self.address = address
|
||||
self.field_path = field_path
|
||||
@@ -0,0 +1,78 @@
|
||||
# ***** BEGIN GPL LICENSE BLOCK *****
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or
|
||||
# modify it under the terms of the GNU General Public License
|
||||
# as published by the Free Software Foundation; either version 2
|
||||
# of the License, or (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software Foundation,
|
||||
# Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
|
||||
#
|
||||
# ***** END GPL LICENCE BLOCK *****
|
||||
#
|
||||
# (c) 2009, At Mind B.V. - Jeroen Bakker
|
||||
# (c) 2014, Blender Foundation - Campbell Barton
|
||||
# (c) 2018, Blender Foundation - Sybren A. Stüvel
|
||||
import logging
|
||||
import os
|
||||
import pathlib
|
||||
import struct
|
||||
import typing
|
||||
|
||||
from . import dna_io, exceptions
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BlendFileHeader:
|
||||
"""
|
||||
BlendFileHeader represents the first 12 bytes of a blend file.
|
||||
|
||||
It contains information about the hardware architecture, which is relevant
|
||||
to the structure of the rest of the file.
|
||||
"""
|
||||
structure = struct.Struct(b'7s1s1s3s')
|
||||
|
||||
def __init__(self, fileobj: typing.IO[bytes], path: pathlib.Path) -> None:
|
||||
log.debug("reading blend-file-header %s", path)
|
||||
fileobj.seek(0, os.SEEK_SET)
|
||||
header = fileobj.read(self.structure.size)
|
||||
values = self.structure.unpack(header)
|
||||
|
||||
self.magic = values[0]
|
||||
|
||||
pointer_size_id = values[1]
|
||||
if pointer_size_id == b'-':
|
||||
self.pointer_size = 8
|
||||
elif pointer_size_id == b'_':
|
||||
self.pointer_size = 4
|
||||
else:
|
||||
raise exceptions.BlendFileError('invalid pointer size %r' % pointer_size_id, path)
|
||||
|
||||
endian_id = values[2]
|
||||
if endian_id == b'v':
|
||||
self.endian = dna_io.LittleEndianTypes
|
||||
self.endian_str = b'<' # indication for struct.Struct()
|
||||
elif endian_id == b'V':
|
||||
self.endian = dna_io.BigEndianTypes
|
||||
self.endian_str = b'>' # indication for struct.Struct()
|
||||
else:
|
||||
raise exceptions.BlendFileError('invalid endian indicator %r' % endian_id, path)
|
||||
|
||||
version_id = values[3]
|
||||
self.version = int(version_id)
|
||||
|
||||
def create_block_header_struct(self) -> struct.Struct:
|
||||
"""Create a Struct instance for parsing data block headers."""
|
||||
return struct.Struct(b''.join((
|
||||
self.endian_str,
|
||||
b'4sI',
|
||||
b'I' if self.pointer_size == 4 else b'Q',
|
||||
b'II',
|
||||
)))
|
||||
@@ -0,0 +1,70 @@
|
||||
# ***** BEGIN GPL LICENSE BLOCK *****
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or
|
||||
# modify it under the terms of the GNU General Public License
|
||||
# as published by the Free Software Foundation; either version 2
|
||||
# of the License, or (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software Foundation,
|
||||
# Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
|
||||
#
|
||||
# ***** END GPL LICENCE BLOCK *****
|
||||
#
|
||||
# (c) 2009, At Mind B.V. - Jeroen Bakker
|
||||
# (c) 2014, Blender Foundation - Campbell Barton
|
||||
# (c) 2018, Blender Foundation - Sybren A. Stüvel
|
||||
import typing
|
||||
|
||||
from blender_asset_tracer import cdefs
|
||||
from . import BlendFileBlock
|
||||
from .dna import FieldPath
|
||||
|
||||
|
||||
def listbase(block: typing.Optional[BlendFileBlock], next_path: FieldPath = b'next') \
|
||||
-> typing.Iterator[BlendFileBlock]:
|
||||
"""Generator, yields all blocks in the ListBase linked list."""
|
||||
while block:
|
||||
yield block
|
||||
next_ptr = block[next_path]
|
||||
if next_ptr == 0:
|
||||
break
|
||||
block = block.bfile.dereference_pointer(next_ptr)
|
||||
|
||||
|
||||
def sequencer_strips(sequence_editor: BlendFileBlock) \
|
||||
-> typing.Iterator[typing.Tuple[BlendFileBlock, int]]:
|
||||
"""Generator, yield all sequencer strip blocks with their type number.
|
||||
|
||||
Recurses into meta strips, yielding both the meta strip itself and the
|
||||
strips contained within it.
|
||||
|
||||
See blender_asset_tracer.cdefs.SEQ_TYPE_xxx for the type numbers.
|
||||
"""
|
||||
|
||||
def iter_seqbase(seqbase) -> typing.Iterator[typing.Tuple[BlendFileBlock, int]]:
|
||||
for seq in listbase(seqbase):
|
||||
seq.refine_type(b'Sequence')
|
||||
seq_type = seq[b'type']
|
||||
yield seq, seq_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)
|
||||
|
||||
sbase = sequence_editor.get_pointer((b'seqbase', b'first'))
|
||||
yield from iter_seqbase(sbase)
|
||||
|
||||
|
||||
def modifiers(object_block: BlendFileBlock) -> typing.Iterator[BlendFileBlock]:
|
||||
"""Generator, yield the object's modifiers."""
|
||||
|
||||
# 'ob->modifiers[...]'
|
||||
mods = object_block.get_pointer((b'modifiers', b'first'))
|
||||
yield from listbase(mods, next_path=(b'modifier', b'next'))
|
||||
Reference in New Issue
Block a user