Skip to content

platforms

Classes

fastvideo.platforms.Platform

Functions

fastvideo.platforms.Platform.get_attn_backend_cls classmethod 
get_attn_backend_cls(selected_backend: AttentionBackendEnum | None, head_size: int, dtype: dtype) -> str

Get the attention backend class of a device.

Source code in fastvideo/platforms/interface.py 
115
116
117
118
119
@classmethod
def get_attn_backend_cls(cls, selected_backend: AttentionBackendEnum | None,
                         head_size: int, dtype: torch.dtype) -> str:
    """Get the attention backend class of a device."""
    return ""
fastvideo.platforms.Platform.get_cpu_architecture classmethod 
get_cpu_architecture() -> CpuArchEnum

Get the CPU architecture of the current platform.

Source code in fastvideo/platforms/interface.py 
244
245
246
247
@classmethod
def get_cpu_architecture(cls) -> CpuArchEnum:
    """Get the CPU architecture of the current platform."""
    return CpuArchEnum.UNSPECIFIED
fastvideo.platforms.Platform.get_current_memory_usage classmethod 
get_current_memory_usage(device: Device | None = None) -> float

Return the memory usage in bytes.

Source code in fastvideo/platforms/interface.py 
228
229
230
231
232
233
234
235
@classmethod
def get_current_memory_usage(cls,
                             device: torch.types.Device | None = None
                             ) -> float:
    """
    Return the memory usage in bytes.
    """
    raise NotImplementedError
fastvideo.platforms.Platform.get_device_capability classmethod 
get_device_capability(device_id: int = 0) -> DeviceCapability | None

Stateless version of :func:torch.cuda.get_device_capability.

Source code in fastvideo/platforms/interface.py 
121
122
123
124
125
126
127
@classmethod
def get_device_capability(
    cls,
    device_id: int = 0,
) -> DeviceCapability | None:
    """Stateless version of :func:`torch.cuda.get_device_capability`."""
    return None
fastvideo.platforms.Platform.get_device_communicator_cls classmethod 
get_device_communicator_cls() -> str

Get device specific communicator class for distributed communication.

Source code in fastvideo/platforms/interface.py 
237
238
239
240
241
242
@classmethod
def get_device_communicator_cls(cls) -> str:
    """
    Get device specific communicator class for distributed communication.
    """
    return "fastvideo.distributed.device_communicators.base_device_communicator.DeviceCommunicatorBase"  # noqa
fastvideo.platforms.Platform.get_device_name classmethod 
get_device_name(device_id: int = 0) -> str

Get the name of a device.

Source code in fastvideo/platforms/interface.py 
152
153
154
155
@classmethod
def get_device_name(cls, device_id: int = 0) -> str:
    """Get the name of a device."""
    raise NotImplementedError
fastvideo.platforms.Platform.get_device_total_memory classmethod 
get_device_total_memory(device_id: int = 0) -> int

Get the total memory of a device in bytes.

Source code in fastvideo/platforms/interface.py 
162
163
164
165
@classmethod
def get_device_total_memory(cls, device_id: int = 0) -> int:
    """Get the total memory of a device in bytes."""
    raise NotImplementedError
fastvideo.platforms.Platform.get_device_uuid classmethod 
get_device_uuid(device_id: int = 0) -> str

Get the uuid of a device, e.g. the PCI bus ID.

Source code in fastvideo/platforms/interface.py 
157
158
159
160
@classmethod
def get_device_uuid(cls, device_id: int = 0) -> str:
    """Get the uuid of a device, e.g. the PCI bus ID."""
    raise NotImplementedError
fastvideo.platforms.Platform.get_torch_device classmethod 
get_torch_device()

Check if the current platform supports torch device.

Source code in fastvideo/platforms/interface.py 
174
175
176
177
178
179
@classmethod
def get_torch_device(cls):
    """
    Check if the current platform supports torch device.
    """
    raise NotImplementedError
fastvideo.platforms.Platform.has_device_capability classmethod 
has_device_capability(capability: tuple[int, int] | int, device_id: int = 0) -> bool

Test whether this platform is compatible with a device capability.

The capability argument can either be:

  • A tuple (major, minor).
  • An integer <major><minor>. (See :meth:DeviceCapability.to_int)
Source code in fastvideo/platforms/interface.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@classmethod
def has_device_capability(
    cls,
    capability: tuple[int, int] | int,
    device_id: int = 0,
) -> bool:
    """
    Test whether this platform is compatible with a device capability.

    The ``capability`` argument can either be:

    - A tuple ``(major, minor)``.
    - An integer ``<major><minor>``. (See :meth:`DeviceCapability.to_int`)
    """
    current_capability = cls.get_device_capability(device_id=device_id)
    if current_capability is None:
        return False

    if isinstance(capability, tuple):
        return current_capability >= capability

    return current_capability.to_int() >= capability
fastvideo.platforms.Platform.inference_mode classmethod 
inference_mode()

A device-specific wrapper of torch.inference_mode.

This wrapper is recommended because some hardware backends such as TPU do not support torch.inference_mode. In such a case, they will fall back to torch.no_grad by overriding this method.

Source code in fastvideo/platforms/interface.py 
181
182
183
184
185
186
187
188
189
@classmethod
def inference_mode(cls):
    """A device-specific wrapper of `torch.inference_mode`.

    This wrapper is recommended because some hardware backends such as TPU
    do not support `torch.inference_mode`. In such a case, they will fall
    back to `torch.no_grad` by overriding this method.
    """
    return torch.inference_mode(mode=True)
fastvideo.platforms.Platform.is_async_output_supported classmethod 
is_async_output_supported(enforce_eager: bool | None) -> bool

Check if the current platform supports async output.

Source code in fastvideo/platforms/interface.py 
167
168
169
170
171
172
@classmethod
def is_async_output_supported(cls, enforce_eager: bool | None) -> bool:
    """
    Check if the current platform supports async output.
    """
    raise NotImplementedError
fastvideo.platforms.Platform.is_cuda_alike 
is_cuda_alike() -> bool

Stateless version of :func:torch.cuda.is_available.

Source code in fastvideo/platforms/interface.py 
105
106
107
def is_cuda_alike(self) -> bool:
    """Stateless version of :func:`torch.cuda.is_available`."""
    return self._enum in (PlatformEnum.CUDA, PlatformEnum.ROCM)
fastvideo.platforms.Platform.seed_everything classmethod 
seed_everything(seed: int | None = None) -> None

Set the seed of each random module. torch.manual_seed will set seed on all devices.

Loosely based on: https://github.com/Lightning-AI/pytorch-lightning/blob/2.4.0/src/lightning/fabric/utilities/seed.py#L20

Source code in fastvideo/platforms/interface.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
@classmethod
def seed_everything(cls, seed: int | None = None) -> None:
    """
    Set the seed of each random module.
    `torch.manual_seed` will set seed on all devices.

    Loosely based on: https://github.com/Lightning-AI/pytorch-lightning/blob/2.4.0/src/lightning/fabric/utilities/seed.py#L20
    """
    if seed is not None:
        random.seed(seed)
        np.random.seed(seed)
        torch.manual_seed(seed)
        torch.cuda.manual_seed_all(seed)
fastvideo.platforms.Platform.verify_model_arch classmethod 
verify_model_arch(model_arch: str) -> None

Verify whether the current platform supports the specified model architecture.

  • This will raise an Error or Warning based on the model support on the current platform.
  • By default all models are considered supported.
Source code in fastvideo/platforms/interface.py 
205
206
207
208
209
210
211
212
213
214
215
@classmethod
def verify_model_arch(cls, model_arch: str) -> None:
    """
    Verify whether the current platform supports the specified model
    architecture.

    - This will raise an Error or Warning based on the model support on
    the current platform.
    - By default all models are considered supported.
    """
    pass
fastvideo.platforms.Platform.verify_quantization classmethod 
verify_quantization(quant: str) -> None

Verify whether the quantization is supported by the current platform.

Source code in fastvideo/platforms/interface.py 
217
218
219
220
221
222
223
224
225
226
@classmethod
def verify_quantization(cls, quant: str) -> None:
    """
    Verify whether the quantization is supported by the current platform.
    """
    if cls.supported_quantization and \
        quant not in cls.supported_quantization:
        raise ValueError(
            f"{quant} quantization is currently not supported in "
            f"{cls.device_name}.")

Functions

fastvideo.platforms.cpu_platform_plugin 

cpu_platform_plugin() -> str | None

Detect if CPU platform should be used.

Source code in fastvideo/platforms/__init__.py 
90
91
92
93
def cpu_platform_plugin() -> str | None:
    """Detect if CPU platform should be used."""
    # CPU is always available as a fallback
    return "fastvideo.platforms.cpu.CpuPlatform"

fastvideo.platforms.mps_platform_plugin 

mps_platform_plugin() -> str | None

Detect if MPS (Metal Performance Shaders) is available on macOS.

Source code in fastvideo/platforms/__init__.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
def mps_platform_plugin() -> str | None:
    """Detect if MPS (Metal Performance Shaders) is available on macOS."""
    is_mps = False

    try:
        import torch
        if torch.backends.mps.is_available():
            is_mps = True
            logger.info("MPS (Metal Performance Shaders) is available")
    except Exception as e:
        logger.info("MPS detection failed: %s", e)

    return "fastvideo.platforms.mps.MpsPlatform" if is_mps else None

Modules

fastvideo.platforms.cpu

Classes

fastvideo.platforms.cpu.CpuPlatform

Bases: Platform

Functions
fastvideo.platforms.cpu.CpuPlatform.get_cpu_architecture classmethod 
get_cpu_architecture() -> CpuArchEnum

Get the CPU architecture.

Source code in fastvideo/platforms/cpu.py 
17
18
19
20
21
22
23
24
25
26
@classmethod
def get_cpu_architecture(cls) -> CpuArchEnum:
    """Get the CPU architecture."""
    machine = platform.machine().lower()
    if machine in ("x86_64", "amd64", "i386", "i686"):
        return CpuArchEnum.X86
    elif machine in ("arm64", "aarch64"):
        return CpuArchEnum.ARM
    else:
        return CpuArchEnum.UNSPECIFIED

fastvideo.platforms.cuda

Code inside this file can safely assume cuda platform, e.g. importing pynvml. However, it should not initialize cuda context.

Classes

fastvideo.platforms.cuda.CudaPlatformBase

Bases: Platform

Functions
fastvideo.platforms.cuda.CudaPlatformBase.get_torch_device classmethod 
get_torch_device()

Return torch.cuda

Source code in fastvideo/platforms/cuda.py 
111
112
113
114
115
116
@classmethod
def get_torch_device(cls):
    """
    Return torch.cuda
    """
    return torch.cuda
fastvideo.platforms.cuda.NvmlCudaPlatform

Bases: CudaPlatformBase

Functions
fastvideo.platforms.cuda.NvmlCudaPlatform.is_full_nvlink classmethod 
is_full_nvlink(physical_device_ids: list[int]) -> bool

query if the set of gpus are fully connected by nvlink (1 hop)

Source code in fastvideo/platforms/cuda.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
@classmethod
@with_nvml_context
def is_full_nvlink(cls, physical_device_ids: list[int]) -> bool:
    """
    query if the set of gpus are fully connected by nvlink (1 hop)
    """
    handles = [
        pynvml.nvmlDeviceGetHandleByIndex(i) for i in physical_device_ids
    ]
    for i, handle in enumerate(handles):
        for j, peer_handle in enumerate(handles):
            if i < j:
                try:
                    p2p_status = pynvml.nvmlDeviceGetP2PStatus(
                        handle,
                        peer_handle,
                        pynvml.NVML_P2P_CAPS_INDEX_NVLINK,
                    )
                    if p2p_status != pynvml.NVML_P2P_STATUS_OK:
                        return False
                except pynvml.NVMLError:
                    logger.exception(
                        "NVLink detection failed. This is normal if"
                        " your machine has no NVLink equipped.")
                    return False
    return True

Functions

fastvideo.platforms.interface

Classes

fastvideo.platforms.interface.DeviceCapability

Bases: NamedTuple

Functions
fastvideo.platforms.interface.DeviceCapability.to_int 
to_int() -> int

Express device capability as an integer <major><minor>.

It is assumed that the minor version is always a single digit.

Source code in fastvideo/platforms/interface.py 
49
50
51
52
53
54
55
56
def to_int(self) -> int:
    """
    Express device capability as an integer ``<major><minor>``.

    It is assumed that the minor version is always a single digit.
    """
    assert 0 <= self.minor < 10
    return self.major * 10 + self.minor
fastvideo.platforms.interface.Platform
Functions
fastvideo.platforms.interface.Platform.get_attn_backend_cls classmethod 
get_attn_backend_cls(selected_backend: AttentionBackendEnum | None, head_size: int, dtype: dtype) -> str

Get the attention backend class of a device.

Source code in fastvideo/platforms/interface.py 
115
116
117
118
119
@classmethod
def get_attn_backend_cls(cls, selected_backend: AttentionBackendEnum | None,
                         head_size: int, dtype: torch.dtype) -> str:
    """Get the attention backend class of a device."""
    return ""
fastvideo.platforms.interface.Platform.get_cpu_architecture classmethod 
get_cpu_architecture() -> CpuArchEnum

Get the CPU architecture of the current platform.

Source code in fastvideo/platforms/interface.py 
244
245
246
247
@classmethod
def get_cpu_architecture(cls) -> CpuArchEnum:
    """Get the CPU architecture of the current platform."""
    return CpuArchEnum.UNSPECIFIED
fastvideo.platforms.interface.Platform.get_current_memory_usage classmethod 
get_current_memory_usage(device: Device | None = None) -> float

Return the memory usage in bytes.

Source code in fastvideo/platforms/interface.py 
228
229
230
231
232
233
234
235
@classmethod
def get_current_memory_usage(cls,
                             device: torch.types.Device | None = None
                             ) -> float:
    """
    Return the memory usage in bytes.
    """
    raise NotImplementedError
fastvideo.platforms.interface.Platform.get_device_capability classmethod 
get_device_capability(device_id: int = 0) -> DeviceCapability | None

Stateless version of :func:torch.cuda.get_device_capability.

Source code in fastvideo/platforms/interface.py 
121
122
123
124
125
126
127
@classmethod
def get_device_capability(
    cls,
    device_id: int = 0,
) -> DeviceCapability | None:
    """Stateless version of :func:`torch.cuda.get_device_capability`."""
    return None
fastvideo.platforms.interface.Platform.get_device_communicator_cls classmethod 
get_device_communicator_cls() -> str

Get device specific communicator class for distributed communication.

Source code in fastvideo/platforms/interface.py 
237
238
239
240
241
242
@classmethod
def get_device_communicator_cls(cls) -> str:
    """
    Get device specific communicator class for distributed communication.
    """
    return "fastvideo.distributed.device_communicators.base_device_communicator.DeviceCommunicatorBase"  # noqa
fastvideo.platforms.interface.Platform.get_device_name classmethod 
get_device_name(device_id: int = 0) -> str

Get the name of a device.

Source code in fastvideo/platforms/interface.py 
152
153
154
155
@classmethod
def get_device_name(cls, device_id: int = 0) -> str:
    """Get the name of a device."""
    raise NotImplementedError
fastvideo.platforms.interface.Platform.get_device_total_memory classmethod 
get_device_total_memory(device_id: int = 0) -> int

Get the total memory of a device in bytes.

Source code in fastvideo/platforms/interface.py 
162
163
164
165
@classmethod
def get_device_total_memory(cls, device_id: int = 0) -> int:
    """Get the total memory of a device in bytes."""
    raise NotImplementedError
fastvideo.platforms.interface.Platform.get_device_uuid classmethod 
get_device_uuid(device_id: int = 0) -> str

Get the uuid of a device, e.g. the PCI bus ID.

Source code in fastvideo/platforms/interface.py 
157
158
159
160
@classmethod
def get_device_uuid(cls, device_id: int = 0) -> str:
    """Get the uuid of a device, e.g. the PCI bus ID."""
    raise NotImplementedError
fastvideo.platforms.interface.Platform.get_torch_device classmethod 
get_torch_device()

Check if the current platform supports torch device.

Source code in fastvideo/platforms/interface.py 
174
175
176
177
178
179
@classmethod
def get_torch_device(cls):
    """
    Check if the current platform supports torch device.
    """
    raise NotImplementedError
fastvideo.platforms.interface.Platform.has_device_capability classmethod 
has_device_capability(capability: tuple[int, int] | int, device_id: int = 0) -> bool

Test whether this platform is compatible with a device capability.

The capability argument can either be:

  • A tuple (major, minor).
  • An integer <major><minor>. (See :meth:DeviceCapability.to_int)
Source code in fastvideo/platforms/interface.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@classmethod
def has_device_capability(
    cls,
    capability: tuple[int, int] | int,
    device_id: int = 0,
) -> bool:
    """
    Test whether this platform is compatible with a device capability.

    The ``capability`` argument can either be:

    - A tuple ``(major, minor)``.
    - An integer ``<major><minor>``. (See :meth:`DeviceCapability.to_int`)
    """
    current_capability = cls.get_device_capability(device_id=device_id)
    if current_capability is None:
        return False

    if isinstance(capability, tuple):
        return current_capability >= capability

    return current_capability.to_int() >= capability
fastvideo.platforms.interface.Platform.inference_mode classmethod 
inference_mode()

A device-specific wrapper of torch.inference_mode.

This wrapper is recommended because some hardware backends such as TPU do not support torch.inference_mode. In such a case, they will fall back to torch.no_grad by overriding this method.

Source code in fastvideo/platforms/interface.py 
181
182
183
184
185
186
187
188
189
@classmethod
def inference_mode(cls):
    """A device-specific wrapper of `torch.inference_mode`.

    This wrapper is recommended because some hardware backends such as TPU
    do not support `torch.inference_mode`. In such a case, they will fall
    back to `torch.no_grad` by overriding this method.
    """
    return torch.inference_mode(mode=True)
fastvideo.platforms.interface.Platform.is_async_output_supported classmethod 
is_async_output_supported(enforce_eager: bool | None) -> bool

Check if the current platform supports async output.

Source code in fastvideo/platforms/interface.py 
167
168
169
170
171
172
@classmethod
def is_async_output_supported(cls, enforce_eager: bool | None) -> bool:
    """
    Check if the current platform supports async output.
    """
    raise NotImplementedError
fastvideo.platforms.interface.Platform.is_cuda_alike 
is_cuda_alike() -> bool

Stateless version of :func:torch.cuda.is_available.

Source code in fastvideo/platforms/interface.py 
105
106
107
def is_cuda_alike(self) -> bool:
    """Stateless version of :func:`torch.cuda.is_available`."""
    return self._enum in (PlatformEnum.CUDA, PlatformEnum.ROCM)
fastvideo.platforms.interface.Platform.seed_everything classmethod 
seed_everything(seed: int | None = None) -> None

Set the seed of each random module. torch.manual_seed will set seed on all devices.

Loosely based on: https://github.com/Lightning-AI/pytorch-lightning/blob/2.4.0/src/lightning/fabric/utilities/seed.py#L20

Source code in fastvideo/platforms/interface.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
@classmethod
def seed_everything(cls, seed: int | None = None) -> None:
    """
    Set the seed of each random module.
    `torch.manual_seed` will set seed on all devices.

    Loosely based on: https://github.com/Lightning-AI/pytorch-lightning/blob/2.4.0/src/lightning/fabric/utilities/seed.py#L20
    """
    if seed is not None:
        random.seed(seed)
        np.random.seed(seed)
        torch.manual_seed(seed)
        torch.cuda.manual_seed_all(seed)
fastvideo.platforms.interface.Platform.verify_model_arch classmethod 
verify_model_arch(model_arch: str) -> None

Verify whether the current platform supports the specified model architecture.

  • This will raise an Error or Warning based on the model support on the current platform.
  • By default all models are considered supported.
Source code in fastvideo/platforms/interface.py 
205
206
207
208
209
210
211
212
213
214
215
@classmethod
def verify_model_arch(cls, model_arch: str) -> None:
    """
    Verify whether the current platform supports the specified model
    architecture.

    - This will raise an Error or Warning based on the model support on
    the current platform.
    - By default all models are considered supported.
    """
    pass
fastvideo.platforms.interface.Platform.verify_quantization classmethod 
verify_quantization(quant: str) -> None

Verify whether the quantization is supported by the current platform.

Source code in fastvideo/platforms/interface.py 
217
218
219
220
221
222
223
224
225
226
@classmethod
def verify_quantization(cls, quant: str) -> None:
    """
    Verify whether the quantization is supported by the current platform.
    """
    if cls.supported_quantization and \
        quant not in cls.supported_quantization:
        raise ValueError(
            f"{quant} quantization is currently not supported in "
            f"{cls.device_name}.")

Functions

fastvideo.platforms.mps

Classes

fastvideo.platforms.mps.MpsPlatform

Bases: Platform

Functions
fastvideo.platforms.mps.MpsPlatform.seed_everything classmethod 
seed_everything(seed: int | None = None) -> None

Set the seed for MPS device.

Source code in fastvideo/platforms/mps.py 
65
66
67
68
69
70
71
72
73
74
@classmethod
def seed_everything(cls, seed: int | None = None) -> None:
    """Set the seed for MPS device."""
    if seed is not None:
        import random

        import numpy as np
        random.seed(seed)
        np.random.seed(seed)
        torch.manual_seed(seed)

Functions

fastvideo.platforms.npu

Classes

fastvideo.platforms.npu.NPUPlatform

Bases: Platform

Functions
fastvideo.platforms.npu.NPUPlatform.get_torch_device classmethod 
get_torch_device()

Return torch.npu

Source code in fastvideo/platforms/npu.py 
93
94
95
96
97
98
@classmethod
def get_torch_device(cls):
    """
    Return torch.npu
    """
    return torch.npu

Functions

fastvideo.platforms.rocm

This file is a platform abstraction for ROCm GPUs, adjusted to match the structure and interface of cuda.py.

Classes

Functions