Skip to content

Rec

VideoRecorder

Video Recorder (WIP)

Example

from tolvera import Tolvera, run, VideoRecorder def main(kwargs): tv = Tolvera(kwargs) vid = VideoRecorder(tv, **kwargs) @tv.cleanup def write(): vid.write() @tv.render def _(): vid() tv.px.diffuse(0.99) tv.v.flock(tv.p) tv.px.particles(tv.p, tv.s.species()) return tv.px

Source code in src/tolvera/rec.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@ti.data_oriented
class VideoRecorder:
    """Video Recorder (WIP)

    Example:
        from tolvera import Tolvera, run, VideoRecorder
        def main(**kwargs):
            tv = Tolvera(**kwargs)
            vid = VideoRecorder(tv, **kwargs)
            @tv.cleanup
            def write():
                vid.write()
            @tv.render
            def _():
                vid()
                tv.px.diffuse(0.99)
                tv.v.flock(tv.p)
                tv.px.particles(tv.p, tv.s.species())
                return tv.px
    """
    def __init__(self, tolvera, **kwargs) -> None:
        """Initialise a video recorder for a Tölvera program.

        Args:
            tolvera (Tolvera): Tölvera instance to record.
            f (int): Number of frames to record. Defaults to 16.
            r (int): Ratio of frames to record (tv.ti.fps/r). Defaults to 4.
            c (int): Frame counter. Defaults to 0.
            w (int): Width of video. Defaults to tv.x.
            h (int): Height of video. Defaults to tv.y.
            output_dir (str): Output directory. Defaults to './output'.
            filename (str): Output filename. Defaults to 'output'.
            automatic_build (bool): Automatically build video. Defaults to True.
            build_mp4 (bool): Build mp4. Defaults to True.
            build_gif (bool): Build gif. Defaults to False.
            clean_frames (bool): Clean frames. Defaults to True.
        """
        self.tv = tolvera
        self.f = kwargs.get('f', 16) # number of frames to record
        self.r = kwargs.get('r', 4) # ratio of frames to record (tv.ti.fps/r)
        self.c = kwargs.get('c', 0) # frame counter
        self.w = kwargs.get('w', self.tv.x) # width
        self.h = kwargs.get('h', self.tv.y) # height
        self.output_dir = kwargs.get('', './output')
        self.filename = f"{datetime.now().strftime(DT_FMT)}_{kwargs.get('filename', 'output')}"
        self.automatic_build = kwargs.get('automatic_build', True)
        self.build_mp4 = kwargs.get('build_mp4', True)
        self.build_gif = kwargs.get('build_gif', False)
        self.clean_frames = kwargs.get('clean_frames', True)
        self.framerate = kwargs.get('framerate', 24)
        self.video_manager = ti.tools.VideoManager(output_dir=self.output_dir, video_filename=self.filename, width=self.w, height=self.h, framerate=self.framerate, automatic_build=False)
        self.vid = Pixel.field(shape=(self.tv.x, self.tv.y, self.f))
        self.px = Pixel.field(shape=(self.tv.x, self.tv.y))
        print(f"[VideoRecorder] {self.w}x{self.h} every {self.r} frames {self.f} times to {self.output_dir}/{self.filename}.")

    @ti.kernel
    def rec(self, i: ti.i32):
        """Record the current frame to the video.

        Args:
            i (ti.i32): Frame index.
        """
        for x, y in ti.ndrange(self.tv.x, self.tv.y):
            self.vid[x, y, i].rgba = self.tv.px.px.rgba[x, y]

    @ti.kernel
    def dump(self, i: ti.i32):
        """Dump the current frame to the video.

        Args:
            i (ti.i32): Frame index.
        """
        for x, y in ti.ndrange(self.tv.x, self.tv.y):
            self.px.rgba[x, y] = self.vid[x, y, i].rgba

    def write_frame(self, i: int):
        """Write a frame to the video.

        Args:
            i (int): Frame index.
        """
        self.dump(i)
        self.video_manager.write_frame(self.px.rgba)

    def write(self):
        """Write all frames to the video and build if necessary."""
        print(f"[VideoRecorder] Writing {self.f} frames to {self.filename}")
        for i in tqdm(range(self.f)):
            self.write_frame(i)
        if self.automatic_build:
            print(f"[VideoRecorder] Building {self.filename} with mp4={self.build_mp4} and gif={self.build_gif}")
            self.video_manager.make_video(mp4=self.build_mp4, gif=self.build_gif)
        if self.clean_frames:
            print(f"[VideoRecorder] Cleaning {self.filename} frames")
            self.clean()

    def clean(self):
        """Delete all previous image files in the saved directory.

        Fixed version, see https://github.com/taichi-dev/taichi/issues/8533
        """
        for fn in os.listdir(self.video_manager.frame_directory):
            if fn.endswith(".png") and fn in self.video_manager.frame_fns:
                os.remove(f"{self.video_manager.frame_directory}/{fn}")

    def step(self):
        """Record the current frame and increment the frame counter."""
        i = self.tv.ctx.i[None]
        if i % self.r == 0:
            self.rec(self.c)
            self.c += 1
        if i == self.f*self.r:
            self.tv.ctx.stop()

    def __call__(self, *args, **kwds):
        """Record the current frame and increment the frame counter."""
        self.step()

__call__(*args, **kwds)

Record the current frame and increment the frame counter.

Source code in src/tolvera/rec.py 
141
142
143
def __call__(self, *args, **kwds):
    """Record the current frame and increment the frame counter."""
    self.step()

__init__(tolvera, **kwargs)

Initialise a video recorder for a Tölvera program.

Parameters:

Name Type Description Default
tolvera Tolvera

Tölvera instance to record.

 required
f int

Number of frames to record. Defaults to 16.

 required
r int

Ratio of frames to record (tv.ti.fps/r). Defaults to 4.

 required
c int

Frame counter. Defaults to 0.

 required
w int

Width of video. Defaults to tv.x.

 required
h int

Height of video. Defaults to tv.y.

 required
output_dir str

Output directory. Defaults to './output'.

 required
filename str

Output filename. Defaults to 'output'.

 required
automatic_build bool

Automatically build video. Defaults to True.

 required
build_mp4 bool

Build mp4. Defaults to True.

 required
build_gif bool

Build gif. Defaults to False.

 required
clean_frames bool

Clean frames. Defaults to True.

 required
Source code in src/tolvera/rec.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def __init__(self, tolvera, **kwargs) -> None:
    """Initialise a video recorder for a Tölvera program.

    Args:
        tolvera (Tolvera): Tölvera instance to record.
        f (int): Number of frames to record. Defaults to 16.
        r (int): Ratio of frames to record (tv.ti.fps/r). Defaults to 4.
        c (int): Frame counter. Defaults to 0.
        w (int): Width of video. Defaults to tv.x.
        h (int): Height of video. Defaults to tv.y.
        output_dir (str): Output directory. Defaults to './output'.
        filename (str): Output filename. Defaults to 'output'.
        automatic_build (bool): Automatically build video. Defaults to True.
        build_mp4 (bool): Build mp4. Defaults to True.
        build_gif (bool): Build gif. Defaults to False.
        clean_frames (bool): Clean frames. Defaults to True.
    """
    self.tv = tolvera
    self.f = kwargs.get('f', 16) # number of frames to record
    self.r = kwargs.get('r', 4) # ratio of frames to record (tv.ti.fps/r)
    self.c = kwargs.get('c', 0) # frame counter
    self.w = kwargs.get('w', self.tv.x) # width
    self.h = kwargs.get('h', self.tv.y) # height
    self.output_dir = kwargs.get('', './output')
    self.filename = f"{datetime.now().strftime(DT_FMT)}_{kwargs.get('filename', 'output')}"
    self.automatic_build = kwargs.get('automatic_build', True)
    self.build_mp4 = kwargs.get('build_mp4', True)
    self.build_gif = kwargs.get('build_gif', False)
    self.clean_frames = kwargs.get('clean_frames', True)
    self.framerate = kwargs.get('framerate', 24)
    self.video_manager = ti.tools.VideoManager(output_dir=self.output_dir, video_filename=self.filename, width=self.w, height=self.h, framerate=self.framerate, automatic_build=False)
    self.vid = Pixel.field(shape=(self.tv.x, self.tv.y, self.f))
    self.px = Pixel.field(shape=(self.tv.x, self.tv.y))
    print(f"[VideoRecorder] {self.w}x{self.h} every {self.r} frames {self.f} times to {self.output_dir}/{self.filename}.")

clean()

Delete all previous image files in the saved directory.

Fixed version, see https://github.com/taichi-dev/taichi/issues/8533

Source code in src/tolvera/rec.py 
123
124
125
126
127
128
129
130
def clean(self):
    """Delete all previous image files in the saved directory.

    Fixed version, see https://github.com/taichi-dev/taichi/issues/8533
    """
    for fn in os.listdir(self.video_manager.frame_directory):
        if fn.endswith(".png") and fn in self.video_manager.frame_fns:
            os.remove(f"{self.video_manager.frame_directory}/{fn}")

dump(i)

Dump the current frame to the video.

Parameters:

Name Type Description Default
i i32

Frame index.

 required
Source code in src/tolvera/rec.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
@ti.kernel
def dump(self, i: ti.i32):
    """Dump the current frame to the video.

    Args:
        i (ti.i32): Frame index.
    """
    for x, y in ti.ndrange(self.tv.x, self.tv.y):
        self.px.rgba[x, y] = self.vid[x, y, i].rgba

rec(i)

Record the current frame to the video.

Parameters:

Name Type Description Default
i i32

Frame index.

 required
Source code in src/tolvera/rec.py 
82
83
84
85
86
87
88
89
90
@ti.kernel
def rec(self, i: ti.i32):
    """Record the current frame to the video.

    Args:
        i (ti.i32): Frame index.
    """
    for x, y in ti.ndrange(self.tv.x, self.tv.y):
        self.vid[x, y, i].rgba = self.tv.px.px.rgba[x, y]

step()

Record the current frame and increment the frame counter.

Source code in src/tolvera/rec.py 
132
133
134
135
136
137
138
139
def step(self):
    """Record the current frame and increment the frame counter."""
    i = self.tv.ctx.i[None]
    if i % self.r == 0:
        self.rec(self.c)
        self.c += 1
    if i == self.f*self.r:
        self.tv.ctx.stop()

write()

Write all frames to the video and build if necessary.

Source code in src/tolvera/rec.py 
111
112
113
114
115
116
117
118
119
120
121
def write(self):
    """Write all frames to the video and build if necessary."""
    print(f"[VideoRecorder] Writing {self.f} frames to {self.filename}")
    for i in tqdm(range(self.f)):
        self.write_frame(i)
    if self.automatic_build:
        print(f"[VideoRecorder] Building {self.filename} with mp4={self.build_mp4} and gif={self.build_gif}")
        self.video_manager.make_video(mp4=self.build_mp4, gif=self.build_gif)
    if self.clean_frames:
        print(f"[VideoRecorder] Cleaning {self.filename} frames")
        self.clean()

write_frame(i)

Write a frame to the video.

Parameters:

Name Type Description Default
i int

Frame index.

 required
Source code in src/tolvera/rec.py 
102
103
104
105
106
107
108
109
def write_frame(self, i: int):
    """Write a frame to the video.

    Args:
        i (int): Frame index.
    """
    self.dump(i)
    self.video_manager.write_frame(self.px.rgba)