videoflow.producers package

This package contains a collection of producers that implement the videoflow.core.node.ProducerNode interface

Submodules

videoflow.producers.basic module

class videoflow.producers.basic.IntProducer(start_value: int = 0, end_value: int | None = None, wait_time_in_seconds: float = 0, fps: float = -1, **kwargs)

Bases: ProducerNode

Each time the next method is called, produces the next integer in the sequence between start_value and end_value.

If wait_time_in_seconds is greater than zero, it sleeps for the specified amount of seconds each time next() is called.

If fps is given a value greater than 0, fps overrides the value of wait_time_in_seconds

next() → int

• Returns:

  • next: an integer

videoflow.producers.video module

class videoflow.producers.video.ImageFolderReader(**kwargs)

Bases: ProducerNode

Reads from a folder of images and returns them one by one. Passes through images in alphabetical order.

close() → None

This method is called by the task running after finishing doing all consuming, processing or producing because of and end signal receival. Should be used to close any resources that were opened by the open() method, such as files, tensorflow sessions, etc.

next() → ndarray

Returns next produced element.

Raises StopIteration after the last element has been produced and a call to self.next happens.

open() → None

This method is called by the task runner before doing any consuming, processing or producing. Should be used to open any resources that will be needed during the life of the task, such as opening files, tensorflow sessions, etc.

class videoflow.producers.video.ImageProducer(image_path: str, **kwargs)

Bases: ProducerNode

Reads a single image and produces it

close() → None

This method is called by the task running after finishing doing all consuming, processing or producing because of and end signal receival. Should be used to close any resources that were opened by the open() method, such as files, tensorflow sessions, etc.

next() → ndarray

Returns image in RGB format.

open() → None

This method is called by the task runner before doing any consuming, processing or producing. Should be used to open any resources that will be needed during the life of the task, such as opening files, tensorflow sessions, etc.

videoflow.producers.video.TIMESTAMP_CLOCK = 'clock'

clock stamps each frame with the wall-clock time it was read (live cameras/streams on NTP/PTP-disciplined hosts); position stamps it with the frame’s position on the video’s own timeline (CAP_PROP_POS_MSEC — synchronized recordings played back together).

Type:

Timestamp sources for VideostreamReader

class videoflow.producers.video.VideoDeviceReader(device_id: int, nb_frames: int = -1, nb_retries: int = 0, is_finite: bool = False, **kwargs)

Bases: VideostreamReader

Opens a video capture object and returns subsequent frames from the video device each time next is called.

• Arguments:

  • device_id: id of the video device connected to the computer

  • nb_frames: number of frames to process. -1 means all of them

get_params() → dict

Returns a JSON-serializable dict of keyword arguments sufficient to reconstruct an equivalent node via type(node)(**node.get_params()). This is what lets a worker process (running in its own container, with no access to the object that built the graph) recreate the exact node it is responsible for running.

Default implementation: walks every class in this node’s MRO that declares its own __init__, inspects that __init__’s named parameters (skipping *args/**kwargs), and for each parameter name looks up self._<name> and then self.<name>. This matches the prevailing convention in this codebase of storing constructor arguments verbatim as an underscore-prefixed attribute of the same name.

Subclasses whose constructor arguments aren’t stored under a matching attribute name, or that accept non-serializable arguments (e.g. references to other nodes, as TaskModuleNode does), must override this method.

class videoflow.producers.video.VideoFileReader(video_file: str, swap_channels: bool = False, nb_frames: int = -1, timestamp_source: str = 'position', **kwargs)

Bases: VideostreamReader

Opens a video capture object and returns subsequent frames from the video file each time next is called. - Arguments:

• video_file: path to video file

• swap_channels: If true, swaps from BGR to RGB

• nb_frames: number of frames to process. -1 means all of them

• timestamp_source: defaults to position (the file’s own timeline), which is what synchronized recordings replayed together should align on.

get_params() → dict

Returns a JSON-serializable dict of keyword arguments sufficient to reconstruct an equivalent node via type(node)(**node.get_params()). This is what lets a worker process (running in its own container, with no access to the object that built the graph) recreate the exact node it is responsible for running.

Default implementation: walks every class in this node’s MRO that declares its own __init__, inspects that __init__’s named parameters (skipping *args/**kwargs), and for each parameter name looks up self._<name> and then self.<name>. This matches the prevailing convention in this codebase of storing constructor arguments verbatim as an underscore-prefixed attribute of the same name.

Subclasses whose constructor arguments aren’t stored under a matching attribute name, or that accept non-serializable arguments (e.g. references to other nodes, as TaskModuleNode does), must override this method.

class videoflow.producers.video.VideoFolderReader(**kwargs)

Bases: ProducerNode

Reads videos from a folder of videos and returns the frames of the videos one by one. Passes through videos in alphabetical order.

close() → None

This method is called by the task running after finishing doing all consuming, processing or producing because of and end signal receival. Should be used to close any resources that were opened by the open() method, such as files, tensorflow sessions, etc.

next() → ndarray

Returns next produced element.

Raises StopIteration after the last element has been produced and a call to self.next happens.

open() → None

This method is called by the task runner before doing any consuming, processing or producing. Should be used to open any resources that will be needed during the life of the task, such as opening files, tensorflow sessions, etc.

class videoflow.producers.video.VideoUrlReader(url: str, nb_frames: int = -1, nb_retries: int = 0, is_finite: bool = False, **kwargs)

Bases: VideostreamReader

Opens a video capture object and returns subsequent frames from the video url each time next is called.

• Arguments:

  • device_id: id of the video device connected to the computer

  • nb_frames: number of frames to process. -1 means all of them

get_params() → dict

Returns a JSON-serializable dict of keyword arguments sufficient to reconstruct an equivalent node via type(node)(**node.get_params()). This is what lets a worker process (running in its own container, with no access to the object that built the graph) recreate the exact node it is responsible for running.

Default implementation: walks every class in this node’s MRO that declares its own __init__, inspects that __init__’s named parameters (skipping *args/**kwargs), and for each parameter name looks up self._<name> and then self.<name>. This matches the prevailing convention in this codebase of storing constructor arguments verbatim as an underscore-prefixed attribute of the same name.

Subclasses whose constructor arguments aren’t stored under a matching attribute name, or that accept non-serializable arguments (e.g. references to other nodes, as TaskModuleNode does), must override this method.

videoflow.producers.video.VideofileReader

alias of VideoFileReader

class videoflow.producers.video.VideostreamReader(url_or_deviceid: int | str, swap_channels: bool = True, nb_frames: int = -1, nb_retries: int = 0, is_finite: bool = True, timestamp_source: str = 'clock', **kwargs)

Bases: ProducerNode

Reader of video streams, using cv2

• Arguments:

  • url_or_deviceid: (int or str) The url, filesystem path or id of the video stream.

  • swap_channels: if True, it will change channels from BGR to RGB

  • nb_frames: (int) The number of frames when to stop. -1 never stops

  • nb_retries: (int) If there are errors reading the stream, how many times to retry.

  • timestamp_source: (str) how each frame’s event time is stamped (see TIMESTAMP_SOURCES): clock (default — wall clock at read) or position (the video’s own timeline). The stamp is attached to the published message via ctx.set_event_timestamp so time-aligned joins downstream can synchronize this stream with others.

close() → None

Releases the video stream object

next(ctx: RuntimeContext | None = None) → tuple[int, ndarray]

• Returns:

  • frame no / index : integer value of the frame read

  • frame: np.ndarray of shape (h, w, 3)

When run by a task (which passes the runtime ctx), each frame’s event time is stamped on the published message per timestamp_source.

• Raises:

  • StopIteration: after it finishes reading the videofile or when it reaches the specified number of frames to process, or if it reaches the number of retries wihout success.

open() → None

Opens the video stream