start refacto
This commit is contained in:
@@ -0,0 +1,54 @@
|
||||
"""
|
||||
Util function for this addon
|
||||
"""
|
||||
|
||||
import bpy
|
||||
from . bl_utils import get_addon_prefs
|
||||
|
||||
|
||||
def thumbnail_blend_file(input_blend, output_img):
|
||||
input_blend = Path(input_blend).resolve()
|
||||
output_img = Path(output_img).resolve()
|
||||
|
||||
print(f'Thumbnailing {input_blend} to {output_img}')
|
||||
blender_thumbnailer = Path(bpy.app.binary_path).parent / 'blender-thumbnailer'
|
||||
|
||||
output_img.parent.mkdir(exist_ok=True, parents=True)
|
||||
|
||||
subprocess.call([blender_thumbnailer, str(input_blend), str(output_img)])
|
||||
|
||||
success = output_img.exists()
|
||||
|
||||
if not success:
|
||||
empty_preview = RESOURCES_DIR / 'empty_preview.png'
|
||||
shutil.copy(str(empty_preview), str(output_img))
|
||||
|
||||
return success
|
||||
|
||||
|
||||
def get_active_library():
|
||||
'''Get the pref library properties from the active library of the asset browser'''
|
||||
prefs = get_addon_prefs()
|
||||
lib_ref = bpy.context.space_data.params.asset_library_reference
|
||||
|
||||
#Check for merged library
|
||||
for l in prefs.libraries:
|
||||
if l.name == lib_ref:
|
||||
return l
|
||||
|
||||
|
||||
def update_library_path(self, context):
|
||||
"""Removing all asset libraries and recreate them"""
|
||||
|
||||
addon_prefs = get_addon_prefs()
|
||||
libs = context.preferences.filepaths.asset_libraries
|
||||
|
||||
for i, lib in reversed(list(enumerate(libs))):
|
||||
if (addon_lib := addon_prefs.libraries.get(lib.name)) and addon_lib.path == lib.path:
|
||||
bpy.ops.preferences.asset_library_remove(index=i)
|
||||
|
||||
for addon_lib in addon_prefs.libraries:
|
||||
if not addon_lib.use:
|
||||
continue
|
||||
bpy.ops.preferences.asset_library_add(directory=str(addon_lib.path))
|
||||
libs[-1].name = addon_lib.name
|
||||
@@ -0,0 +1,485 @@
|
||||
|
||||
"""
|
||||
Generic Blender functions
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from fnmatch import fnmatch
|
||||
from typing import Any, List, Iterable, Optional, Tuple
|
||||
Datablock = Any
|
||||
|
||||
import bpy
|
||||
from bpy_extras import asset_utils
|
||||
#from asset_library.constants import RESOURCES_DIR
|
||||
#from asset_library.common.file_utils import no
|
||||
from os.path import abspath
|
||||
import subprocess
|
||||
|
||||
|
||||
class attr_set():
|
||||
'''Receive a list of tuple [(data_path, "attribute" [, wanted value)] ]
|
||||
entering with-statement : Store existing values, assign wanted value (if any)
|
||||
exiting with-statement: Restore values to their old values
|
||||
'''
|
||||
|
||||
def __init__(self, attrib_list):
|
||||
self.store = []
|
||||
# item = (prop, attr, [new_val])
|
||||
for item in attrib_list:
|
||||
prop, attr = item[:2]
|
||||
self.store.append( (prop, attr, getattr(prop, attr)) )
|
||||
|
||||
for item in attrib_list:
|
||||
prop, attr = item[:2]
|
||||
|
||||
if len(item) >= 3:
|
||||
try:
|
||||
setattr(prop, attr, item[2])
|
||||
except TypeError:
|
||||
print(f'Cannot set attribute {attr} to {prop}')
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_value, exc_traceback):
|
||||
self.restore()
|
||||
|
||||
def restore(self):
|
||||
for prop, attr, old_val in self.store:
|
||||
setattr(prop, attr, old_val)
|
||||
|
||||
|
||||
def unique_name(name, names):
|
||||
if name not in names:
|
||||
return name
|
||||
|
||||
i = 1
|
||||
org_name = name
|
||||
while name in names:
|
||||
name = f'{org_name}.{i:03d}'
|
||||
i += 1
|
||||
|
||||
return name
|
||||
|
||||
|
||||
def get_overriden_col(ob, scene=None):
|
||||
scn = scene or bpy.context.scene
|
||||
|
||||
cols = [c for c in bpy.data.collections if scn.user_of_id(c)]
|
||||
|
||||
return next((c for c in cols if ob in c.all_objects[:]
|
||||
if all(not c.override_library for c in get_col_parents(c))), None)
|
||||
|
||||
|
||||
def get_view3d_persp():
|
||||
windows = bpy.context.window_manager.windows
|
||||
view_3ds = [a for w in windows for a in w.screen.areas if a.type == 'VIEW_3D']
|
||||
view_3d = next((a for a in view_3ds if a.spaces.active.region_3d.view_perspective == 'PERSP'), view_3ds[0])
|
||||
return view_3d
|
||||
|
||||
|
||||
def get_viewport():
|
||||
screen = bpy.context.screen
|
||||
|
||||
areas = [a for a in screen.areas if a.type == 'VIEW_3D']
|
||||
areas.sort(key=lambda x : x.width*x.height)
|
||||
|
||||
return areas[-1]
|
||||
|
||||
|
||||
def biggest_asset_browser_area(screen: bpy.types.Screen) -> Optional[bpy.types.Area]:
|
||||
"""Return the asset browser Area that's largest on screen.
|
||||
|
||||
:param screen: context.window.screen
|
||||
|
||||
:return: the Area, or None if no Asset Browser area exists.
|
||||
"""
|
||||
|
||||
def area_sorting_key(area: bpy.types.Area) -> Tuple[bool, int]:
|
||||
"""Return area size in pixels."""
|
||||
return (area.width * area.height)
|
||||
|
||||
areas = list(suitable_areas(screen))
|
||||
if not areas:
|
||||
return None
|
||||
|
||||
return max(areas, key=area_sorting_key)
|
||||
|
||||
|
||||
def suitable_areas(screen: bpy.types.Screen) -> Iterable[bpy.types.Area]:
|
||||
"""Generator, yield Asset Browser areas."""
|
||||
|
||||
for area in screen.areas:
|
||||
space_data = area.spaces[0]
|
||||
if not asset_utils.SpaceAssetInfo.is_asset_browser(space_data):
|
||||
continue
|
||||
yield area
|
||||
|
||||
|
||||
def area_from_context(context: bpy.types.Context) -> Optional[bpy.types.Area]:
|
||||
"""Return an Asset Browser suitable for the given category.
|
||||
|
||||
Prefers the current Asset Browser if available, otherwise the biggest.
|
||||
"""
|
||||
|
||||
space_data = context.space_data
|
||||
if asset_utils.SpaceAssetInfo.is_asset_browser(space_data):
|
||||
return context.area
|
||||
|
||||
# Try the current screen first.
|
||||
browser_area = biggest_asset_browser_area(context.screen)
|
||||
if browser_area:
|
||||
return browser_area
|
||||
|
||||
for win in context.window_manager.windows:
|
||||
if win.screen == context.screen:
|
||||
continue
|
||||
browser_area = biggest_asset_browser_area(win.screen)
|
||||
if browser_area:
|
||||
return browser_area
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def activate_asset(
|
||||
asset: bpy.types.Action, asset_browser: bpy.types.Area, *, deferred: bool
|
||||
) -> None:
|
||||
"""Select & focus the asset in the browser."""
|
||||
|
||||
space_data = asset_browser.spaces[0]
|
||||
assert asset_utils.SpaceAssetInfo.is_asset_browser(space_data)
|
||||
space_data.activate_asset_by_id(asset, deferred=deferred)
|
||||
|
||||
|
||||
def active_catalog_id(asset_browser: bpy.types.Area) -> str:
|
||||
"""Return the ID of the catalog shown in the asset browser."""
|
||||
return params(asset_browser).catalog_id
|
||||
|
||||
|
||||
def get_asset_space_params(asset_browser: bpy.types.Area) -> bpy.types.FileAssetSelectParams:
|
||||
"""Return the asset browser parameters given its Area."""
|
||||
space_data = asset_browser.spaces[0]
|
||||
assert asset_utils.SpaceAssetInfo.is_asset_browser(space_data)
|
||||
return space_data.params
|
||||
|
||||
|
||||
def refresh_asset_browsers():
|
||||
for area in suitable_areas(bpy.context.screen):
|
||||
bpy.ops.asset.library_refresh({"area": area, 'region': area.regions[3]})
|
||||
|
||||
|
||||
def tag_redraw(screen: bpy.types.Screen) -> None:
|
||||
"""Tag all asset browsers for redrawing."""
|
||||
|
||||
for area in suitable_areas(screen):
|
||||
area.tag_redraw()
|
||||
|
||||
# def get_blender_command(file=None, script=None, background=True, **args):
|
||||
# '''Return a Blender Command as a list to be used in a subprocess'''
|
||||
|
||||
# cmd = [bpy.app.binary_path]
|
||||
|
||||
# if file:
|
||||
# cmd += [str(file)]
|
||||
# if background:
|
||||
# cmd += ['--background']
|
||||
# if script:
|
||||
# cmd += ['--python', str(script)]
|
||||
# if args:
|
||||
# cmd += ['--']
|
||||
# for k, v in args.items():
|
||||
# cmd += [f"--{k.replace('_', '-')}", str(v)]
|
||||
|
||||
# return cmd
|
||||
|
||||
def norm_value(value):
|
||||
if isinstance(value, (tuple, list)):
|
||||
values = []
|
||||
for v in value:
|
||||
if not isinstance(v, str):
|
||||
v = json.dumps(v)
|
||||
values.append(v)
|
||||
|
||||
return values
|
||||
|
||||
if isinstance(value, Path):
|
||||
return str(value)
|
||||
|
||||
if not isinstance(value, str):
|
||||
value = json.dumps(value)
|
||||
return value
|
||||
|
||||
|
||||
def norm_arg(arg_name, format=str.lower, prefix='--', separator='-'):
|
||||
arg_name = norm_str(arg_name, format=format, separator=separator)
|
||||
|
||||
return prefix + arg_name
|
||||
|
||||
|
||||
def get_bl_cmd(blender=None, background=False, focus=True, blendfile=None, script=None, **kargs):
|
||||
cmd = [str(blender)] if blender else [bpy.app.binary_path]
|
||||
|
||||
if background:
|
||||
cmd += ['--background']
|
||||
|
||||
if not focus and not background:
|
||||
cmd += ['--no-window-focus']
|
||||
cmd += ['--window-geometry', '5000', '0', '10', '10']
|
||||
|
||||
cmd += ['--python-use-system-env']
|
||||
|
||||
if blendfile:
|
||||
cmd += [str(blendfile)]
|
||||
|
||||
if script:
|
||||
cmd += ['--python', str(script)]
|
||||
|
||||
if kargs:
|
||||
cmd += ['--']
|
||||
for k, v in kargs.items():
|
||||
k = norm_arg(k)
|
||||
v = norm_value(v)
|
||||
|
||||
cmd += [k]
|
||||
if isinstance(v, (tuple, list)):
|
||||
cmd += v
|
||||
else:
|
||||
cmd += [v]
|
||||
|
||||
return cmd
|
||||
|
||||
|
||||
def get_addon_prefs():
|
||||
addon_name = __package__.split('.')[0]
|
||||
return bpy.context.preferences.addons[addon_name].preferences
|
||||
|
||||
|
||||
def get_col_parents(col, root=None, cols=None):
|
||||
'''Return a list of parents collections of passed col
|
||||
root : Pass a collection to search in (recursive)
|
||||
else search in master collection
|
||||
'''
|
||||
if cols is None:
|
||||
cols = []
|
||||
|
||||
if root == None:
|
||||
root = bpy.context.scene.collection
|
||||
|
||||
for sub in root.children:
|
||||
if sub == col:
|
||||
cols.append(root)
|
||||
|
||||
if len(sub.children):
|
||||
cols = get_col_parents(col, root=sub, cols=cols)
|
||||
return cols
|
||||
|
||||
|
||||
def get_overriden_col(ob, scene=None):
|
||||
'''Get the collection use for making the override'''
|
||||
scn = scene or bpy.context.scene
|
||||
|
||||
cols = [c for c in bpy.data.collections if scn.user_of_id(c)]
|
||||
|
||||
return next((c for c in cols if ob in c.all_objects[:]
|
||||
if all(not c.override_library for c in get_col_parents(c))), None)
|
||||
|
||||
|
||||
def load_assets_from(filepath: Path) -> List[Datablock]:
|
||||
if not has_assets(filepath):
|
||||
# Avoid loading any datablocks when there are none marked as asset.
|
||||
return []
|
||||
|
||||
# Append everything from the file.
|
||||
with bpy.data.libraries.load(str(filepath)) as (
|
||||
data_from,
|
||||
data_to,
|
||||
):
|
||||
for attr in dir(data_to):
|
||||
setattr(data_to, attr, getattr(data_from, attr))
|
||||
|
||||
# Iterate over the appended datablocks to find assets.
|
||||
def loaded_datablocks() -> Iterable[Datablock]:
|
||||
for attr in dir(data_to):
|
||||
datablocks = getattr(data_to, attr)
|
||||
for datablock in datablocks:
|
||||
yield datablock
|
||||
|
||||
loaded_assets = []
|
||||
for datablock in loaded_datablocks():
|
||||
if not getattr(datablock, "asset_data", None):
|
||||
continue
|
||||
|
||||
# Fake User is lost when appending from another file.
|
||||
datablock.use_fake_user = True
|
||||
loaded_assets.append(datablock)
|
||||
return loaded_assets
|
||||
|
||||
|
||||
def has_assets(filepath: Path) -> bool:
|
||||
with bpy.data.libraries.load(str(filepath), assets_only=True) as (
|
||||
data_from,
|
||||
_,
|
||||
):
|
||||
for attr in dir(data_from):
|
||||
data_names = getattr(data_from, attr)
|
||||
if data_names:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def copy_frames(start, end, offset, path):
|
||||
for i in range (start, end):
|
||||
src = path.replace('####', f'{i:04d}')
|
||||
dst = src.replace(src.split('_')[-1].split('.')[0], f'{i+offset:04d}')
|
||||
shutil.copy2(src, dst)
|
||||
|
||||
|
||||
def split_path(path) :
|
||||
try :
|
||||
bone_name = path.split('["')[1].split('"]')[0]
|
||||
except :
|
||||
bone_name = None
|
||||
try :
|
||||
prop_name = path.split('["')[2].split('"]')[0]
|
||||
except :
|
||||
prop_name = path.split('.')[-1]
|
||||
|
||||
return bone_name, prop_name
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def load_datablocks(src, names=None, type='objects', link=True, expr=None, assets_only=False) -> list:
|
||||
return_list = not isinstance(names, str)
|
||||
names = names or []
|
||||
|
||||
if not isinstance(names, (list, tuple)):
|
||||
names = [names]
|
||||
|
||||
if isinstance(expr, str):
|
||||
pattern = expr
|
||||
expr = lambda x : fnmatch(x, pattern)
|
||||
|
||||
with bpy.data.libraries.load(str(src), link=link,assets_only=assets_only) as (data_from, data_to):
|
||||
datablocks = getattr(data_from, type)
|
||||
if expr:
|
||||
names += [i for i in datablocks if expr(i)]
|
||||
elif not names:
|
||||
names = datablocks
|
||||
|
||||
setattr(data_to, type, names)
|
||||
|
||||
datablocks = getattr(data_to, type)
|
||||
|
||||
if return_list:
|
||||
return datablocks
|
||||
|
||||
elif datablocks:
|
||||
return datablocks[0]
|
||||
|
||||
"""
|
||||
# --- Collection handling
|
||||
"""
|
||||
|
||||
def col_as_asset(col, verbose=False):
|
||||
if col is None:
|
||||
return
|
||||
if verbose:
|
||||
print('linking:', col.name)
|
||||
pcol = bpy.data.collections.new(col.name)
|
||||
bpy.context.scene.collection.children.link(pcol)
|
||||
pcol.children.link(col)
|
||||
pcol.asset_mark()
|
||||
return pcol
|
||||
|
||||
def load_col(filepath, name, link=True, override=True, rig_pattern=None, context=None):
|
||||
'''Link a collection by name from a file and override if has armature'''
|
||||
|
||||
# with bpy.data.libraries.load(filepath, link=link) as (data_from, data_to):
|
||||
# data_to.collections = [c for c in data_from.collections if c == name]
|
||||
# if not data_to.collections:
|
||||
# return
|
||||
# return data_to.collections[0]
|
||||
context = context or bpy.context
|
||||
|
||||
col = load_datablocks(filepath, name, link=link, type='collections')
|
||||
|
||||
## create instance object
|
||||
inst = bpy.data.objects.new(col.name, None)
|
||||
inst.instance_collection = col
|
||||
inst.instance_type = 'COLLECTION'
|
||||
context.scene.collection.objects.link(inst)
|
||||
|
||||
# make active
|
||||
inst.select_set(True)
|
||||
context.view_layer.objects.active = inst
|
||||
|
||||
## simple object (no armatures)
|
||||
if not link or not override:
|
||||
return inst
|
||||
if not next((o for o in col.all_objects if o.type == 'ARMATURE'), None):
|
||||
return inst
|
||||
|
||||
## Create the override
|
||||
# Search
|
||||
parent_cols = inst.users_collection
|
||||
child_cols = [child for pcol in parent_cols for child in pcol.children]
|
||||
|
||||
params = {'active_object': inst, 'selected_objects': [inst]}
|
||||
try:
|
||||
bpy.ops.object.make_override_library(params)
|
||||
|
||||
## check which collection is new in parents collection
|
||||
asset_col = next((c for pcol in parent_cols for c in pcol.children if c not in child_cols), None)
|
||||
if not asset_col:
|
||||
print('Overriden, but no collection found !!')
|
||||
return
|
||||
|
||||
for ob in asset_col.all_objects:
|
||||
if ob.type != 'ARMATURE':
|
||||
continue
|
||||
if rig_pattern and not fnmatch(ob.name, rig_pattern):
|
||||
continue
|
||||
|
||||
ob.hide_select = ob.hide_viewport = False
|
||||
ob.select_set(True)
|
||||
context.view_layer.objects.active = ob
|
||||
print(ob.name)
|
||||
return ob
|
||||
|
||||
except Exception as e:
|
||||
print(f'Override failed on {col.name}')
|
||||
print(e)
|
||||
|
||||
return inst
|
||||
|
||||
|
||||
def get_preview(asset_path='', asset_name=''):
|
||||
asset_preview_dir = Path(asset_path).parents[1]
|
||||
name = asset_name.lower()
|
||||
return next((f for f in asset_preview_dir.rglob('*') if f.stem.lower().endswith(name)), None)
|
||||
|
||||
def get_object_libraries(ob):
|
||||
if ob is None:
|
||||
return []
|
||||
|
||||
libraries = [ob.library]
|
||||
if ob.data:
|
||||
libraries += [ob.data.library]
|
||||
|
||||
if ob.type in ('MESH', 'CURVE'):
|
||||
libraries += [m.library for m in ob.data.materials if m]
|
||||
|
||||
filepaths = []
|
||||
for l in libraries:
|
||||
if not l or not l.filepath:
|
||||
continue
|
||||
|
||||
absolute_filepath = abspath(bpy.path.abspath(l.filepath, library=l))
|
||||
if absolute_filepath in filepaths:
|
||||
continue
|
||||
|
||||
filepaths.append(absolute_filepath)
|
||||
|
||||
return filepaths
|
||||
+200
@@ -0,0 +1,200 @@
|
||||
|
||||
from pathlib import Path
|
||||
import uuid
|
||||
import bpy
|
||||
|
||||
|
||||
class CatalogItem:
|
||||
"""Represent a single item of a catalog"""
|
||||
def __init__(self, catalog, path=None, name=None, id=None):
|
||||
|
||||
self.catalog = catalog
|
||||
|
||||
self.path = path
|
||||
self.name = name
|
||||
self.id = str(id or uuid.uuid4())
|
||||
|
||||
if isinstance(self.path, Path):
|
||||
self.path = self.path.as_posix()
|
||||
|
||||
if self.path and not self.name:
|
||||
self.name = self.norm_name(self.path)
|
||||
|
||||
def parts(self):
|
||||
return Path(self.name).parts
|
||||
|
||||
def norm_name(self, name):
|
||||
"""Get a norm name from a catalog_path entry"""
|
||||
return name.replace('/', '-')
|
||||
|
||||
def __repr__(self):
|
||||
return f'CatalogItem(name={self.name}, path={self.path}, id={self.id})'
|
||||
|
||||
|
||||
class CatalogContext:
|
||||
"""Utility class to get catalog relative to the current context asset browser area"""
|
||||
|
||||
@staticmethod
|
||||
def poll():
|
||||
return asset_utils.SpaceAssetInfo.is_asset_browser(bpy.context.space_data)
|
||||
|
||||
@property
|
||||
def id(self):
|
||||
if not self.poll():
|
||||
return
|
||||
|
||||
return bpy.context.space_data.params.catalog_id
|
||||
|
||||
@property
|
||||
def item(self):
|
||||
if not self.poll():
|
||||
return
|
||||
|
||||
return self.get(id=self.active_id)
|
||||
|
||||
@property
|
||||
def path(self):
|
||||
if not self.poll():
|
||||
return
|
||||
|
||||
if self.active_item:
|
||||
return self.active_item.path
|
||||
|
||||
return ''
|
||||
|
||||
|
||||
class Catalog:
|
||||
"""Represent the catalog of the blender asset browser library"""
|
||||
def __init__(self, directory=None):
|
||||
|
||||
self.directory = None
|
||||
self._data = {}
|
||||
|
||||
if directory:
|
||||
self.directory = Path(directory)
|
||||
|
||||
self.context = CatalogContext()
|
||||
|
||||
@property
|
||||
def filepath(self):
|
||||
"""Get the filepath of the catalog text file relative to the directory"""
|
||||
if self.directory:
|
||||
return self.directory /'blender_assets.cats.txt'
|
||||
|
||||
def read(self):
|
||||
"""Read the catalog file of the library target directory or of the specified directory"""
|
||||
|
||||
if not self.filepath or not self.filepath.exists():
|
||||
return {}
|
||||
|
||||
self._data.clear()
|
||||
|
||||
print(f'Read catalog from {self.filepath}')
|
||||
for line in self.filepath.read_text(encoding="utf-8").split('\n'):
|
||||
if line.startswith(('VERSION', '#')) or not line:
|
||||
continue
|
||||
|
||||
cat_id, cat_path, cat_name = line.split(':')
|
||||
self._data[cat_id] = CatalogItem(self, name=cat_name, id=cat_id, path=cat_path)
|
||||
|
||||
return self
|
||||
|
||||
def write(self, sort=True):
|
||||
"""Write the catalog file in the library target directory or of the specified directory"""
|
||||
|
||||
if not self.filepath:
|
||||
raise Exception(f'Cannot write catalog {self} no filepath setted')
|
||||
|
||||
lines = ['VERSION 1', '']
|
||||
|
||||
catalog_items = list(self)
|
||||
if sort:
|
||||
catalog_items.sort(key=lambda x : x.path)
|
||||
|
||||
for catalog_item in catalog_items:
|
||||
lines.append(f"{catalog_item.id}:{catalog_item.path}:{catalog_item.name}")
|
||||
|
||||
print(f'Write Catalog at: {self.filepath}')
|
||||
self.filepath.write_text('\n'.join(lines), encoding="utf-8")
|
||||
|
||||
def get(self, path=None, id=None, fallback=None):
|
||||
"""Found a catalog item by is path or id"""
|
||||
if isinstance(path, Path):
|
||||
path = path.as_posix()
|
||||
|
||||
if id:
|
||||
return self._data.get(id)
|
||||
|
||||
for catalog_item in self:
|
||||
if catalog_item.path == path:
|
||||
return catalog_item
|
||||
|
||||
return fallback
|
||||
|
||||
def remove(self, catalog_item):
|
||||
"""Get a CatalogItem with is path and removing it if found"""
|
||||
|
||||
if not isinstance(catalog_item, CatalogItem):
|
||||
catalog_item = self.get(catalog_item)
|
||||
|
||||
if catalog_item:
|
||||
return self._data.pop(catalog_item.id)
|
||||
|
||||
print(f'Warning: {catalog_item} cannot be remove, not in {self}')
|
||||
return None
|
||||
|
||||
def add(self, catalog_path):
|
||||
"""Adding a CatalogItem with the missing parents"""
|
||||
|
||||
# Add missing parents catalog
|
||||
for parent in Path(catalog_path).parents[:-1]:
|
||||
print(parent, self.get(parent))
|
||||
if self.get(parent):
|
||||
continue
|
||||
|
||||
cat_item = CatalogItem(self, path=parent)
|
||||
self._data[cat_item.id] = cat_item
|
||||
|
||||
cat_item = self.get(catalog_path)
|
||||
if not cat_item:
|
||||
cat_item = CatalogItem(self, path=catalog_path)
|
||||
self._data[cat_item.id] = cat_item
|
||||
|
||||
return cat_item
|
||||
|
||||
def update(self, catalogs):
|
||||
'Add or remove catalog entries if on the list given or not'
|
||||
|
||||
catalogs = set(catalogs) # Remove doubles
|
||||
|
||||
added = [c for c in catalogs if not self.get(path=c)]
|
||||
removed = [c.path for c in self if c.path not in catalogs]
|
||||
|
||||
if added:
|
||||
print(f'{len(added)} Catalog Entry Added \n{tuple(c.name for c in added[:10])}...\n')
|
||||
if removed:
|
||||
print(f'{len(removed)} Catalog Entry Removed \n{tuple(c.name for c in removed[:10])}...\n')
|
||||
|
||||
for catalog_item in removed:
|
||||
self.remove(catalog_item)
|
||||
|
||||
for catalog_item in added:
|
||||
self.add(catalog_item)
|
||||
|
||||
def __iter__(self):
|
||||
return self._data.values().__iter__()
|
||||
|
||||
def __getitem__(self, key):
|
||||
if isinstance(key, int):
|
||||
return self._data.values()[key]
|
||||
|
||||
return self._data[key]
|
||||
|
||||
def __contains__(self, item):
|
||||
if isinstance(item, str): # item is the id
|
||||
return item in self._data
|
||||
else:
|
||||
return item in self
|
||||
|
||||
def __repr__(self):
|
||||
return f'Catalog(filepath={self.filepath})'
|
||||
@@ -0,0 +1,318 @@
|
||||
|
||||
"""Generic python functions to make operation on file and names"""
|
||||
|
||||
import fnmatch
|
||||
import json
|
||||
import platform
|
||||
import re
|
||||
import subprocess
|
||||
import unicodedata
|
||||
import os
|
||||
from pathlib import Path
|
||||
import importlib
|
||||
import sys
|
||||
import shutil
|
||||
|
||||
import contextlib
|
||||
|
||||
@contextlib.contextmanager
|
||||
def cd(path):
|
||||
"""Changes working directory and returns to previous on exit."""
|
||||
prev_cwd = Path.cwd()
|
||||
os.chdir(path)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
os.chdir(prev_cwd)
|
||||
|
||||
def install_module(module_name, package_name=None):
|
||||
'''Install a python module with pip or return it if already installed'''
|
||||
try:
|
||||
module = importlib.import_module(module_name)
|
||||
except ModuleNotFoundError:
|
||||
print(f'Installing Module {module_name} ....')
|
||||
|
||||
subprocess.call([sys.executable, '-m', 'ensurepip'])
|
||||
subprocess.call([sys.executable, '-m', 'pip', 'install', package_name or module_name])
|
||||
|
||||
module = importlib.import_module(module_name)
|
||||
|
||||
return module
|
||||
|
||||
def import_module_from_path(path):
|
||||
from importlib import util
|
||||
|
||||
try:
|
||||
path = Path(path)
|
||||
spec = util.spec_from_file_location(path.stem, str(path))
|
||||
mod = util.module_from_spec(spec)
|
||||
|
||||
spec.loader.exec_module(mod)
|
||||
|
||||
return mod
|
||||
except Exception as e:
|
||||
print(f'Cannot import file {path}')
|
||||
print(e)
|
||||
|
||||
def norm_str(string, separator='_', format=str.lower, padding=0):
|
||||
string = str(string)
|
||||
string = string.replace('_', ' ')
|
||||
string = string.replace('-', ' ')
|
||||
string = re.sub('[ ]+', ' ', string)
|
||||
string = re.sub('[ ]+\/[ ]+', '/', string)
|
||||
string = string.strip()
|
||||
|
||||
if format:
|
||||
string = format(string)
|
||||
|
||||
# Padd rightest number
|
||||
string = re.sub(r'(\d+)(?!.*\d)', lambda x : x.group(1).zfill(padding), string)
|
||||
|
||||
string = string.replace(' ', separator)
|
||||
string = unicodedata.normalize('NFKD', string).encode('ASCII', 'ignore').decode("utf-8")
|
||||
|
||||
return string
|
||||
|
||||
def remove_version(filepath):
|
||||
pattern = '_v[0-9]+\.'
|
||||
search = re.search(pattern, filepath)
|
||||
|
||||
if search:
|
||||
filepath = filepath.replace(search.group()[:-1], '')
|
||||
|
||||
return Path(filepath).name
|
||||
|
||||
def is_exclude(name, patterns) -> bool:
|
||||
# from fnmatch import fnmatch
|
||||
if not isinstance(patterns, (list,tuple)) :
|
||||
patterns = [patterns]
|
||||
return any([fnmatch(name, p) for p in patterns])
|
||||
|
||||
def get_last_files(root, pattern=r'_v\d{3}\.\w+', only_matching=False, ex_file=None, ex_dir=None, keep=1, verbose=False) -> list:
|
||||
'''Recursively get last(s) file(s) (when there is multiple versions) in passed directory
|
||||
root -> str: Filepath of the folder to scan.
|
||||
pattern -> str: Regex pattern to group files.
|
||||
only_matching -> bool: Discard files that aren't matched by regex pattern.
|
||||
ex_file -> list : List of fn_match pattern to exclude files.
|
||||
ex_dir -> list : List of fn_match pattern of directory name to skip.
|
||||
keep -> int: Number of lasts versions to keep when there are mutliple versionned files (e.g: 1 keep only last).
|
||||
verbose -> bool: Print infos in console.
|
||||
'''
|
||||
|
||||
files = []
|
||||
if ex_file is None:
|
||||
all_items = [f for f in os.scandir(root)]
|
||||
else:
|
||||
all_items = [f for f in os.scandir(root) if not is_exclude(f.name, ex_file)]
|
||||
|
||||
allfiles = [f for f in all_items if f.is_file()]
|
||||
# Need to sort to effectively group separated key in list
|
||||
allfiles.sort(key=lambda x: x.name)
|
||||
|
||||
dirs = [f for f in all_items if f.is_dir()]
|
||||
|
||||
for i in range(len(allfiles)-1,-1,-1):# fastest way to iterate on index in reverse
|
||||
if not re.search(pattern, allfiles[i].name):
|
||||
if only_matching:
|
||||
allfiles.pop(i)
|
||||
else:
|
||||
files.append(allfiles.pop(i).path)
|
||||
|
||||
# separate remaining files in prefix grouped lists
|
||||
lilist = [list(v) for k, v in groupby(allfiles, key=lambda x: re.split(pattern, x.name)[0])]
|
||||
|
||||
# get only item last of each sorted grouplist
|
||||
for l in lilist:
|
||||
versions = sorted(l, key=lambda x: x.name)[-keep:] # exclude older
|
||||
for f in versions:
|
||||
files.append(f.path)
|
||||
|
||||
if verbose and len(l) > 1:
|
||||
print(f'{root}: keep {str([x.name for x in versions])} out of {len(l)} elements')
|
||||
|
||||
for d in dirs: # recursively treat all detected directory
|
||||
if ex_dir and is_exclude(d.name, ex_dir):
|
||||
# skip folder with excluded name
|
||||
continue
|
||||
files += get_last_files(
|
||||
d.path, pattern=pattern, only_matching=only_matching, ex_file=ex_file, ex_dir=ex_dir, keep=keep)
|
||||
|
||||
return sorted(files)
|
||||
|
||||
def copy_file(src, dst, only_new=False, only_recent=False):
|
||||
if dst.exists():
|
||||
if only_new:
|
||||
return
|
||||
elif only_recent and dst.stat().st_mtime >= src.stat().st_mtime:
|
||||
return
|
||||
|
||||
dst.parent.mkdir(exist_ok=True, parents=True)
|
||||
print(f'Copy file from {src} to {dst}')
|
||||
if platform.system() == 'Windows':
|
||||
subprocess.call(['copy', str(src), str(dst)], shell=True)
|
||||
else:
|
||||
subprocess.call(['cp', str(src), str(dst)])
|
||||
|
||||
def copy_dir(src, dst, only_new=False, only_recent=False, excludes=['.*'], includes=[]):
|
||||
src, dst = Path(src), Path(dst)
|
||||
|
||||
if includes:
|
||||
includes = r'|'.join([fnmatch.translate(x) for x in includes])
|
||||
if excludes:
|
||||
excludes = r'|'.join([fnmatch.translate(x) for x in excludes])
|
||||
|
||||
if dst.is_dir():
|
||||
dst.mkdir(exist_ok=True, parents=True)
|
||||
else:
|
||||
dst.parent.mkdir(exist_ok=True, parents=True)
|
||||
|
||||
if src.is_file():
|
||||
copy_file(src, dst, only_new=only_new, only_recent=only_recent)
|
||||
|
||||
elif src.is_dir():
|
||||
src_files = list(src.rglob('*'))
|
||||
if excludes:
|
||||
src_files = [f for f in src_files if not re.match(excludes, f.name)]
|
||||
|
||||
if includes:
|
||||
src_files = [f for f in src_files if re.match(includes, f.name)]
|
||||
|
||||
dst_files = [dst/f.relative_to(src) for f in src_files]
|
||||
|
||||
for src_file, dst_file in zip(src_files, dst_files) :
|
||||
if src_file.is_dir():
|
||||
dst_file.mkdir(exist_ok=True, parents=True)
|
||||
else:
|
||||
copy_file(src_file, dst_file, only_new=only_new, only_recent=only_recent)
|
||||
|
||||
|
||||
def open_file(filepath, select=False):
|
||||
'''Open a filepath inside the os explorer'''
|
||||
|
||||
if platform.system() == 'Darwin': # macOS
|
||||
cmd = ['open']
|
||||
elif platform.system() == 'Windows': # Windows
|
||||
cmd = ['explorer']
|
||||
if select:
|
||||
cmd += ['/select,']
|
||||
else: # linux variants
|
||||
cmd = ['xdg-open']
|
||||
if select:
|
||||
cmd = ['nemo']
|
||||
|
||||
cmd += [str(filepath)]
|
||||
subprocess.Popen(cmd)
|
||||
|
||||
def open_blender_file(filepath=None):
|
||||
filepath = filepath or bpy.data.filepath
|
||||
|
||||
cmd = sys.argv
|
||||
|
||||
# if no filepath, use command as is to reopen blender
|
||||
if filepath != '':
|
||||
if len(cmd) > 1 and cmd[1].endswith('.blend'):
|
||||
cmd[1] = str(filepath)
|
||||
else:
|
||||
cmd.insert(1, str(filepath))
|
||||
|
||||
subprocess.Popen(cmd)
|
||||
|
||||
def read_file(path):
|
||||
'''Read a file with an extension in (json, yaml, yml, txt)'''
|
||||
|
||||
exts = ('.json', '.yaml', '.yml', '.txt')
|
||||
|
||||
if not path:
|
||||
print('Try to read empty file')
|
||||
|
||||
path = Path(path)
|
||||
if not path.exists():
|
||||
print('File not exist', path)
|
||||
return
|
||||
|
||||
if path.suffix not in exts:
|
||||
print(f'Cannot read file {path}, extension must be in {exts}')
|
||||
return
|
||||
|
||||
txt = path.read_text()
|
||||
data = None
|
||||
|
||||
if path.suffix.lower() in ('.yaml', '.yml'):
|
||||
yaml = install_module('yaml')
|
||||
try:
|
||||
data = yaml.safe_load(txt)
|
||||
except Exception:
|
||||
print(f'Could not load yaml file {path}')
|
||||
return
|
||||
elif path.suffix.lower() == '.json':
|
||||
try:
|
||||
data = json.loads(txt)
|
||||
except Exception:
|
||||
print(f'Could not load json file {path}')
|
||||
return
|
||||
else:
|
||||
data = txt
|
||||
|
||||
return data
|
||||
|
||||
def write_file(path, data, indent=4):
|
||||
'''Read a file with an extension in (json, yaml, yml, text)'''
|
||||
|
||||
exts = ('.json', '.yaml', '.yml', '.txt')
|
||||
|
||||
if not path:
|
||||
print('Try to write empty file')
|
||||
|
||||
path = Path(path)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if path.suffix not in exts:
|
||||
print(f'Cannot read file {path}, extension must be in {exts}')
|
||||
return
|
||||
|
||||
if path.suffix.lower() in ('.yaml', '.yml'):
|
||||
yaml = install_module('yaml')
|
||||
try:
|
||||
path.write_text(yaml.dump(data), encoding='utf8')
|
||||
except Exception as e:
|
||||
print(e)
|
||||
print(f'Could not write yaml file {path}')
|
||||
return
|
||||
elif path.suffix.lower() == '.json':
|
||||
try:
|
||||
path.write_text(json.dumps(data, indent=indent), encoding='utf8')
|
||||
except Exception as e:
|
||||
print(e)
|
||||
print(f'Could not write json file {path}')
|
||||
return
|
||||
else:
|
||||
data = path.write_text(data, encoding='utf8')
|
||||
|
||||
|
||||
def synchronize(src, dst, only_new=False, only_recent=False, clear=False):
|
||||
|
||||
#actionlib_dir = get_actionlib_dir(custom=custom)
|
||||
#local_actionlib_dir = get_actionlib_dir(local=True, custom=custom)
|
||||
|
||||
try:
|
||||
if clear and Path(dst).exists():
|
||||
shutil.rmtree(dst)
|
||||
|
||||
#set_actionlib_dir(custom=custom)
|
||||
|
||||
script = Path(__file__).parent / 'synchronize.py'
|
||||
|
||||
cmd = [
|
||||
sys.executable,
|
||||
script,
|
||||
'--src', str(src),
|
||||
'--dst', str(dst),
|
||||
'--only-new', json.dumps(only_new),
|
||||
'--only-recent', json.dumps(only_recent),
|
||||
]
|
||||
|
||||
subprocess.Popen(cmd)
|
||||
|
||||
except Exception as e:
|
||||
print(e)
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
import re
|
||||
import os
|
||||
from pathlib import Path
|
||||
from fnmatch import fnmatch
|
||||
from glob import glob
|
||||
import string
|
||||
|
||||
|
||||
class TemplateFormatter(string.Formatter):
|
||||
def format_field(self, value, format_spec):
|
||||
if isinstance(value, str):
|
||||
spec, sep = [*format_spec.split(':'), None][:2]
|
||||
|
||||
if sep:
|
||||
value = value.replace('_', ' ')
|
||||
value = value = re.sub(r'([a-z])([A-Z])', rf'\1{sep}\2', value)
|
||||
value = value.replace(' ', sep)
|
||||
|
||||
if spec == 'u':
|
||||
value = value.upper()
|
||||
elif spec == 'l':
|
||||
value = value.lower()
|
||||
elif spec == 't':
|
||||
value = value.title()
|
||||
|
||||
return super().format(value, format_spec)
|
||||
|
||||
class Template:
|
||||
field_pattern = re.compile(r'{(\w+)\*{0,2}}')
|
||||
field_pattern_recursive = re.compile(r'{(\w+)\*{2}}')
|
||||
|
||||
def __init__(self, template):
|
||||
#asset_data_path = Path(lib_path) / ASSETLIB_FILENAME
|
||||
|
||||
self.raw = template
|
||||
self.formatter = TemplateFormatter()
|
||||
|
||||
@property
|
||||
def glob_pattern(self):
|
||||
pattern = self.field_pattern_recursive.sub('**', self.raw)
|
||||
pattern = self.field_pattern.sub('*', pattern)
|
||||
return pattern
|
||||
|
||||
@property
|
||||
def re_pattern(self):
|
||||
pattern = self.field_pattern_recursive.sub('([\\\w -_.\/]+)', self.raw)
|
||||
pattern = self.field_pattern.sub('([\\\w -_.]+)', pattern)
|
||||
pattern = pattern.replace('?', '.')
|
||||
pattern = pattern.replace('*', '.*')
|
||||
|
||||
return re.compile(pattern)
|
||||
|
||||
@property
|
||||
def fields(self):
|
||||
return self.field_pattern.findall(self.raw)
|
||||
#return [f or '0' for f in fields]
|
||||
|
||||
def parse(self, path):
|
||||
|
||||
path = Path(path).as_posix()
|
||||
|
||||
res = self.re_pattern.findall(path)
|
||||
if not res:
|
||||
print('Could not parse {path} with {self.re_pattern}')
|
||||
return {}
|
||||
|
||||
fields = self.fields
|
||||
|
||||
if len(fields) == 1:
|
||||
field_values = res
|
||||
else:
|
||||
field_values = res[0]
|
||||
|
||||
return {k:v for k,v in zip(fields, field_values)}
|
||||
|
||||
def norm_data(self, data):
|
||||
norm_data = {}
|
||||
for k, v in data.items():
|
||||
|
||||
if isinstance(v, Path):
|
||||
v = v.as_posix()
|
||||
|
||||
norm_data[k] = v
|
||||
|
||||
return norm_data
|
||||
|
||||
def format(self, data=None, **kargs):
|
||||
|
||||
data = {**(data or {}), **kargs}
|
||||
|
||||
try:
|
||||
#print('FORMAT', self.raw, data)
|
||||
path = self.formatter.format(self.raw, **self.norm_data(data))
|
||||
except KeyError as e:
|
||||
print(f'Cannot format {self.raw} with {data}, field {e} is missing')
|
||||
return
|
||||
|
||||
path = os.path.expandvars(path)
|
||||
return Path(path)
|
||||
|
||||
def glob(self, directory, pattern=None):
|
||||
'''If pattern is given it need to be absolute'''
|
||||
if pattern is None:
|
||||
pattern = Path(directory, self.glob_pattern).as_posix()
|
||||
|
||||
for entry in os.scandir(directory):
|
||||
entry_path = Path(entry.path)
|
||||
if entry.is_file() and fnmatch(entry_path.as_posix(), pattern):
|
||||
yield entry_path
|
||||
elif entry.is_dir():
|
||||
yield from self.glob(entry.path, pattern)
|
||||
|
||||
def find(self, data, **kargs):
|
||||
pattern = self.format(data, **kargs)
|
||||
|
||||
pattern_str = str(pattern)
|
||||
if '*' not in pattern_str and '?' not in pattern_str:
|
||||
return pattern
|
||||
|
||||
paths = glob(pattern.as_posix())
|
||||
if paths:
|
||||
return Path(paths[0])
|
||||
|
||||
#return pattern
|
||||
|
||||
def __repr__(self):
|
||||
return f'Template({self.raw})'
|
||||
Reference in New Issue
Block a user