diff --git a/pyoutline/outline/layer.py b/pyoutline/outline/layer.py index c079a97da..f2cdd32d9 100644 --- a/pyoutline/outline/layer.py +++ b/pyoutline/outline/layer.py @@ -15,19 +15,26 @@ """Base classes for all outline modules.""" - +from __future__ import annotations from __future__ import absolute_import from __future__ import print_function from __future__ import division -from builtins import str -from builtins import range -from builtins import object -from future.utils import with_metaclass import os import sys import logging import tempfile +from typing import ( + TypedDict, + List, + Optional, + Callable, + Dict, + Any, + Union, + Tuple, + Set, +) import FileSequence @@ -39,12 +46,25 @@ import outline.io import outline.util +if sys.version_info >= (3, 10, 1): + from typing import Literal +else: + from typing_extensions import Literal + +if sys.version_info >= (3, 12): + from typing import override, Unpack +else: + from typing_extensions import override, Unpack + +_LayerT = Literal["Render", "Util", "Post"] -__all__ = ["Layer", - "Frame", - "LayerPreProcess", - "LayerPostProcess", - "OutlinePostCommand"] +__all__ = [ + "Layer", + "Frame", + "LayerPreProcess", + "LayerPostProcess", + "OutlinePostCommand", +] logger = logging.getLogger("outline.layer") @@ -53,59 +73,77 @@ class LayerType(type): """ - A meta-class to wrap the creation of layer objects so they + A metaclass that wraps the creation of Layer objects so they can be added to the current outline. """ + def __call__(cls, *args, **kwargs): - r = super(LayerType, cls).__call__(*args, **kwargs) - if outline.current_outline() and r.get_arg("register"): - outline.current_outline().add_layer(r) + layer = super().__call__(*args, **kwargs) + if outline.current_outline() and layer.get_arg("register"): + outline.current_outline().add_layer(layer) - # Initialize with plugin system. This is imported - # here to get past a circular dependency. + # Initialize with plugin system. This import is done here to avoid + # a circular dependency. # pylint: disable=import-outside-toplevel from outline.plugins import PluginManager + for plugin in PluginManager.get_plugins(): try: - plugin.init(r) + plugin.init(layer) except AttributeError: pass - return r + return layer -class Layer(with_metaclass(LayerType, object)): - """The base class for all outline modules.""" +class _LayerArgs(TypedDict, total=False): + """Typed dict to annotate layer argument names and types.""" - def __init__(self, name, **args): - # pylint: disable=non-parent-init-called - object.__init__(self) + chunk: int + command: List[str] + cores: int + env: Dict[str, str] + limits: List[str] + memory: str + range: str + register: bool + service: str + timeout: int + timeout_llu: int + type: _LayerT + + +class Layer(metaclass=LayerType): + """Base class for all outline modules.""" + + def __init__(self, name: str, **args: Unpack[_LayerArgs]) -> None: + super().__init__() self.__name = name # Contains the args hash. - self.__args = self.get_default_args(args) + self.__args = self.get_default_args(merge=args) # Default the layer type to the Render type as # defined in the constants module - self.__type = None + self.__type: Optional[_LayerT] = None self.set_type(args.get("type", outline.constants.LAYER_TYPES[0])) # A set of arguments that is required before # the Layer can be launched. - self.__req_args = set() + self.__req_args: Set[Tuple[str, Any]] = set() # A list to store what this layer depends on. - self.__depends = [] + self.__depends: List[outline.depend.Depend] = [] - # If this layer is embedded within a another layer + # If this layer is embedded within another layer # the parent value will point to that layer. - self.__parent = None + self.__parent: Optional[Layer] = None # Contains IO objects that are considered input. - self.__input = {} + self.__input: Dict[str, Any] = {} # Contains IO objects that are considered output. - self.__output = {} + self.__output: Dict[str, Any] = {} # A dictionary of environment variables to apply before execute. self.__env = {} @@ -113,7 +151,7 @@ def __init__(self, name, **args): # Children are unregistered layers that are executed # after the parent layer. - self.__children = [] + self.__children: List[Layer] = [] # The default name of the service. self.__service = self.__args.get("service", "shell") @@ -122,27 +160,43 @@ def __init__(self, name, **args): self.__limits = self.__args.get("limits") # The current frame number. - self.__frame = None + self.__frame: Optional[int] = None # Initialize the outline instance. - self.__outline = None + self.__outline: Optional[outline.Outline] = None # Register an event handler. self.__evh = outline.event.EventHandler(self) # Keep an array of all pre-process frames. - self.__preprocess_layers = [] + self.__preprocess_layers: List[LayerPreProcess] = [] logger.debug( - "module %s loaded from %s", self.__class__.__name__, os.path.realpath(__file__)) + "module %s loaded from %s", + self.__class__.__name__, + os.path.realpath(__file__), + ) + + @override + def __str__(self) -> str: + return self.get_name() - def _after_init(self, ol): + @override + def __repr__(self) -> str: + return ( + f"<{self.__class__.__name__} " + f"name='{self.get_name()}' " + f"type='{self.get_type()}' " + f"at {hex(id(self))}>" + ) + + def _after_init(self, ol: outline.Outline) -> None: """ - This method should be implemented by a subclass. Executed - after a layer has been initialized and added to an outline. + This method should be implemented by a subclass. + Executed after a layer has been initialized and added to an outline. """ - def after_init(self, ol): + def after_init(self, ol: outline.Outline) -> None: """ Executed after a layer has been initialized and added to an outline. Emits an event.AFTER_INIT signal. @@ -150,125 +204,125 @@ def after_init(self, ol): self._after_init(ol) self.__evh.emit(outline.event.LayerEvent(outline.event.AFTER_INIT, self)) - def _after_parented(self, parent): + def _after_parented(self, parent: Layer) -> None: """ - This method should be implemented by a subclass. Executed after a - layer has been initialized and added as a child to another layer. + This method should be implemented by a subclass. + Executed after a layer has been initialized and added as a child to another layer. """ - def after_parented(self, parent): + def after_parented(self, parent: Layer) -> None: """ - This method should be implemented by a subclass. Automatically - called if this Layer instance is parented to another layer instance. + Automatically called when this Layer instance is parented to another layer instance. """ self._after_parented(parent) self.__evh.emit(outline.event.LayerEvent(outline.event.AFTER_PARENTED, self)) - def _before_execute(self): + def _before_execute(self) -> None: """ - This method should be implemented by a subclass. Executed before - all execute checks are started. + This method should be implemented by a subclass. + Executed before all execute checks are started. """ - def before_execute(self): + def before_execute(self) -> None: """ Executed before all execute checks are started. """ self._before_execute() self.__evh.emit(outline.event.LayerEvent(outline.event.BEFORE_EXECUTE, self)) - def _after_execute(self): + def _after_execute(self) -> None: """ - This method should be implemened by a subclass. Executed after - the execute() method has been run even if the frame failed. - Used for doing cleanup operations that should run even - after a frame failure. + This method should be implemented by a subclass. + Executed after the execute() method has been run even if the frame failed. + Used for cleanup operations that should run even after a frame failure. """ - def after_execute(self): + def after_execute(self) -> None: """ - Executed after the execute() method has been run even if the - frame failed. Used for doing cleanup operations that should - run even after a frame failure. + Executed after the execute() method has been run even if the frame failed. + Used for cleanup operations that should run even after a frame failure. """ self._after_execute() frames = self.get_local_frame_set(self.__frame) - self.__evh.emit(outline.event.LayerEvent( - outline.event.AFTER_EXECUTE, self, frames=frames)) + self.__evh.emit( + outline.event.LayerEvent(outline.event.AFTER_EXECUTE, self, frames=frames) + ) @staticmethod - def system(cmd, ignore_error=False, frame=None): + def system(cmd: List[str], ignore_error: bool = False, frame: Optional[int] = None): """ - A convenience method for calling io.system(). Shell out - to the given command and wait for it to finish. + Convenience method for calling io.system(). Shell out to the given command + and wait for it to finish. - @see: L{io.system} + :see: L{io.system} :type cmd: list :param cmd: The command to execute. :type ignore_error: boolean - :param ignore_error: Ignore any L{OSError} or shell command failures. + :param ignore_error: If True, ignore any L{OSError} or shell command failures. """ outline.io.system(cmd, ignore_error, frame) - def get_default_args(self, merge=None): + def get_default_args(self, merge: Optional[Unpack[_LayerArgs]] = None): """ - Create and return a default argument hash. Optionally merge + Create and return a default argument hash. Optionally merge the specified dictionary into the result. """ - # No backend specific defaults should be here, those values + # No backend specific defaults should be here; those values # would be defined within the relevant backend module or in # the outline configuration file. - defaults = {} - - # By default all layers are registerd. Registered layers show up - # as discrete layers. Unregisterd layers are generally embedded - # in registered layers. - defaults["register"] = True - - # The default chunk size. - defaults["chunk"] = 1 - - # A null frame range indicates the event - # will default to the overall frame range - # defined in the parent outline. - defaults["range"] = None + defaults: Dict[str, Any] = { + # By default, all layers are registered. Registered layers show up + # as discrete layers. Unregistered layers are generally embedded + # in registered layers. + "register" : True, + # The default chunk size. + "chunk" : 1, + # A null frame range indicates the event + # will default to the overall frame range + # defined in the parent outline. + "range" : None, + } # Now apply any settings found in the configuration file. - # This settings override the procedural defaults set in - # the layer constructur using default_arg method. + # These settings override the procedural defaults set in + # the layer constructor using default_arg method. if outline.config.has_section(self.__class__.__name__): for key, value in outline.config.items(self.__class__.__name__): defaults[key] = value - # Now apply user supplied arguments. These arguments override - # both the defaults and the class condifuration file. + # Now apply user-supplied arguments. These arguments override + # both the defaults and the class configuration file. if merge: defaults.update(merge) return defaults - def get_parent(self): - """Return the parent Layer. """ + def get_parent(self) -> Optional[Layer]: + """Return the parent Layer.""" return self.__parent - def set_parent(self, layer): + def set_parent(self, layer: Layer) -> None: """Set the parent layer.""" - if not isinstance(layer, (Layer)): - raise outline.exception.LayerException("Parent instance must derive from Layer.") + if not isinstance(layer, Layer): + raise outline.exception.LayerException( + "Parent instance must derive from Layer." + ) self.__parent = layer - def add_child(self, layer): + def add_child(self, layer: Layer) -> None: """ Add a child layer to this layer. Child layers are - executed after the parent layer. + executed after the parent layer. """ - if not isinstance(layer, (Layer)): - raise outline.exception.LayerException("Child instances must derive from Layer.") + if not isinstance(layer, Layer): + raise outline.exception.LayerException( + "Child instances must derive from Layer." + ) layer.set_outline(self.get_outline()) layer.set_parent(self) @@ -276,42 +330,43 @@ def add_child(self, layer): self.__children.append(layer) layer.after_parented(self) - def add_event_listener(self, event_type, callback): + def add_event_listener(self, event_type: str, callback: Callable) -> None: """Add an event listener to the layer's event handler.""" self.__evh.add_event_listener(event_type, callback) - def get_event_handler(self): + def get_event_handler(self) -> outline.event.EventHandler: """ Return the layer's internal EventHandler. """ return self.__evh - def get_children(self): + def get_children(self) -> List[Layer]: """Return a list of this layer's child layers.""" return list(self.__children) - def set_env(self, key, value): - """Set an env var to be set before execute.""" + def set_env(self, key: str, value: str) -> None: + """Set an environment variable to be applied before execute.""" if key in self.__env: logger.warning( - "Overwriting outline env var: %s, from %s to %s", key, self.__env[key], value) + "Overwriting outline env var: %s, from %s to %s", key, self.__env[key], value + ) self.__env[str(key)] = str(value) - def get_env(self, key, default=None): - """Get the value of the env var that will be set before execute.""" + def get_env(self, key: str, default: Optional[str] = None) -> Optional[str]: + """Get the value of the environment variable that will be set before execute.""" return self.__env.get(key, default) - def get_envs(self): - """Return all env.""" + def get_envs(self) -> Dict[str, str]: + """Return all environment variables for this layer.""" return self.__env - def get_name(self): + def get_name(self) -> str: """Return the layer name.""" if self.__parent: - return "%s.%s" % (self.__parent.get_name(), self.__name) + return f"{self.__parent.get_name()}.{self.__name}" return self.__name - def set_name(self, name): + def set_name(self, name: str) -> None: """ Set the layer's name. @@ -323,40 +378,40 @@ def set_name(self, name): raise outline.exception.LayerException(msg) self.__name = name - def get_type(self): + def get_type(self) -> _LayerT: """ - Returns the general scope or purpose of the Layer. Allowed + Return the general scope or purpose of the Layer. Allowed types are: - - Render: a general purpose rendering layer, has inputs and outputs. - - Util: a setup/cleanup frame or trival shell command. + - Render: a general-purpose rendering layer, has inputs and outputs. + - Util: a setup/cleanup frame or trivial shell command. - Post: a post layer which is kicked off after all other layers have completed. """ return self.__type - def set_type(self, t): + def set_type(self, t: _LayerT) -> None: """ - Sets the general scope/purpose of this layer. + Set the general scope/purpose of this layer. """ if t not in outline.constants.LAYER_TYPES: - raise outline.exception.LayerException("%s is not a valid layer type: %s" % ( - t, outline.constants.LAYER_TYPES)) + raise outline.exception.LayerException( + f"{t} is not a valid layer type: {outline.constants.LAYER_TYPES}" + ) self.__type = t - def get_outline(self): + def get_outline(self) -> Optional[outline.Outline]: """Return the parent outline object.""" if self.__parent: return self.__parent.get_outline() return self.__outline - def set_outline(self, new_outline): + def set_outline(self, new_outline: outline.Outline) -> None: """Set this layer's parent outline to the given outline object.""" self.__outline = new_outline - def setup(self): - """Setup is run once before the job is launched - to the render farm. This method would be used for - any pre-launch operations that may be required. + def setup(self) -> None: + """Setup is run once before the job is launched to the render farm. + This method would be used for any pre-launch operations that may be required. """ self.check_required_args() self._setup() @@ -367,15 +422,15 @@ def setup(self): # Emit the setup event. self.__evh.emit(outline.event.LayerEvent(outline.event.SETUP, self)) - def _setup(self): + def _setup(self) -> None: """This method should be implemented by a subclass.""" - def _execute(self, frames): + def _execute(self, frames: FileSequence.FrameSet) -> None: """This method should be implemented by a subclass.""" - def execute(self, frame): + def execute(self, frame: int) -> None: """ - Executes the local frame set. This typically happens + Execute the local frame set. This typically happens on a cluster node. :type frame: int @@ -387,7 +442,7 @@ def execute(self, frame): # Set any arg overrides from args_override in the session self.setup_args_override() - # Find the local frame set, ie the frames that this + # Find the local frame set, i.e., the frames that this # instance is responsible for. frames = self.get_local_frame_set(frame) @@ -397,7 +452,7 @@ def execute(self, frame): # Process the python_path option self.__set_python_path() - # load any outputs found in ol:outputs + # Load any outputs found in ol:outputs self.load_outputs() # Run the pre-execute method to setup any variables that could not @@ -410,28 +465,30 @@ def execute(self, frame): # Double check that all required arguments are set. self.check_required_args() - # Check for the existance of required inputs. + # Check for the existence of required inputs. self.check_input(frames) # Set all post set shot environment variables. for env_k, env_v in self.__outline.get_env().items(): if not env_v[1]: - logger.info("Setting post-set shot environment var: %s %s", - env_k, env_v[0]) + logger.info( + "Setting post-set shot environment var: %s %s", env_k, env_v[0] + ) os.environ[env_k] = env_v[0] - # Set all layer specific post set shot env variables + # Set all layer-specific post set shot env variables try: for env_k, env_v in self.__env.items(): - logger.info("Setting post-set shot environment var: %s %s", - env_k, env_v) + logger.info( + "Setting post-set shot environment var: %s %s", env_k, env_v + ) os.environ[str(env_k)] = str(env_v) except AttributeError: pass logger.info("Layer %s executing local frame set %s", self.get_name(), frames) - # Run the subclasses execute method and all child execute methods + # Run the subclass's execute method and all child execute methods self._execute(frames) for child in self.__children: child.execute(frame) @@ -439,11 +496,11 @@ def execute(self, frame): # Run the subclass's _post_execute method self.after_execute() - # Check the existance of required output + # Check the existence of required output self.check_output(frames) # pylint: disable=broad-except,no-member - def setup_args_override(self): + def setup_args_override(self) -> None: """ Load the args_override data from the session and set them as the args. Useful for when you don't know some args until after launching. @@ -453,20 +510,20 @@ def setup_args_override(self): """ try: - args_override = self.get_data('args_override') - logger.warning('Loaded args_override from session to replace args:') + args_override = self.get_data("args_override") + logger.warning("Loaded args_override from session to replace args:") for key, value in args_override.items(): self.set_arg(key, value) # This was necessary because plugins/s3d.py uses get_creator() - if hasattr(self, 'get_creator') and self.get_creator(): + if hasattr(self, "get_creator") and self.get_creator(): self.get_creator().set_arg(key, value) - logger.warning('Replaced arg %s with %s', key, value) + logger.warning("Replaced arg %s with %s", key, value) except outline.exception.SessionException: - logger.debug('args_override not found in session (This is normal)') + logger.debug("args_override not found in session (this is normal)") except Exception as e: - logger.debug('Not loading args_override from session due to %s', e) + logger.debug("Not loading args_override from session due to %s", e) - def set_default_arg(self, key, value): + def set_default_arg(self, key: str, value: Any) -> None: """ Set the value for the given argument if and only if the argument has not already been set. @@ -474,52 +531,53 @@ def set_default_arg(self, key, value): if key not in self.__args: self.__args[key] = value - def get_arg(self, key, default=None): + def get_arg(self, key: str, default: Optional[Any] = None) -> Any: """ Return the value associated with the specified key. If the value does not exist, return default. """ return self.__args.get(key, default) - def set_arg(self, key, value): + def set_arg(self, key: str, value: Any) -> None: """Set the value of key.""" for arg, rtype in self.__req_args: if arg == key and rtype: if not isinstance(value, rtype): - msg = "The arg %s for the %s module must be a %s" - raise outline.exception.LayerException(msg % (arg, - self.__class__.__name__, - rtype)) + msg = ( + f"The arg {arg} for the {self.__class__.__name__} " + f"module must be a {rtype}" + ) + raise outline.exception.LayerException(msg) break self.__args[key] = value - def is_arg_set(self, key): - """Return true if the key exits in the arg hash.""" + def is_arg_set(self, key: str) -> bool: + """Return true if the key exists in the arg hash.""" return key in self.__args - def get_args(self): + def get_args(self) -> Dict[str, Any]: """Return the arg dictionary.""" return dict(self.__args) - def copy_args_from(self, layer, *names): + def copy_args_from(self, layer: Layer, *names: str) -> None: """ Copy args from the given layer into this layer. """ for arg_name in names: self.set_arg(arg_name, layer.get_arg(arg_name)) - def get_path(self): + def get_path(self) -> str: """Return the session path for the layer.""" return self.__outline.get_session().get_path(layer=self) - def get_session(self): + def get_session(self) -> outline.session.Session: """Return outline's session variable.""" return self.__outline.get_session() - def get_service(self): + def get_service(self) -> str: """ - Return the layer's service name. The service + Return the layer's service name. The service is the primary application being run in the layer. @@ -528,9 +586,9 @@ def get_service(self): """ return self.__service - def set_service(self, service): + def set_service(self, service: str) -> None: """ - Set the service for this layer. The service + Set the service for this layer. The service is the name of the primary application being run in the layer. @@ -539,23 +597,23 @@ def set_service(self, service): """ self.__service = service - def get_limits(self): + def get_limits(self) -> Optional[List[str]]: """ - Return a list limits for this layer. - :rtype: string + Return a list of limits for this layer. + :rtype: list :return: list of limits """ return self.__limits - def set_limits(self, limits): + def set_limits(self, limits: List[str]) -> None: """ Set the limits for this layer. - :type limits: string - :param limits: list of Limit names + :type limits: list + :param limits: list of limit names """ self.__limits = limits - def put_data(self, key, value, force=False): + def put_data(self, key: str, value: Any, force: bool = False) -> None: """ Copy a variable into the layer's session. @@ -566,7 +624,7 @@ def put_data(self, key, value, force=False): """ self.__outline.get_session().put_data(key, value, self, force=force) - def get_data(self, key): + def get_data(self, key: str) -> Any: """ Retrieve a previously saved variable from the session. @@ -575,11 +633,11 @@ def get_data(self, key): """ return self.__outline.get_session().get_data(key, self) - def sym_file(self, src, rename=None): + def sym_file(self, src: str, rename: Optional[str] = None) -> str: """ - Symlink the given file into the layer's session path. If + Symlink the given file into the layer's session path. If the optional rename argument is set, the file will be - renamed during the copy. + renamed during the symlink. :type src: string :param src: The path to the source file. @@ -590,13 +648,15 @@ def sym_file(self, src, rename=None): :rtype: str :return: The full path to the new file in the session. """ - return self.__outline.get_session().sym_file(src, - layer=self, - rename=rename) + return self.__outline.get_session().sym_file( + src, + layer=self, + rename=rename, + ) - def put_file(self, src, rename=None): + def put_file(self, src: str, rename: Optional[str] = None) -> str: """ - Copy the given file into the layer's session path. If + Copy the given file into the layer's session path. If the optional rename argument is set, the file will be renamed during the copy. @@ -609,13 +669,15 @@ def put_file(self, src, rename=None): :rtype: str :return: The full path to the new file in the session. """ - return self.__outline.get_session().put_file(src, - layer=self, - rename=rename) + return self.__outline.get_session().put_file( + src, + layer=self, + rename=rename, + ) - def get_file(self, name, check=True, new=False): + def get_file(self, name: str, check: bool = True, new: bool = False) -> str: """ - Retrieve the session path path to the given file. The + Retrieve the session path to the given file. The file does not have to exist. :type name: str @@ -626,41 +688,45 @@ def get_file(self, name, check=True, new=False): the file does not exist. :type new: boolean - :param new: If new is set and the file your getting already - exists, a SessionException is thrown. This ensures - if your getting a new path to open in the session that + :param new: If new is set and the file you're getting already + exists, a SessionException is thrown. This ensures + that when requesting a new path to open in the session the file does not already exist. If new is specified, - check is automatically set to false. + check is automatically set to False. :rtype: str :return: the full path to the file stored under the given name. """ - return self.__outline.get_session().get_file(name, - layer=self, - check=check, - new=new) + return self.__outline.get_session().get_file( + name, + layer=self, + check=check, + new=new, + ) - def require_arg(self, key, rtype=None): + def require_arg(self, key: str, rtype: Optional[Any] = None) -> None: """Require the specified key to be present.""" self.__req_args.add((key, rtype)) - def set_frame_range(self, frame_range): + def set_frame_range(self, frame_range) -> None: """Set the layer's frame range.""" logger.debug( - "layer %s changing range from %s to %s", self.get_name(), self.__args["range"], - frame_range) + "layer %s changing range from %s to %s", + self.get_name(), + self.__args["range"], + frame_range, + ) self.__args["range"] = str(frame_range) # pylint: disable=inconsistent-return-statements def get_frame_range(self): """ - Return the layer's frame range. If the layer and its + Return the layer's frame range. If the layer and its parent outline file have incompatible frame ranges, return None. :rtype: String - :return: The layer's frame range; - + :return: The layer's frame range. """ if self.__args["range"]: rng = self.__args["range"] @@ -671,19 +737,21 @@ def get_frame_range(self): if self.__outline: - # If there is a layer range and an OL range, return - # the intersection. If the intersection cannot be + # If there is a layer range and an outline range, return + # the intersection. If the intersection cannot be # made then a LayerException is thrown. - # If there is just a layer range,return the layer range. - # if there is no layer range but an ol range, return the ol_range. - # if there is neither a layer range or ol range, return a single frame range. + # If there is just a layer range, return the layer range. + # If there is no layer range but an outline range, return the outline range. + # If there is neither a layer range nor an outline range, return a single frame range. if rng and self.__outline.get_frame_range(): ol_rng = FileSequence.FrameSet(self.__outline.get_frame_range()) ly_rng = FileSequence.FrameSet(rng) - intersect = outline.util.intersect_frame_set(ol_rng, ly_rng, normalize=False) + intersect = outline.util.intersect_frame_set( + ol_rng, ly_rng, normalize=False + ) if not intersect: return None @@ -708,15 +776,15 @@ def get_frame_range(self): return rng return DEFAULT_FRAME_RANGE - def get_local_frame_set(self, start_frame): + def get_local_frame_set(self, start_frame: int) -> FileSequence.FrameSet: """ - Set the local frame set. The local frame set is the frame - list that must be handled by the execute() function. The local - frame set can have more than one frame when chunk_size is greater + Compute the local frame set. The local frame set is the frame + list that must be handled by the execute() function. The local + frame set can include more than one frame when chunk_size is greater than 1. :type start_frame: int - :param start_frame: the starting of the frame set. + :param start_frame: the starting frame of the frame set. """ chunk = self.get_chunk_size() @@ -732,7 +800,7 @@ def get_local_frame_set(self, start_frame): # # Now find the index for the current frame and start - # frame there. Find all of frames this instance + # the frame there. Find all frames this instance # is responsible for. # idx = frame_set.index(int(start_frame)) @@ -744,71 +812,80 @@ def get_local_frame_set(self, start_frame): except IndexError: break if not local_frame_set: - raise outline.exception.LayerException( - "Frame %d is outside of the frame range." % start_frame) + msg = f"Frame {start_frame} is outside of the frame range." + raise outline.exception.LayerException(msg) return outline.util.make_frame_set(local_frame_set) - def set_chunk_size(self, size): + def set_chunk_size(self, size: int) -> None: """ - Set the event's chunk size. The chunk size determines how many frames - each execute is going to handle. + Set the event's chunk size. The chunk size determines how many frames + each execute will handle. :type size: int - :param size: The size of the chunks + :param size: The size of the chunks. """ self.__args["chunk"] = int(size) - def get_chunk_size(self): + def get_chunk_size(self) -> int: """ - Return the chunk size + Return the chunk size. :rtype: int :returns: The event's chunk size. """ return int(self.__args["chunk"]) - def depend_previous(self, on_layer): + def depend_previous(self, on_layer: Layer) -> None: """ - Setup a previous frame depende on the given layer. + Setup a previous-frame dependency on the given layer. :type on_layer: L{Layer} :param on_layer: The L{Layer} to depend on. """ self.depend_on(on_layer, outline.depend.DependType.PreviousFrame) - def depend_all(self, on_layer, propigate=False, any_frame=False): + def depend_all(self, on_layer: Layer, propigate: bool = False, any_frame: bool = False) -> None: """ - Setup a layer on layer depend on the given layer. + Setup a layer-on-layer dependency on the given layer. :type on_layer: L{Layer} :param on_layer: The L{Layer} to depend on. :type propigate: boolean - :param propigate: Whether or not to propigate the depend to - other layers. Default to False. + :param propigate: Whether or not to propagate the dependency to + other layers. Defaults to False. :type any_frame: boolean - :param any_frame: Wheaether or not to setup a depend any. - Default to False. - """ - self.depend_on(self.__resolve_layer_name(on_layer), - outline.depend.DependType.LayerOnLayer, propigate, any_frame) - - def depend_on(self, on_layer, depend_type=outline.depend.DependType.FrameByFrame, - propigate=False, any_frame=False): - """ - Setup a frame by frame on layer depend on the given layer. + :param any_frame: Whether or not to setup an 'any-frame' dependency. + Defaults to False. + """ + self.depend_on( + self.__resolve_layer_name(on_layer), + outline.depend.DependType.LayerOnLayer, + propigate, + any_frame, + ) + + def depend_on( + self, + on_layer: Layer, + depend_type: str = outline.depend.DependType.FrameByFrame, + propigate: bool = False, + any_frame: bool = False, + ) -> None: + """ + Setup a frame-by-frame or layer-on-layer dependency on the given layer. :type on_layer: L{Layer} :param on_layer: The L{Layer} to depend on. :type propigate: boolean - :param propigate: Whether or not to propigate the depend to - other layers. Default to False. + :param propigate: Whether or not to propagate the dependency to + other layers. Defaults to False. :type any_frame: boolean - :param any_frame: Wheaether or not to setup a depend any. - Default to False. + :param any_frame: Whether or not to setup an 'any-frame' dependency. + Defaults to False. """ # Check for duplicates. for depend in self.__depends: @@ -828,7 +905,7 @@ def depend_on(self, on_layer, depend_type=outline.depend.DependType.FrameByFrame logger.info("adding depend %s on %s", self, on_layer) # - # Handle the depend any bullshit + # Handle the 'any frame' behavior. # if any_frame or depend_type == outline.depend.DependType.LayerOnAny: if isinstance(self, LayerPreProcess): @@ -837,9 +914,11 @@ def depend_on(self, on_layer, depend_type=outline.depend.DependType.FrameByFrame depend_type = outline.depend.DependType.FrameByFrame any_frame = False for pre in self.get_preprocess_layers(): - pre.depend_on(on_layer, - outline.depend.DependType.LayerOnLayer, - any_frame=True) + pre.depend_on( + on_layer, + outline.depend.DependType.LayerOnLayer, + any_frame=True, + ) depend = outline.depend.Depend(self, on_layer, depend_type, propigate, any_frame) self.__depends.append(depend) @@ -848,30 +927,34 @@ def depend_on(self, on_layer, depend_type=outline.depend.DependType.FrameByFrame for my_preprocess in self.get_preprocess_layers(): for on_preprocess in on_layer.get_preprocess_layers(): # Depend on the layer's pre-process - my_preprocess.depend_on(on_preprocess, outline.depend.DependType.LayerOnLayer) + my_preprocess.depend_on( + on_preprocess, outline.depend.DependType.LayerOnLayer + ) # - # Handle depend propagation. + # Handle dependency propagation. # - # Propagation occurs when a layer A depends on layer B, and - # layer C depends on layer D, but Layer A also depends on Layer + # Propagation occurs when layer A depends on layer B, and + # layer C depends on layer D, but layer A also depends on layer # C, which means layer D must now also depend on layer B. # # Currently this creates a depend-all (LayerOnLayer) between - # the propagated depends. + # the propagated dependencies. # for depend in on_layer.get_depends(): if depend.is_propagated(): for my_depend in self.get_depends(): dependant = my_depend.get_depend_on_layer() logger.info( - "propagating dependency %s -> %s", dependant, depend.get_depend_on_layer()) - dependant.depend_all(depend.get_depend_on_layer(), - propigate=False) + "propagating dependency %s -> %s", + dependant, + depend.get_depend_on_layer(), + ) + dependant.depend_all(depend.get_depend_on_layer(), propigate=False) - def undepend(self, depend): + def undepend(self, depend: outline.depend.Depend) -> None: """ - Remove the given dependency. If the dependency does not exist + Remove the given dependency. If the dependency does not exist no exception is thrown. """ try: @@ -879,12 +962,12 @@ def undepend(self, depend): except Exception as e: logger.warning("failed to remove dependency %s, %s", depend, e) - def get_depends(self): - """Return a tuple of dependencies this layer depends on.""" - # Do not let people muck with the real list. + def get_depends(self) -> List[outline.depend.Depend]: + """Return a list of dependencies this layer depends on.""" + # Do not let callers modify the real list. return self.__depends - def get_dependents(self): + def get_dependents(self) -> List[outline.depend.Depend]: """Return a list of dependencies that depend on this layer.""" result = [] for layer in self.__outline.get_layers(): @@ -893,22 +976,22 @@ def get_dependents(self): result.append(depend) return result - def check_input(self, frame_set=None): + def check_input(self, frame_set: Optional[FileSequence.FrameSet] = None) -> None: """ - Check the existance of all required input. Raise a LayerException + Check the existence of all required inputs. Raise a LayerException if input is missing. """ for name, inpt in self.__input.items(): if not inpt.get_attribute("checked"): continue if not inpt.exists(frame_set): - msg = "Check input failed (%s), the path %s does not exist." - raise outline.exception.LayerException(msg % (name, inpt.get_path())) + msg = f"Check input failed ({name}), the path {inpt.get_path()} does not exist." + raise outline.exception.LayerException(msg) - def check_output(self, frame_set=None): + def check_output(self, frame_set: Optional[FileSequence.FrameSet] = None) -> None: """ - Check the existance of all required output. Raise a LayerException - if input is missing. + Check the existence of all required outputs. Raise a LayerException + if output is missing. """ if self.get_arg("nocheck"): return @@ -916,26 +999,29 @@ def check_output(self, frame_set=None): if not output.get_attribute("checked"): continue if not output.exists(frame_set): - msg = "Check output failed (%s), the path %s does not exist." - raise outline.exception.LayerException(msg % (name, output.get_path())) + msg = ( + f"Check output failed ({name}), " + f"the path {output.get_path()} does not exist." + ) + raise outline.exception.LayerException(msg) - def add_input(self, name, inpt): + def add_input(self, name: str, inpt: Any) -> None: """ Add an input to this layer. """ if not name: - name = "input%d" % len(self.__input) + name = f"input{len(self.__input)}" name = str(name) if name in self.__input: - msg = "An input with the name %s has already been created." - raise outline.exception.LayerException(msg % name) + msg = f"An input with the name {name} has already been created." + raise outline.exception.LayerException(msg) if not isinstance(inpt, outline.io.Path): inpt = outline.io.Path(inpt) self.__input[name] = inpt - def add_output(self, name, output): + def add_output(self, name: str, output: Any) -> None: """ Add an output to this layer. """ @@ -943,15 +1029,15 @@ def add_output(self, name, output): name = "output%d" % len(self.__output) name = str(name) if name in self.__output: - msg = "An output with the name %s has already been created." - raise outline.exception.LayerException(msg % name) + msg = f"An output with the name {name} has already been created." + raise outline.exception.LayerException(msg) if not isinstance(output, outline.io.Path): output = outline.io.Path(output) self.__output[name] = output - def get_inputs(self): + def get_inputs(self) -> Dict[str, outline.io.Path]: """ Return dictionary of registered inputs. @@ -960,7 +1046,7 @@ def get_inputs(self): """ return dict(self.__input) - def get_outputs(self): + def get_outputs(self) -> Dict[str, outline.io.Path]: """ Return dictionary of registered outputs. @@ -969,85 +1055,85 @@ def get_outputs(self): """ return dict(self.__output) - def get_input(self, name): + def get_input(self, name: str) -> outline.io.Path: """ Return the named input. :rtype: outline.io.Path - :return: the assoicated io.Path object from the given name. + :return: the associated io.Path object for the given name. """ try: return self.__input[name] - except: - raise outline.exception.LayerException( - "An input by the name %s does not exist." % name) + except KeyError: + msg = f"An input by the name {name} does not exist." + raise outline.exception.LayerException(msg) - def get_output(self, name): + def get_output(self, name: str) -> outline.io.Path: """ Return the named output. :rtype: outline.io.Path - :return: the assoicated io.Path object from the given name. + :return: the associated io.Path object for the given name. """ try: return self.__output[name] - except: - raise outline.exception.LayerException( - "An output by the name %s does not exist." % name) + except KeyError: + msg = f"An output by the name {name} does not exist." + raise outline.exception.LayerException(msg) - def set_output_attribute(self, name, value): + def set_output_attribute(self, name: str, value: Any) -> None: """ - Set the given attribute on all registered output. + Set the given attribute on all registered outputs. """ logger.debug("Setting output attribute: %s = %s", name, value) for output in self.__output.values(): output.set_attribute(name, value) - def set_input_attribute(self, name, value): + def set_input_attribute(self, name: str, value: Any) -> None: """ - Set the given attribute on all registered input. + Set the given attribute on all registered inputs. """ logger.debug("Setting input attribute: %s = %s", name, value) for output in self.__input.values(): output.set_attribute(name, value) @staticmethod - def get_temp_dir(): + def get_temp_dir() -> str: """ Return the path to the current temporary directory. """ return tempfile.gettempdir() - def check_required_args(self): + def check_required_args(self) -> None: """ - Check on required properies. - If a required property does not exist throw a LayerException. + Check required properties. + If a required property does not exist, raise a LayerException. """ for key, rtype in self.__req_args: if key not in self.__args: - msg = "The %s layer requires the %s property to be set." - raise outline.exception.LayerException(msg % (self, key)) + msg = f"The {self} layer requires the {key} property to be set." + raise outline.exception.LayerException(msg) if rtype: if not isinstance(self.__args[key], rtype): - msg = "The %s layer requires %s to be of the type %s" - raise outline.exception.LayerException(msg % (self, key, rtype)) + msg = f"The {self} layer requires {key} to be of the type {rtype}" + raise outline.exception.LayerException(msg) - def get_preprocess_layers(self): + def get_preprocess_layers(self) -> List[LayerPreProcess]: """ Return all preprocess layers created by this layer. """ return self.__preprocess_layers - def add_preprocess_layer(self, preprocess): + def add_preprocess_layer(self, preprocess: LayerPreProcess) -> None: """ Add a pre-process layer created by this layer. """ self.__preprocess_layers.append(preprocess) - def load_outputs(self): + def load_outputs(self) -> None: """ - Detects the existance of an ol:outputs file and adds each - element of the file into this layer's output hash. The data + Detect the existence of an ol:outputs file and add each + element of the file into this layer's output hash. The data in ol:outputs is usually written from a pre-process. """ if not os.path.exists(os.path.join(self.get_path(), "ol:outputs")): @@ -1055,10 +1141,10 @@ def load_outputs(self): for name, output in self.get_data("ol:outputs").items(): self.add_output(name, output) - def __set_python_path(self): + def __set_python_path(self) -> None: """ - Prepends an array of paths to the python path before execute. - This method is for testing python libraries before release.. + Prepend an array of paths to the python path before execute. + This method is for testing python libraries prior to release. """ if self.get_arg("python_path"): logger.warning("WARNING: PYTHON PATH HAS BEEN ADJUSTED") @@ -1067,7 +1153,7 @@ def __set_python_path(self): return sys.path = self.get_arg("python_path") + sys.path - def __setup_frame_environment(self, frames): + def __setup_frame_environment(self, frames) -> None: """ Set helpful OL_ environment variables. """ @@ -1075,96 +1161,94 @@ def __setup_frame_environment(self, frames): os.environ["OL_LAYER_SESSION_PATH"] = self.get_path() os.environ["OL_LAYER_RANGE"] = str(frames) - def __create_output_paths(self): + def __create_output_paths(self) -> None: """ - Create directories for all registered output if possible. If + Create directories for all registered outputs if possible. If the directory already exists the operation will be skipped. """ for out in self.get_outputs().values(): if out.get_attribute("mkdir"): out.mkdir() - def __resolve_layer_name(self, layer): + def __resolve_layer_name(self, layer: Union[str, Layer]) -> Layer: """ - Resolve a layer name to a layer object + Resolve a layer name to a Layer object. """ - if not isinstance(layer, (Layer)): + if not isinstance(layer, Layer): return self.get_outline().get_layer(str(layer)) return layer - def __str__(self): - return self.get_name() - class Frame(Layer): """ - A frame is a layer with a single frame. The frame number + A Frame is a Layer that represents a single frame. The frame number defaults to the first frame of the job. """ - def __init__(self, name, **args): - Layer.__init__(self, name, **args) - def get_frame_range(self): + @override + def get_frame_range(self) -> str: """ - Return the frame's number. This overrides the - get_frame_range implementation in Layer so it always - returns a single frame. Frames are immune to being - removed from the job if they do not instersect with - the job's frame range. + Return the frame's number. This overrides the get_frame_range + implementation in Layer so it always returns a single frame. + Frames are immune to being removed from the job if they do not + intersect with the job's frame range. :rtype: String :return: The frame number. """ - # An outline's frame range might be None, in that case + # An outline's frame range might be None; in that case # just return the default frame. if self.get_outline().get_frame_range(): seq = FileSequence.FrameSet(self.get_outline().get_frame_range()) return str(seq[0]) return DEFAULT_FRAME_RANGE - def set_frame_range(self, frame_range): + @override + def set_frame_range(self, frame_range) -> None: """ - Calling this method does nothing. + Calling this method does nothing for a Frame. """ class LayerPreProcess(Frame): """ - A subclass of Frame which must run before the specifed + A subclass of Frame which must run before the specified parent layer is unlocked. The resulting object is automatically named - "parent_preprocess". A dependency is automatically setup between + "parent_preprocess". A dependency is automatically set up between the preprocess and its parent. """ - def __init__(self, creator, **args): - Frame.__init__(self, "%s_%s" - % (creator.get_name(), - args.get("suffix","preprocess")), **args) + + def __init__(self, creator, **args: Unpack[_LayerArgs]): + super().__init__(f"{creator.get_name()}_{args.get('suffix', 'preprocess')}", **args) self.__creator = creator - self.__creator.depend_on(self, outline.depend.DependType.LayerOnLayer, propigate=False) + self.__creator.depend_on( + self, outline.depend.DependType.LayerOnLayer, propigate=False + ) self.__creator.add_preprocess_layer(self) self.set_type("Util") self.set_service("preprocess") - def get_creator(self): - """Return the parent layer.""" + def get_creator(self) -> Layer: + """Return the parent layer (the creator).""" return self.__creator - def execute(self, frame): + @override + def execute(self, frame: int) -> None: """ - Perform pre-propcess execute methods and call - the super class's exceute method. + Perform pre-process execute methods and call + the superclass's execute method. """ super(LayerPreProcess, self).execute(frame) self.__save_outputs() - def get_frame_range(self): + @override + def get_frame_range(self) -> Optional[str]: """ - Return the frame's number. This overrides the - get_frame_range implementation in Frame so it always - returns a single frame. If the pre-process's creator - layer has no valid range then the pre-process + Return the frame's number. This overrides the get_frame_range + implementation in Frame so it always returns a single frame. If the + pre-process's creator layer has no valid range then the pre-process range should be None as well. :rtype: String @@ -1177,10 +1261,10 @@ def get_frame_range(self): fs = FileSequence.FrameSet(seq) return str(fs[0]) - def __save_outputs(self): + def __save_outputs(self) -> None: """ - Save outputs setup by the preprocess to the render - layer's session. The render layer will automatically call + Save outputs set up by the preprocess to the render + layer's session. The render layer will automatically call load_outputs before execution which will load in all output data saved out by this method. """ @@ -1188,27 +1272,27 @@ def __save_outputs(self): return logger.info("Saving %d outputs to ol:outputs", len(self.get_outputs())) - self.get_creator().put_data("ol:outputs", - self.get_outputs(), force=True) + self.get_creator().put_data("ol:outputs", self.get_outputs(), force=True) class LayerPostProcess(Frame): """ A subclass of Frame which always runs after the - specified parent. The resulting object is automatically named - "parent_postprocess". A dependency is automatically setup between + specified parent. The resulting object is automatically named + "parent_postprocess". A dependency is automatically set up between the parent and the post process. """ - def __init__(self, creator, propigate=True, **args): - Frame.__init__(self, "%s_postprocess" % creator.get_name(), **args) + + def __init__(self, creator: Layer, propigate: bool = True, **args: Unpack[_LayerArgs]) -> None: + super().__init__(f"{creator.get_name()}_postprocess", **args) self.__creator = creator self.depend_on(creator, outline.depend.DependType.LayerOnLayer, propigate=propigate) self.set_type("Util") - def get_creator(self): - """Return the layer that creeated this Postprocess.""" + def get_creator(self) -> Layer: + """Return the layer that created this post-process.""" return self.__creator @@ -1217,7 +1301,8 @@ class OutlinePostCommand(Frame): A post command is a special frame that kicks off after the outline is complete, even if the outline has failed. """ - def __init__(self, name, **args): - Frame.__init__(self, name, **args) + + def __init__(self, name: str, **args: Unpack[_LayerArgs]): + super().__init__(name, **args) self.set_type("Post") self.set_service("postprocess") diff --git a/pyoutline/pyproject.toml b/pyoutline/pyproject.toml index 527ce019f..09636c8eb 100644 --- a/pyoutline/pyproject.toml +++ b/pyoutline/pyproject.toml @@ -9,7 +9,8 @@ build-backend = "hatchling.build" name = "opencue_pyoutline" dynamic = ["version"] dependencies = [ - "opencue_pycue" + "opencue_pycue", + "typing_extensions; python_version < '3.12'", # Unpack and override ] requires-python = ">3.7" description = "PyOutline is a Python library for building and managing OpenCue job definitions. It provides an API to create outlines that describe jobs, layers, and tasks, making it easier to construct repeatable workflows for rendering pipelines."