[Python] My wrapper decorator which wraps an abstract method is not invoked

I'm creating a decorator to save log/results , but this is introduced to abstract class:

from functools import wraps

load_dotenv()

class Evaluator(ABC):
def __init__(self, evaluation_config, framework_config, **kwargs):

self.evaluation_config = evaluation_config
self.framework_config = framework_config

self.logger = WandbLogger()

@abstractmethod
def _create_metrics(self) -> List[Metric]:
pass

@abstractmethod
def load_evaluation_data(self, data_path):
pass

@staticmethod
def log_results(func):
"""Decorator to log evaluation results."""
print("this is called")
@wraps(func)
def wrapper(self, *args, **kwargs):
print("this is called too!")
result = func(*args, **kwargs)
self.logger.log(evaluation_result=result, config=self.framework_config)
return result

return wrapper

@log_results
@abstractmethod
def __call__(self, *args, **kwargs):
pass



One of the child-class:


class ScoreEvaluator(Evaluator):
def __init__(self, evaluation_config, framework_config, **kwargs):
super().__init__(evaluation_config, framework_config, **kwargs)

self.metric_cls = self._create_metrics()

def _create_metrics(self):
metrics = []
evaluator_classes = Metric.__subclasses__()

for cls in evaluator_classes:
metrics.append(cls())
return metrics

def load_evaluation_data(self, data_path):


return evaluation_data

def evaluate_criterion(self, predictions: List[float], golds: List[float]):

results = {}
for metric in self.metric_cls:
results[metric.__class__.__name__] = metric.compute(predictions, golds)

return results

async def __call__(self, *args, **kwargs):

evaluation_data_path = os.getenv("EVALUATION_DATA_PATH")
evaluation_data = self.load_evaluation_data(evaluation_data_path)
evaluation_data = {key: value for key, value in evaluation_data.items()}

prediction_dict = {
key: await self.score_generator(
question=[item["question"] for item in evaluation_data[key]],
essay=[item["essay"] for item in evaluation_data[key]],
image=(
[item["image"] for item in evaluation_data[key]]
if TaskName.TASK_1.value == key
else None
),
task_name=key,
do_sample=False,
temperature=0,
)
for key in evaluation_data.keys()
}
print(prediction_dict)

evaluation_results = {}
for key, predictions in prediction_dict.items():
predictions = pd.DataFrame(predictions)
golds = pd.DataFrame([item["human_rate"] for item in evaluation_data[key]])
evaluation_results[key] = {
criterion: self.evaluate_criterion(
predictions=predictions[criterion].tolist(), golds=golds[criterion].tolist()
)
for criterion in golds.columns
}

# self.logger.log(evaluation_result=evaluation_results, config=self.framework_config)

return evaluation_results


There is a Factory to create all the Evaluator


class EvaluatorFactory:
def __init__(self, framework_config, evaluation_config, **kwargs):
self.framework_config = framework_config
self.evaluation_config = evaluation_config
self.kwargs = kwargs

self.evaluators = self._create_evaluators()

def _create_evaluators(self):
evaluators = []
evaluator_classes = Evaluator.__subclasses__()

print(evaluators)
for cls in evaluator_classes:
evaluators.append(cls(framework_config=self.framework_config,
evaluation_config=self.evaluation_config,
**self.kwargs))
return evaluators

async def evaluate(self):
evaluation_results = {}
for evaluator in self.evaluators:
results = await evaluator()
evaluation_results[evaluator.__class__.__name__] = results

return evaluation_results

async def main():
evaluator_factory = EvaluatorFactory(framework_config=config.framework_config,
evaluation_config=config.evaluation_config)

result = await evaluator_factory.evaluate()
print(result)
return result


When I run, I noticed that, it just printed:

this is called


`` without printing the print("this is called too!")

I'm so confused. Thank you!


Edit 1

This is how I use it:

async def evaluate(self):
evaluation_results = {}
for evaluator in self.evaluators:
results = await evaluator()
evaluation_results[evaluator.__class__.__name__] = results

return evaluation_results


This is override function in the child-class:

async def __call__(self, *args, **kwargs):
prediction = self.predict(arg, kwargs)
...
return evaluation_results


I feel like this is quite not logic to implement, but the idea is to force every child-class run the wrapper decorator


Edit 2

I think this can be done with MetaClass but I still got error


def log_results(func):
"""Decorator to log evaluation results."""
print("this is called")

@wraps(func)
def wrapper(self, *args, **kwargs):
print("this is called too!")
print(self)
result = func(self, *args, **kwargs)
self.logger.log(evaluation_result=result, config=self.framework_config)
return result
return wrapper

# class EvaluatorMeta(ABCMeta):
# def __init__(cls, name, bases, dct):
# """Modify the class after it is created."""
# super().__init__(name, bases, dct)
# if "__call__" in dct:
# setattr(cls, "__call__", log_results(getattr(cls, "__call__"))) # Apply decorator

class EvaluatorMeta(ABCMeta):
def __init__(cls, name, bases, dct):
"""Only apply `log_results` when a child class of Evaluator is instantiated."""
super().__init__(name, bases, dct) # Call ABCMeta's __init__
print(cls.__base__)
if cls.__base__ is not ABC: # Only modify child classes, not `Evaluator` itself
setattr(cls, "__call__", log_results(getattr(cls, "__call__")))
# cls.__call__ = log_results(cls.__call__)

class Evaluator(ABC, metaclass=EvaluatorMeta)
...



The issue is the wrapper use a property self.logger, which is initialized in the abstract class, and all the sub-classes want to use this property.

No the log_result should be add inside as the method, because result = func -> this func is a call func which need a self.

OMG

Continue reading...