lib.infer package

The infer Package contains objects and utilities for extracting faces from images and videos.

lib.infer.align Module

Handles face landmark detection plugins and runners

class lib.infer.align.Align(plugin: str, re_feeds: int = 0, re_align: bool = False, normalization: Literal['none', 'clahe', 'hist', 'mean'] | None = None, filters: bool = False, compile_model: bool = False, config_file: str | None = None)

Responsible for handling align plugins within the extract pipeline

Parameters:

• plugin (str) – The plugin that this runner is to use

• re_feeds (int) – Number of times to jitter detection bounding box and average the result. Default: 0

• re_align (bool) – True to re-align faces based on their first-pass results. Default: False

• normalization (T.Literal['none', 'clahe', 'hist', 'mean'] | None) – The normalization to perform on aligner input images. Default: None (no normalization)

• filters (bool) – True to enable aligner filters to filter out faces. Default: False

• compile_model (bool) – True to compile any PyTorch models

• config_file (str | None) – Full path to a custom config file to load. None for default config

output_info() → None

Output the counts from the aligner filter

Return type:

None

post_process(batch: ExtractBatch) → None

Post-process the landmark predictions from the model: average any re-feeds, scale back to original frame dimensions, apply any filters

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for post-processing

Return type:

None

pre_process(batch: ExtractBatch) → None

Obtain the adjusted square ROIs from the plugin based off the provided detection bounding boxes. Crop and size the input face images ready for inference from these ROIs

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for pre-processing

Return type:

None

process(batch: ExtractBatch) → None

Perform inference to get results from the aligner

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for processing

Return type:

None

set_normalize_method(method: Literal['none', 'clahe', 'hist', 'mean'] | None) → None

Update the normalization method with the given method

Parameters:

method (Literal['none', 'clahe', 'hist', 'mean'] | None) – The normalization method to use

Return type:

None

class lib.infer.align.AlignedFilter(enabled: bool)

Applies filters to the output of the aligner

Parameters:

enabled (bool) – True to enable filters. False to disable

__call__(batch: ExtractBatch) → None

Apply aligner filters to the given batch

Parameters:

batch (ExtractBatch) – The batch object to perform filtering on with the landmarks populated

Return type:

None

output_counts() → None

If filters are enabled info log the number of faces filtered

Return type:

None

class lib.infer.align.Normalize(method: Literal['none', 'clahe', 'hist', 'mean'])

Handles the normalization of feed images prior to feeding the model

Parameters:

method (T.Literal['none', 'clahe', 'hist', 'mean'])

__call__(images: npt.NDArray[np.uint8]) → npt.NDArray[np.uint8]

Perform the selected normalization method on the batch of model input images

Parameters:

images (npt.NDArray[np.uint8]) – The batch of model input images to be normalized

Return type:

The given images normalized by the chosen method, or the input batch if no method selected

set_method(method: Literal['none', 'clahe', 'hist', 'mean'] | None) → None

Update the normalization method with the given method

Parameters:

method (Literal['none', 'clahe', 'hist', 'mean'] | None) – The normalization method to use

Return type:

None

class lib.infer.align.ReAlign(enabled: bool, plugin: ExtractPlugin, margin: float)

Handles re-aligning faces based on first-pass results

Parameters:

• enabled (bool) – True if realigns are to be performed

• plugin (ExtractPlugin) – The plugin that will be processing re-aligns

• margin (float) – The % amount that re-feed allows bounding box points to drift

__call__(batch: ExtractBatch, landmarks: npt.NDArray[np.float32], iteration: int) → None

Process the outputs from the plugin when re-aligning data

Is called twice.

• First pass: aligns the image based on the first pass landmarks, stores image patches

that next pass’ feed will be generated from and creates ROI boxes for this aligned patch - 2nd pass: Rotates detections back to frame alignment and updates the ROI to correctly scale and shift the alignments back to frame space downstream

Parameters:

• batch (ExtractBatch) – The batch object being processed for re-aligns

• landmarks (npt.NDArray[np.float32]) – The (x, y) detected landmarks for a batch in mean-space

• iteration (int) – The re-align iteration that is being request. Either 1 or 2

Return type:

None

property default_crop_matrices: npt.NDArray[np.float32]

The default crop matrices used for calculating re-feeds

enabled

True if re-aligns are enabled

get_images(matrices: npt.NDArray[np.float32], feeds: int) → npt.NDArray[np.float32]

Obtain the sub-crops from the main image patches based on the roi stored in the batch and populate them to the batch’s data attribute

Parameters:

• matrices (npt.NDArray[np.float32]) – The matrices that define the crops to extract from the expanded patch in shape (N x total_feeds, 3, 3)

• feeds (int) – The number of feeds that are to be made through the model for this batch

Return type:

The aligned images that are to be used for 2nd pass re-align

iterations

The total number of iterations through the align process required for the selected re-align configuration

class lib.infer.align.ReFeed(re_feeds: int)

Handles preparation of images for re-feeding the aligner with minor adjustments to detection bounding boxes, and averaging the result at the end.

Parameters:

re_feeds (int) – The number of re-feeds to be performed.

__call__(matrices: npt.NDArray[np.float32], with_roi: Literal[False], size: int = 0) → npt.NDArray[np.float32]

__call__(matrices: npt.NDArray[np.float32], with_roi: Literal[True] = True, size: int = 0) → tuple[npt.NDArray[np.float32], npt.NDArray[np.int32]]

Obtain an array of adjusted norm to frame matrices based on the number of re-feed iterations that have been selected and the size of the original ROI.

Parameters:

• matrices (npt.NDArray[np.float32]) – A batch of norm to frame transformation matrices to be randomly adjust for re-feeding the model in shape (N, 3, 3)

• with_roi (bool) – True to also return the adjusted ROIs. Default: False

• size (int) – The size of the image patch that the matrix creates if it cannot be derived from the matrices. Default: 0 (derive from matrices)

Returns:

• matrices – The adjusted matrices for taking points from normalized to frame space in shape ((Num re_feeds * N) + 1, 3, 3), in frame contiguous order (Na, Nb, Nc, Na1, Nb1, Nc1…)

• roi – The ((Num re_feeds * N) + 1, 4) roi for each adjusted feed. Returned if with_roi is True

Return type:

npt.NDArray[np.float32] | tuple[npt.NDArray[np.float32], npt.NDArray[np.int32]]

beta

The amount each corner point can move relative to the boxes shortest side

merge(landmarks: npt.NDArray[np.float32]) → npt.NDArray[np.float32]

If re-feeds enabled return the average result from the re-feeds, otherwise the original array

Parameters:

landmarks (npt.NDArray[np.float32]) – The (N x total_feeds, 68, 2) landmarks from the plugin

Return type:

The final (N, 68, 2) landmarks with any re-feeds merged

total_feeds

The total number of feeds through the model for original boxes plus re-feeds

Classes

Align(plugin[, re_feeds, re_align, ...])	Responsible for handling align plugins within the extract pipeline
AlignedFilter(enabled)	Applies filters to the output of the aligner
Normalize(method)	Handles the normalization of feed images prior to feeding the model
ReAlign(enabled, plugin, margin)	Handles re-aligning faces based on first-pass results
ReFeed(re_feeds)	Handles preparation of images for re-feeding the aligner with minor adjustments to detection bounding boxes, and averaging the result at the end.

Class Inheritance Diagram

lib.infer.detect Module

Handles face detection plugins and runners

class lib.infer.detect.Detect(plugin: str, rotation: str | None = None, min_size: int = 0, max_size: int = 0, compile_model: bool = False, config_file: str | None = None)

Responsible for handling Detection plugins within the extract pipeline

Parameters:

• plugin (str) – The plugin that this runner is to use

• None (rotation |) – The rotation arguments. Either a list of angles between 0 and 360 to rotate at or a single step size. Default: None, no rotations

• min_size (int) – Minimum percentage of the frame’s shortest edge to accept as a successful detection along the detection’s longest edge Default: 0 (accept all detections)

• max_size (int) – Maximum percentage of the frame’s shortest edge to accept as a successful detection along the detection’s longest edge Default: 0 (accept all detections)

• compile_model (bool) – True to compile any PyTorch models

• config_file (str | None) – Full path to a custom config file to load. None for default config

• rotation (str | None)

output_info() → None

Output the counts of filtered items

Return type:

None

post_process(batch: ExtractBatch) → None

Perform detection post processing.

If no rotations were requested, any plugin post-processing will be done here.

Detection boxes are:

• stacked into a single array

• scaled back to frame dimensions,

• filtered for faces which fall outside min/max thresholds

• Added to the batch object along with frame to face mapping information.

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for post-processing

Return type:

None

pre_process(batch: ExtractBatch) → None

Perform pre-processing for detection plugins.

• Gets the scale and padding to take the batch of images to model input size

• Formats the image to the correct color order, dtype and scale for the plugin

• Performs any plugin specific pre-processing

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for pre-processing

Return type:

None

process(batch: ExtractBatch) → None

Obtain the output from the plugin’s model.

Executes the plugin’s predict function and stores the output prior to post-processing.

If rotations have been selected, plugin post-processing is done as part of this process as the computed bounding boxes are required for re-feeding the model future rotations

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for processing

Return type:

None

class lib.infer.detect.Rotator(angles: str | None, image_size: int)

Handles pre-calculation of rotation matrices when rotation angles are requested and rotating images for feeding the detector. Handles reversing the rotation for any found detection bounding boxes.

Parameters:

• rotation – List of requested rotation angles in degrees provided in command line arguments

• image_size (int) – The size of the square image to obtain rotation matrices for

• angles (str | None)

enabled

True if rotations are to be performed

rotate(rotation_index: int, images: ndarray) → ndarray | None

Rotate a batch of images by the matrix provided by the given rotation index. Attempts to detect and handle channels first images as well as channels last

Parameters:

• rotation_index (int) – The matrix to use. This will be an incrementing index from an enumerated loop that selects through the matrices stored for each angle

• images (ndarray) – The original, correctly orientated, batch of images to rotate

Returns:

• The batch of image rotated by the angle identified by the given rotation index.

• None if the given rotation index is invalid

Return type:

ndarray | None

un_rotate(indices_angle: npt.NDArray[np.int32], roi: npt.NDArray[np.float32]) → None

Un-rotate the given bounding boxes for the given angle indices and update in place

Parameters:

• indices_angle (npt.NDArray[np.int32]) – The angle indices that correlate to the angle each roi was rotated to to obtain the result

• roi (npt.NDArray[np.float32]) – Ragged array of (B, N, 4) detected bounding discovered at the corresponding angle index

Return type:

None

Classes

Detect(plugin[, rotation, min_size, ...])	Responsible for handling Detection plugins within the extract pipeline
Rotator(angles, image_size)	Handles pre-calculation of rotation matrices when rotation angles are requested and rotating images for feeding the detector.

Class Inheritance Diagram

lib.infer.handler Module

Handles individual plugins within a plugin runner

class lib.infer.handler.ExtractHandler(plugin: str, compile_model: bool = False, config_file: str | None = None)

Handles the execution of a plugin’s pre_process, process and post_process actions

Parameters:

• plugin (str) – The name of the plugin that this handler is to use

• compile_model (bool) – True to compile any PyTorch models

• config_file (str | None) – Full path to a custom config file to load. None for default config

__call__(input_plugin: ExtractHandler | ExtractRunner | None = None, profile: bool = False) → ExtractRunner

Build and start the plugin handler’s runner

Parameters:

• input_plugin (ExtractHandler | ExtractRunner | None) – The input plugin handler or it’s runner that feeds this handler. None if data is to be fed through the handler runner’s put method (ie, the first handler in an extraction chain). Default: None

• profile (bool) – True if the runner is to be profiled, indicating that threads will not be started. Default: False

Return type:

The extract plugin handler’s runner for this handler

property batch_size: int

The batch size of the plugin

do_compile

True if any managed Torch modules are to be compiled

init_model() → None

Load the model, compile it, if requested, and send a warmup batch through. Called either from the main thread, if compiling, or from the inference thread if not.

Return type:

None

output_info() → None

Called after the final item is put to the out queue. Override for plugin runner specific output

Return type:

None

plugin

The extraction plugin that this handler manages

plugin_name

The name of the plugin that is being handled

plugin_type: Literal['detect', 'align', 'mask', 'identity', 'file']

The type of plugin that this handler manages

abstractmethod post_process(batch: ExtractBatch) → None

Perform post-processing on the given batch object, ready for exit from the plugin. Override for plugin type specific behavior

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for post-processing

Return type:

None

abstractmethod pre_process(batch: ExtractBatch) → None

Override to perform plugin type specific behavior for pre-processing on the given batch object, ready for inference.

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for pre-processing

Return type:

None

abstractmethod process(batch: ExtractBatch) → None

Override to plugin type specific processing to get results from the plugin’s inference for the given batch.

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for processing

Return type:

None

processors: tuple[Literal['pre_process', 'process', 'post_process'], ...] = ('pre_process', 'process', 'post_process')

The processors which should have thread’s launched for this handler

property runner: ExtractRunner

The runner that runs this handler

class lib.infer.handler.ExtractHandlerFace(plugin: str, compile_model: bool = False, config_file: str | None = None)

Handles an extract plugin. Extended with methods common to plugins that use aligned face images as input

Parameters:

• plugin (FacePlugin) – The name of the plugin that this runner is to use

• compile_model (bool) – True to compile any PyTorch models

• config_file (str | None) – Full path to a custom config file to load. None for default config

plugin: FacePlugin

The extraction plugin that this handler manages

process(batch: ExtractBatch) → None

Perform inference to get results from the plugin for the given batch. Override for plugin type specific processing

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for processing

Return type:

None

storage_name

The name that the object will be stored with in the alignments file

class lib.infer.handler.FileHandler

A pseudo handler that passes through data when the pipeline is driven entirely by an alignments file (ie: no plugins are being loaded). This is effectively a No-op which allows the data to pass straight from input to output

class Plugin

Dummy plugin with required properties

__call__(input_plugin: ExtractHandler | ExtractRunner | None = None, profile: bool = False) → ExtractRunner

Build and start the plugin handler’s runner. Overridden to ensure that neither an input plugin or profile are set

Parameters:

• input_plugin (ExtractHandler | ExtractRunner | None) – The input plugin handler or it’s runner that feeds this handler. None if data is to be fed through the handler runner’s put method (ie, the first handler in an extraction chain). Must be None for file handler

• profile (bool) – True if the runner is to be profiled, indicating that threads will not be started. Must be False for file handler

Return type:

The extract plugin handler’s runner for this handler

post_process(batch: ExtractBatch) → None

Not applicable for passthrough plugin.

Parameters:

batch (ExtractBatch)

Return type:

None

pre_process(batch: ExtractBatch) → None

Not applicable for passthrough plugin.

Parameters:

batch (ExtractBatch)

Return type:

None

process(batch: ExtractBatch) → None

Not applicable for passthrough plugin.

Parameters:

batch (ExtractBatch)

Return type:

None

processors: tuple[Literal['pre_process', 'process', 'post_process'], ...] = ()

File handler launches no threads

Classes

ExtractHandler(plugin[, compile_model, ...])	Handles the execution of a plugin's pre_process, process and post_process actions
ExtractHandlerFace(plugin[, compile_model, ...])	Handles an extract plugin.
ExtractPlugin(input_size[, batch_size, ...])	Base extract plugin that all plugins must inherit from.
ExtractRunner(handler)	Runs an extract plugin
FaceswapError	Faceswap Error for handling specific errors with useful information.
FileHandler()	A pseudo handler that passes through data when the pipeline is driven entirely by an alignments file (ie: no plugins are being loaded).
LandmarkType(*values)	Enumeration for the landmark types that Faceswap supports
OutOfMemoryError	Exception raised when device is out of memory
PluginLoader()	Retrieve, or get information on, Faceswap plugins

Variables

EXTRACT_RATIOS	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2).
OOM_MESSAGE	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
annotations
logger	A standard logging.logger with additional "verbose" and "trace" levels added.

Class Inheritance Diagram

lib.infer.identity Module

Handles face identity plugins and runners

class lib.infer.identity.Cluster(predictions: ndarray, method: Literal['single', 'centroid', 'median', 'ward'], threshold: float | None = None)

Cluster the outputs from a VGG-Face 2 Model

Parameters:

• predictions (np.ndarray) – A stacked matrix of identity predictions of the shape (N, D) where N is the number of observations and D are the number of dimensions. NB: The given predictions will be overwritten to save memory. If you still require the original values you should take a copy prior to running this method

• method (T.Literal['single', 'centroid', 'median', 'ward']) – The clustering method to use.

• threshold (float | None) – The threshold to start creating bins for. Set to None to disable binning

__call__() → list[tuple[int, int]]

Process the linkages.

Transforms a distance matrix into a sorted distance matrix according to the order implied by the hierarchical tree (dendrogram).

Returns:

• List of indices with the order implied by the hierarchical tree or list of tuples of

• (index, bin) if a binning threshold was provided

Return type:

list[tuple[int, int]]

class lib.infer.identity.FilterLoader(threshold: float, filter_files: list[str] | None, nfilter_files: list[str] | None)

Obtains face embeddings from images and loads the IdentityFilter as part of the extraction pipeline

Parameters:

• threshold (float) – The threshold value for filtering out items. Default: 0.4

• filter_files (list[str] | None) – The list of full paths to the files to use for filtering. Default: None (don’t use filter)

• nfilter_files (list[str] | None) – The list of full paths to the files to use to nfilter. Default: None (don’t use nfilter)

add_identity_plugin(runner: ExtractRunner) → None

Add the identity plugin for updating with embedding information

Parameters:

runner (ExtractRunner) – The identity runner for the pipeline

Return type:

None

enabled

True if identity face filtering is enabled

get_embeddings(pipeline: ExtractRunner) → None

Obtain the embeddings that are to be used for face filtering and add to the identity plugin

Parameters:

pipeline (ExtractRunner) – The extraction pipelines for obtaining embeddings from non-faceswap images

Return type:

None

threshold

The threshold value for filtering out items

class lib.infer.identity.Identity(plugin: str, threshold: float = 0.4, compile_model: bool = False, config_file: str | None = None)

Responsible for handling Identity/Recognition plugins within the extract pipeline

Parameters:

• plugin (str) – The plugin that this runner is to use

• filter_threshold – The threshold to use when filtering faces by identity. Default: 0.4

• compile_model (bool) – True to compile any PyTorch models

• config_file (str | None) – Full path to a custom config file to load. None for default config

• threshold (float)

add_filter_identities(identities: npt.NDArray[np.float32], is_filter: bool) → None

Add the given identities to the identity filter

Parameters:

• identities (npt.NDArray[np.float32]) – The identity embeddings to add to the filter

• is_filter (bool) – True for filter, False for nFilter

Return type:

None

output_info() → None

Output the counts from the identity filter

Return type:

None

post_process(batch: ExtractBatch) → None

Perform recognition post processing.

Obtains the final output from the identity plugin and performs any plugin specific post- processing

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for post-processing

Return type:

None

pre_process(batch: ExtractBatch) → None

Obtain the aligned face images at the requested size, centering and image format. Perform any plugin specific pre-processing

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for pre-processing

Return type:

None

class lib.infer.identity.IdentityFilter(threshold: float, name: str)

Handles filtering of faces based on provided image files

Parameters:

• threshold (float) – The threshold value for filtering out items

• name (str) – The name of the identity plugin running

__call__(batch: ExtractBatch) → None

Apply the identity filter to the given batch

Parameters:

batch (ExtractBatch) – The batch object to perform filtering on with the identities populated

Return type:

None

add_identities(identities: npt.NDArray[np.float32], is_filter: bool) → None

Add the given identities to the filter

Parameters:

• identities (npt.NDArray[np.float32]) – The identity embeddings to add to the filter

• is_filter (bool) – True for filter, False for nFilter

Return type:

None

enabled

True if the identity filter is enabled

output_counts() → None

If filter is enabled info log the number of faces filtered

Return type:

None

threshold

The threshold for accepting a filter result

Classes

Cluster(predictions, method[, threshold])	Cluster the outputs from a VGG-Face 2 Model
FilterLoader(threshold, filter_files, ...)	Obtains face embeddings from images and loads the IdentityFilter as part of the extraction pipeline
Identity(plugin[, threshold, compile_model, ...])	Responsible for handling Identity/Recognition plugins within the extract pipeline
IdentityFilter(threshold, name)	Handles filtering of faces based on provided image files

Class Inheritance Diagram

lib.infer.iterator Module

Iterators for ingesting into and passing data through extract plugin runners

class lib.infer.iterator.ExtractIterator(queue: Queue[QueueItemInT | ExtractSignal], name: str, plugin_type: T.Literal['detect', 'align', 'mask', 'identity', 'file'], batch_size: int, error_state: ErrorState)

Base class for iterators within Faceswap’s extract pipeline

Type Parameters

QueueItemInT

Type of item received from the input queue.

QueueItemOutT

Type yielded by the iterator.

param queue:

The inbound queue to the plugin

param name:

The plugin name and process calling this iterator

param plugin_type:

The type of extractor plugin that this iterator is serving

param batch_size:

The batch size that data should be returned from the iterator

param error_state:

The pipeline threads’ global Error State object

abstractmethod __next__() → QueueItemOutT | ExtractSignal

Override to return the next batch item from the iterator

Returns:

Batch object for pipeline processing, or a final media object when exiting the pipeline.

Return type:

ExtractBatch or FrameFaces or ExtractSignal

Parameters:

• queue (Queue[QueueItemInT | ExtractSignal])

• name (str)

• plugin_type (T.Literal['detect', 'align', 'mask', 'identity', 'file'])

• batch_size (int)

• error_state (ErrorState)

class lib.infer.iterator.InboundIterator(queue: Queue[QueueItemInT | ExtractSignal], name: str, plugin_type: T.Literal['detect', 'align', 'mask', 'identity', 'file'], batch_size: int, error_state: ErrorState)

An iterator that processes ExtractBatch data from a previous plugin and configures it as an input for the current plugin.

An Inbound iterator assumes that the plugin’s batch size are the number of faces (not frames) that it can process at one time. Detect plugins are the only plugins that work with frames rather than faces, but these will always be the input to the pipeline, so will use an InputIterator not an InboundIterator

Parameters:

• queue (Queue[QueueItemInT | ExtractSignal]) – The outbound queue from the previous plugin

• name (str) – The plugin name and process calling this iterator

• plugin_type (T.Literal['detect', 'align', 'mask', 'identity', 'file']) – The type of extractor plugin that this iterator is serving

• batch_size (int) – The batch size that data should be returned from the iterator

• error_state (ErrorState)

__next__() → ExtractBatch | ExtractSignal

Get the next batch of data from the iterator. Depending on the plugin type calling this iterator, a batch object will be returned for the given batch size of frames (for detect plugins) or faces (for all other plugins)

Returns:

A new ExtractBatch object containing the batch to process through the plugin or an ExtractSignal

Return type:

ExtractBatch or ExtractSignal

Raises:

StopIteration – When the input is exhausted

class lib.infer.iterator.InputIterator(queue: Queue[QueueItemInT | ExtractSignal], name: str, plugin_type: T.Literal['detect', 'align', 'mask', 'identity', 'file'], batch_size: int, error_state: ErrorState)

An iterator that processes FrameFaces data that is input to a plugin pipeline to create ExtractBatch objects at the correct batch size for processing through the pipeline’s first plugin

Parameters:

• queue (Queue[QueueItemInT | ExtractSignal]) – The inbound queue to the plugin pipeline

• name (str) – The plugin name and process calling this iterator

• plugin_type (T.Literal['detect', 'align', 'mask', 'identity', 'file']) – The type of extractor plugin that this iterator is serving

• batch_size (int) – The batch size that data should be returned from the iterator

• error_state (ErrorState)

__next__() → ExtractBatch | ExtractSignal

Get the next batch of data from the iterator. Depending on the plugin type calling this iterator, a batch object will be returned for the given batch size of frames (for detect plugins) or faces (for all other plugins)

Returns:

A new Batch object containing the batch to process through the plugin or a signal

Return type:

ExtractBatch or ExtractSignal

Raises:

StopIteration – When the input is exhausted

class lib.infer.iterator.InterimIterator(queue: Queue[QueueItemInT | ExtractSignal], name: str, plugin_type: T.Literal['detect', 'align', 'mask', 'identity', 'file'], batch_size: int, error_state: ErrorState)

An iterator that simply collects interim ExtractBatch objects from the given queue and yields them

Parameters:

• queue (Queue[QueueItemInT | ExtractSignal]) – The inbound queue to the plugin

• name (str) – The plugin name and process calling this iterator

• plugin_type (T.Literal['detect', 'align', 'mask', 'identity', 'file']) – The type of extractor plugin that this iterator is serving

• batch_size (int) – The batch size that data should be returned from the iterator

• error_state (ErrorState)

__next__() → ExtractBatch | ExtractSignal

Get the next batch of data from the iterator

Returns:

The next available ExtractBatch object to process through the plugin or an ExtractSignal

Return type:

ExtractBatch or ExtractSignal

Raises:

StopIteration – When the input is exhausted

class lib.infer.iterator.OutputIterator(queue: Queue[QueueItemInT | ExtractSignal], name: str, plugin_type: T.Literal['detect', 'align', 'mask', 'identity', 'file'], batch_size: int, error_state: ErrorState)

Handles parsing incoming ExtractBatch objects into FrameFaces objects and yielding one frame at a time from the pipeline

Parameters:

• queue (Queue[QueueItemInT | ExtractSignal]) – The output queue from the plugin runner

• name (str) – The plugin name and process calling this iterator

• plugin_type (T.Literal['detect', 'align', 'mask', 'identity', 'file']) – The type of extractor plugin that this iterator is serving

• batch_size (int) – The batch size that data should be returned from the iterator

• error_state (ErrorState)

__next__() → FrameFaces

Get the next batch of data from the iterator

Return type:

A FrameFaces object for a single frame

Raises:

StopIteration – When the input is exhausted

Classes

ExtractIterator(queue, name, plugin_type, ...)	Base class for iterators within Faceswap's extract pipeline
InboundIterator(queue, name, plugin_type, ...)	An iterator that processes ExtractBatch data from a previous plugin and configures it as an input for the current plugin.
InputIterator(queue, name, plugin_type, ...)	An iterator that processes FrameFaces data that is input to a plugin pipeline to create ExtractBatch objects at the correct batch size for processing through the pipeline's first plugin
InterimIterator(queue, name, plugin_type, ...)	An iterator that simply collects interim ExtractBatch objects from the given queue and yields them
OutputIterator(queue, name, plugin_type, ...)	Handles parsing incoming ExtractBatch objects into FrameFaces objects and yielding one frame at a time from the pipeline

Variables

QueueItemInT	Type variable.
QueueItemOutT	Type variable.

Class Inheritance Diagram

lib.infer.mask Module

Handles face masking plugins and runners

class lib.infer.mask.Mask(plugin: str, compile_model: bool = False, config_file: str | None = None)

Responsible for running Masking plugins within the extract pipeline

Parameters:

• plugin (str) – The plugin that this runner is to use

• compile_model (bool) – True to compile any PyTorch models

• config_file (str | None) – Full path to a custom config file to load. None for default config

post_process(batch: ExtractBatch) → None

Perform mask post processing.

Obtains the final output from the mask plugins and masks any part of the face patch that goes out of bounds

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for post-processing

Return type:

None

pre_process(batch: ExtractBatch) → None

Obtain the aligned face images at the requested size, centering and image format. Perform any plugin specific pre-processing

Parameters:

batch (ExtractBatch) – The incoming ExtractBatch to use for pre-processing

Return type:

None

Classes

Mask(plugin[, compile_model, config_file])

Responsible for running Masking plugins within the extract pipeline

Class Inheritance Diagram

lib.infer.objects Module

Objects used for extraction plugins, runners and pipeline

class lib.infer.objects.ExtractBatch(filenames: list[str] = <factory>, images: list[np.ndarray] = <factory>, sources: list[str | None] = <factory>, is_aligned: bool = False, frame_sizes: list[tuple[int, int]] | None=None, frame_metadata: list[PNGSource] | None = None, passthrough: bool = False)

Dataclass for holding a batch flowing through Extraction plugins.

The batch size for post Detector plugins is not the same as the overall batch size. An image may contain 0 or more detected faces, and these need to be split and recombined to be able to utilize a plugin’s internal batch size.

Parameters:

• filenames (list[str]) – The original frame filenames for the batch

• images (list[np.ndarray]) – The original frames

• sources (list[str | None]) – The full path to the source folder or video file. Default: [] (Not provided)

• is_aligned (bool) – True if images contains aligned faces. False if it contains full frames. Default: False

• frame_sizes (list[tuple[int, int]] | None) – The original frame (height, width) dimensions that contained the aligned images when images are aligned faces. Default: None

• frame_metadata (list[PNGSource] | None) – The original frame meta data when aligned faces is True otherwise None

• passthrough (bool) – True` if the contents of this item are meant to pass straight through the extraction pipeline for immediate return

aligned: ExtractBatchAligned = <dataclasses._MISSING_TYPE object>

Holds the face landmarks found for this batch any any aligned data

append(batch: ExtractBatch) → None

Append the data from the given batch object to this batch object

Parameters:

batch (ExtractBatch) – The object containing data to be appended to this object

Return type:

None

apply_mask(mask: ndarray[tuple[Any, ...], dtype[bool]]) → None

Apply a boolean mask to the batch object. True values are kept, False values are discarded

Parameters:

mask (ndarray[tuple[Any, ...], dtype[bool]]) – The boolean mask to apply to the object. Must be of size (num_boxes, )

Return type:

None

bboxes: npt.NDArray[np.int32] = <dataclasses._MISSING_TYPE object>

The bounding boxes found for this batch

data: np.ndarray = <dataclasses._MISSING_TYPE object>

The data for this batch that has been populated by a processing step for ingestion by the next processing step. Internally populated. Cleared at the end of each plugin

filenames: list[str] = <dataclasses._MISSING_TYPE object>

The original frame filenames

frame_ids: npt.NDArray[np.int32] = <dataclasses._MISSING_TYPE object>

A mapping of each box to which frame they came from

frame_metadata: list[PNGSource] | None = None

The original frame metadata when aligned faces is True otherwise None

frame_sizes: list[tuple[int, int]] | None = None

The original frame (heights, widths) when the images are aligned faces

from_detected_faces(faces: list[DetectedFace]) → None

Populate an ExtractBatch with the contents of a DetectedFace object.

Parameters:

faces (list[DetectedFace]) – The DetectedFace objects to populate this batch

Raises:

ValueError – If attempting to add detected faces without pre-populating filename and image or if bounding boxes pre-exist or if more than one frame is held in this batch

Return type:

None

classmethod from_frame_faces(media: FrameFaces) → ExtractBatch

Populate a new ExtractBatch with the contents of an FrameFaces object.

Parameters:

media (FrameFaces) – The FrameFaces to populate this batch from

Return type:

A new ExtractBatch object populated from the given FrameFaces object

identities: dict[str, npt.NDArray[np.float32]] = <dataclasses._MISSING_TYPE object>

The identity matrices for face recognition found for this batch

images: list[np.ndarray] = <dataclasses._MISSING_TYPE object>

The original frames

is_aligned: bool = False

True if images contains aligned faces. False for full frames

property landmark_type: LandmarkType | None

The landmark type found for this batch or None if not populated

property landmarks: ndarray[tuple[Any, ...], dtype[float32]] | None

The face landmarks found for this batch in frame space or None if not populated

property lengths: ndarray[tuple[Any, ...], dtype[int32]]

The number of bboxes that belong to each frame

masks: dict[str, ExtractBatchMask] = <dataclasses._MISSING_TYPE object>

The masks for this batch

matrices: npt.NDArray[np.float32] = <dataclasses._MISSING_TYPE object>

Transformation matrices for taking points from model input space to frame space. Cleared at the end of each plugin

passthrough: bool = False

Whether this item should pass straight through the pipeline for immediate return

sources: list[str | None] = <dataclasses._MISSING_TYPE object>

The full paths to the source folder or video file. None if not provided

class lib.infer.objects.ExtractBatchAligned(landmarks: ndarray[tuple[Any, ...], dtype[float32]] | None = None, landmark_type: LandmarkType | None = None)

Dataclass for working with batches of aligned images

Parameters:

• landmarks (numpy.ndarray[tuple[Any, ...], numpy.dtype[numpy.float32]] | None) – The face landmarks found for this batch in frame space or None if not available. Default: None (to be populated later)

• landmark_type (lib.align.constants.LandmarkType | None) – The type of landmarks that the batch holds or None if not available. Default: None (to be populated later)

append(batch: ExtractBatchAligned) → None

Append the data from the given batch object to this batch object

Parameters:

batch (ExtractBatchAligned) – The object containing data to be appended to this object

Return type:

None

apply_mask(mask: ndarray[tuple[Any, ...], dtype[bool]]) → None

Apply a boolean mask to the batch object. True values are kept, False values are discarded

Parameters:

mask (ndarray[tuple[Any, ...], dtype[bool]]) – The boolean mask to apply to the object. Must be of size (landmarks, )

Return type:

None

landmark_type: LandmarkType | None = None

The type of landmarks that the batch holds

landmarks: ndarray[tuple[Any, ...], dtype[float32]] | None = None

The face landmarks found for this batch in frame space or None if not populated

property landmarks_68: ndarray[tuple[Any, ...], dtype[float32]]

The stored landmarks as 68 point landmarks if supported, or original landmarks if not ( 4 point ROI landmarks)

property landmarks_normalized: ndarray[tuple[Any, ...], dtype[float32]]

The normalized, aligned 68 point landmarks

property matrices: ndarray[tuple[Any, ...], dtype[float32]]

The face alignment matrices to transform from frame space to normalized (0, 1) space

property matrices_face: ndarray[tuple[Any, ...], dtype[float32]]

The alignment matrices to transform from normalized legacy space (0, 1) to normalized face space

property matrices_head: ndarray[tuple[Any, ...], dtype[float32]]

The alignment matrices to transform from normalized legacy space (0, 1) to normalized head space

property offsets_face: ndarray[tuple[Any, ...], dtype[float32]]

The (N, x, y) offsets required to shift from normalized (legacy) centering to face centering

property offsets_head: ndarray[tuple[Any, ...], dtype[float32]]

The (N, x, y) offsets required to shift from normalized (legacy) centering to head centering

property offsets_legacy: ndarray[tuple[Any, ...], dtype[float32]]

The (N, x, y) offsets for normalized (legacy) centering. This is always (0, 0) for all items in the batch

property rotation: ndarray[tuple[Any, ...], dtype[float32]]

The estimated (N, 3, 1) rotation vectors

property translation: ndarray[tuple[Any, ...], dtype[float32]]

The estimated (N, 3, 1) translation vectors

class lib.infer.objects.ExtractBatchMask(centering: CenteringType, matrices: npt.NDArray[np.float32], storage_size: int = 0, masks: npt.NDArray[np.uint8] = <factory>)

Dataclass for holding information about masks produced by the extraction pipeline

Parameters:

• centering (CenteringType) – The centering type of the masks

• matrices (npt.NDArray[np.float32]) – The normalized matrices required to take the masks from (0, 1) to full frame

• storage_size (int) – The pixel size to store the mask at in the alignments file. Default: 0 (must be populated later)

• masks (npt.NDArray[np.uint8]) – The masks for this batch. Default: empty array (must be populated later)

append(mask_batch: ExtractBatchMask) → None

Append the given mask batch object to this batch mask object

Parameters:

mask_batch (ExtractBatchMask) – The object containing data to be appended to this object

Return type:

None

apply_mask(mask: ndarray[tuple[Any, ...], dtype[bool]]) → None

Apply a boolean mask to the batch object. True values are kept, False values are discarded

Parameters:

mask (ndarray[tuple[Any, ...], dtype[bool]]) – The boolean mask to apply to the object. Must be of size (num_masks, )

Return type:

None

centering: CenteringType = <dataclasses._MISSING_TYPE object>

The centering type of the masks

masks: npt.NDArray[np.uint8] = <dataclasses._MISSING_TYPE object>

The masks for this batch

matrices: npt.NDArray[np.float32] = <dataclasses._MISSING_TYPE object>

The normalized matrices required to take the masks from (0, 1) to full frame

storage_size: int = 0

The pixel size to store the mask at in the alignments file

class lib.infer.objects.ExtractSignal(*values)

Signals to send to the extraction pipeline

FLUSH = 1

Flush all queued items

SHUTDOWN = 2

Flush all queued items and shutdown

class lib.infer.objects.FrameFaces(filename: str, image: npt.NDArray[np.uint8], bboxes: npt.NDArray[np.int32] | None = None, landmarks: npt.NDArray[np.float32] | None = None, identities: dict[str, npt.NDArray[np.float32]] | None = None, masks: dict[str, ExtractBatchMask] | None = None, source: str | None = None, is_aligned: bool = False, frame_metadata: PNGSource | None = None, passthrough: bool = False)

An object for holding information about faces in a single frame

Parameters:

• filename (str) – The original file name of the frame

• image (npt.NDArray[np.uint8]) – The original frame or a faceswap aligned face image

• bboxes (npt.NDArray[np.int32] | None) – The (N, Left, Top, Right, Bottom) bounding boxes of the faces in the frame. Default: None (Not provided)

• landmarks (npt.NDArray[np.float32] | None) – The (N, M, 2) landmarks for each face in the frame, in frame space. Default: None (Not provided)

• identities (dict[str, npt.NDArray[np.float32]] | None) – The identity matrices for each face in the frame. Default: None (Not provided)

• masks (dict[str, ExtractBatchMask] | None) – The mask objects for each face in the frame. Default: None (Not provided)

• source (str | None) – The full path to the source folder or video file. Default: None (Not provided)

• is_aligned (bool) – True if the image is an aligned faceswap image otherwise False. Used for face filtering with vggface2. Aligned faceswap images will automatically skip detection, alignment and masking. Default: False

• frame_metadata (PNGSource | None) – The frame metadata for aligned images. None if the image is not an aligned image

• passthrough (bool) – True if this item is meant to be passed straight through the extraction pipeline with no batching or caching. for immediate return. Default: False

aligned

Holds the face landmarks found for this batch any any aligned data

append(batch: FrameFaces) → None

Append the data from the given batch object to this batch object

Parameters:

batch (FrameFaces) – The object containing data to be appended to this object

Return type:

None

bboxes

The (N, Left, Top, Right, Bottom) bounding boxes of the faces in the frame

property detected_faces: list[DetectedFace]

A list of DetectedFace objects within the image

filename

The original file name of the original frame

frame_metadata: PNGSource | None

The frame metadata that has been added from an aligned image. None if metadata has not been added

identities

The identity matrices for each face in the frame

image

The original frame or a faceswap aligned face image

image_shape

The shape of the original frame

property image_size: tuple[int, int]

The (height, width) of the stored image

is_aligned

True if image is an aligned faceswap image otherwise False

property landmarks: ndarray[tuple[Any, ...], dtype[float32]] | None

The (N, M, 2) landmarks for each face in the frame, in frame space

masks

The mask objects for each face in the frame

passthrough

True if the contents of this item are meant to pass straight through the extraction pipeline for immediate return

remove_image() → None

Delete the image and reset image to None.

Return type:

None

source

The full path to the source folder or video file or None if not provided

lib.infer.objects.frame_faces_to_alignment(media: FrameFaces) → list[PNGAlignments]

Convert the faces in a FrameFaces object into a list of dictionaries (one for each face) for serializing into image headers and alignments files

Parameters:

media (FrameFaces)

Return type:

list[PNGAlignments]

Functions

frame_faces_to_alignment(media)

Convert the faces in a FrameFaces object into a list of dictionaries (one for each face) for serializing into image headers and alignments files

Classes

ExtractBatch(filenames, images, sources, ...)	Dataclass for holding a batch flowing through Extraction plugins.
ExtractBatchAligned([landmarks, landmark_type])	Dataclass for working with batches of aligned images
ExtractBatchMask(centering, matrices, ...)	Dataclass for holding information about masks produced by the extraction pipeline
ExtractSignal(*values)	Signals to send to the extraction pipeline
FrameFaces(filename, image[, bboxes, ...])	An object for holding information about faces in a single frame

Class Inheritance Diagram

lib.infer.plugin_utils Module

General utility functions for Faceswap inference

lib.infer.plugin_utils.compile_models(plugin: ExtractPlugin, modules: list[torch.nn.Module]) → None

Compile any Torch modules in the plugin’s model attribute

Parameters:

• plugin (ExtractPlugin) – The plugin containing Torch modules to be compiled

• modules (list[torch.nn.Module]) – The list of Torch modules contained within the plugin’s model attribute

Return type:

None

lib.infer.plugin_utils.get_torch_modules(obj: Any, mod: str | None = None, seen: set[int] | None = None, results: list[Module] | None = None) → list[Module]

Recursively search a plugin’s model attribute to find any parent :class:`torch.nn.Module`s

Parameters:

• obj (Any) – The object to check if it is a torch Module. This should be a plugin’s model attribute

• mod (str | None) – The module that the parent model class belongs to. Default: None (Collected from the first object entered into the recursive function)

• seen (set[int] | None) – A set of seen object IDs to prevent self-recursion. Default: None (Created when the first object enters the recursive function)

• results (list[Module] | None) – List of discovered torch modules. Default: None (Created when the first object enters the recursive function)

Return type:

The list of discovered torch Modules

lib.infer.plugin_utils.random_input_from_plugin(plugin: ExtractPlugin, batch_size: int, channels_last: bool) → np.ndarray

Obtain a random input array from a plugin’s information for the given batch size

Parameters:

• plugin (ExtractPlugin) – The plugin to obtain the input array for

• batch_size (int) – The batch size for the input array

• channels_last (bool) – True if the data should be formatted channels last

Return type:

A random input array in the correct format for the given plugin at the given batch size

lib.infer.plugin_utils.warmup_plugin(plugin: ExtractPlugin, batch_size: int, channels_last: bool | None = None) → bool | None

Warm up a plugin that contains torch modules. If channels_last is None then attempt to send a channels first batch through. If it fails, send a channels last batch through

Parameters:

• plugin (ExtractPlugin) – The plugin to warmup

• batch_size (int) – The batch size to put through the model

• channels_last (bool | None) – The expected channel order of the plugin or None to detect

Returns:

True if the plugin is detected as channels last, False for channels first, None for could not be detected

Return type:

bool

Functions

compile_models(plugin, modules)	Compile any Torch modules in the plugin's model attribute
get_torch_modules(obj[, mod, seen, results])	Recursively search a plugin's model attribute to find any parent :class:`torch.nn.Module`s
random_input_from_plugin(plugin, batch_size, ...)	Obtain a random input array from a plugin's information for the given batch size
warmup_plugin(plugin, batch_size[, ...])	Warm up a plugin that contains torch modules.

lib.infer.profile Module

GPU profiling for throughput optimization

class lib.infer.profile.DataTracker(size: dataclasses.InitVar[int], max_vram: dataclasses.InitVar[float], face_scaling: int, has_detector: bool)

Stores data from the benchmarking process

Parameters:

• size (InitVar[int]) – The number of plugins that data is being tracked for

• face_scaling (int) – The amount of scaling to apply to downstream non-detection plugins

• has_detector (bool) – True if the first plugin in the pipeline is a detector

• max_vram (InitVar[float]) – The maximum amount of total VRAM to allow Cuda to reserve when profiling

add_iterations_row() → None

Add a new row to the iterations list

Return type:

None

add_next_batch_sizes() → None

Add the next batch size configuration to the batch size array based on the output from the last test

Return type:

None

property batch_sizes: npt.NDArray[np.int64]

All batch size combinations that did not OOM

collect_vram() → None

Store the currently allocated and reserved Cuda VRAM stats

Return type:

None

combos_exhausted: bool = False

True if we have run out of possible combinations to attempt

get_samples(index: int | None = None, adjusted: bool = False) → npt.NDArray[np.float64]

Obtain the number of sample processed by each plugin for a certain valid batch size combination

Parameters:

• index (int | None) – The testing index to obtain the samples for or None for all tests

• adjusted (bool) – True to obtain results adjusted for any non-detector scaling. Default: False

Return type:

The number of samples processed by each plugin

get_samples_stats(method: T.Literal['mean', 'min'], index: int | None = None, adjusted: bool = False) → npt.NDArray[np.float64]

Obtain the average or minimum samples processed for all plugins for a certain batch size combination

Parameters:

• method (T.Literal['mean', 'min']) – mean to obtain the mean number of samples for all plugins. min to obtain the minimum number of samples processed by a plugin

• index (int | None) – The testing index to obtain the average samples for or None for all tests. Default: None

• adjusted (bool) – True to obtain results adjusted for any non-detector scaling. Default: False

Return type:

The average number of samples processed by all plugins

property has_oom: bool

True if the last iteration hit an OOM or fell outside our max VRAM threshold

update_iterations(iterations: int, matrix_id: int) → None

Update the iteration count from a plugin runner in a thread-safe way

Parameters:

• iterations (int) – The iteration count for the plugin

• matrix_id (int) – The column id that belongs to the plugin

Return type:

None

vram: list[tuple[int, int]] = <dataclasses._MISSING_TYPE object>

list of (max allocated, max reserved) VRAM for each testing phase

vram_limit: float = <dataclasses._MISSING_TYPE object>

The limit that Cuda reserved memory must remain within

class lib.infer.profile.Events(ready: list[Event])

Holds thread events for communicating between main thread and plugins during benchmarking

Parameters:

ready (list[threading.Event]) – List of events for each plugin in the pipeline to be tested

set_ready(index: int)

Set the ready event for the given index

Parameters:

index (int) – The index of the ready event to set

wait_ready() → None

Wait for all ready events to set their ready flag and clear the flag

Return type:

None

class lib.infer.profile.ModelProfile(plugin: ExtractPlugin, max_batch_size: int = 128, channels_last: bool = False, run_time: int = 10)

Benchmark a single PyTorch GPU plugin for inference

Parameters:

• plugin (ExtractPlugin) – The plugin to benchmark for inference

• max_batch_size (int) – The maximum batch size to benchmark to

• channels_last (bool) – True if the input to the plugin is channels last

• run_time (int) – The amount of time, in seconds, to benchmark the plugin at each batch size

__call__() → None

Runs benchmarking through the plugin, stores the data and outputs stats

Return type:

None

channels_last

True if the plugin expects channels last input

property run_time: int

The amount of time, in seconds, that benchmarks were ran per batch

class lib.infer.profile.Output(plugin_names: list[str], data: DataTracker, run_time: int)

Handles outputting of information at each test step

Parameters:

• plugin_names (list[str]) – The list of plugin names in the order that they are executed

• data (DataTracker) – The DataTracker object that collects stats

• run_time (int) – The amount of time, in seconds, that each test is run

__call__()

Output the latest test stats

class lib.infer.profile.PipelineProfile(plugins: list[ExtractPlugin], error_state: ErrorState, channels_last: list[bool], warmup_time: int, run_time: int, has_detector: bool, face_scaling: int = 2, max_vram: int = 90)

Benchmark multiple PyTorch GPU plugins running simultaneously for inference

Parameters:

• plugins (list[ExtractPlugin]) – The plugins to benchmark for inference

• error_state (ErrorState) – The global FSThread error state object for the pipeline

• channels_last (list[bool]) – List indicating whether each model is channels first or last

• warmup_time (int) – The amount of time, in seconds, to warmup the plugin at each batch size

• run_time (int) – The amount of time, in seconds, to benchmark the plugin at each batch size

• has_detector (bool) – True if the first plugin in the pipeline is a detector

• face_scaling (int) – The amount of scaling to apply to downstream plugins (ie estimate of average number of faces per frame). Default: 2

• max_vram (int) – The maximum percentage of total VRAM to allow Cuda to reserve when profiling, Default: 90

__call__() → None

Runs benchmarking through all plugins concurrently, store the data and output stats

Return type:

None

class lib.infer.profile.Profiler(runner: ExtractRunner)

Profiles plugins within a pipeline

Parameters:

runner (ExtractRunner) – The output runner from an extract pipeline that is to be profiled

__call__() → None

Call the profiler

Return type:

None

Classes

DataTracker(size, max_vram, face_scaling, ...)	Stores data from the benchmarking process
Events(ready)	Holds thread events for communicating between main thread and plugins during benchmarking
ModelProfile(plugin[, max_batch_size, ...])	Benchmark a single PyTorch GPU plugin for inference
Output(plugin_names, data, run_time)	Handles outputting of information at each test step
PipelineProfile(plugins, error_state, ...[, ...])	Benchmark multiple PyTorch GPU plugins running simultaneously for inference
Profiler(runner)	Profiles plugins within a pipeline

Class Inheritance Diagram

lib.infer.runner Module

Handles extract plugins and runners

class lib.infer.runner.ExtractRunner(handler: HandlerT)

Runs an extract plugin

Parameters:

handler (HandlerT) – The plugin handler that this runner will execute

__call__(input_runner: ExtractRunner | None, profile: bool) → None

Build and start the plugin runner

Parameters:

• input_runner (ExtractRunner | None) – The input plugin that feeds this plugin or None if data is to be fed through the runner’s put method.

• profile (bool) – True if the runner is to be profiled, indicating that threads will not be started

Raises:

ValueError – If the input runner has not been called and assigned a UUID or if this runner has already been called

Return type:

None

__next__() → FrameFaces

Obtain the next item from the plugin’s output

Return type:

The media object with populated detected faces for a frame

flush() → None

Flush all data currently within the pipeline

Return type:

None

property handler: HandlerT

The plugin handler that this runner is executing

property out_queue: Queue[ExtractBatch]

The output queue from this plugin runner

put(filename: str, image: ndarray[tuple[Any, ...], dtype[uint8]], detected_faces: list[DetectedFace] | None = None, source: str | None = None, is_aligned: bool = False, frame_metadata: PNGSource | None = None, passthrough: Literal[False] = False) → None

put(filename: str, image: ndarray[tuple[Any, ...], dtype[uint8]], detected_faces: list[DetectedFace] | None = None, source: str | None = None, is_aligned: bool = False, frame_metadata: PNGSource | None = None, *, passthrough: Literal[True]) → FrameFaces

Put a frame into the pipeline.

Note

When a pipeline is built using the __call__ method, this method will always put items into the first plugin in the pipeline

Parameters:

• filename (str) – The filename of the frame

• image (npt.NDArray[np.uint8]) – The loaded frame as UINT8 BGR array

• detected_faces (list[DetectedFace] | None) – The detected face objects for the frame. None if not any. Default: None

• source (str | None) – The full path to the source folder or video file. Default: None (Not provided)

• is_aligned (bool) – True if the image being passed into the pipeline is an aligned faceswap face. Default: False

• frame_metadata (PNGSource | None) – If the image is aligned then the original frame metadata can be added here. Some plugins (eg: mask) require this to be populated for aligned inputs. Default: None

• passthrough (bool) – True if this item is meant to be passed straight through the extraction pipeline with no caching, for immediate return. Default: False

Return type:

If passthrough is True returns the output FrameFaces object, otherwise None

put_direct(filename: str, image: npt.NDArray[np.uint8], detected_faces: list[DetectedFace], is_aligned: bool = False, frame_size: tuple[int, int] | None = None) → ExtractBatch

Put an item directly into this runner’s plugin and return the result

Parameters:

• filename (str) – The filename of the frame

• image (npt.NDArray[np.uint8]) – The loaded frame as UINT8 BGR array

• detected_faces (list[DetectedFace]) – The detected face objects for the frame

• is_aligned (bool) – True if the image being passed into the pipeline is an aligned faceswap face. Default: False

• frame_size (tuple[int, int] | None) – The (height, width) size of the original frame if passing in an aligned image

Raises:

ValueError – If attempting to put an ExtractBatch object to the first runner in the pipeline or if providing an aligned image with insufficient data

Returns:

The output from this plugin for the given input

Return type:

ExtractBatch

put_media(media: FrameFaces) → None | FrameFaces

Put a frame into the pipeline that is within a FrameFaces object.

Note

When a pipeline is built using the __call__ method, this method will always put items into the first plugin in the pipeline

Parameters:

media (FrameFaces) – The FrameFaces object to put into the pipeline

Returns:

• If the FrameFaces’s passthrough is True returns the output FrameFaces object,

• otherwise None

Return type:

None | FrameFaces

register_external_error_state(state: ErrorState) → None

Register an external error state object.

If we are not running any threads (specifically, file handler), the pipeline can hang the calling thread. The error state from the calling thread can be populated here. This can only be called if no threads have been registered for the runner

Parameters:

state (ErrorState) – The ErrorState object to register

Return type:

None

start() → None

Start the threads. Callback for when the profiler has finished executing

Return type:

None

stop() → None

Indicate to the runner that there is no more data to be ingested

Return type:

None

property uuid: str

Unique identifier for plugin ordering and multi-plugin tracking

class lib.infer.runner.PluginThreads(name: str)

Handles the holding of threads that will run a plugin’s various subprocesses.

Parameters:

name (str) – The name of the plugin that the threads are being created for

property enabled: list[str]

The thread names that have been registered within this group

property error_state: ErrorState

The global FSThread error state object

is_alive() → bool

Test if any thread is alive

Return type:

True if any thread is alive otherwise False

join() → None

Join all of the plugin’s threads

Return type:

None

register_external_error_state(state: ErrorState) → None

Register an external error state object.

If we are not running any threads (specifically, file handler), the pipeline can hang the calling thread. The error state from the calling thread can be populated here. This can only be called if no threads have been registered

Parameters:

state (ErrorState) – The ErrorState object to register

Raises:

RuntimeError – If an ErrorState object is registered when this object already contains threads

Return type:

None

register_thread(name: str, target: Callable[[Literal['pre_process', 'process', 'post_process']], None]) → None

Register a thread

Parameters:

• name (str) – The name of the plugin handler’s processor that is running in the thread

• target (Callable[[Literal['pre_process', 'process', 'post_process']], None]) – The function to run within the thread

Return type:

None

start() → None

Start the plugin’s threads

Return type:

None

lib.infer.runner.get_pipeline(runner: ExtractRunner) → list[ExtractRunner]

Obtain a list of runners in order of input to output of the extraction chain that the given runner belongs to

Parameters:

runner (ExtractRunner) – The initialized runner to obtain the full chain for

Return type:

The ordered list of runners if the inference chain that the given runner belongs to

Functions

get_pipeline(runner)

Obtain a list of runners in order of input to output of the extraction chain that the given runner belongs to

Classes

ExtractRunner(handler)	Runs an extract plugin
PluginThreads(name)	Handles the holding of threads that will run a plugin's various subprocesses.

Variables

HandlerT

Type variable.

Class Inheritance Diagram