diart.blocks.aggregation#

Module Contents#

Classes#

AggregationStrategy

Abstract class representing a strategy to aggregate overlapping buffers

HammingWeightedAverageStrategy

Compute the average weighted by the corresponding Hamming-window aligned to each buffer

AverageStrategy

Compute a simple average over the focus region

FirstOnlyStrategy

Instead of aggregating, keep the first focus region in the buffer list

DelayedAggregation

Aggregate aligned overlapping windows of the same duration
class diart.blocks.aggregation.AggregationStrategy(cropping_mode='loose')#

Bases: abc.ABC

Abstract class representing a strategy to aggregate overlapping buffers

Parameters:

cropping_mode (("strict", "loose", "center"), optional) – Defines the mode to crop buffer chunks as in pyannote.core. See https://pyannote.github.io/pyannote-core/reference.html#pyannote.core.SlidingWindowFeature.crop Defaults to “loose”.

static build(name, cropping_mode='loose')#

Build an AggregationStrategy instance based on its name

Parameters:

  • name (typing_extensions.Literal[mean, hamming, first]) –

  • cropping_mode (typing_extensions.Literal[strict, loose, center]) –

Return type:

AggregationStrategy

__call__(buffers, focus)#

Aggregate chunks over a specific region.

Parameters:

  • buffers (list of SlidingWindowFeature, shapes (frames, speakers)) – Buffers to aggregate

  • focus (Segment) – Region to aggregate that is shared among the buffers

Returns:

aggregation – Aggregated values over the focus region

Return type:

SlidingWindowFeature, shape (cropped_frames, speakers)

abstract aggregate(buffers, focus)#
Parameters:

  • buffers (List[pyannote.core.SlidingWindowFeature]) –

  • focus (pyannote.core.Segment) –

Return type:

numpy.ndarray

class diart.blocks.aggregation.HammingWeightedAverageStrategy(cropping_mode='loose')#

Bases: AggregationStrategy

Compute the average weighted by the corresponding Hamming-window aligned to each buffer

Parameters:

cropping_mode (typing_extensions.Literal[strict, loose, center]) –

aggregate(buffers, focus)#
Parameters:

  • buffers (List[pyannote.core.SlidingWindowFeature]) –

  • focus (pyannote.core.Segment) –

Return type:

numpy.ndarray

class diart.blocks.aggregation.AverageStrategy(cropping_mode='loose')#

Bases: AggregationStrategy

Compute a simple average over the focus region

Parameters:

cropping_mode (typing_extensions.Literal[strict, loose, center]) –

aggregate(buffers, focus)#
Parameters:

  • buffers (List[pyannote.core.SlidingWindowFeature]) –

  • focus (pyannote.core.Segment) –

Return type:

numpy.ndarray

class diart.blocks.aggregation.FirstOnlyStrategy(cropping_mode='loose')#

Bases: AggregationStrategy

Instead of aggregating, keep the first focus region in the buffer list

Parameters:

cropping_mode (typing_extensions.Literal[strict, loose, center]) –

aggregate(buffers, focus)#
Parameters:

  • buffers (List[pyannote.core.SlidingWindowFeature]) –

  • focus (pyannote.core.Segment) –

Return type:

numpy.ndarray

class diart.blocks.aggregation.DelayedAggregation(step, latency=None, strategy='hamming', cropping_mode='loose')#

Aggregate aligned overlapping windows of the same duration across sliding buffers with a specific step and latency.

Parameters:

  • step (float) – Shift between two consecutive buffers, in seconds.

  • latency (float, optional) – Desired latency, in seconds. Defaults to step. The higher the latency, the more overlapping windows to aggregate.

  • strategy (("mean", "hamming", "first"), optional) – Specifies how to aggregate overlapping windows. Defaults to “hamming”. “mean”: simple average “hamming”: average weighted by the Hamming window values (aligned to the buffer) “first”: no aggregation, pick the first overlapping window

  • cropping_mode (("strict", "loose", "center"), optional) – Defines the mode to crop buffer chunks as in pyannote.core. See https://pyannote.github.io/pyannote-core/reference.html#pyannote.core.SlidingWindowFeature.crop Defaults to “loose”.

Example

>>> duration = 5
>>> frames = 500
>>> step = 0.5
>>> speakers = 2
>>> start_time = 10
>>> resolution = duration / frames
>>> dagg = DelayedAggregation(step=step, latency=2, strategy="mean")
>>> buffers = [
>>>     SlidingWindowFeature(
>>>         np.random.rand(frames, speakers),
>>>         SlidingWindow(start=(i + start_time) * step, duration=resolution, step=resolution)
>>>     )
>>>     for i in range(dagg.num_overlapping_windows)
>>> ]
>>> dagg.num_overlapping_windows
... 4
>>> dagg(buffers).data.shape
... (51, 2)  # Rounding errors are possible when cropping the buffers
_prepend(output_window, output_region, buffers)#
Parameters:

  • output_window (pyannote.core.SlidingWindowFeature) –

  • output_region (pyannote.core.Segment) –

  • buffers (List[pyannote.core.SlidingWindowFeature]) –

__call__(buffers)#
Parameters:

buffers (List[pyannote.core.SlidingWindowFeature]) –

Return type:

pyannote.core.SlidingWindowFeature