The main class for processing video streams and files.
from earsegmentationai import VideoProcessor
VideoProcessor(
device: str = "cpu",
threshold: float = 0.5,
skip_frames: int = 0,
smooth_masks: bool = False,
temporal_window: int = 5
)
Parameters:
device
(str): Processing device (“cpu” or “cuda:0”)threshold
(float): Binary threshold for mask generation (0.0 to 1.0)skip_frames
(int): Number of frames to skip between processingsmooth_masks
(bool): Apply temporal smoothing to maskstemporal_window
(int): Window size for temporal smoothingProcess video file or camera stream.
def process(
input_source: Union[str, Path, int],
output_path: Optional[Union[str, Path]] = None,
display: bool = False,
save_masks: bool = False,
mask_dir: Optional[Union[str, Path]] = None,
max_frames: Optional[int] = None
) -> VideoStats
Parameters:
input_source
: Input video source:
output_path
(Optional): Path to save output videodisplay
(bool): Display real-time previewsave_masks
(bool): Save individual frame masksmask_dir
(Optional): Directory for saving masksmax_frames
(Optional): Maximum frames to processReturns:
VideoStats
: Processing statisticsStatistics from video processing.
Attributes:
total_frames
(int): Total frames processedframes_with_ears
(int): Frames where ears were detecteddetection_rate
(float): Percentage of frames with earsaverage_fps
(float): Average processing FPStotal_time
(float): Total processing timeoutput_path
(Optional[str]): Path to output videoframe_dimensions
(Tuple[int, int]): Video dimensions (width, height)Specialized class for real-time camera processing.
from earsegmentationai import CameraProcessor
processor = CameraProcessor(
device="cuda:0",
display_fps: bool = True,
record: bool = False
)
Start camera processing.
def start(
camera_id: int = 0,
output_path: Optional[str] = None,
duration: Optional[int] = None
) -> CameraStats
Parameters:
camera_id
(int): Camera device IDoutput_path
(Optional[str]): Path to save recordingduration
(Optional[int]): Recording duration in secondsStop camera processing.
def stop() -> None
from earsegmentationai import VideoProcessor
# Initialize processor
processor = VideoProcessor(device="cuda:0")
# Process video file
stats = processor.process(
"input_video.mp4",
output_path="output_video.mp4"
)
print(f"Processed {stats.total_frames} frames")
print(f"Detection rate: {stats.detection_rate:.1f}%")
print(f"Average FPS: {stats.average_fps:.1f}")
from earsegmentationai import CameraProcessor
# Initialize camera processor
processor = CameraProcessor(
device="cuda:0",
display_fps=True
)
# Start processing (press 'q' to stop)
stats = processor.start(
camera_id=0,
output_path="recording.mp4"
)
# Process with frame skipping for better performance
processor = VideoProcessor(
device="cuda:0",
skip_frames=2, # Process every 3rd frame
smooth_masks=True, # Temporal smoothing
temporal_window=7
)
# Process with all options
stats = processor.process(
"video.mp4",
output_path="output.mp4",
display=True,
save_masks=True,
mask_dir="masks/",
max_frames=1000
)
# Process RTSP stream
processor = VideoProcessor(device="cuda:0")
stats = processor.process(
"rtsp://192.168.1.100:554/stream",
output_path="stream_output.mp4"
)
# Process HTTP stream
stats = processor.process(
"http://example.com/stream.m3u8",
display=True
)
# Custom callback for each frame
def frame_callback(frame_num, has_ear, confidence):
print(f"Frame {frame_num}: Ear={'Yes' if has_ear else 'No'}, Conf={confidence:.2f}")
processor = VideoProcessor(device="cuda:0")
processor.set_callback(frame_callback)
stats = processor.process("video.mp4")
try:
stats = processor.process("video.mp4")
except FileNotFoundError:
print("Video file not found")
except PermissionError:
print("Cannot access camera")
except Exception as e:
print(f"Processing error: {e}")