core.operation

Operating system for fuzzy numbers and arrays.

This module defines the abstract base classes for operations (OperationMixin) and a central registry (OperationScheduler) for managing and dispatching various fuzzy number operations within the AxisFuzzy framework.

Overview

Provides a unified interface for all fuzzy number operations (arithmetic, logic, comparison, etc).
Supports registration and dispatch of per-mtype operation implementations.
Handles preprocessing, validation, and performance monitoring for all operations.
Enables efficient batch operations on Fuzzarray via backend-aware implementations.

Key Classes

OperationMixin: Abstract interface for per-mtype operation implementations.
OperationScheduler: Singleton registry and performance monitor for operations.
Utilities: Registration decorator and access to the global scheduler.

Notes

All operation implementations must inherit from OperationMixin and register via @register_operation.
The scheduler supports dynamic switching of t-norms and collects global performance statistics.
Operator overloading in Fuzznum and Fuzzarray is dispatched via this system.

Examples

# Register a new operation for a custom mtype
@register_operation
class MyAdd(OperationMixin):
    def get_operation_name(self): return 'add'
    def get_supported_mtypes(self): return ['mytype']
    def _execute_binary_op_impl(self, s1, s2, tnorm): ...

class axisfuzzy.core.operation.OperationMixin[source]

Bases: ABC

Abstract base class for fuzzy number operations.

This class defines the common interface and provides utility methods for various operations (e.g., addition, subtraction, comparison) that can be performed on fuzzy numbers. Subclasses must implement the specific logic for each operation type.

Responsibilities

Define operation name and supported mtypes.
Provide preprocessing and validation for operands.
Implement core logic for binary, unary, comparison, and Fuzzarray-level operations.
Record performance statistics for each operation type.

Notes

Subclasses must override the _execute_*_impl methods for supported operations.
Public execute_* methods handle preprocessing, timing, and error handling.

Examples

from axisfuzzy.core import OperationMixin, register_operation, OperationTNorm

@register_operation
class QROFNAddition(OperationMixin):
    def get_operation_name(self) -> str: return 'add'
    def get_supported_mtypes(self) -> list[str]: return ['qrofn']
    def _execute_binary_op_impl(self, s1, s2, tnorm: OperationTNorm):
        md = tnorm.t_conorm(s1.md, s2.md)
        nmd = tnorm.t_norm(s1.nmd, s2.nmd)
        return {'md': md, 'nmd': nmd, 'q': s1.q}

execute_binary_op(strategy_1, strategy_2, tnorm)[source]

Executes a binary operation between two strategy instances.

Handles preprocessing, calls the concrete implementation, and records performance metrics.

Parameters:

strategy_1 (FuzznumStrategy) – The first operand strategy instance.
strategy_2 (FuzznumStrategy) – The second operand strategy instance.
tnorm (OperationTNorm) – T-norm configuration used for the operation.

Returns:

The result of the binary operation, typically a dictionary of component values for a new fuzzy number.

Return type:

Dict[str, Any]

Raises:

NotImplementedError – If _execute_binary_op_impl is not overridden by the subclass.

Examples

from axisfuzzy.fuzztype.qrofs.op import QROFNAddition
from axisfuzzy.core import fuzznum, OperationTNorm

op = QROFNAddition()
a = fuzznum(mtype='qrofn', md=0.5, nmd=0.3, q=2)._strategy_instance
b = fuzznum(mtype='qrofn', md=0.6, nmd=0.2, q=2)._strategy_instance
tnorm = OperationTNorm(norm_type='algebraic', q=2)
res = op.execute_binary_op(a, b, tnorm)
# res is {'md': 0.8, 'nmd': 0.06, 'q': 2}

execute_comparison_op(strategy_1, strategy_2, tnorm)[source]

Executes a comparison operation between two strategy instances.

Handles preprocessing, calls the concrete implementation, and records performance metrics.

Parameters:

strategy_1 (FuzznumStrategy) – The first operand strategy instance.
strategy_2 (FuzznumStrategy) – The second operand strategy instance.
tnorm (OperationTNorm) – T-norm configuration used for the comparison.

Returns:

The result of the comparison, e.g., {'value': True}.

Return type:

Dict[str, bool]

Raises:

NotImplementedError – If _execute_comparison_op_impl is not overridden by the subclass.

execute_fuzzarray_op(fuzzarray_1, other, tnorm)[source]

Executes an operation specifically designed for Fuzzarray instances.

Calls the concrete implementation and records performance metrics. Preprocessing for Fuzzarray operations is typically handled within the _execute_fuzzarray_op_impl itself due to their complex nature.

Parameters:

fuzzarray_1 (Fuzzarray) – The first Fuzzarray instance.
other (Fuzzarray, Fuzznum, scalar, or None) – The second operand.
tnorm (OperationTNorm) – T-norm configuration used for the operation.

Returns:

The result of the Fuzzarray operation (e.g., a new Fuzzarray or a boolean array).

Return type:

Any

Raises:

NotImplementedError – If _execute_fuzzarray_op_impl is not overridden by the subclass.

execute_unary_op_operand(strategy, operand, tnorm)[source]

Executes a unary operation involving a strategy instance and a scalar operand.

Handles preprocessing, calls the concrete implementation, and records performance metrics.

Parameters:

strategy (FuzznumStrategy) – The strategy instance.
operand (int or float) – The scalar operand (e.g., the exponent in a power operation).
tnorm (OperationTNorm) – T-norm configuration used for the operation.

Returns:

The result of the unary operation.

Return type:

Dict[str, Any]

Raises:

NotImplementedError – If _execute_unary_op_operand_impl is not overridden by the subclass.

execute_unary_op_pure(strategy, tnorm)[source]

Executes a pure unary operation on a strategy instance (no additional operand).

Handles preprocessing, calls the concrete implementation, and records performance metrics.

Parameters:

strategy (FuzznumStrategy) – The strategy instance.
tnorm (OperationTNorm) – T-norm configuration used for the operation.

Returns:

The result of the pure unary operation.

Return type:

Dict[str, Any]

Raises:

NotImplementedError – If _execute_unary_op_pure_impl is not overridden by the subclass.

abstractmethod get_operation_name()[source]

Returns the unique name of the operation (e.g., ‘add’, ‘mul’, ‘gt’).

This name is used to identify and retrieve the operation from the OperationScheduler.

Returns:: Operation name used as key in the operation registry.
Return type:: str

Examples

op = QROFNAddition()
print(op.get_operation_name())  # 'add'

abstractmethod get_supported_mtypes()[source]

Returns a list of fuzzy number membership types (mtypes) that this operation implementation supports.

For example, an operation might only be defined for ‘qrofn’ fuzzy numbers, or for both ‘qrofn’ and ‘qrohfn’.

Returns:: mtype strings for which this operation implementation is applicable.
Return type:: list of str

Examples

op = QROFNAddition()
print(op.get_supported_mtypes())  # ['qrofn']

supports(mtype)[source]

Check whether the operation supports a given mtype.

Parameters:: mtype (str) – Membership-type string to check.
Returns:: True if supported, False otherwise.
Return type:: bool

Examples

from axisfuzzy.fuzztype.qrofs.op import QROFNAddition
op = QROFNAddition()
print(op.supports('qrofn'))  # True
print(op.supports('qrohfn')) # False

class axisfuzzy.core.operation.OperationScheduler[source]

Bases: object

A singleton registry and dispatcher for all fuzzy number operations.

The OperationScheduler is the central nervous system for all mathematical and logical operations within the AxisFuzzy framework. It acts as a singleton, accessible via get_registry_operation(), ensuring a single source of truth for how operations are defined, configured, and executed.

Key Responsibilities:

Operation Registration: It maintains a registry mapping an operation’s name (e.g., ‘add’) and a specific fuzzy number type (mtype, e.g., ‘qrofn’) to a concrete OperationMixin implementation. This registration is typically handled automatically by the @register_operation decorator. This decoupled design allows new operations or support for new mtype to be added modularly without altering the core framework.
Operation Dispatch: When an operation is invoked (e.g., fuzznum1 + fuzznum2), the dispatch system queries the scheduler using get_operation(op_name, mtype) to find the correct implementation for the given operands. If no specific implementation is found, it signals that the operation is not supported.
Global T-Norm Configuration: The scheduler holds the global default T-norm configuration (e.g., ‘algebraic’, ‘einstein’). This ensures that all fuzzy operations, by default, use a consistent mathematical basis. This can be changed at runtime via set_t_norm(), allowing for framework-wide adjustments to the underlying logic for experimental purposes.
Performance Monitoring: It includes a built-in, thread-safe performance monitor. For every operation executed through the system, the scheduler records the execution time and updates statistics, such as total call counts and average execution times per operation type. This is invaluable for debugging, performance tuning, and understanding computational costs.

_operations

The core registry, structured as {op_name: {mtype: operation_instance}}.

Type:: dict

_default_t_norm_type

The name of the default T-norm (e.g., ‘algebraic’).

Type:: str

_default_t_norm_params

Parameters for the default T-norm (e.g., {'p': 2} for ‘hamacher’).

Type:: dict

_performance_stats

A dictionary holding all performance metrics.

Type:: dict

_stats_lock

A lock ensuring thread-safe updates to the performance statistics.

Type:: threading.Lock

Examples

While direct interaction is rare for end-users, understanding its internal role is key for developers.

# 1. Access the global scheduler instance
scheduler = get_registry_operation()

# 2. Change the global T-norm for all subsequent operations
scheduler.set_t_norm('hamacher', p=2)

# 3. Retrieve a specific operation implementation
# This is what FuzznumStrategy does internally.
add_op_for_qrofn = scheduler.get_operation('add', 'qrofn')

# 4. Check which operations are available for a given mtype
print(scheduler.get_available_ops('qrofn'))
# Output: ['add', 'sub', 'mul', 'div', 'gt', 'lt', ...]

# 5. After running some operations, get performance stats
# stats = scheduler.get_performance_stats(time_unit='us')
# print(stats['average_times_by_operation_type'])

get_available_ops(mtype)[source]

List all operation names available for a given mtype.

Parameters:: mtype (str) – The membership type to query.
Returns:: A list of available operation names.
Return type:: list of str

get_default_t_norm_config()[source]

Get the current default t-norm configuration.

Returns:: A tuple containing (t_norm_type, params).
Return type:: tuple[str, dict]

get_operation(op_name, mtype)[source]

Retrieve a registered operation implementation.

Parameters:

op_name (str) – The name of the operation (e.g., ‘add’).
mtype (str) – The membership type (e.g., ‘qrofn’).

Returns:

The registered operation instance, or None if not found.

Return type:

OperationMixin or None

get_performance_stats(time_unit='us')[source]

Obtains global performance statistics for all operations.

The statistics include total operations, total time, average time per operation, and counts/average times grouped by operation name and type. Time values are converted to the specified unit and rounded.

Parameters:: time_unit (str, optional) – Unit for displaying time. Supported: ‘s’, ‘ms’, ‘us’, ‘ns’. Defaults to ‘us’.
Returns:: A dictionary containing performance statistics.
Return type:: Dict[str, Any]
Raises:: ValueError – If an unsupported time_unit is provided.

register(operation)[source]

Registers an OperationMixin instance with the scheduler.

Operations are registered based on their op_name and supported mtype`s. A warning is issued if an operation for a specific `mtype is already registered.

Parameters:: operation (OperationMixin) – The operation instance to register.
Return type:: None

reset_performance_stats()[source]

Resets all collected performance statistics to their initial state.

This method is thread-safe.

Return type:: None

set_t_norm(t_norm_type, **params)[source]

Change the scheduler’s default t-norm configuration.

Available t-norm types include:

Archimedean T-Norms:

‘algebraic’: Algebraic product t-norm, T(a,b) = a × b
‘lukasiewicz’: Łukasiewicz t-norm, T(a,b) = max(0, a + b - 1)
‘einstein’: Einstein t-norm, T(a,b) = (a × b) / (1 + (1-a) × (1-b))
‘hamacher’: Hamacher t-norm family, requires parameter hamacher_param (gamma)
‘yager’: Yager t-norm family, requires parameter yager_param (p)
‘schweizer_sklar’: Schweizer-Sklar t-norm family, requires parameter sklar_param (p)
‘dombi’: Dombi t-norm family, requires parameter dombi_param (p)
‘aczel_alsina’: Aczel-Alsina t-norm family, requires parameter aa_param (p)
‘frank’: Frank t-norm family, requires parameter frank_param (s)

Non-Archimedean T-Norms:

‘minimum’: Minimum t-norm, T(a,b) = min(a,b), the most commonly used standard t-norm
‘drastic’: Drastic product t-norm, extreme operation under boundary conditions
‘nilpotent’: Nilpotent t-norm, T(a,b) = min(a,b) when a+b > 1, otherwise 0

Parameters:

t_norm_type (str) – Name of the t-norm implementation (e.g., ‘algebraic’, ‘einstein’).
**params (dict) – Implementation-specific parameters (e.g., p=2 for ‘hamacher’).

Examples

scheduler = get_registry_operation()
scheduler.set_t_norm('hamacher', p=2)

axisfuzzy.core.operation.get_registry_operation()[source]

Access the global OperationScheduler singleton.

Returns:: The module-level scheduler instance.
Return type:: OperationScheduler

axisfuzzy.core.operation.register_operation(cls_or_eager=None, *, eager=True)[source]

Class decorator that auto-registers OperationMixin subclasses with the global scheduler.

This decorator supports two usage patterns: 1. With parentheses: @register_operation() or @register_operation(eager=False) 2. Without parentheses: @register_operation

Parameters:

cls_or_eager (type or None) – When used without parentheses this is the decorated class; otherwise None.
eager (bool, optional) – If True (default), instantiate and register the operation immediately.

Returns:

The original class object.

Return type:

type

Raises:

TypeError – If the decorated object is not a subclass of OperationMixin.
RuntimeError – If eager instantiation and registration fails.

Examples

@register_operation
class MyAddition(OperationMixin):
    def get_operation_name(self): return 'add'
    def get_supported_mtypes(self): return ['my_type']
    # ... implement _execute_*_impl methods ...