asset_library/adapters/adapter.py

772 lines
27 KiB
Python

#from asset_library.common.functions import (norm_asset_datas,)
from asset_library.common.bl_utils import get_addon_prefs, load_datablocks
from asset_library.common.file_utils import read_file, write_file
from asset_library.common.template import Template
from asset_library.constants import (MODULE_DIR, RESOURCES_DIR)
from asset_library import (action, collection, file)
from bpy.types import PropertyGroup
from bpy.props import StringProperty
import bpy
from itertools import groupby
from pathlib import Path
import shutil
import os
import json
import uuid
import time
from functools import partial
import subprocess
from glob import glob
class AssetLibraryAdapter(PropertyGroup):
#def __init__(self):
name = "Base Adapter"
#library = None
@property
def library(self):
prefs = self.addon_prefs
for lib in prefs.libraries:
if lib.adapter == self:
return lib
@property
def bundle_directory(self):
return self.library.library_path
@property
def data_type(self):
return self.library.data_type
@property
def data_types(self):
return self.library.data_types
def get_catalog_path(self, directory=None):
directory = directory or self.bundle_directory
return Path(directory, 'blender_assets.cats.txt')
@property
def cache_file(self):
return Path(self.bundle_directory) / f"blender_assets.{self.library.id}.json"
@property
def tmp_cache_file(self):
return Path(bpy.app.tempdir) / f"blender_assets.{self.library.id}.json"
@property
def diff_file(self):
return Path(bpy.app.tempdir, 'diff.json')
@property
def preview_blend(self):
return MODULE_DIR / self.data_type.lower() / "preview.blend"
@property
def preview_assets_file(self):
return Path(bpy.app.tempdir, "preview_assets_file.json")
@property
def addon_prefs(self):
return get_addon_prefs()
@property
def module_type(self):
lib_type = self.library.data_type
if lib_type == 'ACTION':
return action
elif lib_type == 'FILE':
return file
elif lib_type == 'COLLECTION':
return collection
def to_dict(self):
return {p: getattr(self, p) for p in self.bl_rna.properties.keys() if p !='rna_type'}
@property
def format_data(self):
"""Dict for formating template"""
return dict(self.to_dict(), bundle_dir=self.library.library_path)
def fetch(self):
raise Exception('This method need to be define in the adapter')
def norm_file_name(self, name):
return name.replace(' ', '_')
def read_file(self, file):
return read_file(file)
def write_file(self, file, data):
return write_file(file, data)
def copy_file(self, source, destination):
src = Path(source)
dst = Path(destination)
if not src.exists():
print(f'Cannot copy file {src}: file not exist')
return
dst.parent.mkdir(exist_ok=True, parents=True)
if src == dst:
print(f'Cannot copy file {src}: source and destination are the same')
return
print(f'Copy file from {src} to {dst}')
shutil.copy2(str(src), str(dst))
def load_datablocks(self, src, names=None, type='objects', link=True, expr=None, assets_only=False):
"""Link or append a datablock from a blendfile"""
if type.isupper():
type = f'{type.lower()}s'
return load_datablocks(src, names=names, type=type, link=link, expr=expr, assets_only=assets_only)
def get_asset_data(self, asset):
"""Extract asset information on a datablock"""
return dict(
name=asset.name,
author=asset.asset_data.author,
tags=list(asset.asset_data.tags.keys()),
metadata=dict(asset.asset_data),
description=asset.asset_data.description,
)
def get_asset_relative_path(self, name, catalog):
'''Get a relative path for the asset'''
name = self.norm_file_name(name)
return Path(catalog, name, name).with_suffix('.blend')
def get_active_asset_library(self):
asset_handle = bpy.context.asset_file_handle
prefs = get_addon_prefs()
asset_handle = bpy.context.asset_file_handle
lib = None
if '.library_id' in asset_handle.asset_data:
lib_id = asset_handle.asset_data['.library_id']
lib = next((l for l in prefs.libraries if l.id == lib_id), None)
if not lib:
print(f"No library found for id {lib_id}")
if not lib:
lib = self
return lib
def get_active_asset_path(self):
'''Get the full path of the active asset_handle from the asset brower'''
prefs = get_addon_prefs()
asset_handle = bpy.context.asset_file_handle
lib = self.get_active_asset_library()
if 'filepath' in asset_handle.asset_data:
asset_path = asset_handle.asset_data['filepath']
asset_path = lib.adapter.format_path(asset_path)
else:
asset_path = bpy.types.AssetHandle.get_full_library_path(
asset_handle, bpy.context.asset_library_ref
)
return asset_path
def get_image_path(self, name, catalog, filepath):
raise Exception('Need to be defined in the adapter')
def get_video_path(self, name, catalog, filepath):
raise Exception('Need to be defined in the adapter')
def new_asset(self, asset, asset_data):
raise Exception('Need to be defined in the adapter')
def remove_asset(self, asset, asset_data):
raise Exception('Need to be defined in the adapter')
def format_asset_data(self, data):
"""Get a dict for use in template fields"""
return {
'asset_name': data['name'],
'asset_path': Path(data['filepath']),
'catalog': data['catalog'],
'catalog_name': data['catalog'].replace('/', '_'),
}
def format_path(self, template, data={}, **kargs):
if not template:
return None
if data:
data = self.format_asset_data(dict(data, **kargs))
else:
data = kargs
if template.startswith('.'): #the template is relative
template = Path(data['asset_path'], template).as_posix()
params = dict(
**data,
**self.format_data,
)
return Template(template).format(params).resolve()
def find_path(self, template, data, **kargs):
path = self.format_path(template, data, **kargs)
paths = glob(str(path))
if paths:
return Path(paths[0])
def read_asset_description_file(self, asset_path) -> dict:
"""Read the description file of the asset"""
description_path = self.get_description_path(asset_path)
return self.read_file(description_path)
def write_description_file(self, asset_data, asset_path) -> None:
description_path = self.get_description_path(asset_path)
return write_file(description_path, asset_data)
def write_asset(self, asset, asset_path):
Path(asset_path).parent.mkdir(exist_ok=True, parents=True)
bpy.data.libraries.write(
str(asset_path),
{asset},
path_remap="NONE",
fake_user=True,
compress=True
)
def read_catalog(self, directory=None):
"""Read the catalog file of the library target directory or of the specified directory"""
catalog_path = self.get_catalog_path(directory)
if not catalog_path.exists():
return {}
cat_data = {}
for line in catalog_path.read_text(encoding="utf-8").split('\n'):
if line.startswith(('VERSION', '#')) or not line:
continue
cat_id, cat_path, cat_name = line.split(':')
cat_data[cat_path] = {'id':cat_id, 'name':cat_name}
return cat_data
def write_catalog(self, catalog_data, directory=None):
"""Write the catalog file in the library target directory or of the specified directory"""
catalog_path = self.get_catalog_path(directory)
lines = ['VERSION 1', '']
# Add missing parents catalog
norm_data = {}
for cat_path, cat_data in catalog_data.items():
norm_data[cat_path] = cat_data
for p in Path(cat_path).parents[:-1]:
if p in cat_data or p in norm_data:
continue
norm_data[p.as_posix()] = {'id': str(uuid.uuid4()), 'name': '-'.join(p.parts)}
for cat_path, cat_data in sorted(norm_data.items()):
cat_name = cat_data['name'].replace('/', '-')
lines.append(f"{cat_data['id']}:{cat_path}:{cat_name}")
print(f'Catalog writen at: {catalog_path}')
catalog_path.write_text('\n'.join(lines), encoding="utf-8")
def read_cache(self, cache_path=None):
cache_path = cache_path or self.cache_file
print(f'Read cache from {cache_path}')
return self.read_file(cache_path)
def write_cache(self, asset_descriptions, cache_path=None):
cache_path = cache_path or self.cache_file
print(f'cache file writen to {cache_path}')
return write_file(cache_path, list(asset_descriptions))
def prop_rel_path(self, path, prop):
'''Get a filepath relative to a property of the adapter'''
field_prop = '{%s}/'%prop
prop_value = getattr(self, prop)
prop_value = Path(os.path.expandvars(prop_value)).resolve()
rel_path = Path(path).resolve().relative_to(prop_value).as_posix()
return field_prop + rel_path
def write_preview(self, preview, filepath):
if not preview or not filepath:
return
filepath = Path(filepath)
filepath.parent.mkdir(parents=True, exist_ok=True)
img_size = preview.image_size
px = [0] * img_size[0] * img_size[1] * 4
preview.image_pixels_float.foreach_get(px)
img = bpy.data.images.new(name=filepath.name, width=img_size[0], height=img_size[1], is_data=True, alpha=True)
img.pixels.foreach_set(px)
img.filepath_raw = str(filepath.with_suffix('.png'))
img.file_format = 'PNG'
img.save()
def draw_header(self, layout):
"""Draw the header of the Asset Browser Window"""
#layout.separator()
self.module_type.gui.draw_header(layout)
def draw_context_menu(self, layout):
"""Draw the context menu of the Asset Browser Window"""
self.module_type.gui.draw_context_menu(layout)
def generate_blend_preview(self, asset_description):
asset_name = asset_description['name']
catalog = asset_description['catalog']
asset_path = self.format_path(asset_description['filepath'])
dst_image_path = self.get_image_path(asset_name, asset_path, catalog)
if dst_image_path.exists():
return
# Check if a source image exists and if so copying it in the new directory
src_image_path = asset_description.get('image')
if src_image_path:
src_image_path = self.get_template_path(src_image_path, asset_name, asset_path, catalog)
if src_image_path and src_image_path.exists():
self.copy_file(src_image_path, dst_image_path)
return
print(f'Thumbnailing {asset_path} to {dst_image_path}')
blender_thumbnailer = Path(bpy.app.binary_path).parent / 'blender-thumbnailer'
dst_image_path.parent.mkdir(exist_ok=True, parents=True)
subprocess.call([blender_thumbnailer, str(asset_path), str(dst_image_path)])
success = dst_image_path.exists()
if not success:
empty_preview = RESOURCES_DIR / 'empty_preview.png'
self.copy_file(str(empty_preview), str(dst_image_path))
return success
def generate_asset_preview(self, asset_description):
"""Only generate preview when conforming a library"""
#print('\ngenerate_preview', asset_description['filepath'])
scn = bpy.context.scene
#Creating the preview for collection, object or material
camera = scn.camera
vl = bpy.context.view_layer
data_type = self.data_type #asset_description['data_type']
asset_path = self.format_path(asset_description['filepath'])
# Check if a source video exists and if so copying it in the new directory
if self.library.template_video:
for asset_data in asset_description['assets']:
dst_asset_path = self.get_asset_bundle_path(asset_data)
dst_video_path = self.format_path(self.library.template_video, asset_data, filepath=dst_asset_path) #Template(src_video_path).find(asset_data, asset_path=dst_asset_path, **self.format_data)
if dst_video_path.exists():
print(f'The dest video {dst_video_path} already exist')
continue
src_video_template = asset_data.get('video')
if not src_video_template:
continue
src_video_path = self.find_path(src_video_template, asset_data, filepath=asset_path)#Template(src_video_path).find(asset_data, asset_path=dst_asset_path, **self.format_data)
if src_video_path:
print(f'Copy video from {src_video_path} to {dst_video_path}')
self.copy_file(src_video_path, dst_video_path)
# Check if asset as a preview image or need it to be generated
asset_data_names = {}
if self.library.template_image:
for asset_data in asset_description['assets']:
name = asset_data['name']
dst_asset_path = self.get_asset_bundle_path(asset_data)
dst_image_path = self.format_path(self.library.template_image, asset_data, filepath=dst_asset_path)
if dst_image_path.exists():
print(f'The dest image {dst_image_path} already exist')
continue
# Check if a source image exists and if so copying it in the new directory
src_image_template = asset_data.get('image')
if src_image_template:
src_image_path = self.find_path(src_image_template, asset_data, filepath=asset_path)
if src_image_path:
self.copy_file(src_image_path, dst_image_path)
#print(f'Copy image from {src_image_path} to {dst_image_path}')
return
#Store in a dict all asset_data that does not have preview
asset_data_names[name] = dict(asset_data, image_path=dst_image_path)
if not asset_data_names:
# No preview to generate
return
#print('Making Preview for', asset_data_names)
asset_names = list(asset_data_names.keys())
assets = self.load_datablocks(asset_path, names=asset_names, link=True, type=data_type)
for asset in assets:
if not asset:
continue
asset_data = asset_data_names[asset.name]
image_path = asset_data['image_path']
if asset.preview:
print(f'Writing asset preview to {image_path}')
self.write_preview(asset.preview, image_path)
continue
if data_type == 'COLLECTION':
bpy.ops.object.collection_instance_add(name=asset.name)
bpy.ops.view3d.camera_to_view_selected()
instance = vl.objects.active
#scn.collection.children.link(asset)
scn.render.filepath = str(image_path)
print(f'Render asset {asset.name} to {image_path}')
bpy.ops.render.render(write_still=True)
#instance.user_clear()
asset.user_clear()
bpy.data.objects.remove(instance)
bpy.ops.outliner.orphans_purge(do_local_ids=True, do_linked_ids=True, do_recursive=True)
def generate_previews(self, cache=None):
print('Generate previews')
if cache in (None, ''):
cache = self.fetch()
elif isinstance(cache, (Path, str)):
cache = self.read_cache(cache)
#cache_diff.sort(key=lambda x :x['filepath'])
#blend_groups = groupby(cache_diff, key=lambda x :x['filepath'])
#TODO Support all multiple data_type
for asset_description in cache:
if asset_description.get('type', self.data_type) == 'FILE':
self.generate_blend_preview(asset_description)
else:
self.generate_asset_preview(asset_description)
# filepath = asset_description['filepath']
# asset_datas = asset_description["assets"]
# asset_datas.sort(key=lambda x :x.get('type', self.data_type))
# data_type_groups = groupby(asset_datas, key=lambda x :x.get('type', self.data_type))
# for data_type, same_type_asset_datas in data_type_groups:
# asset_names = [a['name'] for a in same_type_asset_datas]
# self.generate_preview(filepath, asset_names, data_type)
def set_asset_preview(self, asset, asset_data):
'''Load an externalize image as preview for an asset'''
asset_path = self.format_path(asset_data['filepath'])
image_template = asset_data.get('image')
if self.library.template_image:
asset_path = self.get_asset_bundle_path(asset_data)
image_template = self.library.template_image
image_path = self.find_path(image_template, asset_data, filepath=asset_path)
if image_path:
#print(f'Set asset preview for {image_path} for {asset}')
with bpy.context.temp_override(id=asset):
bpy.ops.ed.lib_id_load_custom_preview(
filepath=str(image_path)
)
if asset.preview:
return asset.preview
def set_asset_catalog(self, asset, asset_data, catalog_data):
"""Find the catalog if already exist or create it"""
catalog_name = asset_data['catalog']
catalog = catalog_data.get(catalog_name)
if not catalog:
catalog = {'id': str(uuid.uuid4()), 'name': catalog_name}
catalog_data[catalog_name] = catalog
asset.asset_data.catalog_id = catalog['id']
def set_asset_metadata(self, asset, asset_data):
"""Create custom prop to an asset base on provided data"""
metadata = asset_data.get('metadata', {})
library_id = self.library.id
if 'library_id' in asset_data:
library_id = asset_data['library_id']
metadata['.library_id'] = library_id
metadata['filepath'] = asset_data['filepath']
for k, v in metadata.items():
asset.asset_data[k] = v
def set_asset_tags(self, asset, asset_data):
"""Create asset tags base on provided data"""
if 'tags' in asset_data:
for tag in asset.asset_data.tags[:]:
asset.asset_data.tags.remove(tag)
for tag in asset_data['tags']:
if not tag:
continue
asset.asset_data.tags.new(tag, skip_if_exists=True)
def set_asset_info(self, asset, asset_data):
"""Set asset description base on provided data"""
for key in ('author', 'description'):
if key in asset_data:
setattr(asset.asset_data, key, asset_data.get(key) or '')
def get_asset_bundle_path(self, asset_data):
catalog_parts = asset_data['catalog'].split('/') + [asset_data['name']]
sub_path = catalog_parts[:self.library.blend_depth]
blend_name = sub_path[-1].replace(' ', '_').lower()
return Path(self.bundle_directory, *sub_path, blend_name).with_suffix('.blend')
def bundle(self, cache_diff=None):
"""Group all new assets in one or multiple blends for the asset browser"""
if self.data_type not in ('FILE', 'ACTION', 'COLLECTION'):
print(f'{self.data_type} is not supported yet')
return
catalog_data = self.read_catalog() #TODO remove unused catalog
write_cache = False
if not cache_diff:
# Get list of all modifications
asset_descriptions = self.fetch()
cache, cache_diff = self.diff(asset_descriptions)
# Only write complete cache at the end
write_cache = True
#self.generate_previews(asset_descriptions)
self.write_cache(asset_descriptions, self.tmp_cache_file)
bpy.ops.assetlib.generate_previews(name=self.library.name, cache=str(self.tmp_cache_file))
#print()
#print(cache)
#raise Exception()
elif isinstance(cache_diff, (Path, str)):
cache_diff = json.loads(Path(cache_diff).read_text(encoding='utf-8'))
if self.library.blend_depth == 0:
raise Exception('Blender depth must be 1 at min')
#groups = [(cache_diff)]
else:
cache_diff.sort(key=self.get_asset_bundle_path)
groups = groupby(cache_diff, key=self.get_asset_bundle_path)
total_assets = len(cache_diff)
print(f'total_assets={total_assets}')
if total_assets == 0:
print('No assets found')
return
#data_types = self.data_types
#if self.data_types == 'FILE'
i = 0
#assets_to_preview = []
for blend_path, asset_datas in groups:
#blend_name = sub_path[-1].replace(' ', '_').lower()
#blend_path = Path(self.bundle_directory, *sub_path, blend_name).with_suffix('.blend')
if blend_path.exists():
print(f'Opening existing bundle blend: {blend_path}')
bpy.ops.wm.open_mainfile(filepath=str(blend_path))
else:
print(f'Create new bundle blend to: {blend_path}')
bpy.ops.wm.read_homefile(use_empty=True)
for asset_data in asset_datas:
if total_assets <= 100 or i % int(total_assets / 10) == 0:
print(f'Progress: {int(i / total_assets * 100)+1}')
operation = asset_data.get('operation', 'ADD')
asset = getattr(bpy.data, self.data_types).get(asset_data['name'])
if operation == 'REMOVE':
if asset:
getattr(bpy.data, self.data_types).remove(asset)
else:
print(f'ERROR : Remove Asset: {asset_data["name"]} not found in {blend_path}')
continue
if operation == 'MODIFY' and not asset:
print(f'WARNING: Modifiy Asset: {asset_data["name"]} not found in {blend_path} it will be created')
elif operation == 'ADD' or not asset:
if asset:
#raise Exception(f"Asset {asset_data['name']} Already in Blend")
print(f"Asset {asset_data['name']} Already in Blend")
getattr(bpy.data, self.data_types).remove(asset)
print(f"INFO: Add new asset: {asset_data['name']}")
asset = getattr(bpy.data, self.data_types).new(name=asset_data['name'])
else:
print(f'operation {operation} not supported should be in (ADD, REMOVE, MODIFIED)')
continue
asset.asset_mark()
self.set_asset_preview(asset, asset_data)
#if not asset_preview:
# assets_to_preview.append((asset_data['filepath'], asset_data['name'], asset_data['data_type']))
#if self.externalize_data:
# self.write_preview(preview, filepath)
self.set_asset_catalog(asset, asset_data, catalog_data)
self.set_asset_metadata(asset, asset_data)
self.set_asset_tags(asset, asset_data)
self.set_asset_info(asset, asset_data)
i += 1
#self.write_asset_preview_file()
print(f'Saving Blend to {blend_path}')
blend_path.parent.mkdir(exist_ok=True, parents=True)
bpy.ops.wm.save_as_mainfile(filepath=str(blend_path), compress=True)
if write_cache:
self.write_cache(asset_descriptions)
self.write_catalog(catalog_data)
bpy.ops.wm.quit_blender()
def norm_cache(self, cache):
""" Return a new flat list of asset data
the filepath keys are merge with the assets keys"""
if not cache or not isinstance(cache[0], dict):
return []
new_cache = []
for asset_description in cache:
asset_description = asset_description.copy()
if 'assets' in asset_description:
assets = asset_description.pop('assets')
for asset_data in assets:
new_cache.append({**asset_description, **asset_data})
else:
new_cache.append(asset_description)
return new_cache
def diff(self, asset_descriptions=None):
"""Compare the library cache with it current state and return the difference"""
cache = self.read_cache()
if cache is None:
print(f'Fetch The library {self.library.name} for the first time, might be long...')
cache = []
asset_descriptions = asset_descriptions or self.fetch()
#print('\n-------------------------', cache)
cache = {f"{a['filepath']}/{a['name']}": a for a in self.norm_cache(cache)}
new_cache = {f"{a['filepath']}/{a['name']}" : a for a in self.norm_cache(asset_descriptions)}
assets_added = [v for k, v in new_cache.items() if k not in cache]
assets_removed = [v for k, v in cache.items() if k not in new_cache]
assets_modified = [v for k, v in cache.items() if v not in assets_removed and v!= new_cache[k]]
if assets_added:
print(f'{len(assets_added)} Assets Added \n{tuple(a["name"] for a in assets_added[:10])}\n')
if assets_removed:
print(f'{len(assets_removed)} Assets Removed \n{tuple(a["name"] for a in assets_removed[:10])}\n')
if assets_modified:
print(f'{len(assets_modified)} Assets Modified \n{tuple(a["name"] for a in assets_modified[:10])}\n')
assets_added = [dict(a, operation='ADD') for a in assets_added]
assets_removed = [dict(a, operation='REMOVE') for a in assets_removed]
assets_modified = [dict(a, operation='MODIFY') for a in assets_modified]
cache_diff = assets_added + assets_removed + assets_modified
if not cache_diff:
print('No change in the library')
return new_cache, cache_diff
def draw_prefs(self, layout):
"""Draw the options in the addon preference for this adapter"""
annotations = self.__class__.__annotations__
for k, v in annotations.items():
layout.prop(self, k, text=bpy.path.display_name(k))