Open
Description
This note refers to the use of Python Profilers to gain insights into the execution time of the parts making up an acoupi's task.
We use the cProfile library to better understand how long each of the components of an acoupi task were taking to be executed. It allows us to define what functions could be optimised for faster execution.
By profiling the detection_task within an acoupi program give us a better understanding of the time taken to run a model inference on a recording. We advise users to profile the detection task when looking to implement a new bioacoustics model inside the acoupi framework.
Create a cprofile task for the detection (add cprofile.py in acoupi/tasks/)
import cProfile
import logging
import pstats
from acoupi import data
from acoupi.components import types
from acoupi.tasks.detection import generate_detection_task
def generate_cprofile_detection_task(
store: types.Store,
model: types.Model,
message_store: types.MessageStore,
logger: logging.Logger = logger,
output_cleaners: Optional[List[types.ModelOutputCleaner]] = None,
processing_filters: Optional[List[types.ProcessingFilter]] = None,
message_factories: Optional[List[types.MessageBuilder]] = None,
profile_output: Optional[str] = "cprofile_output.prof",
) -> Callable[[data.Recording], None]:
"""Wrap the detection task in the cProfile."""
detection_task = generate_detection_task(
store=store,
model=model,
message_store=message_store,
logger=logger,
output_cleaners=output_cleaners,
processing_filters=processing_filters,
message_factories=message_factories,
)
def cprofiled_task(recording: data.Recording) -> None:
"""Run the task through the profiler."""
profiler = cProfile.Profile()
profiler.enable()
# Run the original detection task
detection_task(recording)
profiler.disable()
if cprofile_output:
profiler.dump_stats(cprofile_output)
logger.info(f"Profile data saved to {cprofile_output}")
else:
stats = pstats.Stats(profiler)
stats.strip_dirs().sort_stats("cumulative").print_stats()
return cprofiled_taskModify the acoupi program to include the cprofile_task.
detection_task = tasks.generate_cprofile_detection_task(...)