athena package

Subpackages

Submodules

athena.atConstants module

athena.atConstants.PROGRAM_NAME: str = 'Athena': Project name string

athena.atConstants.CHECK: str = 'check': Name constant for the check function to implement in Process Subclasses.

athena.atConstants.FIX: str = 'fix': Name constant for the fix function to implement in Process Subclasses.

athena.atConstants.TOOL: str = 'tool': Name constant for the tool function to implement in Process Subclasses.

athena.atConstants.AVAILABLE_DISPLAY_MODE: Tuple[str, str, str] = ('Alphabetically', 'Category', 'Header'): All currently supported display modes.

athena.atConstants.DEFAULT_CATEGORY: str = 'Other': Default category for checks, this is used as a fallback an Blueprint should define an accurate category.

athena.atConstants.PROGRESSBAR_FORMAT: str = ' %p% - {0}': Format pattern for a QtWidgets.QProgressBar widget. It allows to add a custom text but force display of progress.

athena.atConstants.BLUEPRINT_TEMPLATE = {'arguments': {'': ([], {})}, 'category': '', 'links': ('', '', ''), 'process': '', 'settings': {}, 'statusOverrides': {'': {}}, 'tags': 0}: Template for what’s expected inside a Blueprint’s description.

athena.atConstants.SETTINGS_TEMPLATE = {'allowRequestStop': True, 'checkContextManagers': (), 'disableSelection': False, 'feedbackDisplayWarning': True, 'feedbackDisplayWarningLimit': 100, 'fixContextManagers': (), 'globalContextManagers': (), 'orderFeedbacksByPriority': False, 'recheck': True, 'toolContextManagers': ()}: Template for what’s supported in a Blueprints’ setting.

athena.atConstants.NO_DOCUMENTATION_AVAILABLE: str = '\nNo documentation available for this process.\n': Default text for Process documentation.

athena.atConstants.REPORT_BUG_LINK: str = 'https://github.com/gpijat/athena/issues': Link to the Github’s issues page.

athena.atConstants.TECHNICAL_DOCUMENTATION: str = 'https://athena-sanity.readthedocs.io/en/latest/': Link to the developer’s documentation.

athena.atCore module

class athena.atCore.Process(*args: Any, **kwargs: Any)[source]

Abstract class that serves as the foundation for all Athena Processes.

The Process class acts as the base (abstract) class for all user-defined check processes within the Athena framework. To create a new process, it’s essential to define at least one Thread object as a class attribute. These threads manage various feedbacks that the process will inspect and potentially address.

During the execution of your process’s check(), if any issues are detected, you need to create instances of the Feedback class (or its subclasses) and register them for the appropriate threads using the addFeedback() method. In the fix() method, iterate over these feedback instances and implement any necessary actions to automatically resolve fixable issues.

Additional methods for managing internal feedback are available, allowing you to retrieve feedbacks, iterate over them, or check if there are any feedbacks for a specific Thread.

At the end of the check process, it’s crucial to set the state for each Thread. By default, they are all set to the atStatus._DEFAULT built-in status. To change a Thread status, you can call the setSuccess() or setFail() methods. Alternatively, you can skip the execution of a specific Thread using the setSkipped() method.

Three non-implemented methods must or can be overridden to create a functional process:

check()

fix()

tool()

Communication of progress can be achieved using the setProgress* methods to provide feedback to the user on the current state of the check or fix progress. For this to work, a QProgressBar must be set using setProgressBar.

Several sunder attributes are defined at the class level, providing the ability to replace default class data. For example, defining _name_ gives a name to the process different from the class name obtained from __name__. This allows defining a user-friendly name. The currently available sunder attributes include:

_name_

_doc_

If you wish to create a custom base class for all your processes, ensure that this base class also inherits from Process to be recognized by Athena’s framework. Do not override the __new__ method without using super, or the process will not be set up as intended.

FEEDBACK_CONTAINER_CLASS

Define which type of feedback container will be internally used to register Feedback per Thread. This is intended to be overridden in a subclass if you’re willing to use a different FeedbackContainer implementation.

alias of FeedbackContainer

_name_: str = '': The Process nice name. If not set, will be set to the value of __name__

_doc_: str = '': The Process user’s documentation. If not set, will be set to the value of __doc__

_listen_for_user_interruption: AtEvent = <athena.atEvent.AtEvent object>

Event that allow to notify subscribers when the user is trying to interrupt the process execution.

classmethod threads() → Iterator[Thread][source]

Iterator to access all threads of the Process.

Returns:: Each thread instances for the current Process.
Return type:: Iterator[Thread]

check(*args: Any, **kwargs: Any) → None[source]

This method must be implemented on all Process to register feedbacks and set status for each threads

fix(*args: Any, **kwargs: Any) → None[source]

This method can be implemented to allow an automatic fix for all feedbacks retrieved by the Process check.

tool(*args: Any, **kwargs: Any) → None[source]

This method can be implemented to open a window that can allow the user to manually find or fix the errors.

set_progress_bar(progress_bar: QtWidgets.QProgressBar) → None[source]

This method should be used to setup the Process progress bar widget.

Parameters:: progress_bar (QtWidgets.QProgressBar) – The new progress bar to link to The Process instance.

set_progress(value: bool | None = None, text: bool | None = None) → None[source]

Set the progress value and/or text for the associated QProgressBar.

This method allows setting the progress value and/or text for the QProgressBar associated with the Process instance. It is typically used to provide feedback to the user on the current state of the check or fix progress.

Parameters:

value (bool | None) – The progress value to set. If None, the progress value remains unchanged.
text (bool | None) – The progress text to set. If None, the progress text remains unchanged.

set_progress_value(value: Number) → None[source]

Set the progress value of the Process progress bar if exist.

Parameters:: value (Number) – The value to set the progress to.
Raises:: TypeError – If value is not numeric.

set_progress_text(text: str) → None[source]

Set the label text of the Process progress bar if exist.

Parameters:: text (str) – Text to display in the progressBar.

clear_feedback() → None[source]

Clear all feedback associated with the threads in the process.

This method resets the feedback containers for each thread by creating new instances of FeedbackContainers. It effectively clears any previously registered feedback.

has_feedback(thread) → None[source]

Check if there is feedback registered for a specific thread.

Parameters:: thread – The thread to check for feedback.
Returns:: True if there is feedback for the specified thread, False otherwise.
Return type:: None

add_feedback(thread: Thread, feedback: Feedback) → None[source]

Add feedback to the specified thread’s feedback container.

Parameters:

thread (Thread) – The thread to which the feedback will be added.
feedback (Feedback) – The feedback instance to add.

iter_feedback(thread: Thread) → Iterator[Feedback][source]

Iterate over the feedback instances registered for a specific thread.

Parameters:: thread (Thread) – The thread for which to iterate over feedback.
Returns:: An iterator over the feedback instances for the specified thread.
Return type:: Iterator[Feedback]

feedback_count(thread: Thread) → int[source]

Get the count of feedback instances registered for a specific thread.

Parameters:: thread (Thread) – The thread for which to get the feedback count.
Returns:: The number of feedback instances for the specified thread.
Return type:: int

get_feedback_containers() → Tuple[FeedbackContainer][source]

Get a tuple of all feedback containers associated with the threads.

Returns:: A tuple containing all feedback containers associated with the threads.
Return type:: Tuple[FeedbackContainer]

set_success(thread: Thread) → None[source]

Set the success status for a specific thread’s feedback container.

Parameters:: thread (Thread) – The thread for which to set the success status.

set_fail(thread: Thread) → None[source]

Set the fail status for a specific thread’s feedback container.

Parameters:: thread (Thread) – The thread for which to set the fail status.

set_skipped(thread: Thread) → None[source]

Set the skipped status for a specific thread’s feedback container.

Parameters:: thread (Thread) – The thread for which to set the skipped status.

listen_for_user_interruption() → None[source]

Trigger an user interruption AtEvent to prematurely end process execution.

This method triggers the _listen_for_user_interruption AtEvent, which requires a registered callback to invoke the _register_interruption() method from the Qt Application.

After triggering the event, if there is a callback to invoke the appropriate method, the boolean value for __do_interrupt will be set to True, and an atExceptions.AtProcessExecutionInterrupted exception will be raised. This exception can be handle in the Qt Application to force the Process Status to atStatus._ABORTED.

Raises:: atExceptions.AtProcessExecutionInterrupted – If the interruption is caught and propagated.

Important

When the Process is executing, the Qt Event System stacks QEvents, processing them at the end of the execution, typically after all process executions are finished. To catch user interactions during execution, the Qt Event System must process pending events. The _listen_for_user_interruption Event can trigger a call to QApplication.processEvents, accomplished by adding it as a callback for this Event.

When this method runs, the callbacks are triggered, allowing the processing of pending interactions, such as a key press. A side effect is that the key press becomes part of the Qt UI thread.

To ensure the raise directly impacts the application and ends the process, it must be done from the main thread. Therefore, the _register_interruption() method needs to be called from the Qt Application to toggle the behavior of this method so it can raise. Typically, this method is called in the check or fix method, ensuring that the raise occurs from the main thread.

_register_interruption() → None[source]

Change the value for the private member __do_interrupt to True.

This simply allow to change the value of this private member from outside the class.

class athena.atCore._NumberParameter(default: Number, minimum: Number | None = None, maximum: Number | None = None, keep_in_range: bool = False)[source]

Bases: Parameter

A sub-type of Parameter that can be used to represent a numeric value.

_NumberParameter is a protected class, only used internally, it must not be instantiated due to the abstract nature of the numbers.Number type.

Warning

This implementation of Parameter is not meant to be instantiated, use the following concrete implementations instead:

IntParameter

FloatParameter

TYPE: alias of Number

type_cast(value: object) → Number[source]

Cast the input value to the numbers.Number type.

As the numbers.Number type is not instantiable, this method is meant to work wit sub-types that define an instantiable type for the TYPE. Will convert the input value to the Parameter’s type and make sur it’s in the range if the keep_in_range attribute is set to True.

Parameters:: value (object) – The value to cast to the Parameter’s type.
Returns:: The input value converted to the Parameter’s type.
Return type:: Number

validate(value: Number) → bool[source]

Validate if the input numeric value is withing the Parameter’s value range.

Parameters:: value (Number) – The input value to validate.
Returns:: Whether or not the value is withing the Parameter’s range.
Return type:: bool

class athena.atCore.AtSession(*args, **kwargs)[source]

Bases: object

Singleton class representing the Athena’s running session.

This class provides a single instance that manages the session state, including a registration system and a development mode toggle.

Example

>>> AtSession().dev = True  # Enable development mode.

property register: Register: Lazily creates and returns the Register.

property dev: bool: Get the current state of development mode.

class athena.atCore.ProtoFeedback(feedback: Any, selectable: bool)[source]

Bases: ABC

Abstract base dataclass for Feedback objects used in Athena.

This is the root of all feedback types and only meant to implement common behavior between all subclasses. It implement the default attributes and behavior for iteration.

feedback: Any: Represents the data held by the feedback instance, which can be of any type based on the specific needs of the associated process.

selectable: bool: Determines whether the feedback is selectable. This attribute is not used in comparison, meaning two instances with the same data (feedback) will be considered similar, regardless of the selectable value.

children: List[ProtoFeedback]: Holds references to all child feedback instances. Unlike the feedback itself, this attribute is mutable, and as such, it is not considered in comparison, hashing, or sorting.

abstract select()[source]

Abstract method to select the feedback item.

Raises:: NotImplementedError – This method must be implemented by subclasses.

abstract deselect() → None[source]

Abstract method to deselect the feedback item.

Raises:: NotImplementedError – This method must be implemented by subclasses.

parent(*feedbacks: List[ProtoFeedback]) → None[source]

Add child feedback items to create a hierarchical structure.

Parameters:: *feedbacks (List[ProtoFeedback]) – Child feedback items to be added.

class athena.atCore.FeedbackContainer(feedback: Thread, selectable: bool, status: Status)[source]

Bases: ProtoFeedback

Base class for Feedback Container containing feedbacks for a specific Thread.

The FeedbackContainer is only meant to be used as a root for a Feedback hierarchy, compared to other feedback, it’s “data” is a Thread, for which it contains feedbacks. It’s also able to have a status, that must be set based on the Thread FailStatus or SuccessStatus.

Notes

The FeedbackContainer class implement a really basic selection and deselection process, you should consider reimplementing it if you intend to use a very specific selection mechanism or need to deal with advanced selection/deselection. (= A lot of elements to deal with)

feedback: Thread: The FeedbackContainer data must be the Thread for which it contain feedbacks.

status: Status: The FeedbackContainer status represent the result state of the Thread based on the contained feedbacks

select(replace: bool = True) → bool[source]

Allow selection for the FeedbackContainer.

If it isn’t selectable, this won’t do anything. On the other hand, if it is, the behavior is to call the select method for each Feedback in the container’s children one by one. Also, if replace is set to True, only the first child will replace the current selection while others will be added to it.

Parameters:: replace (bool) – Whether we replace or add to the current selection.
Returns:: The current state of the replace parameter to know if following selection, in subclass implementation of this method must replace or add.
Return type:: bool

deselect() → None[source]

Allow deselection for the FeedbackContainer.

If it isn’t selectable, this won’t do anything. Otherwise, call the deselect method of each child Feedback, one by one.

set_status(status: Status) → None[source]

Change the current FeedbackContainer status to the given status.

Parameter:: status: The new status to set this FeedbackContainer value to.

class athena.atCore.Feedback(feedback: Any, selectable: bool)[source]

Bases: ProtoFeedback

Base class representing a single found Feedback for Athena.

The Feedback is an abstract way for Processes to register what they find. This class serves as the base Feedback class, suitable for most situations but includes no specific behavior for software/OS selection. The methods implement behavior, not actual selection actions. If you need real selection, consider implementing a subclass and overriding the select/deselect methods. Using this Base Feedback as the parent for other subclassed feedback will cascade the selection to them.

Notes

For DCC-specific implementations, refer to athena.software. The list is not exhaustive and emphasizes common and global behavior. If you have specific needs for another software or pipeline-specific behavior, you can implement your own Feedback subclass.

select(replace: bool = True) → bool[source]

Allow selection for the Feedback.

This implementation is a selection cascade mechanism meant to be overridden and/or called as a superclass. If the feedback is selectable, it won’t do anything, as there are no defined selection behaviors for the default Feedback. If the Feedback is not selectable, it cascades to every child and calls their select method one by one.

Parameters:: replace (bool) – Whether to replace or add to the current selection.

Notes

Having a non-selectable feedback can allow displaying data that has no selection capabilities or create a group for sub-feedbacks and select them all at once.

deselect() → None[source]

Allow deselection for the Feedback.

This implementation is a deselection cascade mechanism meant to be overridden and/or called as a superclass. If the feedback is not selectable, it won’t do anything, as there are no defined deselection behaviors for the default Feedback. If the Feedback is selectable, it cascades to every child and calls their deselect method one by one.

Notes

Having a non-selectable feedback can allow displaying data that has no deselection capabilities or create a group for sub-feedbacks and deselect them all at once.

feedback: Any: Represents the data held by the feedback instance, which can be of any type based on the specific needs of the associated process.

selectable: bool: Determines whether the feedback is selectable. This attribute is not used in comparison, meaning two instances with the same data (feedback) will be considered similar, regardless of the selectable value.

children: List[ProtoFeedback]: Holds references to all child feedback instances. Unlike the feedback itself, this attribute is mutable, and as such, it is not considered in comparison, hashing, or sorting.

class athena.atCore.Thread(title: str, fail_status: ~athena.atStatus.FailStatus = <FailStatus: Error (inf)>, success_status: ~athena.atStatus.SuccessStatus = <SuccessStatus: Success (-inf)>, documentation: str | None = None)[source]

Bases: object

Represent a task within an Athena Process.

A Thread in Athena signifies a single responsibility task assigned to a Process. Each Thread entails the responsibility for the Process to perform tests and register Feedbacks if any unexpected behavior is detected.

The Thread encapsulates the task, defining its importance by specifying both failure and success statuses. The fail status denotes failure, and the success status indicates successful completion.

These statuses can be overridden in the Blueprint for enhanced flexibility. It is essential to generate the Thread with its default and common statuses.

Threads should be defined as class attributes in the Process and passed to methods that handle feedbacks.

property title: str

Getter for the Thread’s title.

Returns:: The thread’s title.

property fail_status: FailStatus

Getter for the Thread’s fail status.

Returns:: The thread’s fail status.

property success_status: SuccessStatus

Getter for the Thread’s success status.

Returns:: The thread’s success status..

override_fail_status(status: FailStatus) → None[source]

Override the actual fail status.

Parameters:: status (FailStatus) – The Status class to use as new fail status.

override_success_status(status: SuccessStatus) → None[source]

Override the actual success status.

Parameters:: status (SuccessStatus) – The Status class to use as new success status.

status(state: bool) → Status[source]

Get the Thread’s current status based on given state boolean.

Returns:: Success status if state is true, fail status otherwise.
Return type:: Status

class athena.atCore.Register[source]

Bases: object

The register is a container that allow the user to load and manage blueprints.

After initialization the register will not contain any data and you will need to load the Blueprint you want for your current Athena Session. The role of the register is to act as a loader and data keeper for all existing Blueprints you defined and want your user’s to be able to use for their sanity checking.

load_blueprints_from_package_str(package: str) → None[source]

Load blueprints from a package name.

This method calls the loadBlueprintsFromPackage method with the imported package object.

Parameters:: package (str) – The python formatted path of the package to load blueprints from.

Warning

This method is deprecated as it suppose a specific hierarchy, a feature inherited from the early ages of Athena. You’re not supposed to have a specific hierarchy in you packages to load their blueprints.

load_blueprints_from_package(package: module) → None[source]

Load blueprints from a package object.

This method iterates over the blueprints paths in the package and calls the loadBlueprintFromModulePath method for each one.

Parameters:: package (module) – The package object to load blueprints from.

Warning

This method is #deprecated** as it suppose a specific hierarchy, a feature inherited from the early ages of Athena. You’re not supposed to have a specific hierarchy in you packages to load their blueprints.

load_blueprint_from_module_path(module_path: str) → None[source]

Load a blueprint from a module OS path.

This method converts the module path to a python import path and load blueprints from it..

Parameters:: module_path (str) – The path of the module to load a blueprint from.

load_blueprint_from_python_import_path(import_str: str) → None[source]

Load a blueprint from a module python’s import path.

This method will import the module from the string and load blueprints from it.

Parameters:: import_str (str) – The python import path to the Blueprint’s module to load.

load_blueprint_from_module(module: module) → None[source]

Load a blueprint from a module object.

This method creates a Blueprint object from the module and adds it to the list of blueprints. If a blueprint with the same name already exists, it replaces it with the new one.

Parameters:: module (module) – The module object to load a blueprint from.

clear() → None[source]

Remove all loaded blueprints from this register.

property blueprints: Tuple[Blueprint, ...]

Getter for Register’s blueprints

Returns:: All blueprints in the current register.

property current_blueprint: Blueprint

Getter for current blueprint.

Returns:: The current blueprint.

blueprint_by_name(name: str) → Blueprint | None[source]

Get a blueprints from the Register based on it’s name.

Will try to find the first Blueprint that match the given name. If none are found, nothing is returned.

Parameters:: name (str) – The name of the blueprint to find.
Returns:: The blueprint that match the name, or None if no blueprint match the given name.
Return type:: Blueprint | None

reload() → None[source]

Clear the currently loaded blueprints and reload them to ensure all blueprints are up to date.

This is mostly intended for development usage, as users are not likely to update the Blueprints. It’s the same are reloading the blueprints, but as they are hold in objects inside a list, they need to be rebuilt with the newly loaded instance of the module.

class athena.atCore.Blueprint(module: module)[source]

Bases: object

Configuration for Athena processes, specifying order and settings.

The Blueprint serves as a configuration for Athena processes, defining the order in which processes are executed and providing settings for each process. It requires two members to be defined in the module:

header: A list of names in the desired execution order, representing the sequence of processes.

descriptions: A dictionary where each value in the header contains data to initialize a Processor.

An optional member called settings can be added, which must be a dictionary as well. The values in this dictionary modify the global execution behavior for the current blueprint.

The Blueprint lazily loads all its data on demand to minimize initialization calls. For example, processors are not created until the processors() attribute is called.

Notes

The name of a processor is based on the name of its module. Data can be stored on a Blueprint using the setData method. This allows storing pre-existing widgets or any other type of data.

property name: str

Getter for the Blueprint’s name.

Returns:: The Blueprint’s name.

property module: module

Getter for the Blueprint’s module

Returns:: The Blueprint’s module.

property file: str

Lazy getter for the Blueprint’s module file path.

Returns:: The blueprint’s module file path.

property icon: str

Lazy getter for the Blueprint’s icon path.

Returns:: The Blueprint’s icon path.

Notes

The icon must be a .png file in the same folder as the Blueprint’s module.

property header: Tuple[str, ...]

Lazy getter for the Blueprint’s header.

Returns:: The value for the header attribute in the Blueprint’s module or an empty Tuple.

property descriptions: Dict[str, Dict[str, Any]]

Lazy getter for the Blueprint’s descriptions.

Returns:: The value for the descriptions attribute in the Blueprint’s module or an empty dict.

property settings: Dict[str, Any]

Lazy getter for the Blueprint’s descriptions.

Returns:: The value for the settings attribute in the Blueprint’s module or an empty dict.

property processors: Tuple[Processor, ...]

Lazy getter for the Blueprint’s processors.

This will create all the processors from the Blueprint’s descriptions ordered based on the header and will automatically resolve the links for each description in case this is meant to be used in batch.

Returns:: A tuple containing all Processor for the current Blueprint’s description.

processor_by_name(name: str) → Processor | None[source]

Find a processor from blueprint’s processors based on it’s name.

Parameters:: name (str) – The name of the processor to find.
Returns:: The processor that match the name, or None if no processor match the given name.
Return type:: Processor | None

class athena.atCore.Tag(value)[source]

Bases: IntFlag

Tags are modifiers used by athena to affect the way a process behavior.

Tags are defined in the Blueprint and will be parsed by the Processor which will change it’s configuration based on them. The Process by itself is totally unaware of it’s Tags as they are only used by it’s wrapper, the Processor.

Tags allow Process to be optional, non blocking, hide their checks or even be completely disabled or dependant to another Process using Link.

NO_TAG: int = 0: This is used as default, representing the absence of any tag

DISABLED: int = 1: Define if a process should be disabled (by default it is enabled)

NO_CHECK: int = 2: This tag will remove the check of a process, it will force the is_checkable to False in blueprint.

NO_FIX: int = 4: This tag will remove the fix of a process, it will force the is_fixable to False in blueprint.

NO_TOOL: int = 8: This tag will remove the tool of a process, it will force the has_tool to False in blueprint.

NON_BLOCKING: int = 16: A non blocking process will raise a non blocking error, its error is ignored.

NO_BATCH: int = 32: This process will only be executed in ui.

NO_UI: int = 64: This process will only be executed in batch.

OPTIONAL: int = 17: This tag will set a check optional, an optional process is not checked by default and will.

DEPENDANT: int = 14: A dependent process need links to be run through another process.

class athena.atCore.Link(value)[source]

Bases: Enum

Give access to sentinel objects for each kind of link.

CHECK: object = <object object>: Represent the link to or from a Process’ check method

FIX: object = <object object>: Represent the link to or from a Process’ fix method

TOOL: object = <object object>: Represent the link to or from a Process’ tool method

class athena.atCore.Processor(process: str, category: str | None = None, arguments: Mapping[str, Tuple[Tuple[Any, ...], Mapping[str, Any]]] | None = None, tags: Tag = Tag.NO_TAG, links: Tuple[Tuple[str, Link, Link], ...] | None = None, status_overrides: Mapping[str, Mapping[Type[Status], Status]] | None = None, settings: Mapping[str, Any] | None = None, **kwargs: Any)[source]

Bases: object

Proxy object representing a Process configured from a Blueprint description.

The Processor is constructed based on a Python import string, responsible for instantiating and managing a specific Process. It requires essential data typically specified in a Blueprint’s description. The class internally determines the implemented methods on the Process class, setting up attributes to describe available functionalities.

In addition, the Processor resolves Tags and slightly overrides the initial Process behavior (e.g., removing fix, disabling the Process in batch mode). This adds flexibility to the framework by allowing customization of the Process’s behavior.

The implementation of the Processor class is predominantly lazy, initializing only the necessary arguments during instantiation and deferring expensive computation until they are actually needed, while caching their results. The Processor takes on the responsibility of resolving all Tag, Link, and Status overrides for all Process’ Thread. Additionally, it can be made aware of Blueprint settings through the setting argument or receive extra data for later use.

Overall, the Processor plays a pivotal role in configuring and executing Processes defined in a Blueprint, contributing to the framework’s flexibility and ease of use. Whether it’s for using them in a UI or batch.

property module_name: str

Lazy getter for the Processor’s Process module name.

Returns:: The Processor’s Process module name.

property module: module

Lazy getter that import the Processor’s Process module.

Returns:: The Processor’s Process module object that was imported.

property process_class

Lazy getter for the Processor’s Process class

Returns:: The Processor’s Process class

property process: Type[Process]

Lazy getter for the Processor’s Process class

Returns:: An initialized instance of the Processor’s Process class.

property parameters: Tuple[Parameter, ...]

Lazy getter Processor’s Process’ Parameters.

Returns:: All Parameters objects for the Processor’s Process.

property overridden_methods: List[str]

Lazy getter for the overridden methods of the Processor’s Process class.

Returns:: List of name for each overridden method on the Processor’s Process class compared to Process

property nice_name: str

Lazy getter for the Processor’s Process nice name.

Returns:: A nice name for the Processor’s Process, split if on camelCase.

property docstring: str

Lazy getter for the Processor’s Process docstring

Use the value in the Processor’s _doc_ attribute, if it’s empty, use class docstring __doc__. If it’s also empty, fallback on NO_DOCUMENTATION_AVAILABLE

Returns:: The formatted docstring to be more readable and also display the path of the process.

property has_check_method: bool

Lazy getter to know if the Processor’s Process has a check method.

Returns:: True if the Processor’s Process has a check method, False otherwise.

property has_fix_method: bool

Lazy getter to know if the Processor’s Process has a fix method.

Returns:: True if the Processor’s Process has a fix method, False otherwise.

property has_tool_method: bool

Lazy getter to know if the Processor’s Process has a tool method.

Returns:: True if the Processor’s Process has a tool method, False otherwise.

property raw_name: str

Getter fore the Processor’s Process raw name.

Returns:: The name for the Processor’s Process, this will fallback to the class name if no _name_ is set.

property is_enabled: bool

Getter for the Processor’s state.

Returns:: True if the Processor is enabled, False Otherwise.

property is_checkable: bool

Getter for the Processor’s checkable state.

Returns:: True if the Processor is checkable, False Otherwise.

property is_fixable: bool

Getter for the Processor’s fixable state.

Returns:: True if the Processor is fixable, False Otherwise.

property in_ui: bool

Getter for Processor’s UI state.

Returns:: True if the Processor can be run in UI mode, False Otherwise.

property in_batch: bool

Getter for Processor’s batch state.

Returns:: True if the Processor can be run in batch mode, False Otherwise.

property is_non_blocking: bool

Getter to know whether the Processor’s Process FailStatus are blocking or not.

Returns:: False if the Processor’s Process FailStatus are blocking, True otherwise.

property category: str

Get the Processor’s category.

Returns:: The Processor’s category name.

get_arguments(method: str) → Tuple[List[Any], Dict[str, Any]][source]

Retrieve arguments values for the given method of the Processor’s Process.

Parameters:: method (str) – The method name as str for which to retrieve the arguments and keyword arguments.
Returns:: Tuple containing a list of arguments values and a dict of keyword arguments values.
Return type:: Tuple[List[Any], Dict[str, Any]]

Notes

This method will not raise any error, if no argument is found, return a tuple containing empty list and empty dict.

get_settings(setting: str, default: Any | None = None) → Any[source]

Get the value for a specific setting if it exists.

Parameters:

setting (str) – The setting to get from the Processor’s settings.
default (Any | None) – The default value to return if the Processor does not have any value for this setting.

Returns:

The value for the requested setting or the default value if the requested setting does not exists.

Return type:

Any

get_lowest_fail_status() → FailStatus[source]

Get the lowest Fail status from all Thread from the Processor’s Process.

Returns:: The Lowest Fail Status of the Processor’s Process across all it’s Thread
Return type:: FailStatus

get_lowest_success_status() → SuccessStatus[source]

Get the lowest Success status from all Thread from the Processor’s Process.

Returns:: The Lowest Success Status of the Processor’s Process across all it’s Thread
Return type:: SuccessStatus

check(links: bool = True, do_profiling: bool = False) → Tuple[FeedbackContainer, ...][source]

This is a wrapper for the Processor’s Process check.

This will automatically execute the check method with it’s right parameters and profile the execution if requested. If the links attribute is set to True, the Processor’s Link will also be executed..

Parameters:

links (bool) – Should the wrapper launch the connected links or not.
do_profiling (bool) – Whether the check method must be run with the Processor’s Profiler.

Returns:

All feedback containers for the Processor’s Process post check.

Return type:

Tuple[FeedbackContainer, …]

fix(links: bool = True, do_profiling: bool = False) → Tuple[FeedbackContainer, ...][source]

This is a wrapper for the Processor’s Process fix.

This will automatically execute the fix method with it’s right parameters and profile the execution if requested. If the links attribute is set to True, the Processor’s Link will also be executed..

Parameters:

links (bool) – Should the wrapper launch the connected links or not.
do_profiling (bool) – Whether the fix method must be run with the Processor’s Profiler.

Returns:

All feedback containers for the Processor’s Process post fix.

Return type:

Tuple[FeedbackContainer, …]

tool(links: bool = True, do_profiling: bool = False) → Any[source]

This is a wrapper for the Processor’s Process tool.

This will automatically execute the tool method with it’s right parameters and profile the execution if requested. If the links attribute is set to True, the Processor’s Link will also be executed..

Parameters:

links (bool) – Should the wrapper launch the connected links or not.
do_profiling (bool) – Whether the tool method must be run with the Processor’s Profiler.

Returns:

The value returned by the Processor’s Process tool method. Usually, None, but it could be the tool widget so it can be parented to the UI this Processor’s is runt from.

Return type:

Any

run_links(which: Link) → None[source]

Execute the Processor’s links for the given Link.

Parameters:: which (Link) – Which link we want to run.

get_parameter(parameter: Parameter) → Any[source]

Get the value for the given Parameter object.

Returns:: The current value for the given Parameter object.
Return type:: Any

Notes

Python descriptors are defined on classes but hold different values per instance of the class that owns them, for this reason, we can’t query the value from the Parameter itself. This method query the value as it knows both the class that own the Parameter and the instance for which we want the value.

set_parameter(parameter: Parameter, value: Any) → Any[source]

Set the given value to the given Parameter object.

Returns:: The current value for the given Parameter object after being set. This allows to know the exact value the attribute was set on after it does it’s validation.
Return type:: Any

Notes

Python descriptors are defined on classes but hold different values per instance of the class that owns them, for this reason, we can’t set the value from the Parameter itself. This method query the value as it knows both the class that own the Parameter and the instance for which we want to set the value.

setup_tags() → None[source]

Setup the tags used by this Processor to modify the it’s behavior.

resolve_links(linked_objects: List[Processor | None], check: Link = 'check', fix: Link = 'fix', tool: Link = 'tool') → None[source]

Resolve the links between the given objects and the current Processor.

This need to be called with an ordered list of Processor or None for Processor to skip. (e.g. to skip those that should not be linked because they dont have to be run in batch or ui.)

Parameters:

linked_objects (List[Processor | None]) – List of all objects used to resolve the current Processor’s links. Objects to skip have to be replaced with None.
check (Link) – Name of the method to use as check link on the given objects.
fix (Link) – Name of the method to use as fix link on the given objects.
tool (Link) – Name of the method to use as tool link on the given objects.

set_progress_bar(progress_bar: QtWidgets.QProgressBar) → None[source]

Set the ProgressBar object in the UI to be used by the Processor’s Process.

Made to be called in the ui this method allow to give access to the progress bar in the Processor’s Process to give feedback on the execution progress while performing check and/or fix.

Parameters:: progress_bar (QtWidgets.QProgressBar) – ProgressBar object object to connect to the process to display check and fix progression.

get_data(key: str, default: Any | None = None) → Any[source]

Get the Processor’s Data for the given key or default value if key does not exists.

Parameters:

key (str) – The key to get the data from.
default (Any | None) – The default value to return if the key does not exists.

Returns:

The value at the given key in the Processor’s data if the key exists. Else the default value is returned.

Return type:

Any

set_data(key: str, value: Any) → None[source]

Set the Processor’s Data for the given key

Parameters:

key (str) – The key to set the data for.
value (Any) – The value to store as data for the given key.

interrupt()[source]: Register user interruption for the Processor’s Process

class athena.atCore.Parameter(default: T)[source]

Bases: ABC

Athena Parameter descriptor that represent a variable value to be used in an Athena Process

The Parameter is a python descriptor object that must be defined at the class level. It require a value that will be used both as it’s default value and it’s current value until it’s changed. The purpose of using a Parameter is to have a variable value the user can tweak to modify how the Process behave. It’s possible to implement different subtype of Parameter to manage different Python types, for this, it’s important to implement 2 methods so the Parameter can convert an input value (usually a string) and validate that this value is correct.

TYPE: Type[T]: Represent the type of the Parameter and is used for type casting in the typeCast() method

property name: str

Getter for the Parameter’s nice name.

Basically, as the name is mangled, remove the 2 leading underscores.

Returns:: The Parameter’s nice name without leading underscores.

property default: T

Getter for the Parameter’s default value.

Returns:: The Parameter’s default value.

abstract type_cast(value: Any) → T[source]

Cast the input value to the Parameter’s type set in TYPE.

This method is not implemented on the Parameter abstract base class, it needs to be implemented in each an every subclass that are meant to be instantiated and use in an Athena Process. The input may be of various type so this method should handle different case of type casting. To do so, you can use the the TYPE attribute to convert the input value to the desired type.

Parameters:: value (Any) – The input value to convert to the Parameter’s type.
Raises:: NotImplementedError – This is an abstract method of an abstract base class, it needs to be implemented in subclasses.

abstract validate(value: T) → bool[source]

Validate that the input value is accepted and therefore can be set to the Parameter’s value.

This method is meant to validate that the given value is an acceptable value for the Parameter. For basic Parameter, this test can simply return True, making all values of the Parameter’s Type acceptable.

Parameters:: value (T) – The value of the Parameter’s type to validate.
Raises:: NotImplementedError – This is an abstract method of an abstract base class, it needs to be implemented in subclasses.

class athena.atCore.BoolParameter(default: bool)[source]

Bases: Parameter

A concrete sub-type of Parameter that can be used to represent a boolean value.

TYPE: alias of bool

type_cast(value: Any) → bool[source]

Cast the input value to the bool type.

The casting is simple and support bool, str and int, all value of those type that can be considered True will return True, every other values will be considered False. The accepted values are:

bool: True

str: ‘True’, ‘true’, ‘Yes’, ‘yes’, ‘ok’, ‘1’

int: 1

Parameters:: value (Any) – The value to cast to bool.
Returns:: The input value evaluated as a boolean.
Return type:: bool

validate(value: bool) → bool[source]

Validate that the given bool is valid.

The given value is valid as long as it’s a boolean.

Parameter:: value: The value to validate.

Returns:: Whether or not the input value is valid. (If it’s of the right type.)
Return type:: bool

class athena.atCore.IntParameter(default: Number, minimum: Number | None = None, maximum: Number | None = None, keep_in_range: bool = False)[source]

Bases: _NumberParameter

A concrete sub-type of Parameter that can be used to represent an integer value.

TYPE: alias of int

class athena.atCore.FloatParameter(default: Number, minimum: Number | None = None, maximum: Number | None = None, keep_in_range: bool = False)[source]

Bases: _NumberParameter

A concrete sub-type of Parameter that can be used to represent a floating point value.

TYPE: alias of float

class athena.atCore.StringParameter(default: str, validation: Sequence[str] | None = None, case_sensitive: bool = True)[source]

Bases: Parameter

A concrete sub-type of Parameter that can be used to represent a string value.

TYPE: alias of str

type_cast(value: object) → str[source]

Cast the input value to string.

Parameters:: value (object) – The input value to convert to the Parameter’s type.
Returns:: The string representation for the input value.
Return type:: str

validate(value: str) → bool[source]

Validate if the input value based on the validation if any.

Parameters:: value (str) – The input value to validate.
Returns:: Whether or not the input value respect the validation if any, else True.
Return type:: bool

athena.atExceptions module

exception athena.atExceptions.AthenaException[source]

Bases: BaseException

Base Athena Exception, must be inherited by all Athena’s Exceptions.

exception athena.atExceptions.AtProcessExecutionInterrupted[source]

Bases: AthenaException, RuntimeError

Exception raised when a process execution is interrupted by the user

athena.atStatus module

class athena.atStatus._BuiltInStatus(name, color, level)[source]

Bases: Status

Represent a Built-In Status, can be instantiated to define a new Built-In level

Creating new built-in Status must be reserved for framework level behavior, developers must work with other Status types. Built-in Status usually represent special case and have special behavior recognized by the framework itself or third-party UI/Extensions.

Notes

_BuiltInStatus instances should use special level, 0.0 is common and totally fine for a “default” kind of built-in Status, on the other hand, built-in statuses that represent an exception may benefit from something like nan. At the end, for _BuiltInStatus the level is less important as it’s mostly used to sort Status and built-in are handle differently as they are “specials” by nature.

athena.atStatus._DEFAULT: _BuiltInStatus = <_BuiltInStatus: Default (0.0)>: Default Status, it is set as base status for all Process

athena.atStatus._SKIPPED: _BuiltInStatus = <_BuiltInStatus: Skipped (nan)>: Represent the state of a Process for which execution has been skipped.

athena.atStatus._ABORTED: _BuiltInStatus = <_BuiltInStatus: Aborted (nan)>: Status to represent the state of a Process that has been aborted by user.

athena.atStatus._EXCEPTION: _BuiltInStatus = <_BuiltInStatus: Exception (nan)>: Status for a Process which encountered an Exception and was interupted.

class athena.atStatus.Status(name, color, level)[source]

Bases: ABC

Base Status class from which status inherit From.

A Status represent the result state of an Athena’s Process, it allows to categorise and prioritise it’s state using the level. A status also have a color assigned that may be used in a user interface to display the result to the users.

Important

Every subclass of status must be frozen and ordered dataclass so that comparison operator will be implemented to sort the different statuses.

property name

property color

property level

class athena.atStatus.FailStatus(name, color, level)[source]

Bases: Status

Represent a Fail Status, can be instantiated to define a new Failure level

Notes

FailStatus should use level values greater than 0, the higher the value is, the more critical the Status is.

class athena.atStatus.SuccessStatus(name, color, level)[source]

Bases: Status

Represent a Success Status, can be instantiated to define a new Success level

Notes

SuccessStatus should use level values lower than 0, the lower the value is, the less critical the Status is.

athena.atStatus.SUCCESS: SuccessStatus = <SuccessStatus: Success (-inf)>: Base success status, represent the classic state of a successful Process execution.

athena.atStatus.CORRECT: SuccessStatus = <SuccessStatus: Correct (-1.0)>: Represent a “good enough” Process execution. Still successful but not optimal.

athena.atStatus.WARNING: FailStatus = <FailStatus: Warning (1.0)>: Represent a lightly fail status, that are usually not too problematic as is but that may require attention.

athena.atStatus.ERROR: FailStatus = <FailStatus: Error (inf)>: Base fail status, represent the classic state of a failed Process execution.

athena.atStatus.get_all_statuses() → Tuple[Status, ...][source]

Return all existing Status in a list.

Returns:: All instance of Status subclass defined. Contains the built-in one (defined in this module) as well as user defined Success subclass.
Return type:: Tuple[Status, …]

athena.atStatus.get_status_by_name(name: str) → Status | None[source]

Find Status instance based on it’s name.

Parameters:: name (str) – The name of the status to find.
Returns:: The status that match the name if any, else None.
Return type:: Status | None

athena.atStatus.get_all_fail_status() → Tuple[FailStatus, ...][source]

Get all Fail Status instances.

Returns:: All Fail Statuses instances.
Return type:: Tuple[FailStatus, …]

athena.atStatus.get_all_success_status() → Tuple[SuccessStatus, ...][source]

Get all Success Status instances.

Returns:: All Success Statuses instances.
Return type:: Tuple[SuccessStatus, …]

athena.atStatus.lowest_fail_status() → FailStatus[source]

Get the lowest Fail Status instance based on Status.level.

Returns:: The Fail Status with the lowest level.
Return type:: FailStatus

athena.atStatus.highest_fail_status() → FailStatus[source]

Get the highest Fail Status instance based on Status.level.

Returns:: The Fail Status with the highest level.
Return type:: FailStatus

athena.atStatus.lowest_success_status() → SuccessStatus[source]

Get the lowest Success Status instance based on Status.level.

Returns:: The Success Status with the lowest level.
Return type:: SuccessStatus

athena.atStatus.highest_success_status() → SuccessStatus[source]

Get the highest Success Status instance based on Status.level.

Returns:: The Success Status with the highest level.
Return type:: SuccessStatus

athena.atUtils module

athena.atUtils.iter_blueprints_path(package: str, software: str = 'standalone', verbose: bool = False) → Iterator[str][source]

Retrieve available envs from imported packages.

Retrieve the currently imported packages path that match the pattern to works with this tool: {program}_{prod} Then, generate the usual path to the env using the package, the current software for the first sub package and env to the desired package.

Parameters:

package (str) – This is the string path to a python package.
software (str) – The software for which to get envs. (default: ‘standalone’)
verbose (bool) – Define if the function should log information about its process.

Returns:

Return a dict containing all envs for the given package and software. The key is the env and the value is a dict containing the imported module object for the env and its str path.

Return type:

Iterator[str]

athena.atUtils.get_packages() → Tuple[str, ...][source]

Get all packages that match the tool convention pattern.

Loop through all modules in sys.modules.keys() and package those who match the tool convention pattern that is {PROGRAM_NAME}_*

Parameters:: verbose – Define if the function should log information about its process. (default: False)
Returns:: Return a dict containing all package that match the pattern of the tool The key is the prod and the value is a dict containing the module object and its str path.
Return type:: Tuple[str, …]

Deprecated since version 0.1.0-beta.2.

athena.atUtils.import_process_module_from_path(process_str_path: str) → module[source]

Import the Process module from the given process python import string.

Parameters:: process_str_path (str) – Python import string to a Process class.
Returns:: The imported module that contains the Process’ class.
Raises:: ImportError – If the imported Process module is missing the mentioned Process class.
Return type:: module

athena.atUtils.get_software() → str[source]

Get the current software from which the tool is executed.

Fallback on different instruction an try to get the current running software. If no software are retrieved, return the default value.

Returns:: The current software if any are find else the default value.
Return type:: str

Deprecated since version 0.1.0-beta.2.

athena.atUtils.get_os() → str[source]

Get the current used OS platform.

If the Process Platform is Darwin, return MacOs instead for simplicity.

Returns:: The current os name.
Return type:: str

athena.atUtils.python_import_path_from_path(path: str) → str[source]

Generate a python import string path from a system path to a python Module or Package.

Iterate through all directories in the path and include it in the python import string if it contains an __init__.py module.

Parameters:: path (str) – The system path to convert to a python import string path.
Returns:: The python import string path generated from the given system path.
Raises:: IOError – If the given system path does not exists.
Return type:: str

athena.atUtils.import_from_str(module_str: str, verbose: bool = False) → module[source]

Try to import the module from the given string

Parameters:

module_str (str) – Path to a module to import.
verbose (bool) – Define if the function should log information about its process. (default: False)

Returns:

The loaded Module or None if fail.

Return type:

module

athena.atUtils.reload_module(module: module) → module[source]

Reload the given module object using the right reload function, whether it’s from imp or importLib.

Python 3.4 and above use the reload function from the importLib module while previous version use whether the reload function from the imp module (>3.0 <3.4) or the built-in function directly (<3.0).

Parameter:: module: The module to reload.

Returns:: The reloaded module given to the function, for convenience.
Return type:: module

Notes

This function exists as an helper to simplify Athena’s module reloading without having to deal with which reload function to use.

athena.atUtils.module_from_str(python_code: str, name: str = 'DummyAthenaModule') → module[source]

Build a Module object with the given str as it’s code.

This will build a python module object and set’s it’s code so it acts as a normal module and can be loaded into Athena’s Register.

Parameters:

python_code (str) – The Python code for the module as a string.
name (str) – Name for the created module object.

Returns:

A python module that contains the given code and can be used the same way any module can.

Return type:

module

Deprecated since version 0.1.0-beta.2.

athena.atUtils.import_path_str_exist(import_str: str) → bool[source]

Tells whether or not the given python import string is valid or not.

Parameters:: import_str (str) – The python import string path to a module or package.
Returns:: Whether or not the given import string is valid and could be imported.
Return type:: bool

athena.atUtils.get_overridden_methods(instance: T, cls: Type[T]) → Dict[str, function][source]

Get all methods that have been overridden from a the given instance and the given type.

Parameters:

instance (T) – An instance of a subclass of cls.
cls (Type[T]) – An object type to compare the instance to.

Returns:

All method that have been overridden from the instance in the given class.

Return type:

Dict[str, function]

athena.atUtils.camel_case_split(to_split: str) → str[source]

Format a string write with camelCase convention into a string with space.

Parameters:: to_split (str) – The camelCase string to split and format.
Returns:: The given string camelCase string splitted from upper cases with whitespaces instead.
Return type:: str

class athena.atUtils.Singleton[source]

Bases: type

Singleton Metaclass to implement the design pattern in different class definition.

athena.atUtils.create_new_athena_package_hierarchy(root_directory: str) → None[source]

Create a new Athena’s package hierarchy at the given root directory.

Parameters:: root_directory (str) – The root directory at which to create a new Athena’s package hierarchy.

Deprecated since version 0.1.0-beta.2.

athena.atUtils.map_mapping(function: Callable[[T], R], mapping: Mapping[T]) → Mapping[R][source]

Execute the given function on all values inside the given Mapping object.

Parameters:

function (Callable[[T], R]) – The function to call for each value and sub-value of the given mapping.
mapping (Mapping[T]) – The Mapping object to recursively iterate on and execute the function for each of it’s values.

Returns:

Equivalent of the input mapping with all values and sub-values modified through the given function.

Return type:

Mapping[R]

Notes

This method will iterate recursively on the mapping and it’s values from top to bottom.

athena.atUtils.map_sequence(function: Callable[[T], R], sequence: Sequence[T]) → Sequence[R][source]

Execute the given function on all values inside the given Sequence object.

Parameters:

function (Callable[[T], R]) – The function to call for each value of the given sequence.
sequence (Sequence[T]) – The Sequence object to recursively iterate on and execute the function for each of it’s values.

Returns:

Equivalent of the input sequence with all values modified through the given function.

Return type:

Sequence[R]

Notes

This method will iterate recursively on the sequence and it’s values from top to bottom.
As str is technically a sub-type of Sequence, the first step is to do an instance check and early return if the sequence is a string. The function will still be called with this string though and the result value returned like it would be for any other type.

athena.atUtils.deep_map(function: Callable[[T], R], collection: Collection[T]) → Collection[R][source]

Execute the given function on all values inside the given Collection object.

This will automatically call map_sequence() or map_mapping() based on the input collection or sub-collection inside it. It should be preferred to those method when you’re not aware of your input collection type or that this type is different from one iteration to the other.

Parameters:

function (Callable[[T], R]) – The function to call for each value and sub-value of the given collection.
collection (Collection[T]) – The Collection object to recursively iterate on and execute the function for each of it’s values.

Returns:

Equivalent of the input collection with all values and sub-values modified through the given function.

Return type:

Collection[R]

Notes

This method will iterate recursively on the collection and it’s values from top to bottom.

athena package

Subpackages

Submodules

athena.atConstants module

athena.atCore module

athena.atExceptions module

athena.atStatus module

athena.atUtils module

Module contents