GithubStar/faceswap
1
0
Fork 0
mirror of https://github.com/deepfakes/faceswap.git synced 2026-10-17 05:21:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

Backported code to support future updates

Browse Source
This commit is contained in:
torzdf
2026-02-27 19:30:27 +00:00
parent 4c2092623d
commit aa40efabf3
15 changed files with 681 additions and 554 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/conf.py
+5
View File
@@ -43,6 +43,11 @@ release = '3.0'
# -- General configuration ---------------------------------------------------
autodoc_typehints = "both"
autodoc_default_options = {
"members": True,
"special-members": "__next__, __call__",
}
# Add any Sphinx extension module names here, as strings. They can be
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
lib/align/aligned_face.py
+3 -3
View File
@@ -14,7 +14,7 @@ import numpy as np
from lib.logger import parse_class_init
from lib.utils import get_module_objects
from .constants import CenteringType, EXTRACT_RATIOS, LandmarkType, _MEAN_FACE
from .constants import CenteringType, EXTRACT_RATIOS, LandmarkType, MEAN_FACE
from .pose import PoseEstimate
logger = logging.getLogger(__name__)
@@ -439,7 +439,7 @@ class AlignedFace(): # pylint:disable=too-many-instance-attributes
used for aligning the image. """
with self._cache.lock("average_distance"):
if not self._cache.average_distance:
mean_face = _MEAN_FACE[self._mean_lookup]
mean_face = MEAN_FACE[self._mean_lookup]
lms = self.normalized_landmarks
if self._landmark_type == LandmarkType.LM_2D_68:
lms = lms[17:] # 68 point landmarks only use core face items
@@ -500,7 +500,7 @@ class AlignedFace(): # pylint:disable=too-many-instance-attributes
lms = self._frame_landmarks
if self._landmark_type == LandmarkType.LM_2D_68:
lms = lms[17:] # 68 point landmarks only use core face items
retval = _umeyama(lms, _MEAN_FACE[self._mean_lookup], True)[0:2]
retval = _umeyama(lms, MEAN_FACE[self._mean_lookup], True)[0:2]
logger.trace("Default matrix: %s", retval) # type:ignore[attr-defined]
return retval
lib/align/alignments.py
+119 -124
View File
@@ -1,9 +1,10 @@
#!/usr/bin/env python3
""" Alignments file functions for reading, writing and manipulating the data stored in a
"""Alignments file functions for reading, writing and manipulating the data stored in a
serialized alignments file. """
from __future__ import annotations
import logging
import os
import sys
import typing as T
from datetime import datetime
@@ -36,7 +37,7 @@ _VERSION = 2.4
# TODO Convert these to Dataclasses
class MaskAlignmentsFileDict(T.TypedDict):
""" Typed Dictionary for storing Masks. """
"""Typed Dictionary for storing Masks."""
mask: bytes
affine_matrix: list[float] | np.ndarray
interpolator: int
@@ -45,8 +46,8 @@ class MaskAlignmentsFileDict(T.TypedDict):
class PNGHeaderAlignmentsDict(T.TypedDict):
""" Base Dictionary for storing a single faces' Alignment Information in Alignments files and
PNG Headers. """
"""Base Dictionary for storing a single faces' Alignment Information in Alignments files and
PNG Headers."""
x: int
y: int
w: int
@@ -57,12 +58,12 @@ class PNGHeaderAlignmentsDict(T.TypedDict):
class AlignmentFileDict(PNGHeaderAlignmentsDict):
""" Typed Dictionary for storing a single faces' Alignment Information in alignments files. """
"""Typed Dictionary for storing a single faces' Alignment Information in alignments files."""
thumb: np.ndarray | None
class PNGHeaderSourceDict(T.TypedDict):
""" Dictionary for storing additional meta information in PNG headers """
"""Dictionary for storing additional meta information in PNG headers."""
alignments_version: float
original_filename: str
face_index: int
@@ -72,19 +73,19 @@ class PNGHeaderSourceDict(T.TypedDict):
class AlignmentDict(T.TypedDict):
""" Dictionary for holding all of the alignment information within a single alignment file """
"""Dictionary for holding all of the alignment information within a single alignment file."""
faces: list[AlignmentFileDict]
video_meta: dict[str, float | int]
class PNGHeaderDict(T.TypedDict):
""" Dictionary for storing all alignment and meta information in PNG Headers """
"""Dictionary for storing all alignment and meta information in PNG Headers."""
alignments: PNGHeaderAlignmentsDict
source: PNGHeaderSourceDict
class Alignments(): # pylint:disable=too-many-public-methods
""" The alignments file is a custom serialized ``.fsa`` file that holds information for each
"""The alignments file is a custom serialized ``.fsa`` file that holds information for each
frame for a video or series of images.
Specifically, it holds a list of faces that appear in each frame. Each face contains
@@ -96,9 +97,9 @@ class Alignments(): # pylint:disable=too-many-public-methods
Parameters
----------
folder: str
folder
The folder that contains the alignments ``.fsa`` file
filename: str, optional
filename
The filename of the ``.fsa`` alignments file. If not provided then the given folder will be
checked for a default alignments file filename. Default: "alignments"
"""
@@ -117,62 +118,61 @@ class Alignments(): # pylint:disable=too-many-public-methods
@property
def frames_count(self) -> int:
""" int: The number of frames that appear in the alignments :attr:`data`. """
"""The number of frames that appear in the alignments :attr:`data`."""
retval = len(self._data)
logger.trace(retval) # type:ignore[attr-defined]
return retval
@property
def faces_count(self) -> int:
""" int: The total number of faces that appear in the alignments :attr:`data`. """
"""The total number of faces that appear in the alignments :attr:`data`"""
retval = sum(len(val["faces"]) for val in self._data.values())
logger.trace(retval) # type:ignore[attr-defined]
return retval
@property
def file(self) -> str:
""" str: The full path to the currently loaded alignments file. """
"""The full path to the currently loaded alignments file."""
return self._io.file
@property
def data(self) -> dict[str, AlignmentDict]:
""" dict: The loaded alignments :attr:`file` in dictionary form. """
"""The loaded alignments :attr:`file` in dictionary form."""
return self._data
@property
def have_alignments_file(self) -> bool:
""" bool: ``True`` if an alignments file exists at location :attr:`file` otherwise
``False``. """
"""``True`` if an alignments file exists at location :attr:`file` otherwise ``False``."""
return self._io.have_alignments_file
@property
def hashes_to_frame(self) -> dict[str, dict[str, int]]:
""" dict: The SHA1 hash of the face mapped to the frame(s) and face index within the frame
that the hash corresponds to.
"""The SHA1 hash of the face mapped to the frame(s) and face index within the frame that
the hash corresponds to.
Notes
-----
This method is depractated and exists purely for updating legacy hash based alignments
This method is deprecated and exists purely for updating legacy hash based alignments
to new png header storage in :class:`lib.align.update_legacy_png_header`.
"""
return self._legacy.hashes_to_frame
@property
def hashes_to_alignment(self) -> dict[str, AlignmentFileDict]:
""" dict: The SHA1 hash of the face mapped to the alignment for the face that the hash
corresponds to. The structure of the dictionary is:
"""The SHA1 hash of the face mapped to the alignment for the face that the hash
corresponds to.
Notes
-----
This method is depractated and exists purely for updating legacy hash based alignments
This method is deprecated and exists purely for updating legacy hash based alignments
to new png header storage in :class:`lib.align.update_legacy_png_header`.
"""
return self._legacy.hashes_to_alignment
@property
def mask_summary(self) -> dict[str, int]:
""" dict: The mask type names stored in the alignments :attr:`data` as key with the number
of faces which possess the mask type as value. """
"""The mask type names stored in the alignments :attr:`data` as key with the number of
faces which possess the mask type as value."""
masks: dict[str, int] = {}
for val in self._data.values():
for face in val["faces"]:
@@ -184,8 +184,8 @@ class Alignments(): # pylint:disable=too-many-public-methods
@property
def video_meta_data(self) -> dict[str, list[int] | list[float] | None]:
""" dict: The frame meta data stored in the alignments file. If data does not exist in the
alignments file then ``None`` is returned for each Key """
"""The frame meta data stored in the alignments file. If data does not exist in the
alignments file then ``None`` is returned for each Key"""
retval: dict[str, list[int] | list[float] | None] = {"pts_time": None, "keyframes": None}
pts_time: list[float] = []
keyframes: list[int] = []
@@ -201,35 +201,33 @@ class Alignments(): # pylint:disable=too-many-public-methods
@property
def thumbnails(self) -> Thumbnails:
""" :class:`~lib.align.thumbnails.Thumbnails`: The low resolution thumbnail images that
exist within the alignments file """
"""The low resolution thumbnail images that exist within the alignments file"""
return self._thumbnails
@property
def version(self) -> float:
""" float: The alignments file version number. """
"""float: The alignments file version number. """
return self._io.version
def _load(self) -> dict[str, AlignmentDict]:
""" Load the alignments data from the serialized alignments :attr:`file`.
"""Load the alignments data from the serialized alignments :attr:`file`.
Populates :attr:`_version` with the alignment file's loaded version as well as returning
the serialized data.
Returns
-------
dict:
The loaded alignments data
The loaded alignments data
"""
return self._io.load()
def save(self) -> None:
""" Write the contents of :attr:`data` and :attr:`_meta` to a serialized ``.fsa`` file at
the location :attr:`file`. """
"""Write the contents of :attr:`data` and :attr:`_meta` to a serialized ``.fsa`` file at
the location :attr:`file`."""
return self._io.save()
def backup(self) -> None:
""" Create a backup copy of the alignments :attr:`file`.
"""Create a backup copy of the alignments :attr:`file`.
Creates a copy of the serialized alignments :attr:`file` appending a
timestamp onto the end of the file name and storing in the same folder as
@@ -238,7 +236,7 @@ class Alignments(): # pylint:disable=too-many-public-methods
return self._io.backup()
def save_video_meta_data(self, pts_time: list[float], keyframes: list[int]) -> None:
""" Save video meta data to the alignments file.
"""Save video meta data to the alignments file.
If the alignments file does not have an entry for every frame (e.g. if Extract Every N
was used) then the frame is added to the alignments file with no faces, so that they video
@@ -246,10 +244,10 @@ class Alignments(): # pylint:disable=too-many-public-methods
Parameters
----------
pts_time: list
pts_time
A list of presentation timestamps (`float`) in frame index order for every frame in
the input video
keyframes: list
keyframes
A list of frame indices corresponding to the key frames in the input video
"""
if pts_time[0] != 0:
@@ -288,7 +286,7 @@ class Alignments(): # pylint:disable=too-many-public-methods
@classmethod
def _pad_leading_frames(cls, pts_time: list[float], keyframes: list[int]) -> tuple[list[float],
list[int]]:
""" Calculate the number of frames to pad the video by when the first frame is not
"""Calculate the number of frames to pad the video by when the first frame is not
a key frame.
A somewhat crude method by obtaining the gaps between existing frames and calculating
@@ -297,17 +295,16 @@ class Alignments(): # pylint:disable=too-many-public-methods
Parameters
----------
pts_time: list
pts_time
A list of presentation timestamps (`float`) in frame index order for every frame in
the input video
keyframes: list
keyframes
A list of keyframes (`int`) for the input video
Returns
-------
tuple
The presentation time stamps with extra frames padded to the beginning and the
keyframes adjusted to include the new frames
The presentation time stamps with extra frames padded to the beginning and the keyframes
adjusted to include the new frames
"""
start_pts = pts_time[0]
logger.debug("Video not cut on keyframe. Start pts: %s", start_pts)
@@ -329,37 +326,35 @@ class Alignments(): # pylint:disable=too-many-public-methods
# << VALIDATION >> #
def frame_exists(self, frame_name: str) -> bool:
""" Check whether a given frame_name exists within the alignments :attr:`data`.
"""Check whether a given frame_name exists within the alignments :attr:`data`.
Parameters
----------
frame_name: str
frame_name
The frame name to check. This should be the base name of the frame, not the full path
Returns
-------
bool
``True`` if the given frame_name exists within the alignments :attr:`data`
otherwise ``False``
``True`` if the given frame_name exists within the alignments :attr:`data` otherwise
``False``
"""
retval = frame_name in self._data.keys()
logger.trace("'%s': %s", frame_name, retval) # type:ignore[attr-defined]
return retval
def frame_has_faces(self, frame_name: str) -> bool:
""" Check whether a given frame_name exists within the alignments :attr:`data` and contains
"""Check whether a given frame_name exists within the alignments :attr:`data` and contains
at least 1 face.
Parameters
----------
frame_name: str
frame_name
The frame name to check. This should be the base name of the frame, not the full path
Returns
-------
bool
``True`` if the given frame_name exists within the alignments :attr:`data` and has at
least 1 face associated with it, otherwise ``False``
``True`` if the given frame_name exists within the alignments :attr:`data` and has at least
1 face associated with it, otherwise ``False``
"""
frame_data = self._data.get(frame_name, T.cast(AlignmentDict, {}))
retval = bool(frame_data.get("faces", []))
@@ -367,20 +362,19 @@ class Alignments(): # pylint:disable=too-many-public-methods
return retval
def frame_has_multiple_faces(self, frame_name: str) -> bool:
""" Check whether a given frame_name exists within the alignments :attr:`data` and contains
"""Check whether a given frame_name exists within the alignments :attr:`data` and contains
more than 1 face.
Parameters
----------
frame_name: str
frame_name
The frame_name name to check. This should be the base name of the frame, not the full
path
Returns
-------
bool
``True`` if the given frame_name exists within the alignments :attr:`data` and has more
than 1 face associated with it, otherwise ``False``
``True`` if the given frame_name exists within the alignments :attr:`data` and has more
than 1 face associated with it, otherwise ``False``
"""
if not frame_name:
retval = False
@@ -391,21 +385,20 @@ class Alignments(): # pylint:disable=too-many-public-methods
return retval
def mask_is_valid(self, mask_type: str) -> bool:
""" Ensure the given ``mask_type`` is valid for the alignments :attr:`data`.
"""Ensure the given ``mask_type`` is valid for the alignments :attr:`data`.
Every face in the alignments :attr:`data` must have the given mask type to successfully
pass the test.
Parameters
----------
mask_type: str
mask_type
The mask type to check against the current alignments :attr:`data`
Returns
-------
bool:
``True`` if all faces in the current alignments possess the given ``mask_type``
otherwise ``False``
``True`` if all faces in the current alignments possess the given ``mask_type`` otherwise
``False``
"""
retval = all((face.get("mask") is not None and
face["mask"].get(mask_type) is not None)
@@ -416,36 +409,34 @@ class Alignments(): # pylint:disable=too-many-public-methods
# << DATA >> #
def get_faces_in_frame(self, frame_name: str) -> list[AlignmentFileDict]:
""" Obtain the faces from :attr:`data` associated with a given frame_name.
"""Obtain the faces from :attr:`data` associated with a given frame_name.
Parameters
----------
frame_name: str
frame_name
The frame name to return faces for. This should be the base name of the frame, not the
full path
Returns
-------
list
The list of face dictionaries that appear within the requested frame_name
The list of face dictionaries that appear within the requested frame_name
"""
logger.trace("Getting faces for frame_name: '%s'", frame_name) # type:ignore[attr-defined]
frame_data = self._data.get(frame_name, T.cast(AlignmentDict, {}))
return frame_data.get("faces", T.cast(list[AlignmentFileDict], []))
def count_faces_in_frame(self, frame_name: str) -> int:
""" Return number of faces that appear within :attr:`data` for the given frame_name.
"""Return number of faces that appear within :attr:`data` for the given frame_name.
Parameters
----------
frame_name: str
frame_name
The frame name to return the count for. This should be the base name of the frame, not
the full path
Returns
-------
int
The number of faces that appear in the given frame_name
The number of faces that appear in the given frame_name
"""
frame_data = self._data.get(frame_name, T.cast(AlignmentDict, {}))
retval = len(frame_data.get("faces", []))
@@ -454,20 +445,19 @@ class Alignments(): # pylint:disable=too-many-public-methods
# << MANIPULATION >> #
def delete_face_at_index(self, frame_name: str, face_index: int) -> bool:
""" Delete the face for the given frame_name at the given face index from :attr:`data`.
"""Delete the face for the given frame_name at the given face index from :attr:`data`.
Parameters
----------
frame_name: str
frame_name
The frame name to remove the face from. This should be the base name of the frame, not
the full path
face_index: int
face_index
The index number of the face within the given frame_name to remove
Returns
-------
bool
``True`` if a face was successfully deleted otherwise ``False``
``True`` if a face was successfully deleted otherwise ``False``
"""
logger.debug("Deleting face %s for frame_name '%s'", face_index, frame_name)
face_index = int(face_index)
@@ -480,21 +470,20 @@ class Alignments(): # pylint:disable=too-many-public-methods
return True
def add_face(self, frame_name: str, face: AlignmentFileDict) -> int:
""" Add a new face for the given frame_name in :attr:`data` and return it's index.
"""Add a new face for the given frame_name in :attr:`data` and return it's index.
Parameters
----------
frame_name: str
frame_name
The frame name to add the face to. This should be the base name of the frame, not the
full path
face: dict
face
The face information to add to the given frame_name, correctly formatted for storing in
:attr:`data`
Returns
-------
int
The index of the newly added face within :attr:`data` for the given frame_name
The index of the newly added face within :attr:`data` for the given frame_name
"""
logger.debug("Adding face to frame_name: '%s'", frame_name)
if frame_name not in self._data:
@@ -505,16 +494,16 @@ class Alignments(): # pylint:disable=too-many-public-methods
return retval
def update_face(self, frame_name: str, face_index: int, face: AlignmentFileDict) -> None:
""" Update the face for the given frame_name at the given face index in :attr:`data`.
"""Update the face for the given frame_name at the given face index in :attr:`data`.
Parameters
----------
frame_name: str
frame_name
The frame name to update the face for. This should be the base name of the frame, not
the full path
face_index: int
face_index
The index number of the face within the given frame_name to update
face: dict
face
The face information to update to the given frame_name at the given face_index,
correctly formatted for storing in :attr:`data`
"""
@@ -522,13 +511,13 @@ class Alignments(): # pylint:disable=too-many-public-methods
self._data[frame_name]["faces"][face_index] = face
def filter_faces(self, filter_dict: dict[str, list[int]], filter_out: bool = False) -> None:
""" Remove faces from :attr:`data` based on a given filter list.
"""Remove faces from :attr:`data` based on a given filter list.
Parameters
----------
filter_dict: dict
filter_dict
Dictionary of source filenames as key with a list of face indices to filter as value.
filter_out: bool, optional
filter_out
``True`` if faces should be removed from :attr:`data` when there is a corresponding
match in the given filter_dict. ``False`` if faces should be kept in :attr:`data` when
there is a corresponding match in the given filter_dict, but removed if there is no
@@ -551,11 +540,11 @@ class Alignments(): # pylint:disable=too-many-public-methods
del frame_data["faces"][face_idx]
def update_from_dict(self, data: dict[str, AlignmentDict]) -> None:
""" Replace all alignments with the contents of the given dictionary
"""Replace all alignments with the contents of the given dictionary
Parameters
----------
data: dict[str, AlignmentDict]
data
The alignments, in correctly formatted dictionary form, to be populated into this
:class:`Alignments`
"""
@@ -564,7 +553,7 @@ class Alignments(): # pylint:disable=too-many-public-methods
# << GENERATORS >> #
def yield_faces(self) -> Generator[tuple[str, list[AlignmentFileDict], int, str], None, None]:
""" Generator to obtain all faces with meta information from :attr:`data`. The results
"""Generator to obtain all faces with meta information from :attr:`data`. The results
are yielded by frame.
Notes
@@ -573,14 +562,14 @@ class Alignments(): # pylint:disable=too-many-public-methods
Yields
------
frame_name: str
frame_name
The frame name that the face belongs to. This is the base name of the frame, as it
appears in :attr:`data`, not the full path
faces: list
faces
The list of face `dict` objects that exist for this frame
face_count: int
face_count
The number of faces that exist within :attr:`data` for this frame
frame_fullname: str
frame_fullname
The full path (folder and filename) for the yielded frame
"""
for frame_fullname, val in self._data.items():
@@ -592,13 +581,13 @@ class Alignments(): # pylint:disable=too-many-public-methods
yield frame_name, val["faces"], face_count, frame_fullname
def update_legacy_has_source(self, filename: str) -> None:
""" Update legacy alignments files when we have the source filename available.
"""Update legacy alignments files when we have the source filename available.
Updates here can only be performed when we have the source filename
Parameters
----------
filename: str:
filename
The filename/folder of the original source images/video for the current alignments
"""
updates = [updater.is_updated for updater in (VideoExtension(self, filename), )]
@@ -608,15 +597,15 @@ class Alignments(): # pylint:disable=too-many-public-methods
class _IO():
""" Class to handle the saving/loading of an alignments file.
"""Class to handle the saving/loading of an alignments file.
Parameters
----------
alignments: :class:'~Alignments`
alignments
The parent alignments class that these IO operations belong to
folder: str
folder
The folder that contains the alignments ``.fsa`` file
filename: str
filename
The filename of the ``.fsa`` alignments file.
"""
def __init__(self, alignments: Alignments, folder: str, filename: str) -> None:
@@ -628,36 +617,34 @@ class _IO():
@property
def file(self) -> str:
""" str: The full path to the currently loaded alignments file. """
"""The full path to the currently loaded alignments file."""
return self._file
@property
def version(self) -> float:
""" float: The alignments file version number. """
"""The alignments file version number."""
return self._version
@property
def have_alignments_file(self) -> bool:
""" bool: ``True`` if an alignments file exists at location :attr:`file` otherwise
``False``. """
"""``True`` if an alignments file exists at location :attr:`file` otherwise ``False``."""
retval = os.path.exists(self._file)
logger.trace(retval) # type:ignore[attr-defined]
return retval
def _get_location(self, folder: str, filename: str) -> str:
""" Obtains the location of an alignments file.
"""Obtains the location of an alignments file.
Parameters
----------
folder: str
folder
The folder that the alignments file is located in
filename: str
filename
The filename of the alignments file
Returns
-------
str
The full path to the alignments file
The full path to the alignments file
"""
logger.debug("Getting location: (folder: '%s', filename: '%s')", folder, filename)
noext_name, extension = os.path.splitext(filename)
@@ -673,8 +660,8 @@ class _IO():
return location
def update_legacy(self) -> None:
""" Check whether the alignments are legacy, and if so update them to current alignments
format. """
"""Check whether the alignments are legacy, and if so update them to current alignments
format."""
updates = [updater.is_updated for updater in (FileStructure(self._alignments),
LandmarkRename(self._alignments),
ListToNumpy(self._alignments),
@@ -685,36 +672,44 @@ class _IO():
self.save()
def update_version(self) -> None:
""" Update the version of the alignments file to the latest version """
"""Update the version of the alignments file to the latest version"""
self._version = _VERSION
logger.info("Updating alignments file to version %s", self._version)
def load(self) -> dict[str, AlignmentDict]:
""" Load the alignments data from the serialized alignments :attr:`file`.
"""Load the alignments data from the serialized alignments :attr:`file`.
Populates :attr:`_version` with the alignment file's loaded version as well as returning
the serialized data.
Returns
-------
dict:
The loaded alignments data
The loaded alignments data
"""
logger.debug("Loading alignments")
if not self.have_alignments_file:
raise FaceswapError(f"Error: Alignments file not found at {self._file}")
raise FaceswapError(f"Alignments file not found at {self._file}")
logger.info("Reading alignments from: '%s'", self._file)
data = self._serializer.load(self._file)
meta = data.get("__meta__", {"version": 1.0})
self._version = meta["version"]
if self._version < 2.0:
logger.error("This alignments file was generated with a very old legacy extraction "
"method.")
logger.error("Updating these very old files is no longer supported.")
logger.error("To update to a more recent, supported format, you should run the "
"alignments tool's 'extract' job with this file in Faceswap v2.3: "
"https://github.com/deepfakes/faceswap/releases/tag/v2.3.0")
sys.exit(1)
data = data.get("__data__", data)
logger.debug("Loaded alignments")
return data
def save(self) -> None:
""" Write the contents of :attr:`data` and :attr:`_meta` to a serialized ``.fsa`` file at
the location :attr:`file`. """
"""Write the contents of :attr:`data` and :attr:`_meta` to a serialized ``.fsa`` file at
the location :attr:`file`."""
logger.debug("Saving alignments")
logger.info("Writing alignments to: '%s'", self._file)
data = {"__meta__": {"version": self._version},
@@ -723,7 +718,7 @@ class _IO():
logger.debug("Saved alignments")
def backup(self) -> None:
""" Create a backup copy of the alignments :attr:`file`.
"""Create a backup copy of the alignments :attr:`file`.
Creates a copy of the serialized alignments :attr:`file` appending a
timestamp onto the end of the file name and storing in the same folder as
@@ -733,10 +728,10 @@ class _IO():
if not os.path.isfile(self._file):
logger.debug("No alignments to back up")
return
now = datetime.now().strftime("%Y%m%d_%H%M%S")
now = datetime.now().strftime("%Y-%m-%d_%H.%M.%S")
src = self._file
split = os.path.splitext(src)
dst = f"{split[0]}_{now}{split[1]}"
dst = f"{split[0]}_bk_{now}{split[1]}"
idx = 1
while True:
if not os.path.exists(dst):
lib/align/constants.py
+1 -1
View File
@@ -51,7 +51,7 @@ class LandmarkType(Enum):
return shapes[shape]
_MEAN_FACE: dict[LandmarkType, np.ndarray] = {
MEAN_FACE: dict[LandmarkType, np.ndarray] = {
LandmarkType.LM_2D_4: np.array(
[[0.0, 0.0], [1.0, 0.0], [1.0, 1.0], [0.0, 1.0]]), # Clockwise from TL
LandmarkType.LM_2D_51: np.array([
lib/align/pose.py
+202 -56
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
""" Holds estimated pose information for a faceswap aligned face """
"""Holds estimated pose information for a faceswap aligned face """
from __future__ import annotations
import logging
@@ -11,22 +11,62 @@ import numpy as np
from lib.logger import parse_class_init
from lib.utils import get_module_objects
from .constants import _MEAN_FACE, LandmarkType
from .constants import MEAN_FACE, LandmarkType
logger = logging.getLogger(__name__)
if T.TYPE_CHECKING:
import numpy.typing as npt
from .constants import CenteringType
class PoseEstimate():
""" Estimates pose from a generic 3D head model for the given 2D face landmarks.
_CORE_LMS = np.array([6, 7, 8, 9, 10, 17, 21, 22, 26, 31, 32, 33, 34,
35, 36, 39, 42, 45, 48, 50, 51, 52, 54, 56, 57, 58], dtype="int32")
"""The indices used from 68 point landmarks to align to a 3D head"""
_DISTORTION_COEFFICIENTS = np.zeros((4, 1), dtype="float32")
"""The distortion co-efficient for 3D point estimation (assumes no lens distortion)"""
_MEAN_FACE3D = MEAN_FACE[LandmarkType.LM_3D_26]
"""The (26, 3) 3D landmark points for a "mean" head in 3D normalized space"""
_CENTER_OFFSETS: dict[CenteringType, npt.NDArray[np.float32]] = {
"legacy": np.array([0.0, 0.0, 0.0], dtype="float32"),
"head": np.array([0.0, 0.0, -2.3], dtype="float32"),
"face": np.array([0.0, -1.5, 4.2], dtype="float32")
}
"""The offsets required to shift the center point of a head in 3D space relative to legacy
centering"""
def get_camera_matrix(focal_length: int = 4) -> np.ndarray:
"""Obtain an estimate of a camera matrix in normalized space
Parameters
----------
landmarks: :class:`numpy.ndarry`
focal_length
The focal length to obtain the matrix for. Default: 4
Returns
-------
An estimated camera matrix
"""
focal_length = 4
camera_matrix = np.array([[focal_length, 0, 0.5],
[0, focal_length, 0.5],
[0, 0, 1]], dtype="double")
logger.trace("camera_matrix: %s", camera_matrix) # type:ignore[attr-defined]
return camera_matrix
class PoseEstimate():
"""Estimates pose from a generic 3D head model for the given 2D face landmarks.
Parameters
----------
landmarks
The original 68 point landmarks aligned to 0.0 - 1.0 range
landmarks_type: :class:`~LandmarksType`
landmarks_type
The type of landmarks that are generating this face
References
@@ -38,14 +78,13 @@ class PoseEstimate():
def __init__(self, landmarks: np.ndarray, landmarks_type: LandmarkType) -> None:
logger.trace(parse_class_init(locals())) # type:ignore[attr-defined]
self._distortion_coefficients = np.zeros((4, 1)) # Assuming no lens distortion
self._xyz_2d: np.ndarray | None = None
if landmarks_type != LandmarkType.LM_2D_68:
self._log_once("Pose estimation is not available for non-68 point landmarks. Pose and "
"offset data will all be returned as the incorrect value of '0'")
self._landmarks_type = landmarks_type
self._camera_matrix = self._get_camera_matrix()
self._camera_matrix = get_camera_matrix()
self._rotation, self._translation = self._solve_pnp(landmarks)
self._offset = self._get_offset()
self._pitch_yaw_roll: tuple[float, float, float] = (0, 0, 0)
@@ -53,8 +92,8 @@ class PoseEstimate():
@property
def xyz_2d(self) -> np.ndarray:
""" :class:`numpy.ndarray` projected (x, y) coordinates for each x, y, z point at a
constant distance from adjusted center of the skull (0.5, 0.5) in the 2D space. """
"""projected (x, y) coordinates for each x, y, z point at a constant distance from adjusted
center of the skull (0.5, 0.5) in the 2D space."""
if self._xyz_2d is None:
xyz = cv2.projectPoints(np.array([[6., 0., -2.3],
[0., 6., -2.3],
@@ -62,86 +101,70 @@ class PoseEstimate():
self._rotation,
self._translation,
self._camera_matrix,
self._distortion_coefficients)[0].squeeze()
_DISTORTION_COEFFICIENTS)[0].squeeze()
self._xyz_2d = xyz - self._offset["head"]
return self._xyz_2d
@property
def offset(self) -> dict[CenteringType, np.ndarray]:
""" dict: The amount to offset a standard 0.0 - 1.0 umeyama transformation matrix for a
from the center of the face (between the eyes) or center of the head (middle of skull)
rather than the nose area. """
"""The amount to offset a standard 0.0 - 1.0 Umeyama transformation matrix from the center
of the face (between the eyes) or center of the head (middle of skull) rather than the nose
area."""
return self._offset
@property
def pitch(self) -> float:
""" float: The pitch of the aligned face in eular angles """
"""The pitch of the aligned face in Eular angles"""
if not any(self._pitch_yaw_roll):
self._get_pitch_yaw_roll()
return self._pitch_yaw_roll[0]
@property
def yaw(self) -> float:
""" float: The yaw of the aligned face in eular angles """
"""The yaw of the aligned face in Eular angles"""
if not any(self._pitch_yaw_roll):
self._get_pitch_yaw_roll()
return self._pitch_yaw_roll[1]
@property
def roll(self) -> float:
""" float: The roll of the aligned face in eular angles """
"""The roll of the aligned face in Eular angles"""
if not any(self._pitch_yaw_roll):
self._get_pitch_yaw_roll()
return self._pitch_yaw_roll[2]
@classmethod
def _log_once(cls, message: str) -> None:
""" Log a warning about unsupported landmarks if a message has not already been logged """
"""Log a warning about unsupported landmarks if a message has not already been logged"""
if cls._logged_once:
return
logger.warning(message)
cls._logged_once = True
def _get_pitch_yaw_roll(self) -> None:
""" Obtain the yaw, roll and pitch from the :attr:`_rotation` in eular angles. """
"""Obtain the yaw, roll and pitch from the :attr:`_rotation` in Eular angles."""
proj_matrix = np.zeros((3, 4), dtype="float32")
proj_matrix[:3, :3] = cv2.Rodrigues(self._rotation)[0]
euler = cv2.decomposeProjectionMatrix(proj_matrix)[-1]
self._pitch_yaw_roll = T.cast(tuple[float, float, float], tuple(euler.squeeze()))
logger.trace("yaw_pitch: %s", self._pitch_yaw_roll) # type:ignore[attr-defined]
@classmethod
def _get_camera_matrix(cls) -> np.ndarray:
""" Obtain an estimate of the camera matrix based off the original frame dimensions.
Returns
-------
:class:`numpy.ndarray`
An estimated camera matrix
"""
focal_length = 4
camera_matrix = np.array([[focal_length, 0, 0.5],
[0, focal_length, 0.5],
[0, 0, 1]], dtype="double")
logger.trace("camera_matrix: %s", camera_matrix) # type:ignore[attr-defined]
return camera_matrix
def _solve_pnp(self, landmarks: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
""" Solve the Perspective-n-Point for the given landmarks.
"""Solve the Perspective-n-Point for the given landmarks.
Takes 2D landmarks in world space and estimates the rotation and translation vectors
in 3D space.
Parameters
----------
landmarks: :class:`numpy.ndarry`
landmarks
The original 68 point landmark co-ordinates relating to the original frame
Returns
-------
rotation: :class:`numpy.ndarray`
rotation
The solved rotation vector
translation: :class:`numpy.ndarray`
translation
The solved translation vector
"""
if self._landmarks_type != LandmarkType.LM_2D_68:
@@ -149,43 +172,166 @@ class PoseEstimate():
rotation = np.array([[0.0], [0.0], [0.0]])
translation = rotation.copy()
else:
points = landmarks[[6, 7, 8, 9, 10, 17, 21, 22, 26, 31, 32, 33, 34,
35, 36, 39, 42, 45, 48, 50, 51, 52, 54, 56, 57, 58]]
_, rotation, translation = cv2.solvePnP(_MEAN_FACE[LandmarkType.LM_3D_26],
points = landmarks[_CORE_LMS]
_, rotation, translation = cv2.solvePnP(_MEAN_FACE3D,
points,
self._camera_matrix,
self._distortion_coefficients,
_DISTORTION_COEFFICIENTS,
flags=cv2.SOLVEPNP_ITERATIVE)
logger.trace("points: %s, rotation: %s, translation: %s", # type:ignore[attr-defined]
points, rotation, translation)
return rotation, translation
def _get_offset(self) -> dict[CenteringType, np.ndarray]:
""" Obtain the offset between the original center of the extracted face to the new center
def _get_offset(self) -> dict[CenteringType, npt.NDArray[np.float32]]:
"""Obtain the offset between the original center of the extracted face to the new center
of the head in 2D space.
Returns
-------
:class:`numpy.ndarray`
The x, y offset of the new center from the old center.
The x, y offset of the new center from the old center.
"""
offset: dict[CenteringType, np.ndarray] = {"legacy": np.array([0.0, 0.0])}
legacy = np.array([0.0, 0.0], dtype="float32")
offset: dict[CenteringType, npt.NDArray[np.float32]] = {}
if self._landmarks_type != LandmarkType.LM_2D_68:
offset["face"] = np.array([0.0, 0.0])
offset["head"] = np.array([0.0, 0.0])
offset["legacy"] = legacy
offset["face"] = np.array([0.0, 0.0], dtype="float32")
offset["head"] = np.array([0.0, 0.0], dtype="float32")
else:
points: dict[T.Literal["face", "head"], tuple[float, ...]] = {"head": (0.0, 0.0, -2.3),
"face": (0.0, -1.5, 4.2)}
for key, pnts in points.items():
center = cv2.projectPoints(np.array([pnts]).astype("float32"),
for key, points in _CENTER_OFFSETS.items():
if key == "legacy":
offset[key] = legacy
continue
center = cv2.projectPoints(np.array([points]).astype("float32"),
self._rotation,
self._translation,
self._camera_matrix,
self._distortion_coefficients)[0].squeeze()
_DISTORTION_COEFFICIENTS)[0].squeeze().astype("float32")
logger.trace("center %s: %s", key, center) # type:ignore[attr-defined]
offset[key] = center - np.array([0.5, 0.5])
offset[key] = center - np.array([0.5, 0.5], dtype="float32")
logger.trace("offset: %s", offset) # type:ignore[attr-defined]
return offset
class Batch3D:
"""Functions to perform 3D space calculations on batches """
_camera_matrix = get_camera_matrix()
_legacy_offset = np.array([[0.0, 0.0]], dtype="float32")
_to_center_shift = np.array([[0.5, 0.5]], dtype="float32")
@classmethod
def solve_pnp(cls, landmarks: npt.NDArray[np.float32]) -> npt.NDArray[np.float32]:
"""Estimate rotation and translation from a mean 3D head model
Parameters
----------
landmarks
The (N, 68, 2) 2D normalized landmark points to obtain the rotation and translation
vectors for
Returns
-------
The rotation and translation vectors for the given landmarks in format:
```
(rotation, N, 3, 1
translation, N, 3, 1)
```
"""
core_lms = np.ascontiguousarray(landmarks[:, _CORE_LMS])
retval = np.array([cv2.solvePnP(_MEAN_FACE3D,
lms,
cls._camera_matrix,
_DISTORTION_COEFFICIENTS,
flags=cv2.SOLVEPNP_ITERATIVE)[1:]
for lms in core_lms]).astype("float32").swapaxes(0, 1)
return retval
@classmethod
def rodrigues(cls, vectors: npt.NDArray[np.float32]) -> npt.NDArray[np.float32]:
"""Perform batch conversion of rotation vectors to rotation matrices
Parameters
----------
vectors
The (N, 3, 1) rotation vectors to convert
Returns
-------
The (N, 3, 3) rotation matrices
"""
vectors = vectors.reshape(-1, 3)
theta = np.linalg.norm(vectors, axis=1, keepdims=True)
units = vectors / (theta + 1e-12)
k = np.zeros((vectors.shape[0], 3, 3), dtype="float32")
k[:, 0, 1] = -units[:, 2]
k[:, 0, 2] = units[:, 1]
k[:, 1, 0] = units[:, 2]
k[:, 1, 2] = -units[:, 0]
k[:, 2, 0] = -units[:, 1]
k[:, 2, 1] = units[:, 0]
ident = np.eye(3, dtype="float32")
retval = ident + np.sin(theta)[:, None] * k + (1 - np.cos(theta))[:, None] * (k @ k)
return retval
@classmethod
def project_points(cls,
points: npt.NDArray[np.float32],
rotation_vectors: npt.NDArray[np.float32],
translation_vectors: npt.NDArray[np.float32]) -> npt.NDArray[np.float32]:
"""Batch protection of points from 3D space to 2D space
Parameters
----------
points
The (N, M, 3) points to project
rotation_vectors
The (N, 3, 1) rotation vectors for projection
translation_vectors
The (N, 3, 1) translation vectors for projection
Returns
-------
The (N, M, 2) projected points in 2D space
"""
rot = cls.rodrigues(rotation_vectors)
x_cam = np.einsum('nij,nmj->nmi', rot, points) + translation_vectors.swapaxes(1, 2)
x_y = x_cam[..., :2] / x_cam[..., 2: 3]
cam = cls._camera_matrix
retval = np.empty_like(x_y)
retval[:, :, 0] = cam[0, 0] * x_y[..., 0] + cam[0, 2]
retval[:, :, 1] = cam[1, 1] * x_y[..., 1] + cam[1, 2]
return retval
@classmethod
def get_offsets(cls,
centering: CenteringType,
rotation_vectors: npt.NDArray[np.float32],
translation_vectors: npt.NDArray[np.float32]) -> npt.NDArray[np.float32]:
"""Obtain the offset for moving normalized 68 point landmarks from legacy centering
Parameters
----------
centering
The centering type to obtain the offset for
rotation_vectors
The (N, 3, 1) batch of rotation vectors to receive offsets for
translation_vectors
The (N, 3, 1) batch of translation vectors to receive offsets for
Returns
-------
The (N, 2) offsets for the given rotation/translation vector
"""
batch_size = rotation_vectors.shape[0]
if centering == "legacy":
return np.broadcast_to(cls._legacy_offset, (batch_size, 2))
points3d = np.broadcast_to(_CENTER_OFFSETS[centering][None], (batch_size, 3))
offsets = cls.project_points(points3d[:, None, :],
rotation_vectors,
translation_vectors)[:, 0]
return offsets - cls._to_center_shift
__all__ = get_module_objects(__name__)
lib/gpu_stats/_base.py
+53 -54
View File
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
""" Parent class for obtaining Stats for various GPU/TPU backends. All GPU Stats should inherit
from the :class:`_GPUStats` class contained here. """
"""Parent class for obtaining Stats for various GPU/TPU backends. All GPU Stats should inherit
from the :class:`_GPUStats` class contained here."""
import logging
@@ -17,51 +17,60 @@ class GPUInfo():
Attributes:
----------
vram: list[int]
vram
List of integers representing the total VRAM available on each GPU, in MB.
vram_free: list[int]
vram_free
List of integers representing the free VRAM available on each GPU, in MB.
driver: str
driver
String representing the driver version being used for the GPUs.
devices: list[str]
devices
List of strings representing the names of each GPU device.
devices_active: list[int]
devices_active
List of integers representing the indices of the active GPU devices.
"""
vram: list[int]
"""List of integers representing the total VRAM available on each GPU, in MB."""
vram_free: list[int]
"""List of integers representing the free VRAM available on each GPU, in MB."""
driver: str
"""String representing the driver version being used for the GPUs."""
devices: list[str]
"""List of strings representing the names of each GPU device."""
devices_active: list[int]
"""List of integers representing the indices of the active GPU devices."""
@dataclass
class BiggestGPUInfo():
""" Dataclass for holding GPU Information about the card with most available VRAM.
"""Dataclass for holding GPU Information about the card with most available VRAM.
Attributes
----------
card_id: int
card_id
Integer representing the index of the GPU device.
device: str
device
The name of the device
free: float
free
The amount of available VRAM on the GPU
total: float
total
the total amount of VRAM on the GPU
"""
card_id: int
"""Integer representing the index of the GPU device."""
device: str
"""The name of the device"""
free: float
"""The amount of available VRAM on the GPU"""
total: float
"""the total amount of VRAM on the GPU"""
class _GPUStats():
""" Parent class for collecting GPU device information.
"""Parent class for collecting GPU device information.
Parameters:
-----------
log : bool, optional
log
Flag indicating whether or not to log debug messages. Default: `True`.
"""
@@ -90,22 +99,22 @@ class _GPUStats():
@property
def device_count(self) -> int:
"""int: The number of GPU devices discovered on the system. """
"""The number of GPU devices discovered on the system."""
return self._device_count
@property
def cli_devices(self) -> list[str]:
""" list[str]: Formatted index: name text string for each GPU """
"""Formatted index: name text string for each GPU"""
return [f"{idx}: {device}" for idx, device in enumerate(self._device_names)]
@property
def exclude_all_devices(self) -> bool:
""" bool: ``True`` if all GPU devices have been explicitly disabled otherwise ``False`` """
"""``True`` if all GPU devices have been explicitly disabled otherwise ``False``"""
return all(idx in _EXCLUDE_DEVICES for idx in range(self._device_count))
@property
def sys_info(self) -> GPUInfo:
""" :class:`GPUInfo`: The GPU Stats that are required for system information logging """
"""The GPU Stats that are required for system information logging"""
return GPUInfo(vram=self._vram,
vram_free=self._get_free_vram(),
driver=self._driver,
@@ -113,14 +122,14 @@ class _GPUStats():
devices_active=self._active_devices)
def _log(self, level: str, message: str) -> None:
""" If the class has been initialized with :attr:`log` as `True` then log the message
"""If the class has been initialized with :attr:`log` as `True` then log the message
otherwise skip logging.
Parameters
----------
level: str
level
The log level to log at
message: str
message
The message to log
"""
if self._logger is None:
@@ -129,25 +138,24 @@ class _GPUStats():
logger(message)
def _initialize(self) -> None:
""" Override to initialize the GPU device handles and any other necessary resources. """
"""Override to initialize the GPU device handles and any other necessary resources."""
self._is_initialized = True
def _shutdown(self) -> None:
""" Override to shutdown the GPU device handles and any other necessary resources. """
"""Override to shutdown the GPU device handles and any other necessary resources."""
self._is_initialized = False
def _get_device_count(self) -> int:
""" Override to obtain the number of GPU devices
"""Override to obtain the number of GPU devices
Returns
-------
int
The total number of GPUs connected to the PC
The total number of GPUs connected to the PC
"""
raise NotImplementedError()
def _get_active_devices(self) -> list[int]:
""" Obtain the indices of active GPUs (those that have not been explicitly excluded in
"""Obtain the indices of active GPUs (those that have not been explicitly excluded in
the command line arguments).
Notes
@@ -156,77 +164,69 @@ class _GPUStats():
Returns
-------
list
The list of device indices that are available for Faceswap to use
The list of device indices that are available for Faceswap to use
"""
devices = [idx for idx in range(self._device_count) if idx not in _EXCLUDE_DEVICES]
self._log("debug", f"Active GPU Devices: {devices}")
return devices
def _get_handles(self) -> list:
""" Override to obtain GPU specific device handles for all connected devices.
"""Override to obtain GPU specific device handles for all connected devices.
Returns
-------
list
The device handle for each connected GPU
The device handle for each connected GPU
"""
raise NotImplementedError()
def _get_driver(self) -> str:
""" Override to obtain the GPU specific driver version.
"""Override to obtain the GPU specific driver version.
Returns
-------
str
The GPU driver currently in use
The GPU driver currently in use
"""
raise NotImplementedError()
def _get_device_names(self) -> list[str]:
""" Override to obtain the names of all connected GPUs. The quality of this information
"""Override to obtain the names of all connected GPUs. The quality of this information
depends on the backend and OS being used, but it should be sufficient for identifying
cards.
Returns
-------
list
List of device names for connected GPUs as corresponding to the values in
:attr:`_handles`
List of device names for connected GPUs as corresponding to the values in :attr:`_handles`
"""
raise NotImplementedError()
def _get_vram(self) -> list[int]:
""" Override to obtain the total VRAM in Megabytes for each connected GPU.
"""Override to obtain the total VRAM in Megabytes for each connected GPU.
Returns
-------
list
List of `float`s containing the total amount of VRAM in Megabytes for each
connected GPU as corresponding to the values in :attr:`_handles`
List of `float`s containing the total amount of VRAM in Megabytes for each connected GPU
as corresponding to the values in :attr:`_handles`
"""
raise NotImplementedError()
def _get_free_vram(self) -> list[int]:
""" Override to obtain the amount of VRAM that is available, in Megabytes, for each
"""Override to obtain the amount of VRAM that is available, in Megabytes, for each
connected GPU.
Returns
-------
list
List of `float`s containing the amount of VRAM available, in Megabytes, for each
connected GPU as corresponding to the values in :attr:`_handles
List of `float`s containing the amount of VRAM available, in Megabytes, for each connected
GPU as corresponding to the values in :attr:`_handles
"""
raise NotImplementedError()
def get_card_most_free(self) -> BiggestGPUInfo:
""" Obtain statistics for the GPU with the most available free VRAM.
"""Obtain statistics for the GPU with the most available free VRAM.
Returns
-------
:class:`BiggestGpuInfo`
If a GPU is not detected then the **card_id** is returned as ``-1`` and the amount
of free and total RAM available is fixed to 2048 Megabytes.
If a GPU is not detected then the **card_id** is returned as ``-1`` and the amount
of free and total RAM available is fixed to 2048 Megabytes.
"""
if len(self._active_devices) == 0:
retval = BiggestGPUInfo(card_id=-1,
@@ -245,11 +245,10 @@ class _GPUStats():
return retval
def exclude_devices(self, devices: list[int]) -> None:
""" Exclude GPU devices from being used by Faceswap. Override for backend specific logic
"""Exclude GPU devices from being used by Faceswap. Override for backend specific logic
Parameters
----------
devices: list[int]
The GPU device IDS to be excluded
The GPU device IDS to be excluded
"""
raise NotImplementedError
lib/gpu_stats/apple_silicon.py
+27 -35
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
""" Collects and returns Information on available Apple Silicon SoCs in Apple Macs. """
"""Collects and returns Information on available Apple Silicon SoCs in Apple Macs."""
import typing as T
import os
@@ -12,11 +12,11 @@ from lib.utils import FaceswapError, get_module_objects
from ._base import _GPUStats
_metal_initialized: bool = False
_METAL_INITIALIZED: bool = False
class AppleSiliconStats(_GPUStats):
""" Holds information and statistics about Apple Silicon SoC(s) available on the currently
"""Holds information and statistics about Apple Silicon SoC(s) available on the currently
running Apple system.
Notes
@@ -27,7 +27,7 @@ class AppleSiliconStats(_GPUStats):
Parameters
----------
log: bool, optional
log
Whether the class should output information to the logger. There may be occasions where the
logger has not yet been set up when this class is queried. Attempting to log in these
instances will raise an error. If GPU stats are being queried prior to the logger being
@@ -41,7 +41,7 @@ class AppleSiliconStats(_GPUStats):
super().__init__(log=log)
def _initialize(self) -> None:
""" Initialize Metal for Apple Silicon SoC(s).
"""Initialize Metal for Apple Silicon SoC(s).
If :attr:`_is_initialized` is ``True`` then this function just returns performing no
action. Otherwise :attr:`is_initialized` is set to ``True`` after successfully
@@ -57,13 +57,11 @@ class AppleSiliconStats(_GPUStats):
super()._initialize()
def _initialize_metal(self) -> None:
""" Initialize Metal on first call to this class and set global
:attr:``_metal_initialized`` to ``True``. If Metal has already been initialized then return
performing no action.
"""
global _metal_initialized # pylint:disable=global-statement
"""Initialize Metal on first call to this class and set global :attr:``_METAL_INITIALIZED``
to ``True``. If Metal has already been initialized then return performing no action."""
global _METAL_INITIALIZED # pylint:disable=global-statement
if _metal_initialized:
if _METAL_INITIALIZED:
return
self._log("debug", "Performing first time Apple SoC setup.")
@@ -77,10 +75,10 @@ class AppleSiliconStats(_GPUStats):
self._test_torch()
_metal_initialized = True
_METAL_INITIALIZED = True
def _test_torch(self) -> None:
""" Test that torch can execute correctly.
"""Test that torch can execute correctly.
Raises
------
@@ -92,24 +90,23 @@ class AppleSiliconStats(_GPUStats):
self._log("debug",
f"Torch initialization test: (mem_info: {meminfo})")
except RuntimeError as err:
msg = ("An unhandled exception occured initializing the device via Torch "
msg = ("An unhandled exception occurred initializing the device via Torch "
f"Library. Original error: {str(err)}")
raise FaceswapError(msg) from err
def _get_device_count(self) -> int:
""" Detect the number of SoCs attached to the system.
"""Detect the number of SoCs attached to the system.
Returns
-------
int
The total number of SoCs available
The total number of SoCs available
"""
retval = len(self._mps_devices)
self._log("debug", f"GPU Device count: {retval}")
return retval
def _get_handles(self) -> list:
""" Obtain the device handles for all available Apple Silicon SoCs.
"""Obtain the device handles for all available Apple Silicon SoCs.
Notes
-----
@@ -118,15 +115,14 @@ class AppleSiliconStats(_GPUStats):
Returns
-------
list
The list of indices for available Apple Silicon SoCs
The list of indices for available Apple Silicon SoCs
"""
handles = list(range(self._device_count))
self._log("debug", f"GPU Handles found: {handles}")
return handles
def _get_driver(self) -> str:
""" Obtain the Apple Silicon driver version currently in use.
"""Obtain the Apple Silicon driver version currently in use.
Notes
-----
@@ -135,34 +131,31 @@ class AppleSiliconStats(_GPUStats):
Returns
-------
str
The current SoC driver version
The current SoC driver version
"""
driver = "Not Applicable"
self._log("debug", f"GPU Driver: {driver}")
return driver
def _get_device_names(self) -> list[str]:
""" Obtain the list of names of available Apple Silicon SoC(s) as identified in
"""Obtain the list of names of available Apple Silicon SoC(s) as identified in
:attr:`_handles`.
Returns
-------
list
The list of available Apple Silicon SoC names
The list of available Apple Silicon SoC names
"""
names = [d.type for d in self._mps_devices]
self._log("debug", f"GPU Devices: {names}")
return names
def _get_vram(self) -> list[int]:
""" Obtain the VRAM in Megabytes for each available Apple Silicon SoC(s) as identified in
"""Obtain the VRAM in Megabytes for each available Apple Silicon SoC(s) as identified in
:attr:`_handles`.
Returns
-------
list
The RAM in Megabytes for each available Apple Silicon SoC
The RAM in Megabytes for each available Apple Silicon SoC
"""
vram = [int((torch.mps.driver_allocated_memory() / self._device_count) / (1024 * 1024))
for _ in range(self._device_count)]
@@ -170,14 +163,13 @@ class AppleSiliconStats(_GPUStats):
return vram
def _get_free_vram(self) -> list[int]:
""" Obtain the amount of VRAM that is available, in Megabytes, for each available Apple
"""Obtain the amount of VRAM that is available, in Megabytes, for each available Apple
Silicon SoC.
Returns
-------
list
List of `float`s containing the amount of RAM available, in Megabytes, for each
available SoC as corresponding to the values in :attr:`_handles
List of `float`s containing the amount of RAM available, in Megabytes, for each available
SoC as corresponding to the values in :attr:`_handles
"""
vram = [int((psutil.virtual_memory().available / self._device_count) / (1024 * 1024))
for _ in range(self._device_count)]
@@ -185,11 +177,11 @@ class AppleSiliconStats(_GPUStats):
return vram
def exclude_devices(self, devices: list[int]) -> None:
""" Apple-Silicon does not support excluding devices
"""Apple-Silicon does not support excluding devices
Parameters
----------
devices: list[int]
devices
The GPU device IDS to be excluded
"""
self._log("warning", "Apple Silicon does not support excluding GPUs. This option has been "
lib/gpu_stats/cpu.py
+17 -24
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
""" Dummy functions for running faceswap on CPU. """
"""Dummy functions for running faceswap on CPU."""
from lib.utils import get_module_objects
@@ -7,7 +7,7 @@ from ._base import _GPUStats
class CPUStats(_GPUStats):
""" Holds information and statistics about the CPU on the currently running system.
"""Holds information and statistics about the CPU on the currently running system.
Notes
-----
@@ -19,7 +19,7 @@ class CPUStats(_GPUStats):
Parameters
----------
log: bool, optional
log
Whether the class should output information to the logger. There may be occasions where the
logger has not yet been set up when this class is queried. Attempting to log in these
instances will raise an error. If GPU stats are being queried prior to the logger being
@@ -28,85 +28,78 @@ class CPUStats(_GPUStats):
"""
def _get_device_count(self) -> int:
""" Detect the number of GPUs attached to the system. Always returns zero for CPU
"""Detect the number of GPUs attached to the system. Always returns zero for CPU
backends.
Returns
-------
int
The total number of GPUs connected to the PC
The total number of GPUs connected to the PC
"""
retval = 0
self._log("debug", f"GPU Device count: {retval}")
return retval
def _get_handles(self) -> list:
""" Obtain the device handles for all connected GPUs.
"""Obtain the device handles for all connected GPUs.
Returns
-------
list
An empty list for CPU Backends
An empty list for CPU Backends
"""
handles: list = []
self._log("debug", f"GPU Handles found: {len(handles)}")
return handles
def _get_driver(self) -> str:
""" Obtain the driver version currently in use.
"""Obtain the driver version currently in use.
Returns
-------
str
An empty string for CPU backends
An empty string for CPU backends
"""
driver = ""
self._log("debug", f"GPU Driver: {driver}")
return driver
def _get_device_names(self) -> list[str]:
""" Obtain the list of names of connected GPUs as identified in :attr:`_handles`.
"""Obtain the list of names of connected GPUs as identified in :attr:`_handles`.
Returns
-------
list
An empty list for CPU backends
An empty list for CPU backends
"""
names: list[str] = []
self._log("debug", f"GPU Devices: {names}")
return names
def _get_vram(self) -> list[int]:
""" Obtain the RAM in Megabytes for the running system.
"""Obtain the RAM in Megabytes for the running system.
Returns
-------
list
An empty list for CPU backends
An empty list for CPU backends
"""
vram: list[int] = []
self._log("debug", f"GPU VRAM: {vram}")
return vram
def _get_free_vram(self) -> list[int]:
""" Obtain the amount of RAM that is available, in Megabytes, for the running system.
"""Obtain the amount of RAM that is available, in Megabytes, for the running system.
Returns
-------
list
An empty list for CPU backends
An empty list for CPU backends
"""
vram: list[int] = []
self._log("debug", f"GPU VRAM free: {vram}")
return vram
def exclude_devices(self, devices: list[int]) -> None:
""" CPU does not support excluding devices
"""CPU does not support excluding devices
Parameters
----------
devices: list[int]
The GPU device IDS to be excluded
The GPU device IDS to be excluded
"""
self._log("warning", "CPU does not support excluding GPUs. This option has been ignored")
lib/gpu_stats/nvidia.py
+25 -33
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
""" Collects and returns Information on available Nvidia GPUs. """
"""Collects and returns Information on available Nvidia GPUs. """
import os
import pynvml # pylint:disable=import-error
@@ -10,7 +10,7 @@ from ._base import _GPUStats, _EXCLUDE_DEVICES
class NvidiaStats(_GPUStats):
""" Holds information and statistics about Nvidia GPU(s) available on the currently
"""Holds information and statistics about Nvidia GPU(s) available on the currently
running system.
Notes
@@ -20,7 +20,7 @@ class NvidiaStats(_GPUStats):
Parameters
----------
log: bool, optional
log
Whether the class should output information to the logger. There may be occasions where the
logger has not yet been set up when this class is queried. Attempting to log in these
instances will raise an error. If GPU stats are being queried prior to the logger being
@@ -29,7 +29,7 @@ class NvidiaStats(_GPUStats):
"""
def _initialize(self) -> None:
""" Initialize PyNVML for Nvidia GPUs.
"""Initialize PyNVML for Nvidia GPUs.
If :attr:`_is_initialized` is ``True`` then this function just returns performing no
action. Otherwise :attr:`is_initialized` is set to ``True`` after successfully
@@ -54,7 +54,7 @@ class NvidiaStats(_GPUStats):
f"Error: {str(err)}")
raise FaceswapError(msg) from err
except Exception as err: # pylint:disable=broad-except
msg = ("An unhandled exception occured reading from the Nvidia Machine Learning "
msg = ("An unhandled exception occurred reading from the Nvidia Machine Learning "
f"Library. Original error: {str(err)}")
raise FaceswapError(msg) from err
@@ -62,18 +62,17 @@ class NvidiaStats(_GPUStats):
super()._initialize()
def _shutdown(self) -> None:
""" Cleanly close access to NVML and set :attr:`_is_initialized` back to ``False``. """
"""Cleanly close access to NVML and set :attr:`_is_initialized` back to ``False``. """
self._log("debug", "Shutting down NVML")
pynvml.nvmlShutdown()
super()._shutdown()
def _get_device_count(self) -> int:
""" Detect the number of GPUs attached to the system.
"""Detect the number of GPUs attached to the system.
Returns
-------
int
The total number of GPUs connected to the PC
The total number of GPUs connected to the PC
"""
try:
retval = pynvml.nvmlDeviceGetCount()
@@ -85,14 +84,13 @@ class NvidiaStats(_GPUStats):
return retval
def _get_active_devices(self) -> list[int]:
""" Obtain the indices of active GPUs (those that have not been explicitly excluded by
"""Obtain the indices of active GPUs (those that have not been explicitly excluded by
CUDA_VISIBLE_DEVICES environment variable or explicitly excluded in the command line
arguments).
Returns
-------
list
The list of device indices that are available for Faceswap to use
The list of device indices that are available for Faceswap to use
"""
# pylint:disable=duplicate-code
devices = super()._get_active_devices()
@@ -104,12 +102,11 @@ class NvidiaStats(_GPUStats):
return devices
def _get_handles(self) -> list:
""" Obtain the device handles for all connected Nvidia GPUs.
"""Obtain the device handles for all connected Nvidia GPUs.
Returns
-------
list
The list of pointers for connected Nvidia GPUs
The list of pointers for connected Nvidia GPUs
"""
handles = [pynvml.nvmlDeviceGetHandleByIndex(i)
for i in range(self._device_count)]
@@ -117,12 +114,11 @@ class NvidiaStats(_GPUStats):
return handles
def _get_driver(self) -> str:
""" Obtain the Nvidia driver version currently in use.
"""Obtain the Nvidia driver version currently in use.
Returns
-------
str
The current GPU driver version
The current GPU driver version
"""
try:
driver = pynvml.nvmlSystemGetDriverVersion()
@@ -133,12 +129,11 @@ class NvidiaStats(_GPUStats):
return driver
def _get_device_names(self) -> list[str]:
""" Obtain the list of names of connected Nvidia GPUs as identified in :attr:`_handles`.
"""Obtain the list of names of connected Nvidia GPUs as identified in :attr:`_handles`.
Returns
-------
list
The list of connected Nvidia GPU names
The list of connected Nvidia GPU names
"""
names = [pynvml.nvmlDeviceGetName(handle)
for handle in self._handles]
@@ -146,13 +141,12 @@ class NvidiaStats(_GPUStats):
return names
def _get_vram(self) -> list[int]:
""" Obtain the VRAM in Megabytes for each connected Nvidia GPU as identified in
"""Obtain the VRAM in Megabytes for each connected Nvidia GPU as identified in
:attr:`_handles`.
Returns
-------
list
The VRAM in Megabytes for each connected Nvidia GPU
The VRAM in Megabytes for each connected Nvidia GPU
"""
vram = [pynvml.nvmlDeviceGetMemoryInfo(handle).total / (1024 * 1024)
for handle in self._handles]
@@ -160,14 +154,13 @@ class NvidiaStats(_GPUStats):
return vram
def _get_free_vram(self) -> list[int]:
""" Obtain the amount of VRAM that is available, in Megabytes, for each connected Nvidia
"""Obtain the amount of VRAM that is available, in Megabytes, for each connected Nvidia
GPU.
Returns
-------
list
List of `float`s containing the amount of VRAM available, in Megabytes, for each
connected GPU as corresponding to the values in :attr:`_handles
List of `float`s containing the amount of VRAM available, in Megabytes, for each connected
GPU as corresponding to the values in :attr:`_handles
"""
is_initialized = self._is_initialized
if not is_initialized:
@@ -183,18 +176,17 @@ class NvidiaStats(_GPUStats):
return vram
def exclude_devices(self, devices: list[int]) -> None:
""" Exclude GPU devices from being used by Faceswap. Sets the CUDA_VISIBLE_DEVICES
"""Exclude GPU devices from being used by Faceswap. Sets the CUDA_VISIBLE_DEVICES
environment variable. This must be called before Torch/Keras are imported
Parameters
----------
devices: list[int]
The GPU device IDS to be excluded
The GPU device IDS to be excluded
"""
# pylint:disable=duplicate-code
if not devices:
return
self._log("debug", f"Excluding GPU indicies: {devices}")
self._log("debug", f"Excluding GPU indices: {devices}")
_EXCLUDE_DEVICES.extend(devices)
@@ -204,7 +196,7 @@ class NvidiaStats(_GPUStats):
if d not in _EXCLUDE_DEVICES)
env_vars = [f"{k}: {v}" for k, v in os.environ.items() if k.lower().startswith("cuda")]
self._log("debug", f"Cuda environmet variables: {env_vars}")
self._log("debug", f"Cuda environment variables: {env_vars}")
__all__ = get_module_objects(__name__)
lib/gpu_stats/rocm.py
+30 -39
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
""" Collects and returns Information about connected AMD GPUs for ROCm using sysfs and from
"""Collects and returns Information about connected AMD GPUs for ROCm using sysfs and from
modinfo
As no ROCm compatible hardware was available for testing, this just returns information on all AMD
@@ -211,11 +211,11 @@ _DEVICE_LOOKUP = { # ref: https://gist.github.com/roalercon/51f13a387f3754615cc
class ROCm(_GPUStats):
""" Holds information and statistics about GPUs connected using sysfs
"""Holds information and statistics about GPUs connected using sysfs
Parameters
----------
log: bool, optional
log
Whether the class should output information to the logger. There may be occasions where the
logger has not yet been set up when this class is queried. Attempting to log in these
instances will raise an error. If GPU stats are being queried prior to the logger being
@@ -229,37 +229,35 @@ class ROCm(_GPUStats):
super().__init__(log=log)
def _from_sysfs_file(self, path: str) -> str:
""" Obtain the value from a sysfs file. On permission error or file doesn't exist, log and
"""Obtain the value from a sysfs file. On permission error or file doesn't exist, log and
return empty value
Parameters
----------
path: str
path
The path to a sysfs file to obtain the value from
Returns
-------
str
The obtained value from the given path
The obtained value from the given path
"""
if not os.path.isfile(path):
self._log("debug", f"File '{path}' does not exist. Returning empty string")
return ""
try:
with open(path, "r", encoding="utf-8", errors="ignore") as sysfile:
val = sysfile.read().strip()
with open(path, "r", encoding="utf-8", errors="ignore") as sys_file:
val = sys_file.read().strip()
except PermissionError:
self._log("debug", f"Permission error accessing file '{path}'. Returning empty string")
val = ""
return val
def _get_sysfs_paths(self) -> list[str]:
""" Obtain a list of sysfs paths to AMD branded GPUs connected to the system
"""Obtain a list of sysfs paths to AMD branded GPUs connected to the system
Returns
-------
list[str]
List of full paths to the sysfs entries for connected AMD GPUs
List of full paths to the sysfs entries for connected AMD GPUs
"""
base_dir = "/sys/class/drm/"
@@ -286,7 +284,7 @@ class ROCm(_GPUStats):
return retval
def _initialize(self) -> None:
""" Initialize sysfs for ROCm backend.
"""Initialize sysfs for ROCm backend.
If :attr:`_is_initialized` is ``True`` then this function just returns performing no
action.
@@ -303,12 +301,11 @@ class ROCm(_GPUStats):
super()._initialize()
def _get_device_count(self) -> int:
""" The number of AMD cards found in sysfs
"""The number of AMD cards found in sysfs
Returns
-------
int
The total number of GPUs available
The total number of GPUs available
"""
if self._is_wsl:
retval = torch.cuda.device_count()
@@ -318,13 +315,12 @@ class ROCm(_GPUStats):
return retval
def _get_handles(self) -> list:
""" The sysfs doesn't use device handles, so we just return the list of the sysfs locations
"""The sysfs doesn't use device handles, so we just return the list of the sysfs locations
per card
Returns
-------
list
The list of all discovered GPUs
The list of all discovered GPUs
"""
if self._is_wsl:
handles = list(range(self._device_count))
@@ -334,12 +330,11 @@ class ROCm(_GPUStats):
return handles
def _get_driver(self) -> str:
""" Obtain the driver versions currently in use from modinfo
"""Obtain the driver versions currently in use from modinfo
Returns
-------
str
The current AMDGPU driver versions
The current AMDGPU driver versions
"""
if self._is_wsl:
retval = "unknown (wsl2)"
@@ -364,12 +359,11 @@ class ROCm(_GPUStats):
return retval
def _get_device_names(self) -> list[str]:
""" Obtain the list of names of connected GPUs as identified in :attr:`_handles`.
"""Obtain the list of names of connected GPUs as identified in :attr:`_handles`.
Returns
-------
list
The list of connected AMD GPU names
The list of connected AMD GPU names
"""
retval = []
for device in self._handles:
@@ -403,14 +397,13 @@ class ROCm(_GPUStats):
return retval
def _get_active_devices(self) -> list[int]:
""" Obtain the indices of active GPUs (those that have not been explicitly excluded by
"""Obtain the indices of active GPUs (those that have not been explicitly excluded by
HIP_VISIBLE_DEVICES environment variable or explicitly excluded in the command line
arguments).
Returns
-------
list
The list of device indices that are available for Faceswap to use
The list of device indices that are available for Faceswap to use
"""
devices = super()._get_active_devices()
env_devices = os.environ.get("HIP_VISIBLE_DEVICES")
@@ -421,13 +414,12 @@ class ROCm(_GPUStats):
return devices
def _get_vram(self) -> list[int]:
""" Obtain the VRAM in Megabytes for each connected AMD GPU as identified in
"""Obtain the VRAM in Megabytes for each connected AMD GPU as identified in
:attr:`_handles`.
Returns
-------
list
The VRAM in Megabytes for each connected Nvidia GPU
The VRAM in Megabytes for each connected Nvidia GPU
"""
retval = []
for device in self._handles:
@@ -446,14 +438,13 @@ class ROCm(_GPUStats):
return retval
def _get_free_vram(self) -> list[int]:
""" Obtain the amount of VRAM that is available, in Megabytes, for each connected AMD
"""Obtain the amount of VRAM that is available, in Megabytes, for each connected AMD
GPU.
Returns
-------
list
List of `float`s containing the amount of VRAM available, in Megabytes, for each
connected GPU as corresponding to the values in :attr:`_handles
List of `float`s containing the amount of VRAM available, in Megabytes, for each connected
GPU as corresponding to the values in :attr:`_handles
"""
retval = []
total_vram = self._get_vram()
@@ -480,17 +471,17 @@ class ROCm(_GPUStats):
return retval
def exclude_devices(self, devices: list[int]) -> None:
""" Exclude GPU devices from being used by Faceswap. Sets the HIP_VISIBLE_DEVICES
"""Exclude GPU devices from being used by Faceswap. Sets the HIP_VISIBLE_DEVICES
environment variable. This must be called before Torch/Keras are imported
Parameters
----------
devices: list[int]
devices
The GPU device IDS to be excluded
"""
if not devices:
return
self._log("debug", f"Excluding GPU indicies: {devices}")
self._log("debug", f"Excluding GPU indices: {devices}")
_EXCLUDE_DEVICES.extend(devices)
@@ -500,7 +491,7 @@ class ROCm(_GPUStats):
if d not in _EXCLUDE_DEVICES)
env_vars = [f"{k}: {v}" for k, v in os.environ.items() if k.lower().startswith("hip")]
self._log("debug", f"HIP environmet variables: {env_vars}")
self._log("debug", f"HIP environment variables: {env_vars}")
__all__ = get_module_objects(__name__)
lib/gui/command.py
+2 -2
View File
@@ -146,10 +146,10 @@ class CommandTab(ttk.Frame): # pylint:disable=too-many-ancestors
def add_frame_separator(self):
""" Add a separator between top and bottom frames """
logger.debug("Add frame seperator")
logger.debug("Add frame separator")
sep = ttk.Frame(self, height=2, relief=tk.RIDGE)
sep.pack(fill=tk.X, pady=(5, 0), side=tk.TOP)
logger.debug("Added frame seperator")
logger.debug("Added frame separator")
class ActionFrame(ttk.Frame): # pylint:disable=too-many-ancestors
lib/gui/display_page.py
+1 -1
View File
@@ -92,7 +92,7 @@ class DisplayPage(ttk.Frame): # pylint:disable=too-many-ancestors
def add_frame_separator(self):
""" Add a separator between top and bottom frames """
logger.debug("Adding frame seperator")
logger.debug("Adding frame separator")
sep = ttk.Frame(self, height=2, relief=tk.RIDGE)
sep.pack(fill=tk.X, pady=(5, 0), side=tk.BOTTOM)
lib/image.py
+7 -2
View File
@@ -1211,12 +1211,17 @@ class ImagesLoader(ImageIO):
video then this is a list of dummy filenames as corresponding to an alignments file """
return self._file_list
def add_skip_list(self, skip_list):
@property
def processed_file_list(self) -> list[str]:
"""A list of files in the source location with any files that will be skipped removed"""
return [f for i, f in enumerate(self._file_list) if i not in self._skip_list]
def add_skip_list(self, skip_list: list[int]):
""" Add a skip list to this :class:`ImagesLoader`
Parameters
----------
skip_list: list
skip_list: list[int]
A list of indices corresponding to the frame indices that should be skipped by the
:func:`load` function.
"""
lib/logger.py
+127 -107
View File
@@ -1,6 +1,8 @@
#!/usr/bin/python
""" Logging Functions for Faceswap. """
"""Logging Functions for Faceswap."""
# NOTE: Don't import non stdlib packages. This module is accessed by setup.py
from __future__ import annotations
import collections
import logging
from logging.handlers import RotatingFileHandler
@@ -16,64 +18,70 @@ from datetime import datetime
from lib.utils import get_module_objects
if T.TYPE_CHECKING:
import numpy as np
# Add our custom levels to logger
for new_level in (("VERBOSE", 15), ("TRACE", 5)):
level_name, level_num = new_level
level_map = logging.getLevelNamesMapping()
if level_name in level_map:
continue
logging.addLevelName(level_num, level_name)
level_map[level_name] = level_num
class FaceswapLogger(logging.Logger):
""" A standard :class:`logging.logger` with additional "verbose" and "trace" levels added. """
def __init__(self, name: str) -> None:
for new_level in (("VERBOSE", 15), ("TRACE", 5)):
level_name, level_num = new_level
if hasattr(logging, level_name):
continue
logging.addLevelName(level_num, level_name)
setattr(logging, level_name, level_num)
super().__init__(name)
"""A standard :class:`logging.logger` with additional "verbose" and "trace" levels added. """
def verbose(self, msg: str, *args, **kwargs) -> None:
# pylint:disable=wrong-spelling-in-docstring
""" Create a log message at severity level 15.
"""Create a log message at severity level 15.
Parameters
----------
msg: str
msg
The log message to be recorded at Verbose level
args: tuple
args
Standard logging arguments
kwargs: dict
kwargs
Standard logging key word arguments
"""
if self.isEnabledFor(15):
kwargs.setdefault("stacklevel", 2)
self._log(15, msg, args, **kwargs)
def trace(self, msg: str, *args, **kwargs) -> None:
# pylint:disable=wrong-spelling-in-docstring
""" Create a log message at severity level 5.
"""Create a log message at severity level 5.
Parameters
----------
msg: str
msg
The log message to be recorded at Trace level
args: tuple
args
Standard logging arguments
kwargs: dict
kwargs
Standard logging key word arguments
"""
if self.isEnabledFor(5):
kwargs.setdefault("stacklevel", 2)
self._log(5, msg, args, **kwargs)
class ColoredFormatter(logging.Formatter):
""" Overrides the stand :class:`logging.Formatter` to enable colored labels for message level
"""Overrides the stand :class:`logging.Formatter` to enable colored labels for message level
labels on supported platforms
Parameters
----------
fmt: str
fmt
The format string for the message as a whole
pad_newlines: bool, Optional
pad_newlines
If ``True`` new lines will be padded to appear in line with the log message, if ``False``
they will be left aligned
kwargs: dict
kwargs
Standard :class:`logging.Formatter` keyword arguments
"""
def __init__(self, fmt: str, pad_newlines: bool = False, **kwargs) -> None:
@@ -89,13 +97,12 @@ class ColoredFormatter(logging.Formatter):
@classmethod
def _get_color_compatibility(cls) -> bool:
""" Return whether the system supports color ansi codes. Most OSes do other than Windows
"""Return whether the system supports color ansi codes. Most OSes do other than Windows
below Windows 10 version 1511.
Returns
-------
bool
``True`` if the system supports color ansi codes otherwise ``False``
``True`` if the system supports color ansi codes otherwise ``False``
"""
if platform.system().lower() != "windows":
return True
@@ -108,20 +115,19 @@ class ColoredFormatter(logging.Formatter):
return False
def _get_newline_padding(self, pad_newlines: bool, fmt: str) -> int:
""" Parses the format string to obtain padding for newlines if requested
"""Parses the format string to obtain padding for newlines if requested
Parameters
----------
fmt: str
fmt
The format string for the message as a whole
pad_newlines: bool, Optional
pad_newlines
If ``True`` new lines will be padded to appear in line with the log message, if
``False`` they will be left aligned
Returns
-------
int
The amount of padding to apply to the front of newlines
The amount of padding to apply to the front of newlines
"""
if not pad_newlines:
return 0
@@ -134,36 +140,34 @@ class ColoredFormatter(logging.Formatter):
return sum(pads) + spaces
def _get_sample_time_string(self) -> int:
""" Obtain a sample time string and calculate correct padding.
"""Obtain a sample time string and calculate correct padding.
This may be inaccurate when ticking over an integer from single to double digits, but that
shouldn't be a huge issue.
Returns
-------
int
The length of the formatted date-time string
The length of the formatted date-time string
"""
sample_time = time.time()
date_format = self.datefmt if self.datefmt else self.default_time_format
datestring = time.strftime(date_format, logging.Formatter.converter(sample_time))
date_string = time.strftime(date_format, logging.Formatter.converter(sample_time))
if not self.datefmt and self.default_msec_format:
msecs = (sample_time - int(sample_time)) * 1000
datestring = self.default_msec_format % (datestring, msecs)
return len(datestring)
m_secs = (sample_time - int(sample_time)) * 1000
date_string = self.default_msec_format % (date_string, m_secs)
return len(date_string)
def format(self, record: logging.LogRecord) -> str:
""" Color the log message level if supported otherwise return the standard log message.
"""Color the log message level if supported otherwise return the standard log message.
Parameters
----------
record: :class:`logging.LogRecord`
record
The incoming log record to be formatted for entry into the logger.
Returns
-------
str
The formatted log message
The formatted log message
"""
formatted = super().format(record)
levelname = record.levelname
@@ -178,7 +182,7 @@ class ColoredFormatter(logging.Formatter):
class FaceswapFormatter(logging.Formatter):
""" Overrides the standard :class:`logging.Formatter`.
"""Overrides the standard :class:`logging.Formatter`.
Strip newlines from incoming log messages.
@@ -186,17 +190,16 @@ class FaceswapFormatter(logging.Formatter):
"""
def format(self, record: logging.LogRecord) -> str:
""" Strip new lines from log records and rewrite certain warning messages to debug level.
"""Strip new lines from log records and rewrite certain warning messages to debug level.
Parameters
----------
record : :class:`logging.LogRecord`
record
The incoming log record to be formatted for entry into the logger.
Returns
-------
str
The formatted log message
The formatted log message
"""
record.message = record.getMessage()
record = self._lower_external(record)
@@ -224,20 +227,19 @@ class FaceswapFormatter(logging.Formatter):
@classmethod
def _lower_external(cls, record: logging.LogRecord) -> logging.LogRecord:
""" Some external libs log at a higher level than we would really like, so lower their
"""Some external libs log at a higher level than we would really like, so lower their
log level.
Specifically: Matplotlib font properties
Parameters
----------
record: :class:`logging.LogRecord`
record
The log record to check for rewriting
Returns
----------
:class:`logging.LogRecord`
The log rewritten or untouched record
The log rewritten or untouched record
"""
if (record.levelno == 20 and record.funcName == "__init__"
and record.module == "font_manager"):
@@ -253,11 +255,11 @@ class RollingBuffer(collections.deque):
crash log. """
def write(self, buffer: str) -> None:
""" Splits lines from the incoming buffer and writes them out to the rolling buffer.
"""Splits lines from the incoming buffer and writes them out to the rolling buffer.
Parameters
----------
buffer: str
buffer
The log messages to write to the rolling buffer
"""
for line in buffer.rstrip().splitlines():
@@ -265,15 +267,15 @@ class RollingBuffer(collections.deque):
class TqdmHandler(logging.StreamHandler):
""" Overrides :class:`logging.StreamHandler` to use :func:`tqdm.tqdm.write` rather than writing
"""Overrides :class:`logging.StreamHandler` to use :func:`tqdm.tqdm.write` rather than writing
to :func:`sys.stderr` so that log messages do not mess up tqdm progress bars. """
def emit(self, record: logging.LogRecord) -> None:
""" Format the incoming message and pass to :func:`tqdm.tqdm.write`.
"""Format the incoming message and pass to :func:`tqdm.tqdm.write`.
Parameters
----------
record : :class:`logging.LogRecord`
record
The incoming log record to be formatted for entry into the logger.
"""
# tqdm is imported here as it won't be installed when setup.py is running
@@ -283,17 +285,16 @@ class TqdmHandler(logging.StreamHandler):
def _set_root_logger(loglevel: int = logging.INFO) -> logging.Logger:
""" Setup the root logger.
"""Setup the root logger.
Parameters
----------
loglevel: int, optional
loglevel
The log level to set the root logger to. Default :attr:`logging.INFO`
Returns
-------
:class:`logging.Logger`
The root logger for Faceswap
The root logger for Faceswap
"""
rootlogger = logging.getLogger()
rootlogger.setLevel(loglevel)
@@ -302,21 +303,21 @@ def _set_root_logger(loglevel: int = logging.INFO) -> logging.Logger:
def log_setup(loglevel, log_file: str, command: str, is_gui: bool = False) -> None:
""" Set up logging for Faceswap.
"""Set up logging for Faceswap.
Sets up the root logger, the formatting for the crash logger and the file logger, and sets up
the crash, file and stream log handlers.
Parameters
----------
loglevel: str
loglevel
The requested log level that Faceswap should be run at.
log_file: str
log_file
The location of the log file to write Faceswap's log to
command: str
command
The Faceswap command that is being run. Used to dictate whether the log file should
have "_gui" appended to the filename or not.
is_gui: bool, optional
is_gui
Whether Faceswap is running in the GUI or not. Dictates where the stream handler should
output messages to. Default: ``False``
"""
@@ -349,24 +350,23 @@ def _file_handler(loglevel,
log_file: str,
log_format: FaceswapFormatter,
command: str) -> RotatingFileHandler:
""" Add a rotating file handler for the current Faceswap session. 1 backup is always kept.
"""Add a rotating file handler for the current Faceswap session. 1 backup is always kept.
Parameters
----------
loglevel: str
loglevel
The requested log level that messages should be logged at.
log_file: str
log_file
The location of the log file to write Faceswap's log to
log_format: :class:`FaceswapFormatter:
log_format
The formatting to store log messages as
command: str
command
The Faceswap command that is being run. Used to dictate whether the log file should
have "_gui" appended to the filename or not.
Returns
-------
:class:`logging.RotatingFileHandler`
The logging file handler
The logging file handler
"""
if log_file:
filename = log_file
@@ -385,21 +385,20 @@ def _file_handler(loglevel,
def _stream_handler(loglevel: int, is_gui: bool) -> logging.StreamHandler | TqdmHandler:
""" Add a stream handler for the current Faceswap session. The stream handler will only ever
"""Add a stream handler for the current Faceswap session. The stream handler will only ever
output at a maximum of VERBOSE level to avoid spamming the console.
Parameters
----------
loglevel: int
loglevel
The requested log level that messages should be logged at.
is_gui: bool, optional
is_gui
Whether Faceswap is running in the GUI or not. Dictates where the stream handler should
output messages to.
Returns
-------
:class:`TqdmHandler` or :class:`logging.StreamHandler`
The stream handler to use
The stream handler to use
"""
# Don't set stdout to lower than verbose
loglevel = max(loglevel, 15)
@@ -418,19 +417,18 @@ def _stream_handler(loglevel: int, is_gui: bool) -> logging.StreamHandler | Tqdm
def _stream_setup_handler(loglevel: int) -> logging.StreamHandler:
""" Add a stream handler for faceswap's setup.py script
"""Add a stream handler for faceswap's setup.py script
This stream handler outputs a limited set of easy to use information using colored labels
if available. It will only ever output at a minimum of INFO level
Parameters
----------
loglevel: int
loglevel
The requested log level that messages should be logged at.
Returns
-------
:class:`logging.StreamHandler`
The stream handler to use
The stream handler to use
"""
loglevel = max(loglevel, 15)
log_format = ColoredFormatter("%(levelname)-8s %(message)s", pad_newlines=True)
@@ -441,18 +439,17 @@ def _stream_setup_handler(loglevel: int) -> logging.StreamHandler:
def _crash_handler(log_format: FaceswapFormatter) -> logging.StreamHandler:
""" Add a handler that stores the last 100 debug lines to :attr:'_DEBUG_BUFFER' for use in
"""Add a handler that stores the last 100 debug lines to :attr:'_DEBUG_BUFFER' for use in
crash reports.
Parameters
----------
log_format: :class:`FaceswapFormatter:
log_format
The formatting to store log messages as
Returns
-------
:class:`logging.StreamHandler`
The crash log handler
The crash log handler
"""
log_crash = logging.StreamHandler(_DEBUG_BUFFER)
log_crash.setFormatter(log_format)
@@ -461,33 +458,31 @@ def _crash_handler(log_format: FaceswapFormatter) -> logging.StreamHandler:
def get_loglevel(loglevel: str) -> int:
""" Check whether a valid log level has been supplied, and return the numeric log level that
"""Check whether a valid log level has been supplied, and return the numeric log level that
corresponds to the given string level.
Parameters
----------
loglevel: str
loglevel
The loglevel that has been requested
Returns
-------
int
The numeric representation of the given loglevel
The numeric representation of the given loglevel
"""
numeric_level = getattr(logging, loglevel.upper(), None)
numeric_level = logging.getLevelNamesMapping()[loglevel.upper()]
if not isinstance(numeric_level, int):
raise ValueError(f"Invalid log level: {loglevel}")
return numeric_level
def crash_log() -> str:
""" On a crash, write out the contents of :func:`_DEBUG_BUFFER` containing the last 100 lines
"""On a crash, write out the contents of :func:`_DEBUG_BUFFER` containing the last 100 lines
of debug messages to a crash report in the root Faceswap folder.
Returns
-------
str
The filename of the file that contains the crash report
The filename of the file that contains the crash report
"""
original_traceback = traceback.format_exc().encode("utf-8")
path = os.path.dirname(os.path.realpath(sys.argv[0]))
@@ -505,18 +500,44 @@ def crash_log() -> str:
return filename
def _process_value(value: T.Any) -> T.Any:
""" Process the values from a local dict and return in a loggable format
def format_array(array: np.ndarray) -> str:
"""Format arrays to be suitable for logging
Parameters
----------
value: Any
array
The array to be formatted for logging
Returns
-------
String representation of an array for logging
"""
try:
import numpy as np # pylint:disable=import-outside-toplevel
except ImportError:
return repr(array)
if np.prod(array.shape) <= 10:
retval = "np.array("
if array.dtype == "object":
retval += f"{[x.tolist() for x in array]}"
else:
retval += str(array.tolist())
return f"{retval}, dtype='{array.dtype}')"
return f"<array(shape{array.shape}, '{array.dtype}')>"
def _process_value(value: T.Any) -> T.Any:
"""Process the values from a local dict and return in a format suitable for logging
Parameters
----------
value
The dictionary value
Returns
-------
Any
The original or ammended value
The original or amended value
"""
if isinstance(value, (list, tuple, set)) and len(value) > 10:
return f'[type: "{type(value).__name__}" len: {len(value)}'
@@ -526,26 +547,25 @@ def _process_value(value: T.Any) -> T.Any:
except ImportError:
return repr(value)
if isinstance(value, np.ndarray) and np.prod(value.shape) > 10:
return f'[type: "{type(value).__name__}" shape: {value.shape}, dtype: "{value.dtype}"]'
if isinstance(value, np.ndarray):
return format_array(value)
return repr(value)
def parse_class_init(locals_dict: dict[str, T.Any]) -> str:
""" Parse a locals dict from a class and return in a format suitable for logging
"""Parse a locals dict from a class and return in a format suitable for logging
Parameters
----------
locals_dict: dict[str, T.Any]
locals_dict
A locals() dictionary from a newly initialized class
Returns
-------
str
The locals information suitable for logging
The locals information suitable for logging
"""
delimit = {k: _process_value(v)
for k, v in locals_dict.items() if k != "self"}
for k, v in locals_dict.items() if k not in ("self", "__class__")}
dsp = ", ".join(f"{k}={v}" for k, v in delimit.items())
dsp = f"({dsp})" if dsp else ""
return f"Initializing {locals_dict['self'].__class__.__name__}{dsp}"
@@ -555,8 +575,8 @@ _OLD_FACTORY = logging.getLogRecordFactory()
def _faceswap_logrecord(*args, **kwargs) -> logging.LogRecord:
""" Add a flag to :class:`logging.LogRecord` to not strip formatting from particular
records. """
"""Add a flag to :class:`logging.LogRecord` to not strip formatting from particular
records."""
record = _OLD_FACTORY(*args, **kwargs)
record.strip_spaces = True # type:ignore
return record
lib/system/sysinfo.py
+62 -73
View File
@@ -1,5 +1,5 @@
#!/usr/bin python3
""" Obtain information about the running system, environment and GPU. """
"""Obtain information about the running system, environment and GPU."""
import json
import os
@@ -22,7 +22,7 @@ except ImportError:
class _SysInfo():
""" Obtain information about the System, Python and GPU """
"""Obtain information about the System, Python and GPU"""
def __init__(self) -> None:
self._state_file = _State().state_file
self._configs = _Configs().configs
@@ -36,40 +36,40 @@ class _SysInfo():
@property
def _ram_free(self) -> int:
""" int : The amount of free RAM in bytes. """
"""The amount of free RAM in bytes."""
if psutil is None:
return -1
return psutil.virtual_memory().free
@property
def _ram_total(self) -> int:
""" int : The amount of total RAM in bytes. """
"""The amount of total RAM in bytes."""
if psutil is None:
return -1
return psutil.virtual_memory().total
@property
def _ram_available(self) -> int:
""" int : The amount of available RAM in bytes. """
"""The amount of available RAM in bytes."""
if psutil is None:
return -1
return psutil.virtual_memory().available
@property
def _ram_used(self) -> int:
""" int : The amount of used RAM in bytes. """
"""The amount of used RAM in bytes."""
if psutil is None:
return -1
return psutil.virtual_memory().used
@property
def _fs_command(self) -> str:
""" str : The command line command used to execute faceswap. """
"""The command line command used to execute faceswap."""
return " ".join(sys.argv)
@property
def _conda_version(self) -> str:
""" str : The installed version of Conda, or `N/A` if Conda is not installed. """
"""The installed version of Conda, or `N/A` if Conda is not installed."""
if not self._system.is_conda:
return "N/A"
with Popen("conda --version", shell=True, stdout=PIPE, stderr=PIPE) as conda:
@@ -81,7 +81,7 @@ class _SysInfo():
@property
def _git_commits(self) -> str:
""" str : The last 5 git commits for the currently running Faceswap. """
"""The last 5 git commits for the currently running Faceswap."""
commits = git.get_commits(3)
if not commits:
return "Not Found"
@@ -89,14 +89,14 @@ class _SysInfo():
@property
def _cuda_versions(self) -> str:
""" str : The globally installed Cuda versions"""
"""The globally installed Cuda versions"""
if not self._cuda.versions:
return "No global Cuda versions found"
return ", ".join(".".join(str(x) for x in v) for v in self._cuda.versions)
@property
def _cuda_version(self) -> str:
""" str : The installed CUDA version. """
"""The installed CUDA version."""
if self._cuda.version == (0, 0):
retval = "No global version found"
if self._system.is_conda:
@@ -106,7 +106,7 @@ class _SysInfo():
@property
def _cudnn_versions(self) -> str:
""" str : The installed cuDNN versions. """
"""The installed cuDNN versions."""
if not self._cuda.cudnn_versions:
retval = "No global version found"
if self._system.is_conda:
@@ -121,25 +121,24 @@ class _SysInfo():
@property
def _rocm_version(self) -> str:
""" str : The default ROCm version """
"""The default ROCm version"""
if self._rocm.version == (0, 0, 0):
return "No default ROCm version found"
return ".".join(str(x) for x in self._rocm.version)
@property
def _rocm_versions(self) -> str:
""" str : The installed ROCm versions """
"""The installed ROCm versions"""
if not self._rocm.versions:
return "No ROCm versions found"
return ", ".join(".".join(str(x) for x in v) for v in self._rocm.versions)
def _get_gpu_info(self) -> GPUInfo:
""" Obtain GPU Stats. If an error is raised, swallow the error, and add to GPUInfo output
"""Obtain GPU Stats. If an error is raised, swallow the error, and add to GPUInfo output
Returns
-------
:class:`~lib.gpu_stats.GPUInfo`
The information on connected GPUs
The information on connected GPUs
"""
if GPUStats is None:
return GPUInfo(vram=[],
@@ -159,12 +158,11 @@ class _SysInfo():
return retval
def _format_ram(self) -> str:
""" Format the RAM stats into Megabytes to make it more readable.
"""Format the RAM stats into Megabytes to make it more readable.
Returns
-------
str
The total, available, used and free RAM displayed in Megabytes
The total, available, used and free RAM displayed in Megabytes
"""
retval = []
for name in ("total", "available", "used", "free"):
@@ -174,13 +172,12 @@ class _SysInfo():
return ", ".join(retval)
def full_info(self) -> str:
""" Obtain extensive system information stats, formatted into a human readable format.
"""Obtain extensive system information stats, formatted into a human readable format.
Returns
-------
str
The system information for the currently running system, formatted for output to
console or a log file.
The system information for the currently running system, formatted for output to console or
a log file.
"""
retval = "\n============ System Information ============\n"
sys_info = {"backend": get_backend(),
@@ -226,26 +223,25 @@ class _SysInfo():
def get_sysinfo() -> str:
""" Obtain extensive system information stats, formatted into a human readable format.
"""Obtain extensive system information stats, formatted into a human readable format.
If an error occurs obtaining the system information, then the error message is returned
instead.
Returns
-------
str
The system information for the currently running system, formatted for output to
console or a log file.
The system information for the currently running system, formatted for output to console or a
log file.
"""
try:
retval = _SysInfo().full_info()
except Exception as err: # pylint:disable=broad-except
retval = f"Exception occured trying to retrieve sysinfo: {str(err)}"
retval = f"Exception occurred trying to retrieve sysinfo: {str(err)}"
raise
return retval
class _Configs(): # pylint:disable=too-few-public-methods
""" Parses the config files in /faceswap/config and outputs the information stored within them
"""Parses the config files in /faceswap/config and outputs the information stored within them
in a human readable format. """
def __init__(self) -> None:
@@ -253,62 +249,59 @@ class _Configs(): # pylint:disable=too-few-public-methods
self.configs = self._get_configs()
def _get_configs(self) -> str:
""" Obtain the formatted configurations from the config folder.
"""Obtain the formatted configurations from the config folder.
Returns
-------
str
The current configuration in the config files formatted in a human readable format
The current configuration in the config files formatted in a human readable format
"""
try:
config_files = [os.path.join(self.config_dir, cfile)
for cfile in os.listdir(self.config_dir)
if os.path.basename(cfile) == ".faceswap"
or os.path.splitext(cfile)[1] == ".ini"]
config_files = [os.path.join(self.config_dir, c_file)
for c_file in os.listdir(self.config_dir)
if os.path.basename(c_file) == ".faceswap"
or os.path.splitext(c_file)[1] == ".ini"]
return self._parse_configs(config_files)
except FileNotFoundError:
return ""
def _parse_configs(self, config_files: list[str]) -> str:
""" Parse the given list of config files into a human readable format.
"""Parse the given list of config files into a human readable format.
Parameters
----------
config_files : list[str]
config_files
A list of paths to the faceswap config files
Returns
-------
str
The current configuration in the config files formatted in a human readable format
The current configuration in the config files formatted in a human readable format
"""
formatted = ""
for cfile in config_files:
fname = os.path.basename(cfile)
ext = os.path.splitext(cfile)[1]
for c_file in config_files:
fname = os.path.basename(c_file)
ext = os.path.splitext(c_file)[1]
formatted += f"\n--------- {fname} ---------\n"
if ext == ".ini":
formatted += self._parse_ini(cfile)
formatted += self._parse_ini(c_file)
elif fname == ".faceswap":
formatted += self._parse_json(cfile)
formatted += self._parse_json(c_file)
return formatted
def _parse_ini(self, config_file: str) -> str:
""" Parse an ``.ini`` formatted config file into a human readable format.
"""Parse an ``.ini`` formatted config file into a human readable format.
Parameters
----------
config_file : str
config_file
The path to the config.ini file
Returns
-------
str
The current configuration in the config file formatted in a human readable format
The current configuration in the config file formatted in a human readable format
"""
formatted = ""
with open(config_file, "r", encoding="utf-8", errors="replace") as cfile:
for line in cfile.readlines():
with open(config_file, "r", encoding="utf-8", errors="replace") as c_file:
for line in c_file.readlines():
line = line.strip()
if line.startswith("#") or not line:
continue
@@ -320,21 +313,20 @@ class _Configs(): # pylint:disable=too-few-public-methods
return formatted
def _parse_json(self, config_file: str) -> str:
""" Parse an ``.json`` formatted config file into a formatted string.
"""Parse an ``.json`` formatted config file into a formatted string.
Parameters
----------
config_file : str
config_file
The path to the config.json file
Returns
-------
dict
The current configuration in the config file formatted as a python dictionary
The current configuration in the config file formatted as a python dictionary
"""
formatted: str = ""
with open(config_file, "r", encoding="utf-8", errors="replace") as cfile:
conf_dict = json.load(cfile)
with open(config_file, "r", encoding="utf-8", errors="replace") as c_file:
conf_dict = json.load(c_file)
for key in sorted(conf_dict.keys()):
formatted += self._format_text(key, conf_dict[key])
return formatted
@@ -345,21 +337,20 @@ class _Configs(): # pylint:disable=too-few-public-methods
Parameters
----------
key : str
key
The label for this display item
value : str
value
The value for this display item
Returns
-------
str
The formatted key value pair for display
The formatted key value pair for display
"""
return f"{key.strip() + ':':<25} {value.strip()}\n"
class _State(): # pylint:disable=too-few-public-methods
""" Parses the state file in the current model directory, if the model is training, and
"""Parses the state file in the current model directory, if the model is training, and
formats the content into a human readable format. """
def __init__(self) -> None:
self._model_dir = self._get_arg("-m", "--model-dir")
@@ -368,18 +359,17 @@ class _State(): # pylint:disable=too-few-public-methods
@property
def _is_training(self) -> bool:
""" bool : ``True`` if this function has been called during a training session
otherwise ``False``. """
"""``True`` if this function has been called during a training session otherwise
``False``."""
return len(sys.argv) > 1 and sys.argv[1].lower() == "train"
@staticmethod
def _get_arg(*args: str) -> str | None:
""" Obtain the value for a given command line option from sys.argv.
"""Obtain the value for a given command line option from sys.argv.
Returns
-------
str or ``None``
The value of the given command line option, if it exists, otherwise ``None``
The value of the given command line option, if it exists, otherwise ``None``
"""
cmd = sys.argv
for opt in args:
@@ -390,12 +380,11 @@ class _State(): # pylint:disable=too-few-public-methods
return None
def _get_state_file(self) -> str:
""" Parses the model's state file and compiles the contents into a human readable string.
"""Parses the model's state file and compiles the contents into a human readable string.
Returns
-------
str
The state file formatted into a human readable format
The state file formatted into a human readable format
"""
if not self._is_training or self._model_dir is None or self._trainer is None:
return ""
@@ -404,8 +393,8 @@ class _State(): # pylint:disable=too-few-public-methods
return ""
retval = "\n\n=============== State File =================\n"
with open(fname, "r", encoding="utf-8", errors="replace") as sfile:
retval += sfile.read()
with open(fname, "r", encoding="utf-8", errors="replace") as s_file:
retval += s_file.read()
return retval