"""Ways to operate on and ask questions of nxt layers.
"""
# Built-in
import os
import json
import copy
import logging
import time
import sys
from collections import OrderedDict
# Internal
if sys.version_info[0] == 2:
import nxt_io
import nxt_path
import nxt_node
from constants import GRAPH_VERSION
from runtime import Console
else:
from . import nxt_io
from . import nxt_path
from . import nxt_node
from .constants import GRAPH_VERSION
from .runtime import Console
from . import UNTITLED
logger = logging.getLogger(__name__)
[docs]class SAVE_KEY(object):
VERSION = 'version'
ALIAS = 'alias'
MUTE = 'mute'
SOLO = 'solo'
REFERENCES = 'references'
COMP_ORVERRIDES = 'comp_overrides'
COLOR = 'color'
META_DATA = 'meta_data'
NODES = 'nodes'
CWD = 'cwd'
FILEPATH = 'filepath'
REAL_PATH = 'real_path'
ATTRS = 'attrs'
NAME = 'name'
LOCK = 'lock'
[docs]class LAYERS(object):
COMP = '<display>'
TARGET = '<target>'
TOP = '<top>'
[docs]class AUTHORING:
CREATE = 'create'
REFERENCE = 'reference'
ABOVE = 0
BELOW = 1
[docs]class LayerReturnTypes(object):
Node = 'Node' # Return a node or list of nodes
Path = 'Path' # Return a node path or list of node paths
NodeTable = 'NodeTable' # Return a list of lists [[NodePath, Node]]
NameDict = 'NameDict' # Return a dict where k is node NAME and v is node
PathDict = 'PathDict' # Return a dict where k is node PATH and v is node
Boolean = 'Boolean' # Return if there is at least one child
[docs]class SpecLayer(object):
"""Layer object representing layer data from disk.
"""
RETURNS = LayerReturnTypes
[docs] @classmethod
def load_from_filepath(cls, filepath):
"""Given a filepath, return a SpecLayer loaded from that path.
:param filepath: path to file to load
:type filepath: str
:return: Loaded SpecLayer
:rtype: SpecLayer
"""
layer_data = nxt_io.load_file_data(filepath)
return cls.load_from_layer_data(layer_data)
[docs] @classmethod
def load_from_layer_data(cls, layer_data):
return cls(layer_data)
def __init__(self, layer_data=None):
"""Create a SpecLayer, optionally from layer data.
:param layer_data: layer data dictionary, defaults to None
:type layer_data: dict, optional
"""
if layer_data is None:
layer_data = {}
self._node_table = []
self._cached_children = {}
self._cached_implied_children = {}
# Whether _cached_children holds an entry for every path in the
# layer. See _build_child_cache.
self._bulk_child_cache_valid = False
# Ancestor path -> implied child paths, for every path in the layer.
# See _build_implied_index.
self._implied_index = {}
self._implied_index_valid = False
self._implied_index_len = -1
self._name = UNTITLED
self._layer_idx = 0
self.mute = False
self.solo = False
self.filepath = layer_data.get(SAVE_KEY.FILEPATH)
self.real_path = layer_data.get(SAVE_KEY.REAL_PATH)
self.cwd = layer_data.get(SAVE_KEY.CWD)
if self.filepath:
file_name = os.path.basename(self.filepath)
else:
file_name = UNTITLED
self.alias = layer_data.get(SAVE_KEY.ALIAS, file_name)
self.comp_overrides = layer_data.get(SAVE_KEY.COMP_ORVERRIDES, {})
self.sub_layer_paths = []
self.sub_layers = []
for layer_path in layer_data.get(SAVE_KEY.REFERENCES, []):
if not layer_path:
continue
self.sub_layer_paths += [layer_path]
self.sub_layers += [{SAVE_KEY.FILEPATH: layer_path}]
self.parent_layer = layer_data.get('parent_layer', None)
meta_data = layer_data.get(SAVE_KEY.META_DATA, {})
# Fixme: Shouldn't this use the same method as solo/mute/lock?
self.positions = meta_data.get(META_DATA_KEY.POSITIONS, {})
self.collapse = meta_data.get(META_DATA_KEY.COLLAPSE, {})
self.aliases = meta_data.get(META_DATA_KEY.ALIASES, {})
self.colors = meta_data.get(META_DATA_KEY.COLORS, {})
self.color = layer_data.get(SAVE_KEY.COLOR, None)
self.lock = layer_data.get(SAVE_KEY.LOCK, False)
self.locks = meta_data.get(META_DATA_KEY.LOCKS, {})
self._input_metadata = meta_data
self.spec_list = []
self._nodes_path_as_key = {}
self._nodes_node_as_key = {}
self._construct_node_specs(layer_data)
self.refresh()
[docs] def reload_from_data(self, layer_data):
"""Replace everything this layer holds with what the data says.
The layer object is kept. A stage holds its layers by object and
so does everything looking at one -- the target layer, the
display layer, the layer tree, the source layer recorded on every
node -- so building a new layer and swapping it in would leave
all of those pointing at a layer that is not in the graph.
What is not in the data is what does not belong to it: where the
layer sits in the stack, which layer references it, which file it
came from, and whether it is locked, which is a local opinion
nobody writes to a file. Those are carried over.
What a layer references does come from the data, because that is
what the file says this layer is built on, and the layers already
loaded behind the references that survive are kept. Putting the
stage's stack back in step with the new list -- loading what
arrived, letting go of what went, and stacking them in the order
the file gives -- is `Stage.settle_references`, because the stack
is the stage's to hold, not one layer's.
:param layer_data: clean dict of an nxt save file
:type layer_data: dict
"""
wired = {}
for sub_layer_data in self.sub_layers:
sub_layer = sub_layer_data.get('layer')
if sub_layer is not None:
wired[sub_layer_data[SAVE_KEY.FILEPATH]] = sub_layer
layer_idx = self._layer_idx
layer_data = dict(layer_data)
layer_data['parent_layer'] = self.parent_layer
for key, fallback in ((SAVE_KEY.FILEPATH, self.filepath),
(SAVE_KEY.REAL_PATH, self.real_path),
(SAVE_KEY.CWD, self.cwd),
(SAVE_KEY.LOCK, self.lock)):
if layer_data.get(key) is None:
layer_data[key] = fallback
self.__init__(layer_data)
self._layer_idx = layer_idx
for sub_layer_data in self.sub_layers:
sub_layer = wired.get(sub_layer_data[SAVE_KEY.FILEPATH])
if sub_layer is not None:
sub_layer_data['layer'] = sub_layer
[docs] def get_cwd(self):
path = getattr(self, SAVE_KEY.CWD)
if path:
return path
path = self.real_path
if path:
return os.path.dirname(path)
return os.getcwd()
def _construct_node_specs(self, layer_data):
self.spec_list = []
self._nodes_path_as_key = {}
self._nodes_node_as_key = {}
nodes = order_nodes_dict(layer_data.get(SAVE_KEY.NODES, {}))
for node_path, node_data in nodes.items():
parent_path = nxt_path.get_parent_path(node_path)
root_name = nxt_path.node_name_from_node_path(node_path)
node_data['name'] = root_name
node_data[nxt_node.INTERNAL_ATTRS.SOURCE_LAYER] = self.real_path
root_spec_node = nxt_node.create_spec_node(node_data, self,
parent_path=parent_path)
root_parent_path = getattr(root_spec_node,
nxt_node.INTERNAL_ATTRS.PARENT_PATH)
root_node_path = nxt_path.join_node_paths(root_parent_path,
root_name)
self.spec_list += [root_spec_node]
self._nodes_path_as_key[root_node_path] = root_spec_node
self._nodes_node_as_key[root_spec_node] = root_node_path
self.clear_node_child_cache(root_node_path)
[docs] def refresh(self):
"""Re-populates the node table, and sets nodes' source layer.
"""
self._node_table = []
for path, node in self._nodes_path_as_key.items():
for attr in get_node_local_attr_names(path, [self]):
source_attr = attr + nxt_node.META_ATTRS.SOURCE
setattr(node, source_attr, (self.real_path, path))
node_ns = nxt_path.str_path_to_node_namespace(path)
self._node_table += [[node_ns, node]]
self.clear_node_child_cache(path)
self.sort_node_table()
[docs] def sort_node_table(self):
self._node_table = sort_multidimensional_list(self._node_table,
sort_by_idx=0)
[docs] def layer_idx(self):
return self._layer_idx
[docs] def lookup(self, node_path):
return self._nodes_path_as_key.get(node_path)
[docs] def node_exists(self, node_path):
return bool(self.lookup(node_path))
[docs] def get_exec_in(self, node_path):
node = self.lookup(node_path)
parent_path = getattr(node, nxt_node.INTERNAL_ATTRS.PARENT_PATH)
if node and parent_path == nxt_path.WORLD:
try:
return getattr(node, nxt_node.INTERNAL_ATTRS.EXECUTE_IN)
except AttributeError:
pass
return None
[docs] def get_node_path(self, node):
return self._nodes_node_as_key.get(node)
[docs] def ancestors(self, node_path, return_type=LayerReturnTypes.Node,
include_implied=False):
if include_implied and return_type != LayerReturnTypes.Path:
raise TypeError('When including implied nodes, {} is an '
'unsupported return type'.format(return_type))
if include_implied:
return nxt_path.all_ancestor_paths(node_path)
ancestor_nodes = []
ancestor_paths = []
node_table = []
name_dict = {}
if sys.version_info[0] == 2:
node_path_is_str = isinstance(node_path, basestring)
else:
node_path_is_str = isinstance(node_path, str)
if not node_path_is_str:
node_path = self.get_node_path(node_path)
node = self.lookup(node_path)
parent_path = getattr(node, nxt_node.INTERNAL_ATTRS.PARENT_PATH)
if node:
while parent_path != nxt_path.WORLD:
_node = self.lookup(parent_path)
if _node is None:
parent_path = nxt_path.get_parent_path(parent_path)
continue
if return_type == LayerReturnTypes.Boolean:
return True
elif return_type == LayerReturnTypes.Node:
ancestor_nodes += [_node]
elif return_type == LayerReturnTypes.Path:
ancestor_paths += [parent_path]
elif return_type == LayerReturnTypes.NodeTable:
node_table += [[parent_path, _node]]
elif return_type == LayerReturnTypes.NameDict:
key = getattr(node, nxt_node.INTERNAL_ATTRS.NAME)
name_dict[key] = _node
node = _node
ppath_attr = nxt_node.INTERNAL_ATTRS.PARENT_PATH
parent_path = getattr(node, ppath_attr)
if return_type == LayerReturnTypes.Node:
return ancestor_nodes
elif return_type == LayerReturnTypes.Path:
return ancestor_paths
elif return_type == LayerReturnTypes.NodeTable:
return node_table
elif return_type == LayerReturnTypes.NameDict:
return name_dict
elif return_type == LayerReturnTypes.Boolean:
return False
[docs] def children(self, node_path=nxt_path.WORLD,
return_type=LayerReturnTypes.Node, ordered=False,
include_implied=False):
if include_implied and return_type != LayerReturnTypes.Path:
raise ValueError('When including implied, can only return path. '
'Nothing else exists for implicit nodes.')
children_nodes = []
children_paths = []
node_table = []
name_dict = {}
if not node_path:
if return_type in (LayerReturnTypes.Node, LayerReturnTypes.Path,
LayerReturnTypes.NodeTable):
return []
elif return_type is LayerReturnTypes.NameDict:
return {}
elif return_type is LayerReturnTypes.Boolean:
return False
else:
logger.error('Invalid return type provided')
return None
child_order = []
implied_children = []
# Look up real children cache
children_cache = self._cached_children.get(node_path)
if node_path == nxt_path.WORLD:
children_cache = None
elif children_cache is None and not self._bulk_child_cache_valid:
# Filling this cache one path at a time walked every node in the
# layer per lookup, making a comp O(n^2) in node count. Fill it
# for every path in one pass instead.
self._build_child_cache()
children_cache = self._cached_children.get(node_path)
if children_cache is not None:
children_nodes = children_cache[LayerReturnTypes.Node][:]
children_paths = children_cache[LayerReturnTypes.Path][:]
node_table = children_cache[LayerReturnTypes.NodeTable][:]
name_dict = copy.copy(children_cache[LayerReturnTypes.NameDict])
cache_real = False
else:
self._cached_children[node_path] = {LayerReturnTypes.Node:
children_nodes,
LayerReturnTypes.Path:
children_paths,
LayerReturnTypes.NodeTable:
node_table,
LayerReturnTypes.NameDict:
name_dict}
cache_real = True
# Lookup implied cache
cache_implied = False
if include_implied:
if node_path is nxt_path.WORLD:
implied_c_cache = None
else:
implied_c_cache = self._cached_implied_children.get(node_path)
if implied_c_cache is not None:
implied_children = implied_c_cache[LayerReturnTypes.Path][:]
if return_type is LayerReturnTypes.Boolean and not cache_real:
return bool(implied_children + children_paths)
elif node_path is not nxt_path.WORLD:
# Same answer the scan below would give, without walking
# every node in the layer for this one path.
implied_children = self._implied_children_for(node_path)
path_implied = {LayerReturnTypes.Path: implied_children}
self._cached_implied_children[node_path] = path_implied
else:
# The world path is never served from the index, it is
# always recomputed. See _build_implied_index.
path_implied = {LayerReturnTypes.Path: implied_children}
self._cached_implied_children[node_path] = path_implied
cache_implied = True
re_cache = cache_implied or cache_real
if re_cache:
# Hoisted out of the loop below, these only depend on node_path.
# Mirrors nxt_path.is_ancestor, which treats the world path as an
# ancestor of everything by identity.
node_path_is_world = node_path is nxt_path.WORLD
ancestor_prefix = node_path + nxt_path.NODE_SEP
trim_depth = nxt_path.get_path_depth(node_path) + 1
seen_implied = set()
trim = nxt_path.trim_to_depth
parent_attr = nxt_node.INTERNAL_ATTRS.PARENT_PATH
name_attr = nxt_node.INTERNAL_ATTRS.NAME
for path, node in self._nodes_path_as_key.items():
if cache_real:
parent_path = getattr(node, parent_attr)
# Compared by value. This used to be an identity check,
# which silently missed children whose parent path was an
# equal but distinct string. The bulk index above matches
# by value too, so both paths agree.
if parent_path == node_path:
children_nodes += [node]
children_paths += [path]
node_table += [[path, node]]
name_dict[getattr(node, name_attr)] = node
if not include_implied or not cache_implied:
continue
if not node_path_is_world:
if not path.startswith(ancestor_prefix):
continue
trimmed = trim(path, trim_depth)
if trimmed not in seen_implied:
seen_implied.add(trimmed)
implied_children += [trimmed]
if cache_real:
k = LayerReturnTypes.Path
self._cached_children[node_path][k] = children_paths[:]
if include_implied:
for imp in implied_children:
if imp not in children_paths:
children_paths += [imp]
if return_type is LayerReturnTypes.Boolean:
return bool(children_paths)
if ordered:
node = self.lookup(node_path)
if include_implied and not node:
child_order = []
else:
co_attr = nxt_node.INTERNAL_ATTRS.CHILD_ORDER
child_order = getattr(node, co_attr)
if child_order:
ordered_child_nodes = []
ordered_child_paths = []
ordered_node_table = []
for child_name in child_order:
# return type NODE
for n in children_nodes:
c_name = getattr(n, nxt_node.INTERNAL_ATTRS.NAME)
if c_name is child_name:
ordered_child_nodes += [n]
# return type PATH
for p in children_paths:
if nxt_path.node_name_from_node_path(p) == child_name:
ordered_child_paths += [p]
# return type TABLE
for item in node_table:
p, n = item
if nxt_path.node_name_from_node_path(p) == child_name:
ordered_node_table += [item]
# return type NODE
for n in children_nodes:
if n not in ordered_child_nodes:
ordered_child_nodes += [n]
children_nodes = ordered_child_nodes
# return type PATH
for p in children_paths:
if p not in ordered_child_paths:
ordered_child_paths += [p]
children_paths = ordered_child_paths
# return type TABLE
for item in node_table:
p, n = item
if p not in ordered_node_table:
ordered_node_table += [item]
node_table = ordered_node_table
if return_type is LayerReturnTypes.Node:
return children_nodes
elif return_type is LayerReturnTypes.Path:
return children_paths
elif return_type is LayerReturnTypes.NodeTable:
return node_table
elif return_type is LayerReturnTypes.NameDict:
return name_dict
elif return_type is LayerReturnTypes.Boolean:
return False
else:
logger.error('Invalid return type provided')
return None
def _implied_children_for(self, node_path):
"""Implied child paths of a path, from the bulk index.
An implied child is a path a node's descendants pass through but
that has no node of its own, so /a has an implied child /a/b when
only /a/b/c exists.
:param node_path: path to get the implied children of
:type node_path: str
:return: list of node paths
:rtype: list
"""
if (not self._implied_index_valid
or self._implied_index_len != len(self._nodes_path_as_key)):
self._build_implied_index()
return self._implied_index.get(node_path, [])[:]
def _build_implied_index(self):
"""Map every ancestor path in the layer to its implied children.
Built in one pass over the nodes. Answering this per path meant
testing every node in the layer for ancestry on every lookup, and
compositing asks for the implied children of most paths.
Nodes are visited in the order they appear in the layer and each
bucket keeps first seen order, which is the order the per path scan
this replaced produced.
"""
index = {}
seen = {}
sep = nxt_path.NODE_SEP
for path in self._nodes_path_as_key:
if path == nxt_path.WORLD:
continue
parts = path.split(sep)
# parts[0] is empty for a rooted path, so a path of depth d has
# d + 1 parts. Ancestors run from depth 1 up to the path's
# parent, the world path is excluded because children() always
# recomputes it.
for depth in range(1, len(parts) - 1):
ancestor = sep.join(parts[:depth + 1])
trimmed = sep.join(parts[:depth + 2])
bucket = index.get(ancestor)
if bucket is None:
index[ancestor] = bucket = []
seen[ancestor] = set()
ancestor_seen = seen[ancestor]
if trimmed not in ancestor_seen:
ancestor_seen.add(trimmed)
bucket += [trimmed]
self._implied_index = index
self._implied_index_len = len(self._nodes_path_as_key)
self._implied_index_valid = True
def _build_child_cache(self):
"""Populate the real children cache for every path in the layer in
a single pass over the nodes.
children() used to fill this cache one path at a time, and each fill
walked every node in the layer. Compositing asks for the children of
most paths, so that made a comp quadratic in node count.
"""
cache = {}
def new_entry():
return {LayerReturnTypes.Node: [],
LayerReturnTypes.Path: [],
LayerReturnTypes.NodeTable: [],
LayerReturnTypes.NameDict: {}}
# Childless paths need an entry too, or they keep missing the cache.
for path in self._nodes_path_as_key:
if path != nxt_path.WORLD:
cache[path] = new_entry()
for path, node in self._nodes_path_as_key.items():
parent_path = getattr(node, nxt_node.INTERNAL_ATTRS.PARENT_PATH)
# The world path is never cached, children() always rebuilds it.
if not parent_path or parent_path == nxt_path.WORLD:
continue
entry = cache.get(parent_path)
if entry is None:
# Parent has no node of its own, it is an implied path.
entry = new_entry()
cache[parent_path] = entry
entry[LayerReturnTypes.Node] += [node]
entry[LayerReturnTypes.Path] += [path]
entry[LayerReturnTypes.NodeTable] += [[path, node]]
key = getattr(node, nxt_node.INTERNAL_ATTRS.NAME)
entry[LayerReturnTypes.NameDict][key] = node
self._cached_children = cache
self._bulk_child_cache_valid = True
[docs] def get_cached_child_paths(self, parent_path):
"""Get the cached child paths for the given parent path. If no cache
exists, None is returned.
:param parent_path: Path to node that has children
:type parent_path: str
:return: List of node paths or None
:rtype: list | None
"""
return self._cached_children.get(parent_path,
{}).get(LayerReturnTypes.Path)
[docs] def descendants(self, node_path=nxt_path.WORLD,
return_type=LayerReturnTypes.Path, ordered=False,
include_implied=False):
if return_type not in (LayerReturnTypes.Node, LayerReturnTypes.Path,
LayerReturnTypes.NodeTable,):
raise TypeError('Unsupported return type {}'.format(return_type))
if node_path == nxt_path.WORLD:
if ordered:
raise ValueError('Cannot get ordered descendants of the world')
return self._world_descendants(return_type,
include_implied=include_implied)
if ordered:
if return_type is not LayerReturnTypes.Path:
raise ValueError('Ordered descedants can only return paths.')
return self._ordered_descendants(node_path,
include_implied=include_implied)
descendants = self.children(node_path, return_type,
include_implied=include_implied)
more_descendants = descendants
while more_descendants:
_temp = []
if return_type == LayerReturnTypes.Path:
for d_pth in more_descendants:
_temp += self.children(d_pth, return_type,
include_implied=include_implied)
elif return_type == LayerReturnTypes.NodeTable:
for d_pth, d in more_descendants:
_temp += self.children(d_pth, return_type,
include_implied=include_implied)
descendants += _temp
more_descendants = _temp
return descendants
def _ordered_descendants(self, node_path, include_implied=False):
desc = []
for child_path in self.children(node_path, ordered=True,
return_type=LayerReturnTypes.Path,
include_implied=include_implied):
desc += [child_path]
desc += self._ordered_descendants(child_path,
include_implied=include_implied)
return desc
def _world_descendants(self, return_type, include_implied=False):
if include_implied and return_type != LayerReturnTypes.Path:
raise ValueError('When including implied, can only return path. '
'Nothing else exists for implicit nodes.')
if return_type == LayerReturnTypes.Path:
paths = list(self._nodes_path_as_key.keys())
if nxt_path.WORLD in paths:
paths.remove(nxt_path.WORLD)
if not include_implied:
return paths
implied_paths = set()
for real_path in paths:
ancest_paths = nxt_path.all_ancestor_paths(real_path)
implied_paths = implied_paths.union(ancest_paths)
if nxt_path.WORLD in implied_paths:
implied_paths.remove(nxt_path.WORLD)
return implied_paths.union(paths)
if return_type == LayerReturnTypes.Node:
nodes = list(self._nodes_node_as_key.keys())
world_node = self._nodes_path_as_key.get(nxt_path.WORLD)
if world_node:
nodes.remove(world_node)
return nodes
if return_type == LayerReturnTypes.NodeTable:
node_table = []
for path, node in self._nodes_path_as_key.items():
if path is nxt_path.WORLD:
continue
node_table += [[path, node]]
return node_table
raise TypeError('Unsupported return type {}'.format(return_type))
[docs] def add_child_to_child_cache(self, parent_path, child_path, child_node):
if not parent_path or parent_path == nxt_path.WORLD:
return
children_cache = self._cached_children.get(parent_path)
if children_cache is not None:
children_nodes = children_cache[LayerReturnTypes.Node][:]
children_paths = children_cache[LayerReturnTypes.Path][:]
node_table = children_cache[LayerReturnTypes.NodeTable][:]
name_dict = copy.copy(children_cache[LayerReturnTypes.NameDict])
else:
children_nodes = []
children_paths = []
node_table = []
name_dict = {}
children_nodes += [child_node]
children_paths += [child_path]
node_table += [[child_path, child_node]]
key = getattr(child_node, nxt_node.INTERNAL_ATTRS.NAME)
name_dict[key] = child_node
cache = {LayerReturnTypes.Node: children_nodes,
LayerReturnTypes.Path: children_paths,
LayerReturnTypes.NodeTable: node_table,
LayerReturnTypes.NameDict: name_dict}
self._cached_children[parent_path] = cache
return
[docs] def remove_child_from_child_cache(self, parent_path, child_path,
child_node):
if parent_path is nxt_path.WORLD:
return
children_cache = self._cached_children.get(parent_path)
if children_cache is None:
return
children_nodes = children_cache[LayerReturnTypes.Node]
children_paths = children_cache[LayerReturnTypes.Path]
node_table = children_cache[LayerReturnTypes.NodeTable]
name_dict = children_cache[LayerReturnTypes.NameDict]
try:
children_nodes.remove(child_node)
except ValueError:
pass
try:
children_paths.remove(child_path)
except ValueError:
pass
try:
node_table.remove([child_path, child_node])
except ValueError:
pass
key = getattr(child_node, nxt_node.INTERNAL_ATTRS.NAME)
try:
name_dict.pop(key)
except KeyError:
pass
[docs] def clear_node_child_cache(self, parent_path):
# The bulk indexes no longer cover every path, rebuild on demand.
self._bulk_child_cache_valid = False
self._implied_index_valid = False
try:
self._cached_children.pop(parent_path)
except KeyError:
pass
try:
self._cached_implied_children.pop(parent_path)
except KeyError:
pass
[docs] def get_exec_order(self, start_path):
if start_path == nxt_path.WORLD:
return []
start_root_path = nxt_path.get_root_path(start_path)
exec_order = []
start_found = False
root_exec_order = self.get_root_exec_order(start_root_path)
for root_path in root_exec_order:
if not start_found and root_path == start_path:
start_found = True
node = self.lookup(root_path)
enabled = nxt_node.get_node_enabled(node)
if enabled is None:
enabled = True
if start_found and not enabled:
continue
if start_found:
exec_order += [root_path]
path_rt_type = LayerReturnTypes.Path
disabled_desc = []
for desc_path in self.descendants(node_path=root_path,
ordered=True,
return_type=path_rt_type):
if not start_found and desc_path == start_path:
start_found = True
desc_node = self.lookup(desc_path)
enabled = nxt_node.get_node_enabled(desc_node)
if enabled is None:
enabled = True
if enabled:
for potential_ancestor in disabled_desc:
if nxt_path.is_ancestor(desc_path, potential_ancestor):
enabled = False
break
if not enabled:
disabled_desc += [desc_path]
continue
if start_found:
exec_order += [desc_path]
return exec_order
[docs] def get_root_exec_order(self, start_root_path):
exec_order = [start_root_path]
prev_root_path = start_root_path
while True:
for root in self.children():
try:
exec_in = getattr(root, nxt_node.INTERNAL_ATTRS.EXECUTE_IN)
match = exec_in == prev_root_path
except AttributeError:
continue
if not match:
continue
prev_root_path = self.get_node_path(root)
exec_order += [prev_root_path]
break # for loop
else:
break # while loop
return exec_order
[docs] def get_muted(self, local=False):
muted = getattr(self, SAVE_KEY.MUTE, False)
if not local:
ref_data = get_comped_layer_overs(self, SAVE_KEY.COMP_ORVERRIDES)
r_dat = ref_data.get(self.filepath, {})
muted = r_dat.get(SAVE_KEY.MUTE, muted)
return muted
[docs] def set_muted(self, state):
self.mute = state
[docs] def set_mute_over(self, layer_path, state):
ref_data = get_comped_layer_overs(self, SAVE_KEY.COMP_ORVERRIDES)
r_dat = ref_data.get(layer_path, {})
r_dat[SAVE_KEY.MUTE] = state
ref_data[layer_path] = r_dat
self.comp_overrides = ref_data
[docs] def get_soloed(self, local=False):
soloed = getattr(self, SAVE_KEY.SOLO, False)
if not local:
ref_data = get_comped_layer_overs(self, SAVE_KEY.COMP_ORVERRIDES)
r_dat = ref_data.get(self.filepath, {})
soloed = r_dat.get(SAVE_KEY.SOLO, soloed)
return soloed
[docs] def set_soloed(self, state):
self.solo = state
[docs] def set_solo_over(self, layer_path, state):
ref_data = get_comped_layer_overs(self, SAVE_KEY.COMP_ORVERRIDES)
r_dat = ref_data.get(layer_path, {})
r_dat[SAVE_KEY.SOLO] = state
ref_data[layer_path] = r_dat
self.comp_overrides = ref_data
[docs] def add_reference(self, layer_path=None, layer=None, insert_idx=None):
"""Given a layer path, a layer object, or both, add a reference to
specified layer on this layer, optionally at a specified insert index.
:param layer_path: Path to use as reference path, defaults to None, \
if None is given, defaults to given `layer.real_path`
:type layer_path: str, optional
:param layer: layer to add as reference layer in this layer
:type layer: SpecLayer
:param insert_idx: index to insert given layer at, defaults to None, \
if None is given, defaults to lowest reference layer.
:type insert_idx: int, optional
:return: index given layer was inserted at.
:rtype: int
:raises ValueError: If not provided with a layer path or layer.
"""
if not (layer_path or layer):
raise ValueError("Must specify layer path or layer to reference.")
if insert_idx is None:
insert_idx = len(self.sub_layer_paths)
if layer_path is None:
layer_path = layer.real_path
self.sub_layer_paths.insert(insert_idx, layer_path)
sub_layer_data = {SAVE_KEY.FILEPATH: layer_path}
if layer is not None:
sub_layer_data['layer'] = layer
self.sub_layers.insert(insert_idx, sub_layer_data)
[docs] def get_references(self):
refs = []
for ref in self.sub_layers:
refs += [ref[SAVE_KEY.FILEPATH]]
return refs
[docs] def rebased_references(self, old_dir, new_dir):
"""How this layer's relative references read from another folder.
A reference relative to this layer's folder means a different file
once the layer is somewhere else, so a copy saved elsewhere would
have lost them. Each one that found its file beside the old folder
is written again from the new one: relative if the file is beside
the new folder too, whole if not. References that are absolute, use
a variable or a token, are found through NXT_FILE_ROOTS, or find
nothing at all are left as written, since none of those depend on
this layer's folder.
:param old_dir: the folder the layer was saved in
:param new_dir: the folder it is being saved to
:return: {reference as written now: reference to write instead}
:rtype: dict
"""
if not old_dir or not new_dir:
return {}
if os.path.normcase(os.path.realpath(old_dir)) == \
os.path.normcase(os.path.realpath(new_dir)):
return {}
renamed = {}
for ref in self.get_references():
if (not ref or ref.startswith('${') or ref.startswith('$')
or ref.startswith('%') or nxt_io.is_absolute(ref)):
continue
beside_old = nxt_path.full_file_expand(ref, start=old_dir)
if not os.path.isfile(beside_old):
continue
found, exists = nxt_io.expand_reference_path(ref, old_dir)
if not exists or os.path.normcase(found) != \
os.path.normcase(beside_old):
# Found through a root, which works from any folder.
continue
new_ref = nxt_io.stored_reference_path(beside_old, new_dir)
if new_ref != ref:
renamed[ref] = new_ref
return renamed
[docs] def rebase_references(self, renamed):
"""Write references, and the comp overrides that belong to them,
under new names. See rebased_references.
:param renamed: {reference as written now: reference to write instead}
:type renamed: dict
"""
for sub_layer_data in self.sub_layers:
ref = sub_layer_data[SAVE_KEY.FILEPATH]
if ref not in renamed:
continue
sub_layer_data[SAVE_KEY.FILEPATH] = renamed[ref]
layer = sub_layer_data.get('layer')
if layer is not None and layer.filepath == ref:
layer.filepath = renamed[ref]
self.sub_layer_paths = [renamed.get(path, path)
for path in self.sub_layer_paths]
for old_ref, new_ref in renamed.items():
if old_ref in self.comp_overrides:
self.comp_overrides[new_ref] = self.comp_overrides.pop(old_ref)
def _get_meta_attr(self, meta_attr, over_dict_attr=None, local=False,
fallback_to_local=True):
"""Private method for getting meta attr opinions, comped or local,
from the layer class.
:param meta_attr: Class attr to get the opinion of (ie color or alias)
:param over_dict_attr: Class attr name which corresponds to a dict of
overrides for the given meta_attr on lower layers
:param local: If Ture the local, non-comped, opinion is returned
:param fallback_to_local: If True and there is no override the local
opinion is returned.
:raises: ValueError
:return: opinion of meta_attr
"""
val = getattr(self, meta_attr, None)
if not local:
both_or_none = (over_dict_attr, not local)
if any(both_or_none) and not all(both_or_none):
raise ValueError('Must provide an over_dict_attr if local is '
'False!')
overs = get_comped_layer_overs(self, over_dict_attr)
if fallback_to_local:
fallback = val
else:
fallback = None
val = overs.get(self.filepath, fallback)
return val
[docs] def get_alias(self, local=False, fallback_to_local=True):
"""Get the layer's alias (nice name). By default the local alias is
returned, if local is False the parent layer's opinion of the alias is
returned.
:param local: If True the local opinion is returned, if False the
strongest override opinion is returned.
:type local: bool
:param fallback_to_local: if True and local is False and there is no
override opinion the local alias is returned. If false and no
override opinion is found None is returned.
:return: string of layer alias or None
"""
return self._get_meta_attr(SAVE_KEY.ALIAS, META_DATA_KEY.ALIASES,
local=local,
fallback_to_local=fallback_to_local)
[docs] def set_alias(self, alias):
self.alias = alias
[docs] def set_alias_over(self, alias):
"""Set a alias override on the root parent of a layer.
:param alias: new alias string
"""
layer_path = self.filepath
root = get_root_parent(self)
if not root:
logger.error('Can not set alias override on top layer!')
if alias is None:
if layer_path in list(root.aliases.keys()):
root.aliases.pop(layer_path)
else:
root.aliases[layer_path] = alias
[docs] def get_color(self, local=True, fallback_to_local=True):
"""Get the layer's color. By default the local color is
retuned, if local is False the parent layer's opinion of the alias is
returned.
:param local: bool
:param fallback_to_local: if True and local is False and there is no
override opinion the local color is returned. If false and no
override opinion is found None is returned.
:return: string of layer alias
"""
return self._get_meta_attr(SAVE_KEY.COLOR, META_DATA_KEY.COLORS,
local=local,
fallback_to_local=fallback_to_local)
[docs] def set_color_over(self, color):
"""Set a color override on the root parent of a layer.
:param color: hex color
"""
layer_path = self.filepath
root = get_root_parent(self)
if not root:
logger.error('Can not set colors override on top layer!')
if color is None:
if layer_path in list(root.colors.keys()):
root.colors.pop(layer_path)
else:
root.colors[layer_path] = color
[docs] def get_locked(self, local=False, fallback_to_local=True):
"""Get the locked state of the layer. By
default the local opinion is returned, if local is False the parent
layer's opinion of the alias is returned.
:param local: If True the local opinion is returned, if False the
strongest override opinion is returned.
:type local: bool
:param fallback_to_local: if True and local is False and there is no
override opinion the local alias is returned. If false and no
override opinion is found None is returned.
:return: bool of layer's lock state
"""
locked = self._get_meta_attr(SAVE_KEY.LOCK, META_DATA_KEY.LOCKS,
local=local,
fallback_to_local=fallback_to_local)
return locked or False
[docs] def set_locked(self, lock):
self.lock = lock
[docs] def set_locked_over(self, lock):
"""Set a lock override on the root parent of a layer.
:param lock: bool
"""
layer_path = self.filepath
root = get_root_parent(self)
if not root:
logger.error('Can not set colors override on top layer!')
if lock is None:
if layer_path in list(root.colors.keys()):
root.locks.pop(layer_path)
else:
root.locks[layer_path] = lock
[docs] def save(self, filepath=None, as_copy=False):
"""Save this layer, optionally to new, given filepath.
Saved to another folder, relative references are written so they
still find the same files from there. See rebased_references.
:param filepath: file path to save to, defaults to None
:type filepath: str, optional
:param as_copy: write a copy and leave this layer as it is, where it
is, rather than moving it to the new file
:type as_copy: bool
"""
filepath = filepath or self.real_path
renamed = {}
if self.real_path and filepath != self.real_path:
renamed = self.rebased_references(os.path.dirname(self.real_path),
os.path.dirname(filepath))
if renamed and not as_copy:
# The layer lives there now, so a plain save later writes the
# same thing.
self.rebase_references(renamed)
# Update graph name
graph_name = os.path.splitext(os.path.basename(filepath))[0]
if self.get_alias(local=True) == UNTITLED and not as_copy:
self.set_alias(graph_name)
save_data = self.get_save_data()
if as_copy:
if save_data.get(SAVE_KEY.ALIAS) == UNTITLED:
save_data[SAVE_KEY.ALIAS] = graph_name
if renamed:
save_data[SAVE_KEY.REFERENCES] = [
renamed.get(ref, ref)
for ref in save_data.get(SAVE_KEY.REFERENCES, [])]
overrides = save_data.get(SAVE_KEY.COMP_ORVERRIDES)
if overrides:
save_data[SAVE_KEY.COMP_ORVERRIDES] = OrderedDict(
(renamed.get(key, key), value)
for key, value in overrides.items())
try:
json.dumps(save_data, indent=4, sort_keys=False)
except TypeError:
logger.error("Failed to save file!")
# TODO: Should raise here, but its out of scope for what I'm
# working on right now.
return
logger.info("Save Data Generated")
nxt_io.save_file_data(save_data=save_data, filepath=filepath)
if filepath != self.real_path and not as_copy:
self.real_path = filepath
self.propegate_real_path()
return save_data
[docs] def get_comp_overrides(self):
return OrderedDict(sorted(self.comp_overrides.items(),
key=lambda x: x[0]))
[docs] def get_save_data(self):
"""Returns save data for this layer.
"""
save_dict = {
SAVE_KEY.VERSION: GRAPH_VERSION.VERSION_STR,
SAVE_KEY.ALIAS: self.get_alias(local=True),
SAVE_KEY.MUTE: self.get_muted(local=True),
SAVE_KEY.SOLO: self.get_soloed(local=True),
SAVE_KEY.REFERENCES: self.get_references(),
SAVE_KEY.COMP_ORVERRIDES: self.get_comp_overrides(),
SAVE_KEY.COLOR: self.color,
SAVE_KEY.META_DATA: self.get_meta_data(),
SAVE_KEY.NODES: self.get_nodes_save_data()}
rm = []
for k, v in save_dict.items():
if v in ((), [], {}):
rm += [k]
for k in rm:
save_dict.pop(k)
return self.order_save_dict(save_dict)
[docs] def order_save_dict(self, save_dict):
"""Given a save dictionary, returns an ordered dictionary where the
save keys are ordered correctly. Leftover keys are put at the end.
:param data: dictionary to order
:type data: dict
:return: ordered version of given dictionary
:rtype: OrderedDcit
"""
# order keys
keys_order = (SAVE_KEY.VERSION, SAVE_KEY.ALIAS, SAVE_KEY.COLOR,
SAVE_KEY.MUTE, SAVE_KEY.SOLO, SAVE_KEY.REFERENCES,
SAVE_KEY.COMP_ORVERRIDES, SAVE_KEY.META_DATA,
SAVE_KEY.NODES, SAVE_KEY.REAL_PATH)
result = OrderedDict()
data_keys = list(save_dict.keys())
for key in keys_order:
if key in data_keys:
result[key] = save_dict[key]
data_keys.remove(key)
# leftovers
for key in data_keys:
result[key] = save_dict[key]
return result
[docs] def get_nodes_save_data(self):
"""Gets an ordered dict of all of the spec nodes
:return: OrderedDict
"""
nodes_dict = OrderedDict()
for spec_node in self.spec_list:
node_data = nxt_node.get_node_as_dict(spec_node)
node_path = self.get_node_path(spec_node)
nodes_dict[node_path] = node_data
return order_nodes_dict(nodes_dict)
[docs] def propegate_real_path(self):
"""Update nodes source layer to current real path
"""
for node in self.descendants(return_type=LayerReturnTypes.Node):
setattr(node, nxt_node.INTERNAL_ATTRS.SOURCE_LAYER, self.real_path)
[docs]class CompLayer(SpecLayer):
[docs] @classmethod
def load_from_layer_data(cls, layer_data):
raise NotImplementedError
def __init__(self):
super(CompLayer, self).__init__({})
self._name = 'CompositeLayer'
self._layer_range = (0, 0)
self._dirty_map = {}
self._sublayer_node_tables = []
self.runtime = False
self.running = False
self._console = Console(_globals={}, _locals={}, node_path='STAGE')
self.cache_layer = CacheLayer()
# Quick and dirty hack to be used by the UI
self.failure = False
[docs] def layer_idx(self):
return self._layer_range[0]
[docs] def get_node_dirties(self, node_path):
"""Get all nodes that depend on the given node path. The given node
path will be the 0th item in the return list.
:param node_path: String of node path
:return: List of string node paths
"""
concerns = self._dirty_map.get(node_path, [])[:]
new_concerns = [node_path] + concerns
while new_concerns:
_tmp = []
for concern in new_concerns:
found = self._dirty_map.get(concern, [])
for item in found:
if item not in concerns + _tmp:
_tmp += [item]
new_concerns = _tmp
concerns += _tmp
return concerns
[docs]class CacheLayer(SpecLayer):
def __init__(self):
super(CacheLayer, self).__init__({})
self.nodes = {}
self.node_time_map = {}
self.node_times = []
[docs] def was_during_node_exec(self, time_float):
"""Checks if given time float is within any executed node's exec
timeframe.
:param time_float: float of time
:return: bool
"""
for enter_time, exit_time in reversed(self.node_times):
if time_float >= enter_time and exit_time is None:
return True
if enter_time <= time_float <= exit_time:
return True
return False
[docs] def set_node_enter_time(self, node_path, t=0.):
if not t:
t = time.time()
node_runs = self.node_time_map.get(node_path, [])
time_entry = [[t, None]]
node_runs += time_entry
self.node_times += time_entry
self.node_time_map[node_path] = node_runs
[docs] def set_node_exit_time(self, node_path, t=0.):
if not t:
t = time.time()
try:
node_runs = self.node_time_map[node_path]
except IndexError:
raise IndexError('No start time set for {}'.format(node_path))
for run_time in node_runs:
if run_time[1] is None:
run_time.pop(1)
run_time += [t]
return
raise ValueError('No exec tuples for {} have a missing end '
'time!'.format(node_path))
[docs] def get_node_run_time(self, node_path, idx=-1):
try:
enter_time, exit_time = self.node_time_map[node_path][idx]
except (KeyError, IndexError):
raise ValueError('No time for {} at index {}'.format(node_path,
idx))
return exit_time - enter_time
[docs] def lookup(self, node_path):
return self.nodes.get(node_path)
[docs] def add_node(self, node_path, node):
cached_code = getattr(node, nxt_node.INTERNAL_ATTRS.CACHED_CODE, '\n')
spl_code = cached_code.split('\n')
setattr(node, nxt_node.INTERNAL_ATTRS.COMPUTE, spl_code)
self.nodes[node_path] = node
[docs] def save(self, filepath):
save_data = self.get_save_data()
nxt_io.save_file_data(save_data=save_data, filepath=filepath)
return save_data
[docs] def get_save_data(self):
save_data = {
'version': GRAPH_VERSION.VERSION_STR,
'nodes': {}
}
for path, node in self.nodes.items():
save_data['nodes'][path] = nxt_node.get_node_as_dict(node)
return save_data
[docs] @classmethod
def load_from_layer_data(cls, layer_data):
new_layer = cls()
result_nodes = {}
name_attr = nxt_node.INTERNAL_ATTRS.NAME
name_key = nxt_node.INTERNAL_ATTRS.as_save_key(name_attr)
for node_path, node_data in layer_data.get('nodes', {}).items():
parent_path = nxt_path.get_parent_path(node_path)
node_name = nxt_path.node_name_from_node_path(node_path)
node_data[name_key] = node_name
spec_node = nxt_node.create_spec_node(node_data, new_layer,
parent_path)
code_lines = getattr(spec_node, nxt_node.INTERNAL_ATTRS.COMPUTE)
code = '\n'.join(code_lines)
setattr(spec_node, nxt_node.INTERNAL_ATTRS.CACHED_CODE, code)
result_nodes[node_path] = spec_node
new_layer.nodes = result_nodes
return new_layer
[docs]def get_active_layers(layers):
"""Given a list of layers, return those that should contribute to comp.
If there are soloed layers, only the soloed layers that are not muted
will be returned. If there are not soloed layers, only the non-muted
layers will be returned.
:param layers: list of layers to filter
:type layers: list
:return: list of layers that should contribute to a comp.
:rtype: list
"""
soloed_layers = get_soloed_layers(layers)
potential_layers = soloed_layers or layers
result_layers = []
for layer in potential_layers:
if layer.get_muted():
continue
result_layers += [layer]
return result_layers
[docs]def get_soloed_layers(layers):
"""Given a list of layers, return only those that are soloed
:param layers: list of layers to filter
:type layers: list
:return: list of layers that are soloed.
:rtype: list
"""
return [layer for layer in layers if layer.get_soloed()]
[docs]def get_muted_layers(layers):
"""Given a list of layers, return only those that are muted
:param layers: list of layers to filter
:type layers: list
:return: list of layers that are muted.
:rtype: list
"""
return [layer for layer in layers if layer.get_muted()]
[docs]def get_node_local_attr_names(node_path, layers):
"""Get all attribute names that are local to the given node path within
the given layers.
NOTE that layers are filtered to active layers.
:param node_path: node path to find local attributes of
:type node_path: str
:param layers: list of layers to look for attributes of given node within
:type layers: list
:return: list of local attributes found in given layers
:rtype: list
"""
local_attrs = []
# TODO should we filter for active layers or expect callers to?
active_layers = get_active_layers(layers)
for layer in active_layers:
node = layer.lookup(node_path)
if not node:
continue
for attr in node.__dict__.keys():
if attr in local_attrs:
continue
if attr in nxt_node.INTERNAL_ATTRS.PROTECTED:
continue
if attr.endswith(nxt_node.META_ATTRS._suffix):
continue
local_attrs += [attr]
return local_attrs
[docs]def get_comped_layer_overs(layer, over_dict_attr):
over_dict = {}
over_dict.update(getattr(layer, over_dict_attr, {}))
parent = layer.parent_layer
while parent:
over_dict.update(getattr(parent, over_dict_attr, {}))
parent = parent.parent_layer
return over_dict
[docs]def get_root_parent(layer):
parent = layer.parent_layer
root = None
while parent:
root = parent
parent = parent.parent_layer
return root
[docs]def order_nodes_dict(node_dict):
"""Given a dictionary mapping node paths to node data, return an ordered
dict where the keys are sorted.
:param node_dict: dictionary mapping node paths to node data
:type node_dict: dict
:return: ordered dctionary with keys sosrted
:rtype: OrderedDict
"""
unsorted_paths = list(node_dict.keys())
sorted_paths = sorted(unsorted_paths)
ordered_nodes = OrderedDict()
for path in sorted_paths:
ordered_nodes[path] = node_dict[path]
return ordered_nodes
[docs]class RankedList(object):
"""List that answers "where is this item" and "put this item at index i"
in log time rather than by scanning.
Stage.sort_instances moves most of its items exactly once, and both
list.index and list.remove scan from the front, which made the sort
quadratic in node count. Items here sit in per position buckets with a
Fenwick tree over the bucket sizes turning a bucket into a global index.
Items are matched by identity. The list scans this replaces matched by
equality, but the items are (trace, node) pairs holding distinct node
classes, so no two of them ever compare equal.
"""
def __init__(self, items):
self._buckets = [[item] for item in items]
self._where = {id(item): i for i, item in enumerate(items)}
size = len(self._buckets)
self._n = size
self._total = size
# Fenwick tree, 1 based, holding the live count of each bucket.
tree = [0] * (size + 1)
for i in range(1, size + 1):
tree[i] += 1
parent = i + (i & -i)
if parent <= size:
tree[parent] += tree[i]
self._tree = tree
def _add(self, bucket, delta):
i = bucket + 1
while i <= self._n:
self._tree[i] += delta
i += i & -i
def _prefix(self, bucket):
"""Number of live items in the buckets before the given one."""
total = 0
i = bucket
while i > 0:
total += self._tree[i]
i -= i & -i
return total
def _find(self, rank):
"""The bucket holding a global rank, and the offset inside it."""
pos = 0
remaining = rank
step = 1
while step * 2 <= self._n:
step *= 2
while step:
nxt = pos + step
if nxt <= self._n and self._tree[nxt] <= remaining:
pos = nxt
remaining -= self._tree[nxt]
step //= 2
return pos, remaining
[docs] def index(self, item):
bucket = self._where[id(item)]
offset = 0
for other in self._buckets[bucket]:
if other is item:
break
offset += 1
return self._prefix(bucket) + offset
[docs] def remove(self, item):
bucket = self._where.pop(id(item))
contents = self._buckets[bucket]
for i, other in enumerate(contents):
if other is item:
del contents[i]
break
self._add(bucket, -1)
self._total -= 1
[docs] def insert(self, rank, item):
if not self._n:
# Nothing to position against, start a bucket.
self._buckets += [[item]]
self._n = 1
self._tree = [0, 1]
self._where[id(item)] = 0
self._total += 1
return
if rank >= self._total:
# Past the end, which list.insert treats as an append.
bucket = self._n - 1
self._buckets[bucket].append(item)
else:
bucket, offset = self._find(rank)
if bucket >= self._n:
bucket = self._n - 1
self._buckets[bucket].append(item)
else:
self._buckets[bucket].insert(offset, item)
self._where[id(item)] = bucket
self._add(bucket, 1)
self._total += 1
[docs] def to_list(self):
out = []
for contents in self._buckets:
out += contents
return out
[docs]def sort_multidimensional_list(multi_list, sort_by_idx):
"""Takes a multi-dimensional list and sorts it by the length of a sub-list
item. The item who's length we sort by is
determined by the idx parameter.
If your input is:
```
sort_multidimensional_list(
[
[[1,2,3], ['a', 'b']],
[[4,5,6], ['a']]
], idx = 1)
```
Your output will be:
```
[
[[1,2,3], ['a']],
[[4,5,6], ['a', 'b']]
]
```
:param multi_list: multi-dimensional list
:param sort_by_idx: Index of sub list item to sort by
:return: list
"""
if not multi_list:
return multi_list
# Sorted in place, callers rely on the mutation as well as the return.
# list.sort is stable, matching the bubble sort this replaced.
multi_list.sort(key=lambda item: len(item[sort_by_idx]))
return multi_list