Поскольку я хочу создавать потоки продолжительностью в час, мне нужно регулярно сбрасывать поток на диск. Кроме того, если моя программа выйдет из строя, файл должен быть действительным.
Вот мой код для класса рекордера, который я вызываю с «test.mkv» в качестве выходного файла.
Код: Выделить всё
class Recorder:
def __init__(self) -> None:
recorder_instances.append(self)
self.frame_count = 0
def start(self, output_file, width, height, fps):
self.output_container = av.open(output_file, mode="w")
self.output_stream = self.output_container.add_stream("libvpx-vp9", rate=fps)
self.output_stream.width = width
self.output_stream.height = height
self.output_stream.options = {
# Add more options as needed
}
def recieved_frame(self, frame):
for packet in self.output_stream.encode(frame):
self.output_container.mux(packet)
def stop(self):
# Flush any remaining packets
for packet in self.output_stream.encode(None):
self.output_container.mux(packet)
# Close the output container
self.output_container.close()
Код: Выделить всё
# Flush any remaining packets
for packet in self.output_stream.encode(None):
self.output_container.mux(packet)
Подробнее здесь: https://stackoverflow.com/questions/789 ... -regularly