Module wtracker.utils.frame_reader
View Source
from __future__ import annotations
import os
import glob
import numpy as np
import cv2 as cv
from wtracker.utils.path_utils import join_paths
class FrameReader:
"""
An class for reading frames from a directory of frame files.
Args:
root_folder (str): The root folder path where the frame files are located.
frame_files (list[str]): A list of frame file names.
read_format (int, optional): The format in which the frames should be read.
Attributes:
root_folder (str): The root folder path where the frame files are located.
frame_shape (tuple[int, ...]): The shape of the frame.
frame_size (tuple[int, int]): The size of the frame.
files (list[str]): The list of file paths.
read_format (int): The read format of the frame reader.
"""
def __init__(
self,
root_folder: str,
frame_files: list[str],
read_format: int = cv.IMREAD_GRAYSCALE,
):
assert os.path.exists(root_folder)
assert len(frame_files) > 0
self._root_folder = root_folder
self._files = frame_files
self._read_format = read_format
self._frame_shape = self._extract_frame_shape()
def _extract_frame_shape(self) -> tuple[int, ...]:
path = join_paths(self.root_folder, self.files[0])
frame = cv.imread(path, self._read_format)
return frame.shape
@staticmethod
def create_from_template(root_folder: str, name_format: str, read_format: int = cv.IMREAD_GRAYSCALE) -> FrameReader:
"""
Creates a FrameReader object from a file name template.
Args:
root_folder (str): The root folder where the frame files are located.
name_format (str): The format of the frame file names.
read_format (int, optional): The format in which the frames should be read.
Returns:
FrameReader: The created FrameReader object.
"""
# get all files matching name format
fmt = name_format.format("[0-9]*")
frame_paths = glob.glob(fmt, root_dir=root_folder)
frame_paths = [f for f in frame_paths if os.path.isfile(join_paths(root_folder, f))]
frame_paths = sorted(frame_paths)
return FrameReader(root_folder, frame_paths, read_format)
@staticmethod
def create_from_directory(root_folder: str, read_format: int = cv.IMREAD_GRAYSCALE) -> FrameReader:
"""
Creates a FrameReader object from a directory.
Args:
root_folder (str): The root folder containing the frame files.
read_format (int, optional): The format in which the frames should be read.
Returns:
FrameReader: The created FrameReader object.
"""
# get all files in root
frame_paths = glob.glob("*.*", root_dir=root_folder)
frame_paths = [f for f in frame_paths if os.path.isfile(join_paths(root_folder, f))]
frame_paths = sorted(frame_paths)
return FrameReader(root_folder, frame_paths, read_format)
@property
def root_folder(self) -> str:
"""
Returns the root folder path.
Returns:
str: The root folder path.
"""
return self._root_folder
@property
def frame_shape(self) -> tuple[int, ...]:
"""
Returns the shape of the frame.
Returns:
tuple[int, ...]: The shape of the frame, in format (h, w, ...).
"""
return self._frame_shape
@property
def frame_size(self) -> tuple[int, int]:
"""
Returns the size of the frame.
Returns:
tuple[int, int]: The shape of the frame, in format (h, w).
"""
return self._frame_shape[:2]
@property
def files(self) -> list[str]:
"""
Returns the list of files associated with the FrameReader object.
Returns:
list[str]: The list of file paths.
"""
return self._files
@property
def read_format(self) -> int:
"""
Returns the read format of the frame reader.
Returns:
int: The read format.
"""
return self._read_format
def __len__(self) -> int:
return len(self._files)
def __getitem__(self, idx: int) -> np.ndarray:
if idx < 0 or idx >= len(self._files):
raise IndexError("index out of bounds")
path = join_paths(self.root_folder, self.files[idx])
frame = cv.imread(path, self._read_format)
return frame.astype(np.uint8, copy=False)
def __iter__(self):
return FrameStream(self)
def make_stream(self):
"""
Creates and returns a FrameStream object using the current instance of FrameReader.
Returns:
FrameStream: A FrameStream object.
"""
return FrameStream(self)
class FrameStream:
"""
A class for streaming frames from a FrameReader object.
This class serves as an iterator for the FrameReader object.
Args:
frame_reader (FrameReader): The frame reader object.
"""
def __init__(self, frame_reader: FrameReader):
self._frame_reader = frame_reader
self._idx = -1
self.frame = None
@property
def index(self) -> int:
"""
The index of the current frame.
"""
return self._idx
def __len__(self):
return len(self._frame_reader)
def __iter__(self):
return self
def __next__(self) -> np.ndarray:
self.progress()
if not self.can_read():
raise StopIteration()
frame = self.read()
return frame
def can_read(self) -> bool:
return self._idx >= 0 and self._idx < len(self._frame_reader)
def seek(self, idx: int) -> bool:
"""
Move the index to the specified position.
Args:
idx (int): The index to seek to.
Returns:
bool: True if the index is within the valid range, False otherwise.
"""
self._idx = idx
self.frame = None
return self.can_read()
def read(self) -> np.ndarray:
"""
Read and return the frame at the current index.
Raises:
IndexError: If the index is out of bounds.
Returns:
np.ndarray: The frame at the current index.
"""
if not self.can_read():
raise IndexError("index out of bounds")
if self.frame is None:
self.frame = self._frame_reader[self._idx]
return self.frame
def progress(self, n: int = 1) -> bool:
"""
Moves the current index forward by the specified number of steps.
Args:
n (int): The number of steps to move forward.
Returns:
bool: True if the index was successfully moved forward, False otherwise.
"""
return self.seek(self._idx + n)
def reset(self):
"""
Resets the frame reader to the beginning of the steam.
"""
self.seek(-1)
class DummyReader(FrameReader):
"""
A dummy frame reader that generates empty frames of a specified resolution.
Args:
num_frames (int): The number of frames to generate.
resolution (tuple[int, int]): The resolution of the frames, in format (h, w).
colored (bool, optional): Whether the frames are colored or grayscale.
"""
def __init__(self, num_frames: int, resolution: tuple[int, int], colored: bool = True):
self.colored = colored
self._resolution = resolution
shape = (*resolution, 3) if colored else resolution
self._frame = np.full(shape, fill_value=255, dtype=np.uint8)
frames = [str(i) for i in range(num_frames)]
super().__init__(".", frame_files=frames)
def __getitem__(self, idx: int) -> np.ndarray:
return self._frame.copy()
def _extract_frame_shape(self) -> tuple[int, ...]:
if self.colored:
return (*self._resolution, 3)
return self._resolution
Classes
DummyReader
class DummyReader(
num_frames: 'int',
resolution: 'tuple[int, int]',
colored: 'bool' = True
)
A dummy frame reader that generates empty frames of a specified resolution.
Attributes
| Name | Type | Description | Default |
|---|---|---|---|
| num_frames | int | The number of frames to generate. | None |
| resolution | tuple[int, int] | The resolution of the frames, in format (h, w). | None |
| colored | bool | Whether the frames are colored or grayscale. | None |
View Source
class DummyReader(FrameReader):
"""
A dummy frame reader that generates empty frames of a specified resolution.
Args:
num_frames (int): The number of frames to generate.
resolution (tuple[int, int]): The resolution of the frames, in format (h, w).
colored (bool, optional): Whether the frames are colored or grayscale.
"""
def __init__(self, num_frames: int, resolution: tuple[int, int], colored: bool = True):
self.colored = colored
self._resolution = resolution
shape = (*resolution, 3) if colored else resolution
self._frame = np.full(shape, fill_value=255, dtype=np.uint8)
frames = [str(i) for i in range(num_frames)]
super().__init__(".", frame_files=frames)
def __getitem__(self, idx: int) -> np.ndarray:
return self._frame.copy()
def _extract_frame_shape(self) -> tuple[int, ...]:
if self.colored:
return (*self._resolution, 3)
return self._resolution
Ancestors (in MRO)
- wtracker.utils.frame_reader.FrameReader
Static methods
create_from_directory
def create_from_directory(
root_folder: 'str',
read_format: 'int' = 0
) -> 'FrameReader'
Creates a FrameReader object from a directory.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| root_folder | str | The root folder containing the frame files. | None |
| read_format | int | The format in which the frames should be read. | None |
Returns:
| Type | Description |
|---|---|
| FrameReader | The created FrameReader object. |
View Source
@staticmethod
def create_from_directory(root_folder: str, read_format: int = cv.IMREAD_GRAYSCALE) -> FrameReader:
"""
Creates a FrameReader object from a directory.
Args:
root_folder (str): The root folder containing the frame files.
read_format (int, optional): The format in which the frames should be read.
Returns:
FrameReader: The created FrameReader object.
"""
# get all files in root
frame_paths = glob.glob("*.*", root_dir=root_folder)
frame_paths = [f for f in frame_paths if os.path.isfile(join_paths(root_folder, f))]
frame_paths = sorted(frame_paths)
return FrameReader(root_folder, frame_paths, read_format)
create_from_template
def create_from_template(
root_folder: 'str',
name_format: 'str',
read_format: 'int' = 0
) -> 'FrameReader'
Creates a FrameReader object from a file name template.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| root_folder | str | The root folder where the frame files are located. | None |
| name_format | str | The format of the frame file names. | None |
| read_format | int | The format in which the frames should be read. | None |
Returns:
| Type | Description |
|---|---|
| FrameReader | The created FrameReader object. |
View Source
@staticmethod
def create_from_template(root_folder: str, name_format: str, read_format: int = cv.IMREAD_GRAYSCALE) -> FrameReader:
"""
Creates a FrameReader object from a file name template.
Args:
root_folder (str): The root folder where the frame files are located.
name_format (str): The format of the frame file names.
read_format (int, optional): The format in which the frames should be read.
Returns:
FrameReader: The created FrameReader object.
"""
# get all files matching name format
fmt = name_format.format("[0-9]*")
frame_paths = glob.glob(fmt, root_dir=root_folder)
frame_paths = [f for f in frame_paths if os.path.isfile(join_paths(root_folder, f))]
frame_paths = sorted(frame_paths)
return FrameReader(root_folder, frame_paths, read_format)
Instance variables
files
Returns the list of files associated with the FrameReader object.
frame_shape
Returns the shape of the frame.
frame_size
Returns the size of the frame.
read_format
Returns the read format of the frame reader.
root_folder
Returns the root folder path.
Methods
make_stream
def make_stream(
self
)
Creates and returns a FrameStream object using the current instance of FrameReader.
Returns:
| Type | Description |
|---|---|
| FrameStream | A FrameStream object. |
View Source
def make_stream(self):
"""
Creates and returns a FrameStream object using the current instance of FrameReader.
Returns:
FrameStream: A FrameStream object.
"""
return FrameStream(self)
FrameReader
class FrameReader(
root_folder: 'str',
frame_files: 'list[str]',
read_format: 'int' = 0
)
An class for reading frames from a directory of frame files.
Attributes
| Name | Type | Description | Default |
|---|---|---|---|
| root_folder | str | The root folder path where the frame files are located. | None |
| frame_files | list[str] | A list of frame file names. | None |
| read_format | int | The format in which the frames should be read. | None |
| root_folder | str | The root folder path where the frame files are located. | None |
| frame_shape | tuple[int, ...] | The shape of the frame. | None |
| frame_size | tuple[int, int] | The size of the frame. | None |
| files | list[str] | The list of file paths. | None |
| read_format | int | The read format of the frame reader. | None |
View Source
class FrameReader:
"""
An class for reading frames from a directory of frame files.
Args:
root_folder (str): The root folder path where the frame files are located.
frame_files (list[str]): A list of frame file names.
read_format (int, optional): The format in which the frames should be read.
Attributes:
root_folder (str): The root folder path where the frame files are located.
frame_shape (tuple[int, ...]): The shape of the frame.
frame_size (tuple[int, int]): The size of the frame.
files (list[str]): The list of file paths.
read_format (int): The read format of the frame reader.
"""
def __init__(
self,
root_folder: str,
frame_files: list[str],
read_format: int = cv.IMREAD_GRAYSCALE,
):
assert os.path.exists(root_folder)
assert len(frame_files) > 0
self._root_folder = root_folder
self._files = frame_files
self._read_format = read_format
self._frame_shape = self._extract_frame_shape()
def _extract_frame_shape(self) -> tuple[int, ...]:
path = join_paths(self.root_folder, self.files[0])
frame = cv.imread(path, self._read_format)
return frame.shape
@staticmethod
def create_from_template(root_folder: str, name_format: str, read_format: int = cv.IMREAD_GRAYSCALE) -> FrameReader:
"""
Creates a FrameReader object from a file name template.
Args:
root_folder (str): The root folder where the frame files are located.
name_format (str): The format of the frame file names.
read_format (int, optional): The format in which the frames should be read.
Returns:
FrameReader: The created FrameReader object.
"""
# get all files matching name format
fmt = name_format.format("[0-9]*")
frame_paths = glob.glob(fmt, root_dir=root_folder)
frame_paths = [f for f in frame_paths if os.path.isfile(join_paths(root_folder, f))]
frame_paths = sorted(frame_paths)
return FrameReader(root_folder, frame_paths, read_format)
@staticmethod
def create_from_directory(root_folder: str, read_format: int = cv.IMREAD_GRAYSCALE) -> FrameReader:
"""
Creates a FrameReader object from a directory.
Args:
root_folder (str): The root folder containing the frame files.
read_format (int, optional): The format in which the frames should be read.
Returns:
FrameReader: The created FrameReader object.
"""
# get all files in root
frame_paths = glob.glob("*.*", root_dir=root_folder)
frame_paths = [f for f in frame_paths if os.path.isfile(join_paths(root_folder, f))]
frame_paths = sorted(frame_paths)
return FrameReader(root_folder, frame_paths, read_format)
@property
def root_folder(self) -> str:
"""
Returns the root folder path.
Returns:
str: The root folder path.
"""
return self._root_folder
@property
def frame_shape(self) -> tuple[int, ...]:
"""
Returns the shape of the frame.
Returns:
tuple[int, ...]: The shape of the frame, in format (h, w, ...).
"""
return self._frame_shape
@property
def frame_size(self) -> tuple[int, int]:
"""
Returns the size of the frame.
Returns:
tuple[int, int]: The shape of the frame, in format (h, w).
"""
return self._frame_shape[:2]
@property
def files(self) -> list[str]:
"""
Returns the list of files associated with the FrameReader object.
Returns:
list[str]: The list of file paths.
"""
return self._files
@property
def read_format(self) -> int:
"""
Returns the read format of the frame reader.
Returns:
int: The read format.
"""
return self._read_format
def __len__(self) -> int:
return len(self._files)
def __getitem__(self, idx: int) -> np.ndarray:
if idx < 0 or idx >= len(self._files):
raise IndexError("index out of bounds")
path = join_paths(self.root_folder, self.files[idx])
frame = cv.imread(path, self._read_format)
return frame.astype(np.uint8, copy=False)
def __iter__(self):
return FrameStream(self)
def make_stream(self):
"""
Creates and returns a FrameStream object using the current instance of FrameReader.
Returns:
FrameStream: A FrameStream object.
"""
return FrameStream(self)
Descendants
- wtracker.utils.frame_reader.DummyReader
Static methods
create_from_directory
def create_from_directory(
root_folder: 'str',
read_format: 'int' = 0
) -> 'FrameReader'
Creates a FrameReader object from a directory.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| root_folder | str | The root folder containing the frame files. | None |
| read_format | int | The format in which the frames should be read. | None |
Returns:
| Type | Description |
|---|---|
| FrameReader | The created FrameReader object. |
View Source
@staticmethod
def create_from_directory(root_folder: str, read_format: int = cv.IMREAD_GRAYSCALE) -> FrameReader:
"""
Creates a FrameReader object from a directory.
Args:
root_folder (str): The root folder containing the frame files.
read_format (int, optional): The format in which the frames should be read.
Returns:
FrameReader: The created FrameReader object.
"""
# get all files in root
frame_paths = glob.glob("*.*", root_dir=root_folder)
frame_paths = [f for f in frame_paths if os.path.isfile(join_paths(root_folder, f))]
frame_paths = sorted(frame_paths)
return FrameReader(root_folder, frame_paths, read_format)
create_from_template
def create_from_template(
root_folder: 'str',
name_format: 'str',
read_format: 'int' = 0
) -> 'FrameReader'
Creates a FrameReader object from a file name template.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| root_folder | str | The root folder where the frame files are located. | None |
| name_format | str | The format of the frame file names. | None |
| read_format | int | The format in which the frames should be read. | None |
Returns:
| Type | Description |
|---|---|
| FrameReader | The created FrameReader object. |
View Source
@staticmethod
def create_from_template(root_folder: str, name_format: str, read_format: int = cv.IMREAD_GRAYSCALE) -> FrameReader:
"""
Creates a FrameReader object from a file name template.
Args:
root_folder (str): The root folder where the frame files are located.
name_format (str): The format of the frame file names.
read_format (int, optional): The format in which the frames should be read.
Returns:
FrameReader: The created FrameReader object.
"""
# get all files matching name format
fmt = name_format.format("[0-9]*")
frame_paths = glob.glob(fmt, root_dir=root_folder)
frame_paths = [f for f in frame_paths if os.path.isfile(join_paths(root_folder, f))]
frame_paths = sorted(frame_paths)
return FrameReader(root_folder, frame_paths, read_format)
Instance variables
files
Returns the list of files associated with the FrameReader object.
frame_shape
Returns the shape of the frame.
frame_size
Returns the size of the frame.
read_format
Returns the read format of the frame reader.
root_folder
Returns the root folder path.
Methods
make_stream
def make_stream(
self
)
Creates and returns a FrameStream object using the current instance of FrameReader.
Returns:
| Type | Description |
|---|---|
| FrameStream | A FrameStream object. |
View Source
def make_stream(self):
"""
Creates and returns a FrameStream object using the current instance of FrameReader.
Returns:
FrameStream: A FrameStream object.
"""
return FrameStream(self)
FrameStream
class FrameStream(
frame_reader: 'FrameReader'
)
A class for streaming frames from a FrameReader object.
This class serves as an iterator for the FrameReader object.
Attributes
| Name | Type | Description | Default |
|---|---|---|---|
| frame_reader | FrameReader | The frame reader object. | None |
View Source
class FrameStream:
"""
A class for streaming frames from a FrameReader object.
This class serves as an iterator for the FrameReader object.
Args:
frame_reader (FrameReader): The frame reader object.
"""
def __init__(self, frame_reader: FrameReader):
self._frame_reader = frame_reader
self._idx = -1
self.frame = None
@property
def index(self) -> int:
"""
The index of the current frame.
"""
return self._idx
def __len__(self):
return len(self._frame_reader)
def __iter__(self):
return self
def __next__(self) -> np.ndarray:
self.progress()
if not self.can_read():
raise StopIteration()
frame = self.read()
return frame
def can_read(self) -> bool:
return self._idx >= 0 and self._idx < len(self._frame_reader)
def seek(self, idx: int) -> bool:
"""
Move the index to the specified position.
Args:
idx (int): The index to seek to.
Returns:
bool: True if the index is within the valid range, False otherwise.
"""
self._idx = idx
self.frame = None
return self.can_read()
def read(self) -> np.ndarray:
"""
Read and return the frame at the current index.
Raises:
IndexError: If the index is out of bounds.
Returns:
np.ndarray: The frame at the current index.
"""
if not self.can_read():
raise IndexError("index out of bounds")
if self.frame is None:
self.frame = self._frame_reader[self._idx]
return self.frame
def progress(self, n: int = 1) -> bool:
"""
Moves the current index forward by the specified number of steps.
Args:
n (int): The number of steps to move forward.
Returns:
bool: True if the index was successfully moved forward, False otherwise.
"""
return self.seek(self._idx + n)
def reset(self):
"""
Resets the frame reader to the beginning of the steam.
"""
self.seek(-1)
Descendants
- wtracker.sim.view_controller.ViewController
Instance variables
index
The index of the current frame.
Methods
can_read
def can_read(
self
) -> 'bool'
View Source
def can_read(self) -> bool:
return self._idx >= 0 and self._idx < len(self._frame_reader)
progress
def progress(
self,
n: 'int' = 1
) -> 'bool'
Moves the current index forward by the specified number of steps.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| n | int | The number of steps to move forward. | None |
Returns:
| Type | Description |
|---|---|
| bool | True if the index was successfully moved forward, False otherwise. |
View Source
def progress(self, n: int = 1) -> bool:
"""
Moves the current index forward by the specified number of steps.
Args:
n (int): The number of steps to move forward.
Returns:
bool: True if the index was successfully moved forward, False otherwise.
"""
return self.seek(self._idx + n)
read
def read(
self
) -> 'np.ndarray'
Read and return the frame at the current index.
Returns:
| Type | Description |
|---|---|
| np.ndarray | The frame at the current index. |
Raises:
| Type | Description |
|---|---|
| IndexError | If the index is out of bounds. |
View Source
def read(self) -> np.ndarray:
"""
Read and return the frame at the current index.
Raises:
IndexError: If the index is out of bounds.
Returns:
np.ndarray: The frame at the current index.
"""
if not self.can_read():
raise IndexError("index out of bounds")
if self.frame is None:
self.frame = self._frame_reader[self._idx]
return self.frame
reset
def reset(
self
)
Resets the frame reader to the beginning of the steam.
View Source
def reset(self):
"""
Resets the frame reader to the beginning of the steam.
"""
self.seek(-1)
seek
def seek(
self,
idx: 'int'
) -> 'bool'
Move the index to the specified position.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| idx | int | The index to seek to. | None |
Returns:
| Type | Description |
|---|---|
| bool | True if the index is within the valid range, False otherwise. |
View Source
def seek(self, idx: int) -> bool:
"""
Move the index to the specified position.
Args:
idx (int): The index to seek to.
Returns:
bool: True if the index is within the valid range, False otherwise.
"""
self._idx = idx
self.frame = None
return self.can_read()