Components#

`acoupi.components` #

Acoupi components.

Component Types#

`acoupi.components.types` #

Module containing the types used by the acoupi.

Attributes#

`P = ParamSpec('P')` `module-attribute` #

Classes#

`AudioRecorder` #

Bases: ABC

Record audio from the microphone.

Functions#

`record(deployment)` `abstractmethod` #

Record audio from the microphone and return the recording.

Parameters:

Name	Type	Description	Default
`deployment`	`Deployment`	The deployment to use for recording the audio.	required

Returns:

Type	Description
`Recording`	The recording that was made. Object containing a temporary recording path, along with recording details such as datetime, duration, and samplerate.

Notes

The recording path should be saved in temporary memory until it gets processed. The path will be updated after the recording has been processed.

`MessageBuilder` #

Bases: ABC, Generic[P]

Create messages from input data.

See Also

See the module message_factories for concrete implementations of the MessageBuilder.

DetectionThresholdMessageBuilder: Filters detections by a score threshold.
FullModelOutputMessageBuilder: No filtering. Format the entire model output.

Functions#

`build_message(*args, **kwargs)` `abstractmethod` #

Will build a message or return None depending on the input data.

Returns:

Type	Description
`Optional[Message]`	The assembled message or None if the input data is not suitable.

`MessageStore` #

Bases: ABC

Keeps track of messages that have been produced and sent.

Functions#

`get_unsent_messages()` `abstractmethod` #

Get the recordings that have not been synced to the server.

Returns:

Type	Description
`List[Message]`	A list of unsent messages.

`store_message(message)` `abstractmethod` #

Register a message with the store.

Parameters:

Name	Type	Description	Default
`message`	`Message`	Store a message.	required

`store_response(response)` `abstractmethod` #

Register a message response with the store.

Parameters:

Name	Type	Description	Default
`response`	`Response`	Store the server response.	required

`Messenger` #

Bases: ABC

Send messages.

The Messenger is responsible for sending messages to a remote server according to communication protocol.

See Also

See the module messenger for concrete implementations of the Messenger.

MQTTMessenger: Send messages using the MQTT protocol.
HTTPMessenger: Send messages using the HTTP POST Request.

Functions#

`send_message(message)` `abstractmethod` #

Send the message to a remote server.

Parameters:

Name	Type	Description	Default
`message`	`Message`	The message to send.	required

Returns:

Type	Description
`Response`	A response containing the message, status, content, and received time.

`Model` #

Bases: Protocol

Model for making predictions.

The Model is responsible for running the model on the audio file and returning the predicted detections.

Detections should be returned as a list of Detection objects.

Functions#

`run(recording)` `abstractmethod` #

Run the model on the audio file and return the result.

Can optionally use deployment info to enhance predictions.

Parameters:

Name	Type	Description	Default
`recording`	`Recording`	The recording to process.	required

Returns:

Type	Description
`model_output`	The model output containing the detections.

`ModelOutputCleaner` #

Bases: ABC

Clean the model output.

The ModelOutputCleaner is responsible for cleaning the model output. This can include removing detections that are too short, too long, have a specific label or low confidence.

Notes

Model output cleaners are particularly useful when using a pre-trained model that produces irrelevant predictions. This component helps prune those predictions to make them more relevant to the task at hand.

See Also

See the module output_cleaners for a concrete implementation of the ModelOutputCleaner.

ThresholdDetectionCleaner: Keeps only the classifcations and dectections that are equal or higher than a threshold.

Functions#

`clean(model_output)` `abstractmethod` #

Clean the model output.

This method will remove any predicted tag or detection that does not meet the specified criteria.

Parameters:

Name	Type	Description	Default
`model_output`	`ModelOutput`	The model output to clean.	required

Returns:

Type	Description
`ModelOutput`	The cleaned model output.

`ProcessingFilter` #

Bases: ABC

Determine if a recording should be processed by a model.

The ProcessingFilter is responsible for determining if a recording should be processed by the model.

Functions#

`should_process_recording(recording)` `abstractmethod` #

Determine if the recording should be processed by the model.

Parameters:

Name	Type	Description	Default
`recording`	`Recording`	The recording to check.	required

Returns:

Type	Description
`should_process`	True if the recording should be processed, False otherwise.

`RecordingCondition` #

Bases: ABC

Decide if a recording should be made.

Will only do a recording if the RecordingCondition is met.

See Also

See the module recording_conditions for concrete implementations of the RecordingCondition.

IsInInterval: Record if the current time is within a specified interval.

Functions#

`should_record()` `abstractmethod` #

Determine if a recording should be made.

Returns:

Type	Description
`bool`	True if a recording should be made, False otherwise.

`RecordingSavingFilter` #

Bases: ABC

The Recording Saving Filter is responsible for determining if a recording should be saved.

Notes

The RecordingSavingFilter is responsible for determining if a recording should be saved. The RecordingSavingFilter is used by the management task. If the boolean value returned is True, the recording will be saved. If False, the recording will be deleted.

See Also

See saving_filters for concrete implementations of the RecordingSavingFilter.

After_DawnDuskTimeInterval / Before_DawnDuskTimeInterval: Save recordings if they falls withing a specified time interval happening after or before astronomical dawn and dusk.
SavingThreshold: Save recordings if any of the detection and classification tag score associated to the recording model output is higher or equal than a specified threshold.
SaveIfInInterval: Save recordings if the recording falls within a specified interval.
FrequencySchedule: Save recordings if the recording falls within the specified frequency schedule.

Functions#

`should_save_recording(recording, model_outputs=None)` `abstractmethod` #

Determine if a recording should be saved.

Parameters:

Name	Type	Description	Default
`recording`	`Recording`	The recording to check.	required
`model_outputs`	`Optional[List[ModelOutput]]`	The model outputs associated to the recording. Used in some implementations when the decision to save a recording depends on the model outputs, rather the recording itself.	`None`

Returns:

Type	Description
`bool`	True if the recording should be saved, False otherwise.

`RecordingSavingManager` #

Bases: ABC

The Recording SavingManager is responsible for saving recordings.

Notes

The RecordingSavingManager is responsible for saving recordings. The RecordingSavingManager is used by the management task. The RecordingSavingManager is used to save recordings to the correct path.

See Also

See the module saving_managers for concrete implementations of the RecordingSavingManager.

SaveRecordingManager: Save recordings to a specified directory according to the model outputs.
DateFileManager: Save recordings to directories based on the date of the recording.

Functions#

`save_recording(recording, model_outputs=None)` `abstractmethod` #

Save the recording.

Parameters:

Name	Type	Description	Default
`recording`	`Recording`	The recording to save.	required
`model_outputs`	`Optional[List[ModelOutput]]`	The model outputs associated to the recording. Used to determined where and how to save the recording.	`None`

Returns:

Type	Description
`Optional[Path]`	The path to the saved recording. Returns None if this manager is not responsible for saving the recording.

`RecordingScheduler` #

Bases: ABC

Manage time between recordings.

The RecordingScheduler is responsible for determining the when recording should be made.

See Also

See the module recording_schedulers for a concrete implementation of the RecordingScheduler.

IntervalScheduler: Record at a fixed interval.

Functions#

`time_until_next_recording(time=None)` `abstractmethod` #

Provide the number of seconds until the next recording.

Parameters:

Name	Type	Description	Default
`time`	`Optional[datetime]`	The time to use for determining the next recording, by default None.	`None`

Returns:

Type	Description
`float`	The number of seconds until the next recording. Will return 0 if a recording should be made immediately.

`Store` #

Bases: ABC

The Store is responsible for storing the detections locally.

The store keeps track of all the recordings, detections, and deployments made.

Functions#

`get_current_deployment()` `abstractmethod` #

Get the current deployment from the local filesystem.

`get_recordings(ids)` `abstractmethod` #

Get a list recordings from the store by their ids.

Each recording is returned with the full list of model outputs registered.

Parameters:

Name	Type	Description	Default
`ids`	`List[UUID]`	The ids of the recordings to get.	required

Returns:

Type	Description
`A list of tuples of the recording and the model outputs.`

`get_recordings_by_path(paths)` `abstractmethod` #

Get a list recordings from the store by their paths.

Each recording is returned with the full list of model outputs registered.

Parameters:

Name	Type	Description	Default
`paths`	`List[Path]`	The paths of the recordings to get.	required

Returns:

Type	Description
`A list of tuples of the recording and the model outputs.`

`store_deployment(deployment)` `abstractmethod` #

Store the deployment locally.

`store_model_output(model_output)` `abstractmethod` #

Store the model output locally.

`store_recording(recording, deployment=None)` `abstractmethod` #

Store the recording locally.

Parameters:

Name	Type	Description	Default
`recording`	`Recording`	The recording to store.	required
`deployment`	`Optional[Deployment]`	The deployment associated with the recording, by default None.	`None`

`update_deployment(deployment)` `abstractmethod` #

Update the deployment.

`update_recording_path(recording, path)` `abstractmethod` #

Update the path of the recording.

`Summariser` #

Bases: ABC

Summarise model outputs.

The Summariser is responsible for summarising model outputs (i.e., detections) into a message.

See Also

See the module summarisers for concrete implementations of the Summariser.

StatisticsDetectionsSummariser: Summarises detections by calculating the mean, min, max, and count of classification probabilities for each species.
ThresholdsDetectionsSummariser: Count the number of detections for each species that falls into three thresholds different bands: low, medium, and high.

Functions#

`build_summary(now)` `abstractmethod` #

Build a summary.

Parameters:

Name	Type	Description	Default
`now`	`datetime`	The time of the summary.	required

Returns:

Type	Description
`Message`	The summary as a data.Message object. The message should be in JSON format.

Concrete Components#

`acoupi.components.audio_recorder` #

Implementation of AudioRecorder for acoupi.

An AudioRecorder is used to record audio files. The PyAudioRecorder is implemented as class that inherit from AudioRecorder. The class should implement the record() method which return a temporary audio file based on the dataclass Recording.

The audio recorder takes argument related to the audio device. It specifies the acoutics parameters of recording an audio file. These are the samplerate, the duration, the number of audio_channels, the chunk size, and the index of the audio device. The index of the audio device corresponds to the index of the USB port the device is connected to. The audio recorder return a temporary .wav file.

Attributes#

`TMP_PATH = Path('/run/shm/')` `module-attribute` #

Classes#

`MicrophoneConfig` #

Bases: BaseModel

Attributes#

`audio_channels: int = 1` `class-attribute` `instance-attribute` #

`device_name: str` `instance-attribute` #

`samplerate: int = 48000` `class-attribute` `instance-attribute` #

Functions#

`setup(args, prompt=True, prefix='')` `classmethod` #

Set up the microphone configuration.

`PyAudioRecorder(duration, samplerate, audio_channels, device_name, chunksize=2048, audio_dir=TMP_PATH, logger=None)` #

Bases: AudioRecorder

Component that records fixed duration audio to a file.

Attributes#

`audio_channels: int = audio_channels` `instance-attribute` #

The number of audio channels.

`audio_dir: Path = audio_dir` `instance-attribute` #

The directory where to store the created recordings.

`chunksize: int = chunksize` `instance-attribute` #

`device_name: str = device_name` `instance-attribute` #

The name of the input audio device.

`duration: float = duration` `instance-attribute` #

The duration of the audio file in seconds.

`logger = logger` `instance-attribute` #

`sample_width = pyaudio.get_sample_size(pyaudio.paInt16)` `instance-attribute` #

`samplerate: int = samplerate` `instance-attribute` #

The samplerate of the audio file in Hz.

Functions#

`check()` #

Check if the audio recorder is compatible with the config.

`get_input_device(p)` #

Get the input device.

`get_recording_data(duration=None, num_chunks=None)` #

`record(deployment)` #

Record an audio file.

Returns:

Type	Description
`data.Recording: A Recording object containing the temporary path of the file.`

`save_recording(data, path)` #

Save the recording to a file.

Functions#

`parse_microphone_config(args, prompt=True, prefix='')` #

`acoupi.components.message_factories` #

Message factories for acoupi.

Message factories are responsible for building messages from model outputs. Messages are intended to be sent to remote servers using communication protocols (e.g., MQTT, HTTP) for further processing, storage, or analysis.

The message factories are useful to filter outputs from the model according to various criteria, avoiding sending unnecessary information to a server, when connectivity is limited. For example, message factories can be used to filter detections with low score.

Message factories are implemented as classes that inherit from MessageBuilder. The class should implement the build_message method, which takes a model output and returns a message. The message should be a JSON string containing the information to be sent to the remote server.

Classes#

`DetectionThresholdMessageBuilder(detection_threshold)` #

Bases: MessageBuilder

A MessageBuilder that builds message from model outputs.

This message builder builds a message from a model output. The created message will contain the full model output as a JSON string. This includes information about the model used, the recording including deployment info, and the detections and predicted tags that meet the threshold.

Attributes#

`detection_threshold = detection_threshold` `instance-attribute` #

Functions#

`build_message(model_output)` #

Build a message with only detections meeting threshold.

Parameters:

Name	Type	Description	Default
`self`		The minimum detection score required for a detection to be included in the message.	required
`self`		A list of detections from the model_output.detections to be filtered.	required
`model_output`	`ModelOutput`	The model output to build the message from.	required

Returns:

Type	Description
`A message containing the model output or None if no valid detections.`

Examples:

>>> model_output = data.ModelOutput(
...     data.Detection(
...         detection_score=0.5,
...         tags=[
...             data.PredictedTag(
...                 tag=data.Tag(
...                     key="species", value="species_1"
...                 ),
...                 confidence_score=0.4,
...             )
...         ],
...     )
... )
>>> message_builder = DetectionThresholdMessageBuilder(
...     detection_threshold=0.6
... )
>>> message_builder.build_message(model_output)
None

>>> model_output = data.ModelOutput(
...     data.Detection(
...         detection_score=0.9,
...         tags=[
...             data.PredictedTag(
...                 tag=data.Tag(
...                     key="species", value="species_1"
...                 ),
...                 confidence_score=0.9,
...             )
...         ],
...     )
... )
>>> message_builder = DetectionThresholdMessageBuilder(
...     detection_threshold=0.6
... )
>>> message_builder.build_message(model_output)
Message(content='{"name_model": "TestModel", "recording": {"path": "recording.wav", "deployment": {}, "tags": [], "detections": [{"detection_score": 0.9, "location": {}, "tags": [{"tag": {"key": "species", "value": "species_1"}, "confidence_score": 0.9}]}]}')

`filter_detections(detections)` #

Remove detections with low score.

`FullModelOutputMessageBuilder` #

Bases: MessageBuilder

A MessageBuilder that builds message from model outputs.

This message builder builds a message from a model output. The created message will contain the full model output as a JSON string.

Functions#

`build_message(model_output)` #

Build a message from a recording and model outputs.

Parameters:

Name	Type	Description	Default
`model_output`	`ModelOutput`	The model output to build the message from.	required

Returns:

Type	Description
`A message containing the full model output.`

`acoupi.components.messengers` #

Messengers for acoupi.

Messengers are responsible for sending messages to external services. The messengers are templates illustrating how to send messages using different communication protocols (e.g., MQTT, HTTP).

The messengers are implemented as classes that inherit from Messenger. The class should implement the send_message method, which takes a message and sends it to the external service. The class should also implement the check method, which checks the connection status of the messenger.

The MQTTMessenger sends messages using the MQTT protocol. The HTTPMessenger sends messages using HTTP POST requests.

Classes#

`HTTPConfig` #

Bases: BaseModel

Attributes#

`base_url: str` `instance-attribute` #

`content_type: str = 'application/json'` `class-attribute` `instance-attribute` #

`timeout: int = 5` `class-attribute` `instance-attribute` #

`HTTPMessenger(base_url, base_params=None, headers=None, timeout=5, content_type='application/json', logger=None)` #

Bases: Messenger

Messenger that sends messages via HTTP POST requests.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	The URL to send messages to. This should include the protocol (e.g. http:// or https://) and the hostname (e.g. localhost) and the path (e.g. /api/endpoint).	required
`base_params`	`Optional[dict]`	Base parameters to send with each request, by default None.	`None`
`headers`	`Optional[dict]`	Headers to send with each request, by default None.	`None`
`timeout`	`int`	Seconds to wait for a response before timing out, by default 5.	`5`
`content_type`	`str`	The content type to send with each request, by default "application/json".	`'application/json'`

Attributes#

`base_params: dict = base_params or {}` `instance-attribute` #

Base parameters to send with each request.

`base_url: str = base_url` `instance-attribute` #

The base URL to send messages to.

`content_type = self.headers['Content-Type']` `instance-attribute` #

`headers: dict = headers or {}` `instance-attribute` #

Headers to send with each request.

`logger = logger` `instance-attribute` #

`timeout: int = timeout` `instance-attribute` #

Timeout for sending messages in seconds.

Functions#

`check()` #

Check the connection status of the HTTP client.

Raises:

Type	Description
`HealthCheckError`	If the connection is not successful. This could be due to a connection error or if the POST method is not allowed.

`from_config(config, logger=None)` `classmethod` #

Create an HTTPMessenger from a configuration object.

`send_message(message)` #

Send a recording message through a HTTP POST request.

`MQTTConfig` #

Bases: BaseModel

Attributes#

`host: str` `instance-attribute` #

`password: Optional[SecretStr] = None` `class-attribute` `instance-attribute` #

`port: int = 1884` `class-attribute` `instance-attribute` #

`timeout: int = 5` `class-attribute` `instance-attribute` #

`topic: str = 'acoupi'` `class-attribute` `instance-attribute` #

`username: str` `instance-attribute` #

Functions#

`dump_password(value)` #

`MQTTMessenger(host, topic, port=1884, username=None, password=None, timeout=5, logger=None)` #

Bases: Messenger

Messenger that sends messages via MQTT.

Parameters:

Name	Type	Description	Default
`host`	`str`	The host to connect to. Example: "mqtt.localhost.org".	required
`username`	`str`	The username to authenticate with.	`None`
`topic`	`str`	The topic to send messages to. Example: "org/survey/device_00/".	required
`port`	`int`	The port to connect to, by default 1884.	`1884`
`password`	`Optional[SecretStr]`	The password to authenticate with, by default None.	`None`

Notes

Will use the device ID as the client ID.

Attributes#

`client: mqtt.Client = mqtt.Client(callback_api_version=CallbackAPIVersion.VERSION2, client_id=self.client_id, clean_session=False)` `instance-attribute` #

The MQTT client.

`client_id = get_device_id()` `instance-attribute` #

`host = host` `instance-attribute` #

`logger: logging.Logger = logger` `instance-attribute` #

`port = port` `instance-attribute` #

`timeout: int = timeout` `instance-attribute` #

Timeout for sending messages.

`topic: str = topic` `instance-attribute` #

The MQTT topic to send messages to.

Functions#

`check()` #

Check the connection status of the MQTT client.

Raises:

Type	Description
`HealthCheckError`	If the connection is not successful. This could be due to a connection error or an authentication failure.

`check_connection()` #

Check the connection status of the MQTT client.

`from_config(config, logger=None)` `classmethod` #

Create an MQTTMessenger from a configuration object.

`send_message(message)` #

Send a recording message.

Parameters:

Name	Type	Description	Default
`message`	`Message`	The message to send.	required

Returns:

Type	Description
`Response`	A response containing the message, status, content, and received time.

Examples:

>>> message = data.Message(
...     content="hello world",
... )
>>> messenger = MQTTMessenger(
...     host="mqtt.localhost.org",
...     username="mqttusername",
...     topic="org/survey/device_00",
...     clientid="org/survey/device_00",
... )
>>> messenger.send_message(message)
>>> data.Response(
...     message=data.Message(content="{}"),
...     status=ResponseStatus.SUCCESS,
...     content="MQTT_ERR_SUCCESS",
...     received_on=datetime.datetime(),
... )

Functions#

`acoupi.components.models` #

`acoupi.components.stores` #

Storages for acoupi.

Storages are used to store recordings and detections locally. The stores keep track of the recordings and detections that have been made. The stored data can be retrieved later to be sent to a remote server.

Storages are implemented as classes that inherit from Storage. The class should implement the methods for storing and retrieving data, as well as the methods for retrieving the current deployment and the recordings and detections for a given deployment. See the Storage class for more details.

Classes#

`SqliteStore(db_path)` #

Bases: Store

Sqlite store implementation.

The store is used to store the recordings, detections and deployments locally. The data is stored in a sqlite database file in the given path.

Under the hood, the store uses the Pony ORM to interact with the database. The database schema is defined in the database module and contains the following tables:

Deployment: Contains the deployment information. Each deployment is associated with a device, and has a start datetime. The deployment can also have a latitude and longitude associated with it.
Recording: Contains the recording information. Each recording is associated with a deployment, and has a datetime, duration, samplerate and number of audio_channels.
PredictedTag: Contains the predicted tag information. Each predicted tag has a key, value and score.
Detection: Contains the detection information. Each detection consists of a location, score and a list of predicted tags.
ModelOutput: Contains the model output information. Each model output has the model name, the list of predicted tags at the recording level, and a list of detections.

The store is thread-safe, and can be used from multiple threads simultaneously.

Notes

The ID of the deployment, recording, model output and detection is a UUID field. Note that sqlite stores UUIDs as a BLOB, so when querying the database with SQL, you should use the hex function to convert the UUID to a string.

Example queries:

.. code-block:: sql

-- Find a specific deployment by UUID
SELECT * FROM Deployment WHERE hex(id) = '00000000000000000000000000000000';

-- Get all deployment UUIDs as strings
SELECT hex(id) FROM Deployment;

Will create a database file at the given path if it does not exist.

Args: db_path: Path to the database file. Can be set to :memory: to use an in-memory database.

Attributes#

`database: orm.Database = orm.Database()` `instance-attribute` #

The Pony ORM database object.

`db_path: Path = db_path` `instance-attribute` #

Path to the database file.

`models: db_types.BaseModels = create_base_models(self.database)` `instance-attribute` #

The Pony ORM models.

Functions#

`get_current_deployment()` #

Get the current deployment.

The current deployment is the one with the latest started_on datetime.

If no deployment is found, a new deployment will be registered with the current datetime, and the latitude and longitude set to None.

Returns:

Type	Description
`The current deployment`

`get_detections(ids=None, model_output_ids=None, score_gt=None, score_lt=None, model_names=None, after=None, before=None)` #

Get a list of detections from the store based on their model_output ids.

`get_model_outputs(after=None, before=None, ids=None, recording_ids=None, model_names=None, detection_ids=None, limit=None)` #

Get a list of model outputs from the store by their id.

Args: start_time: The time to start the search from. end_time: The time to end the search at.

Returns:

Type	Description
`List of model_outputs matching the created_on datetime.`

`get_predicted_tags(detection_ids=None, after=None, before=None, score_gt=None, score_lt=None, keys=None, values=None)` #

Get a list of predicted tags from the store based on their detection ids.

`get_recordings(ids)` #

Get a list recordings from the store by their ids.

Each recording is returned with the full list of model outputs registered.

Args: ids: The ids of the recordings to get.

Returns:

Type	Description
`A list of tuples of the recording and the model outputs.`

`get_recordings_by_path(paths)` #

Get a list of recordings from the store by their paths.

Args: paths: The paths of the recordings to get.

Returns:

Type	Description
`List of tuples of the recording and the corresponding model outputs.`

`store_deployment(deployment)` #

Store the deployment locally.

Args: deployment: The deployment to store

`store_model_output(model_output)` #

Store the model output locally.

`store_recording(recording)` #

Store the recording locally.

If the deployment is not provided, the current deployment will be used.

Args: recording: The recording to store

`update_deployment(deployment)` #

`update_recording_path(recording, path)` #

Update the path of a recording.

Args: recording: The recording to update. path: The new path.

`acoupi.components.message_stores` #

`acoupi.components.output_cleaners` #

ModelOutput cleaners for acoupi.

Model Output Cleaners are responsible for cleaning the outputs of a model (i.e., detections) that do not meet certain criteria. This can include removing low confidence tags and detections, or detections and tags that have a specific labels. For example, the ThresholdDetectionCleaner removes any predictions (i.e., detections and tags) with a score below a threshold.

The ModelOutputCleaner is implemented as a class that inherits from ModelOutputCleaner. The class should implement the clean method, which takes a data.ModelOutput object and returns a cleaned data.ModelOutput object. The modeloutput that does not meet the criteria are removed.

The ModelOutputCleaner is used in the detection task to clean the outputs of the model BEFORE storing them in the store. The ModelOutputCleaner is passed to the detection task as a list of ModelOutputCleaner objects. This allows to use multiple ModelOutputCleaners to clean the model output.

Classes#

`ThresholdDetectionCleaner(detection_threshold)` #

Bases: ModelOutputCleaner

Keeps predictions with a score higher than a threshold.

This class implements a model output cleaner that removes any predictions with a score below a threshold. This includes removing low confidence tags and detections.

Attributes#

`detection_threshold: float = detection_threshold` `instance-attribute` #

The threshold to use to define when a detection is stored.

Functions#

`clean(model_output)` #

Clean the model output.

Parameters:

Name	Type	Description	Default
`model_output`	`ModelOutput`	The model output to clean.	required

Returns:

Type	Description
`ModelOutput`	The cleaned model output.

Examples:

>>> model_output = data.ModelOutput(
...     detections=[
...         data.Detection(
...             detection_score=0.8,
...             tags=[
...                 data.PredictedTag(
...                     tag=data.Tag(
...                         key="species", value="species_1"
...                     ),
...                     confidence_score=0.7,
...                 )
...             ],
...             tags=[
...                 data.PredictedTag(
...                     tag=data.Tag(
...                         key="species", value="species_2"
...                     ),
...                     confidence_score=0.4,
...                 )
...             ],
...         )
...     ]
... )
>>> cleaner = ThresholdDetectionCleaner(detection_threshold=0.6)
>>> model_output = cleaner.clean(model_output)
>>> assert model_output == data.ModelOutput(
...     detections=[
...         data.Detection(
...             detection_score=0.8,
...             tags=[
...                 data.PredictedTag(
...                     tag=data.Tag(
...                         key="species", value="species_1"
...                     ),
...                     confidence_score=0.7,
...                 )
...             ],
...         )
...     ]
... )

`clean_detection(detection)` #

Remove tags with low score from detection.

`get_clean_detections(detections)` #

Remove detections with low score.

`get_clean_tags(tags)` #

Remove tags with low score.

`acoupi.components.processing_filters` #

Processing filters for the Acoupi project.

Processing filters are used to determine if a recording should be processed by the model. This is useful for example if you want to only process recordings that satisfy certain criteria, such as a minimum duration or that surpass a certain amplitude threshold. This can be used to reduce the amount of computational resources required to process a large number of recordings.

Processing filters are implemented as classes that inherit from ProcessingFilter. The class should implement the should_process_recording method, which takes a Recording object and returns a boolean indicating if the recording should be processed by the model.

Keep in mind that the should_process_recording method is called for every recording, so it should be as efficient as possible.

Classes#

`TrivialProcessingFilter` #

Bases: ProcessingFilter

A ProcessingFilter that always returns True.

Functions#

`should_process_recording(recording)` #

Determine if the recording should be processed by the model.

`acoupi.components.recording_conditions` #

Recording conditions for acoupi.

Recording conditions are used to determine if a recording should be made according to a specific condition. This is useful for example if you want to only record during specific times of day, such as between 8am and 5pm, or if you want to record during specific days of the week, such as only on weekdays, or if you want to record based on the value of a sensor (e.g., readings from temperature or luminosity sensors).

Recording conditions are implemented as classes that inherit from RecordingCondition. The class should implement the should_record method, which returns a boolean indicating if a recording should be made.

Classes#

`DawnTimeInterval(duration, timezone)` #

Bases: RecordingCondition

A RecordingCondition that records only during the dawn time interval.

Parameters:

Name	Type	Description	Default
`duration`	`float`	The duration of time (in minutes) before and after dawntime.	required
`timezone`	`tzinfo`	The timezone that the dawn time is in.	required

Attributes#

`duration: float = duration` `instance-attribute` #

The duration of time (in minutes) before and after dawntime.

`timezone: datetime.tzinfo = timezone` `instance-attribute` #

The timezone that the dawn time is in.

Functions#

`should_record()` #

Determine if a recording should be made.

Returns:

Type	Description
`bool`	True if the current time is within the dawn time interval. False otherwise.

Examples:

>>> dawn_time = time(6, 0)
>>> duration = 30
>>> timezone = "Europe/London"
>>> time = datetime(2024, 1, 1, 6, 15, 0, tzinfo=timezone)
>>> DawnTimeInterval(
...     dawn_time, duration, timezone
... ).should_record(time)
True

>>> dawn_time = time(6, 0)
>>> duration = 30
>>> timezone = "Europe/London"
>>> time = datetime(2024, 1, 1, 5, 45, 0, tzinfo=timezone)
>>> DawnTimeInterval(
...     dawn_time, duration, timezone
... ).should_record(time)
False

`IsInInterval(interval, timezone)` #

Bases: RecordingCondition

A RecordingCondition that records only during a specific interval of time.

This class checks whether the current time falls withing a specific interval. If the current time is within the interval, recording is allowed.

Parameters:

Name	Type	Description	Default
`interval`	`TimeInterval`	An object containing a start and end time (datetime.time). The interval of time where audio recordings are allowed.	required
`timezone`	`tzinfo`	The timezone that the interval is in. This ensures that the interval is calculated correctly across different timezones.	required

Attributes#

`interval = interval` `instance-attribute` #

`timezone = timezone` `instance-attribute` #

Functions#

`should_record()` #

Determine if a recording should be made.

Returns:

Type	Description
`bool`	True if the current time falls within the interval. False otherwise.

Notes

Uses the current time as provided by the system clock.

Examples:

>>> interval = TimeInterval(start=time(8, 0), end=time(17, 0))
>>> timezone = "Europe/London"
>>> time = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone)
>>> IsInInterval(interval, timezone).should_record(time)
True

>>> interval = TimeInterval(start=time(8, 0), end=time(17, 0))
>>> timezone = "Europe/London"
>>> time = datetime(2024, 1, 1, 18, 0, 0, tzinfo=timezone)
>>> IsInInterval(interval, timezone).should_record(time)
False

`IsInIntervals(intervals, timezone)` #

Bases: RecordingCondition

A RecordManager that records during multiple intervals of time.

Parameters:

Name	Type	Description	Default
`intervals`	`List[TimeInterval]`	of Interval objects.	required
`timezone`	`tzinfo`	should be made.	required

Attributes#

`intervals = intervals` `instance-attribute` #

`timezone = timezone` `instance-attribute` #

Functions#

`should_record()` #

Determine if a recording should be made.

`acoupi.components.recording_schedulers` #

Recording Schedulers for acoupi.

Recording schedulers are used to determine how often recordings should be made. This is useful for example if you want to record at a constant interval, or if you want to record at a variable interval, such as every 10 minutes during the day and every 30 minutes at night.

Recording schedulers are implemented as classes that inherit from RecordingScheduler. The class should implement the time_until_next_recording method, which returns the time in seconds until the next recording should be made.

Classes#

`IntervalScheduler(timeinterval)` #

Bases: RecordingScheduler

Will wait for a constant amount of time between each recording.

Parameters:

Name	Type	Description	Default
`timeinterval`	`float`	The interval between each recording. In seconds.	required

Attributes#

`interval: float` `instance-attribute` #

The interval between each recording. In seconds.

`timeinterval = timeinterval` `instance-attribute` #

Functions#

`time_until_next_recording(time=None)` #

Provide the number of seconds until the next recording.

Parameters:

Name	Type	Description	Default
`time`	`Optional[datetime]`	The time to use for determining the next recording, by default None.	`None`

Returns:

Type	Description
`float`	The number of seconds until the next recording. Will return 0 if a recording should be made immediately.

`acoupi.components.saving_filters` #

Recording Saving Filters for acoupi.

RecordingSavingFilters are used to determine if a recording should be saved based on specific criteria. These filters can be used to save recordings based on time intervals, detection probabilities, classification probabilities, classification tag values, and more.

The Recording SavingFilters are implemented as classes that inherit from the RecordingSavingFilter Implementation of the RecordingSavingFilters should implement the should_save_recording method, which takes a recording object and a list of model outputs, and returns a boolean value.

The RecordingSavingFilters are used in the acoupi.tasks.management module to determine if a recording should be saved based on the output of the models and the filters provided. If a recording pass filters, it will be kept and stored in a directory specified by the RecordingSavingManager. If a recording does not pass the filters, it is deleted.

The RecordingSavingFilters are optional and can be ignored if no recordings should be saved.

Classes#

`After_DawnDuskTimeInterval(duration, timezone)` #

Bases: RecordingSavingFilter

An after dawn and dusk time RecordingSavingFilter.

Attributes#

`duration: float = duration` `instance-attribute` #

The duration (in minutes) before dawn and dusk where recordings will be saved.

`timezone: datetime.tzinfo = timezone` `instance-attribute` #

The timezone to use when determining dawntime and dusktime.

Functions#

`should_save_recording(recording, model_outputs=None)` #

Save a recording if it falls within the specified interval, after dawn and dusk.

Notes

The dawn and dusk times are calculated using the astral library. The sun function returns the dawn and dusk times for a specific location, datetime and timezone. This information is used to determine the interval after dawn and dusk dusk, and whether the current recording falls within this interval.

Examples:

>>> DawnTime GMT: 2024-01-01 07:26:00+00:00
>>> DuskTime GMT: 2024-01-01 16:42:00+00:00
>>> duration = 30
>>> timezone = "Europe/London"

>>> saving_filter = After_DawnDuskTimeInterval(
...     duration, timezone
... )
>>> recording = data.Recording(
...     datetime=datetime.datetime(
...         2024, 1, 1, 7, 0, 0, tzinfo=timezone
...     )
... )
>>> assert saving_filter.should_save_recording(recording)
False

>>> saving_filter = After_DawnDuskTimeInterval(
...     duration, timezone
... )
>>> recording = data.Recording(
...     datetime=datetime.datetime(
...         2024, 1, 1, 17, 0, 0, tzinfo=timezone
...     )
... )
>>> assert saving_filter.should_save_recording(recording)
True

`Before_DawnDuskTimeInterval(duration, timezone)` #

Bases: RecordingSavingFilter

A before dawn and dusk time RecordingSavingFilter.

Attributes#

`duration = duration` `instance-attribute` #

`timezone = timezone` `instance-attribute` #

Functions#

`should_save_recording(recording, model_outputs=None)` #

Save a recording if it falls within the specified interval, before dawn and dusk.

Notes

The dawn and dusk times are calculated using the astral library. The sun function returns the dawn and dusk times for a specific location, datetime and timezone. This information is used to determine the interval before dawn and dusk dusk, and whether the current recording falls within this interval.

Examples:

>>> DawnTime GMT: 2024-01-01 07:26:00+00:00
>>> DuskTime GMT: 2024-01-01 16:42:00+00:00
>>> duration = 30
>>> timezone = "Europe/London"

>>> saving_filter = Before_DawnDuskTimeInterval(
...     duration, timezone
... )
>>> recording = data.Recording(
...     datetime=datetime.datetime(
...         2024, 1, 1, 7, 0, 0, tzinfo=timezone
...     )
... )
>>> assert saving_filter.should_save_recording(recording)
True

>>> saving_filter = Before_DawnDuskTimeInterval(
...     duration, timezone
... )
>>> recording = data.Recording(
...     datetime=datetime.datetime(
...         2024, 1, 1, 17, 0, 0, tzinfo=timezone
...     )
... )
>>> assert saving_filter.should_save_recording(recording)
False

`DetectionTagValue(values)` #

Bases: RecordingSavingFilter

A RecordingSavingFilter that keeps recordings with specific tag values.

Attributes#

`values: List[str] = values` `instance-attribute` #

The tag values to focus on.

Functions#

`has_confident_tagvalues(model_output)` #

Determine if a model output has a confident tag values.

An output is considered confident if any of its tag value (e.g., species_name) is in the values list.

Parameters:

Name	Type	Description	Default
`model_output`	`ModelOutput`	The model output of the recording containing detections and tags.	required

Returns:

Type	Description
`bool`

`should_save_recording(recording, model_outputs=None)` #

Save a recording if it contains any confident tag values.

`DetectionTags(tags, saving_threshold=0.5)` #

Bases: RecordingSavingFilter

A RecordingSavingFilter that keeps recordings with selected tags.

This filter will keep recordings that contain confident tag predictions that are in the tags list.

Attributes#

`saving_threshold: float = saving_threshold` `instance-attribute` #

The score threshold to use.

`tags: List[data.Tag] = tags` `instance-attribute` #

The tags to focus on.

Functions#

`has_confident_tag(model_output)` #

Determine if a model output has a confident tag.

An output is considered confident if any of its tags or detections have a score greater than or equal to the threshold.

Parameters:

Name	Type	Description	Default
`model_output`	`ModelOutput`	The model output of the recording containing detections and tags.	required

Returns:

Type	Description
`bool`

`should_save_recording(recording, model_outputs=None)` #

Save a recording if it contains any confident tags or detections.

`FrequencySchedule(duration, frequency)` #

Bases: RecordingSavingFilter

A frequency schedule RecordingSavingFilter.

Attributes#

`duration: int = duration` `instance-attribute` #

The duration of time (in minutes) where recordings will be saved.

`frequency: int = frequency` `instance-attribute` #

The frequency of time (in minutes) where recordings will be saved.

Functions#

`should_save_recording(recording, model_outputs=None)` #

Determine if a recording should be saved.

`SaveIfInInterval(interval, timezone)` #

Bases: RecordingSavingFilter

A time interval RecordingSavingFilter.

Attributes#

`interval: data.TimeInterval = interval` `instance-attribute` #

The interval of time where recordings will be saved.

`timezone: datetime.tzinfo = timezone` `instance-attribute` #

The timezone to use when determining if recording should be saved.

Functions#

`should_save_recording(recording, model_outputs=None)` #

Save a recording if it falls within the specified interval.

Examples:

>>> interval = data.TimeInterval(
...     start=datetime.time(21, 30), end=datetime.time(23, 00)
... )
>>> timezone = datetime.timezone.utc
>>> filter = SaveIfInInterval(interval, timezone)
>>> recording = data.Recording(
...     datetime=datetime.datetime(
...         2024, 1, 1, 22, 0, 0, tzinfo=timezone
...     )
... )
>>> filter.should_save_recording(recording)
True

`SavingThreshold(saving_threshold)` #

Bases: RecordingSavingFilter

A SavingThreshold RecordingSavingFilter.

Attributes#

`saving_threshold: float = saving_threshold` `instance-attribute` #

The score threshold to use.

Functions#

`has_confident_model_output(model_output)` #

Determine if a model output has confident detections or tags.

An output is considered confident if any of its detection score or classification tag score is greater than or equal to the threshold.

Parameters:

Name	Type	Description	Default
`model_output`	`ModelOutput`	The model output of the recording containing detections and tags.	required

Returns:

Type	Description
`bool`	True if any detection or classification score is above the saving threshold. False if no detection or classification score is above the saving threshold.

`should_save_recording(recording, model_outputs=None)` #

Save a recording if it contains any confident detections or tags.

`acoupi.components.saving_managers` #

Saving managers for the recordings and detections of acoupi.

Saving managers are used to determine where and how the recordings and detections of an audio file should be saved. This is helpful to handle recordings files and detections outputs. Recordings and detections outputs can be saved into a specific format (i.e, .wav files, .csv files) and at a specific location (i.e, rpi memory, external hardrive, folder XX/YY).

The SavingManagers are implemented as class that inherit from RecordingSavingManager. The classes should implement the save_recording method. The save_recording method takes a recording object and a list of model outputs as input and returns the path where the recording should be saved.

The save_recording method is called by the file management task to determine where the recording should be saved. The file management task is responsible for moving recordings from the memory to the disk, and remove recordings that are no longer needed.

The SavingManagers are optional and can be ingored if the no recordings are saved.

Classes#

`BaseFileManager(directory, logger=None)` #

Bases: RecordingSavingManager, ABC

Base implementation for simple recording saving managers.

This class can be used to implement simple recording saving managers that do not use model outputs to determine where the recording should be saved.

All recordings are saved in a directory specified in the constructor and the relative path is determined by the get_file_path method.

Attributes#

`directory: Path = directory` `instance-attribute` #

Directory where the files are stored.

`logger = logger` `instance-attribute` #

Functions#

`get_file_path(recording)` `abstractmethod` #

Get the path where the file of a recording should be stored.

The path must be relative to the directory specified in the constructor.

Parameters:

Name	Type	Description	Default
`recording`	`Recording`	Recording to get the path for.	required

Returns:

Type	Description
`Path of the file.`

`save_recording(recording, model_outputs=None)` #

Save a recording to a file.

Parameters:

Name	Type	Description	Default
`recording`	`Recording`	Recording to save.	required

Returns:

Type	Description
`Path of the saved file.`

`DateFileManager(directory, logger=None)` #

Bases: BaseFileManager

FileManager that uses the date to organise the recordings.

The recordings are organised in directories of the form

YYYY/MM/DD/

where YYYY is the year, MM is the month and DD is the day. The files are named using the time of the recording and its ID. The format is

HHMMSS_ID.wav

All the files are stored in a single directory that is specified in the constructor.

Functions#

`get_file_path(recording)` #

Get the path where the file of a recording should be stored.

Parameters:

Name	Type	Description	Default
`recording`	`Recording`	Recording to get the path for.	required

Returns:

Type	Description
`Path of the file.`

`IDFileManager(directory, logger=None)` #

Bases: BaseFileManager

FileManager that uses the ID of the recording to organise the files.

The recordings are saved in a single directory that is specified in the constructor. The files are named using the ID of the recording. The format is

ID.wav

Functions#

`get_file_path(recording)` #

Get the the path where the file of a recording should be stored.

`SaveRecordingManager(dirpath, dirpath_true=None, dirpath_false=None, timeformat='%Y%m%d_%H%M%S', detection_threshold=0.6, saving_threshold=0.3, logger=None)` #

Bases: RecordingSavingManager

A Recording SavingManager that save audio recordings.

Attributes#

`detection_threshold: float = detection_threshold` `instance-attribute` #

Threshold determining if a recording contains confident detections.

`dirpath: Path = dirpath` `instance-attribute` #

Directory path to save recordings.

`dirpath_false: Path = dirpath_false` `instance-attribute` #

Directory path to save recordings if audio recording contain no confident detections (i.e., below the detection threshold).

`dirpath_true: Path = dirpath_true` `instance-attribute` #

Directory path to save recordings if audio recording contains confident detections (i.e., above the detection threshold).

`logger = logger` `instance-attribute` #

`saving_threshold: float = saving_threshold` `instance-attribute` #

Threshold determining if recordings should be saved (i.e., regardless of confident or unconfident detections).

`timeformat: str = timeformat` `instance-attribute` #

Datetime format to use to name the recording file path.

Functions#

`get_saving_recording_path(model_outputs)` #

Determine where the recording should be saved.

Parameters:

Name	Type	Description	Default
`model_outputs`	`Optional[List[ModelOutput]]`	List of model outputs containing detections and tags.	required

Returns:

Type	Description
`Path`	Path where the recording should be saved.

`save_recording(recording, model_outputs=None)` #

Save a recording to a file.

Examples:

>>> dirpath = Path("path/to/save")
>>> dirpath_true = Path("path/to/save/confident_detections")
>>> dirpath_false = Path("path/to/save/unconfident_detections")
>>> detection_threshold = 0.8
>>> saving_threshold = 0.3

>>> model_outputs = [
...     data.ModelOutput(tags=[data.Tag(confidence_score=0.7)]),
...     data.ModelOutput(
...         detections=[data.Detection(detection_score=0.6)]
...     ),
... ]
>>> saving_directory = self.get_saving_recording_path(
...     model_outputs
... )
>>> assert saving_directory == dirpath_false

`acoupi.components.summariser` #

Summariser for acoupi.

Summarisers are responsible for summarising model outputs (i.e., detections). Summarisers output a summary of type data.Message. The StatisticsDetectionsSummariser summarises the detections by calculating the mean, min, max, and count of classifications probabilities for each species. The ThresholdsDetectionsSummariser summarises the detections by calculating the count and mean of classifications probabilities for each species that fall within a low, mid, and high threshold.

The message output by the Summarisers is then used by the Messenger to send the summary to a remote server. Summarisers are implemented as classes that inherit from Summariser. Implemntation of the Summarisers should refer to the database, where the classifications probabilities are stored. The class should implement the build_summary method, which takes a datetime.datetime object and returns a message in JSON format.

Classes#

`StatisticsDetectionsSummariser(store, interval=3600)` #

Bases: Summariser

Summarises detections by calculating the mean, min, max, and count of classification probabilities for each species.

Attributes#

`interval: datetime.timedelta = interval` `instance-attribute` #

The interval to get the detections from in seconds.

`store: SqliteStore = store` `instance-attribute` #

The store to get the detections from.

Functions#

`build_summary(now)` #

Build a message from a summary.

Parameters:

Name	Type	Description	Default
`now`	`datetime`	The current time to get the detections from.	required
`self`		The predicted tags from the store associated with the audio recordings that falls within the time interval (current_time - interval_minute).	required

Returns:

Type	Description
`Message`	A message containing the summary of the detections. The summary includes the mean, min, max, and count of classification probabilities for each species.

Examples:

>>> store = SqliteStore("test.db")
>>> summariser = StatisticsDetectionsSummariser(store)
>>> now = datetime.datetime.now()
>>> summariser.build_summary(now)
... summary_message = data.Message(
...     content='{
...         "species_1": {
...             "mean": 0.5,
...             "min": 0.1,
...             "max": 0.9,
...             "count": 10},
...         "species_2": {
...             "mean": 0.6,
...             "min": 0.2,
...             "max": 0.8,
...             "count": 20},
...         "timeinterval": {
...             "starttime": "2021-01-01T00:00:00",
...             "endtime": "2021-01-01T00:10:00"}
...     }'
... )

`ThresholdsDetectionsSummariser(store, interval=3600, low_band_threshold=0.1, mid_band_threshold=0.5, high_band_threshold=0.9)` #

Bases: Summariser

Summariser that summarises detections by classification score thresholds.

Parameters:

Name	Type	Description	Default
`low_band_threshold`	`float`	The lower threshold for the classification score, by default 0.1.	`0.1`
`mid_band_threshold`	`float`	The middle threshold for the classification score, by default 0.5.	`0.5`
`high_band_threshold`	`float`	The higher threshold for the classification score, by default 0.9.	`0.9`

Attributes#

`high_band_threshold = high_band_threshold` `instance-attribute` #

`interval: datetime.timedelta = interval` `instance-attribute` #

The interval to get the detections from in seconds.

`low_band_threshold = low_band_threshold` `instance-attribute` #

`mid_band_threshold = mid_band_threshold` `instance-attribute` #

`store: SqliteStore = store` `instance-attribute` #

The store to get the detections from.

Functions#

`build_summary(now)` #

Build a message from a summary.

Parameters:

Name	Type	Description	Default
`now`	`datetime`	The current time to get the detections from.	required
`self`		The predicted tags from the store associated with the audio recordings that falls within the time interval (current_time - interval_minute).	required

Returns:

Type	Description
`Message`	A message containing the summary of the detections. The summary includes the count and mean of classification probabilities for each species that fall within a low, mid, and high threshold.

Examples:

>>> store = SqliteStore("test.db")
>>> summariser = ThresholdsDetectionsSummariser(store)
>>> now = datetime.datetime.now()
>>> summariser.build_summary(now)
... summary_message = data.Message(
...     content='{
...         "species_1": {
...             "count_low_threshold": 10, "count_mid_threshold": 20, "count_high_threshold": 30,
...             "mean_low_threshold": 0.1, "mean_mid_threshold": 0.5, "mean_high_threshold": 0.9},
...         "species_2": {
...             "count_low_threshold": 15, "count_mid_threshold": 25, "count_high_threshold": 35,
...             "mean_low_threshold": 0.2, "mean_mid_threshold": 0.6, "mean_high_threshold": 0.8},
...         "timeinterval": {
...             "starttime": "2021-01-01T00:00:00",
...             "endtime": "2021-01-01T00:10:00"}
...    }'
... )

Components#

acoupi.components #

Component Types#

acoupi.components.types #

Attributes#

P = ParamSpec('P') module-attribute #

Classes#

AudioRecorder #

Functions#

record(deployment) abstractmethod #

MessageBuilder #

Functions#

build_message(*args, **kwargs) abstractmethod #

MessageStore #

Functions#

get_unsent_messages() abstractmethod #

store_message(message) abstractmethod #

store_response(response) abstractmethod #

Messenger #

Functions#

send_message(message) abstractmethod #

Model #

Functions#

run(recording) abstractmethod #

ModelOutputCleaner #

Functions#

clean(model_output) abstractmethod #

ProcessingFilter #

Functions#

should_process_recording(recording) abstractmethod #

RecordingCondition #

Functions#

should_record() abstractmethod #

RecordingSavingFilter #

Functions#

should_save_recording(recording, model_outputs=None) abstractmethod #

RecordingSavingManager #

Functions#

save_recording(recording, model_outputs=None) abstractmethod #

RecordingScheduler #

Functions#

time_until_next_recording(time=None) abstractmethod #

Store #

Functions#

get_current_deployment() abstractmethod #

get_recordings(ids) abstractmethod #

get_recordings_by_path(paths) abstractmethod #

store_deployment(deployment) abstractmethod #

store_model_output(model_output) abstractmethod #

store_recording(recording, deployment=None) abstractmethod #

update_deployment(deployment) abstractmethod #

update_recording_path(recording, path) abstractmethod #

Summariser #

Functions#

build_summary(now) abstractmethod #

Concrete Components#

acoupi.components.audio_recorder #

Attributes#

TMP_PATH = Path('/run/shm/') module-attribute #

Classes#

MicrophoneConfig #

Attributes#

audio_channels: int = 1 class-attribute instance-attribute #

device_name: str instance-attribute #

samplerate: int = 48000 class-attribute instance-attribute #

Functions#

setup(args, prompt=True, prefix='') classmethod #

PyAudioRecorder(duration, samplerate, audio_channels, device_name, chunksize=2048, audio_dir=TMP_PATH, logger=None) #

Attributes#

audio_channels: int = audio_channels instance-attribute #

audio_dir: Path = audio_dir instance-attribute #

chunksize: int = chunksize instance-attribute #

device_name: str = device_name instance-attribute #

duration: float = duration instance-attribute #

logger = logger instance-attribute #

sample_width = pyaudio.get_sample_size(pyaudio.paInt16) instance-attribute #

samplerate: int = samplerate instance-attribute #

Functions#

check() #

get_input_device(p) #

`acoupi.components` #

`acoupi.components.types` #

`P = ParamSpec('P')` `module-attribute` #

`AudioRecorder` #

`record(deployment)` `abstractmethod` #

`MessageBuilder` #

`build_message(*args, **kwargs)` `abstractmethod` #

`MessageStore` #

`get_unsent_messages()` `abstractmethod` #

`store_message(message)` `abstractmethod` #

`store_response(response)` `abstractmethod` #

`Messenger` #

`send_message(message)` `abstractmethod` #

`Model` #

`run(recording)` `abstractmethod` #

`ModelOutputCleaner` #

`clean(model_output)` `abstractmethod` #

`ProcessingFilter` #

`should_process_recording(recording)` `abstractmethod` #

`RecordingCondition` #

`should_record()` `abstractmethod` #

`RecordingSavingFilter` #

`should_save_recording(recording, model_outputs=None)` `abstractmethod` #

`RecordingSavingManager` #

`save_recording(recording, model_outputs=None)` `abstractmethod` #

`RecordingScheduler` #

`time_until_next_recording(time=None)` `abstractmethod` #

`Store` #

`get_current_deployment()` `abstractmethod` #

`get_recordings(ids)` `abstractmethod` #

`get_recordings_by_path(paths)` `abstractmethod` #

`store_deployment(deployment)` `abstractmethod` #

`store_model_output(model_output)` `abstractmethod` #

`store_recording(recording, deployment=None)` `abstractmethod` #

`update_deployment(deployment)` `abstractmethod` #

`update_recording_path(recording, path)` `abstractmethod` #

`Summariser` #

`build_summary(now)` `abstractmethod` #

`acoupi.components.audio_recorder` #

`TMP_PATH = Path('/run/shm/')` `module-attribute` #

`MicrophoneConfig` #

`audio_channels: int = 1` `class-attribute` `instance-attribute` #

`device_name: str` `instance-attribute` #

`samplerate: int = 48000` `class-attribute` `instance-attribute` #

`setup(args, prompt=True, prefix='')` `classmethod` #

`PyAudioRecorder(duration, samplerate, audio_channels, device_name, chunksize=2048, audio_dir=TMP_PATH, logger=None)` #

`audio_channels: int = audio_channels` `instance-attribute` #

`audio_dir: Path = audio_dir` `instance-attribute` #

`chunksize: int = chunksize` `instance-attribute` #

`device_name: str = device_name` `instance-attribute` #

`duration: float = duration` `instance-attribute` #

`logger = logger` `instance-attribute` #

`sample_width = pyaudio.get_sample_size(pyaudio.paInt16)` `instance-attribute` #

`samplerate: int = samplerate` `instance-attribute` #

`check()` #

`get_input_device(p)` #

`get_recording_data(duration=None, num_chunks=None)` #

`record(deployment)` #

`save_recording(data, path)` #

`parse_microphone_config(args, prompt=True, prefix='')` #

`acoupi.components.message_factories` #

`DetectionThresholdMessageBuilder(detection_threshold)` #

`detection_threshold = detection_threshold` `instance-attribute` #

`build_message(model_output)` #

`filter_detections(detections)` #

`FullModelOutputMessageBuilder` #

`build_message(model_output)` #

`acoupi.components.messengers` #

`HTTPConfig` #

`base_url: str` `instance-attribute` #

`content_type: str = 'application/json'` `class-attribute` `instance-attribute` #

`timeout: int = 5` `class-attribute` `instance-attribute` #

`HTTPMessenger(base_url, base_params=None, headers=None, timeout=5, content_type='application/json', logger=None)` #

`base_params: dict = base_params or {}` `instance-attribute` #

`base_url: str = base_url` `instance-attribute` #

`content_type = self.headers['Content-Type']` `instance-attribute` #

`headers: dict = headers or {}` `instance-attribute` #

`logger = logger` `instance-attribute` #

`timeout: int = timeout` `instance-attribute` #

`check()` #