mirror of
https://github.com/ychalier/datamoshing.git
synced 2026-06-16 12:31:19 +02:00
initial commit
This commit is contained in:
Yohan Chalier
+12
@@ -0,0 +1,12 @@
|
||||
__pycache__/
|
||||
*.pyc
|
||||
|
||||
# media files
|
||||
|
||||
*.avi
|
||||
*.yuv
|
||||
*.mp4
|
||||
*.jpg
|
||||
*.png
|
||||
*.mov
|
||||
*.h264
|
||||
@@ -0,0 +1,20 @@
|
||||
# Datamoshing
|
||||
|
||||
This repository contains Python scripts to perform some datamoshing effects.
|
||||
|
||||
For details about how this works and what it does, please see [this blog article](https://chalier.fr/blog/datamoshing).
|
||||
|
||||
## Contents
|
||||
|
||||
Script | Description
|
||||
------ | -----------
|
||||
[Drop h264 I-Frames](drop-h264-iframes/) | Removes every reference frames from a video, except the first one.
|
||||
[Optical Flow Transfer](optical-flow-transfer/) | Transfer optical flow from one video to an image.
|
||||
|
||||
## Examples
|
||||
|
||||
Here are some videos made using those scripts:
|
||||
|
||||
<video src="https://i.imgur.com/bOHT26q.mp4" autoplay loop mute controls></video>
|
||||
|
||||
<video src="https://i.imgur.com/pt6Sq7A.mp4" autoplay loop mute controls></video>
|
||||
@@ -0,0 +1,87 @@
|
||||
# Drop h264 I-Frames
|
||||
|
||||
Removes every reference frames from a video, except the first one.
|
||||
|
||||
## Getting Started
|
||||
|
||||
### Prerequisites
|
||||
|
||||
You'll need a working installation of [Python 3](https://www.python.org/), and [FFmpeg](https://ffmpeg.org/). Make sure they are in PATH.
|
||||
|
||||
### Installation
|
||||
|
||||
Download or clone this repository:
|
||||
|
||||
```console
|
||||
git clone https://github.com/ychalier/datamoshing.git
|
||||
cd datamoshing/drop-h264-iframes/
|
||||
```
|
||||
|
||||
Install the requirements:
|
||||
|
||||
```console
|
||||
pip -m install requirements.txt
|
||||
```
|
||||
|
||||
### Basic usage
|
||||
|
||||
Simply execute the main script:
|
||||
|
||||
```console
|
||||
python drop_h264_iframes.py full <source-video> <output-video>
|
||||
```
|
||||
|
||||
## Usage
|
||||
|
||||
Source and output videos can be of any type, as they will be converted to h264, using FFmpeg, during the process.
|
||||
|
||||
You can pass different actions strings to perform different things:
|
||||
|
||||
Action | Description
|
||||
------ | -----------
|
||||
`preprocess` | Convert the source video to h264 and analyse its NAL units
|
||||
`rebuild` | Rebuild a video from the h264 source and the NAL units analysis
|
||||
`full` | Pipeline doing `preprocess` and `rebuild`
|
||||
`split` | Analyse the NAL units of a h264 video
|
||||
`probe` | Use FFprobe to analyse frames of a video
|
||||
|
||||
### Arguments
|
||||
|
||||
You can pass FFmpeg compression arguments that will be applied during conversion to h264. They will impact the result of the datamoshing.
|
||||
|
||||
Argument | Default | Description
|
||||
-------- | ------- | -----------
|
||||
`--sc-threshold` | `40` | Scenecut detection threshold
|
||||
`--g` | `250` | Maximum frames between two I-Frames
|
||||
`--keyint-min` | `25` | Minimum frames between two I-Frames
|
||||
`--bf` | `0` | Maximum frames between two B-Frames
|
||||
`--profile` | `main` | Encoding profile
|
||||
`--level` | `4.0` | Quality level
|
||||
`--crf` | `23` | Balance between quality and compression
|
||||
|
||||
### Tips
|
||||
|
||||
For a good effect, you may want to concatenate videos clips together. This can be done using FFmpeg.
|
||||
|
||||
1. Create a text file listing the clips to concatenate:
|
||||
```text
|
||||
file 'first.mp4'
|
||||
file 'second.mp4'
|
||||
file 'third.mp4'
|
||||
```
|
||||
2. Execute the following command:
|
||||
```console
|
||||
ffmpeg -f concat -safe 0 -i .\list.txt -c copy output.mp4
|
||||
```
|
||||
|
||||
There's more information on [FFmpeg's documentation](https://trac.ffmpeg.org/wiki/Concatenate).
|
||||
|
||||
## Example
|
||||
|
||||
<video src="https://i.imgur.com/bOHT26q.mp4" autoplay loop mute controls></video>
|
||||
|
||||
I wrote some details [on my blog](https://chalier.fr/blog/datamoshing#droppingreferenceframes).
|
||||
|
||||
## Known Issues
|
||||
|
||||
The resulting video may stutter if too many frames are dropped. It it less noticeable with high FPS videos. Unless I get to time travel, I will probably not address it.
|
||||
@@ -0,0 +1,250 @@
|
||||
import subprocess
|
||||
import argparse
|
||||
import shutil
|
||||
import csv
|
||||
import os
|
||||
import tempfile
|
||||
import json
|
||||
import tqdm
|
||||
|
||||
|
||||
NAL_UNIT_TYPES = {
|
||||
0: "Unspecified",
|
||||
1: "Coded slice of a non-IDR picture",
|
||||
2: "Coded slice data partition A",
|
||||
3: "Coded slice data partition B",
|
||||
4: "Coded slice data partition C",
|
||||
5: "Coded slice of an IDR picture",
|
||||
6: "Supplemental enhancement information (SEI)",
|
||||
7: "Sequence parameter set",
|
||||
8: "Picture parameter set",
|
||||
9: "Access unit delimiter",
|
||||
10: "End of sequence",
|
||||
11: "End of stream",
|
||||
12: "Filler data",
|
||||
13: "Sequence parameter set extension",
|
||||
14: "Prefix NAL unit",
|
||||
15: "Subset sequence parameter set",
|
||||
16: "Reserved",
|
||||
17: "Reserved",
|
||||
18: "Reserved",
|
||||
19: "Coded slice of an auxiliary coded picture without partitioning",
|
||||
20: "Coded slice extension",
|
||||
21: "Coded slice extension for depth view components",
|
||||
22: "Reserved",
|
||||
23: "Reserved",
|
||||
24: "Unspecified",
|
||||
25: "Unspecified",
|
||||
26: "Unspecified",
|
||||
27: "Unspecified",
|
||||
28: "Unspecified",
|
||||
29: "Unspecified",
|
||||
30: "Unspecified",
|
||||
31: "Unspecified",
|
||||
}
|
||||
|
||||
|
||||
def setup_directory(path):
|
||||
if os.path.isdir(path):
|
||||
shutil.rmtree(path)
|
||||
os.makedirs(path)
|
||||
|
||||
|
||||
def encode_h264(input_path, output_path, compression_options):
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
"-loglevel",
|
||||
"quiet",
|
||||
"-stats",
|
||||
"-hide_banner",
|
||||
"-i",
|
||||
input_path,
|
||||
"-an",
|
||||
"-vcodec",
|
||||
"libx264",
|
||||
]
|
||||
for key, value in compression_options.items():
|
||||
cmd += [key, value]
|
||||
cmd += [
|
||||
output_path,
|
||||
"-y"
|
||||
]
|
||||
subprocess.Popen(cmd).wait()
|
||||
|
||||
|
||||
def probe(input_path, output_path):
|
||||
path = os.path.join(output_path, "probe.json")
|
||||
output_file = open(path, "w")
|
||||
process = subprocess.Popen(
|
||||
[
|
||||
"ffprobe",
|
||||
"-v",
|
||||
"quiet",
|
||||
"-pretty",
|
||||
"-print_format",
|
||||
"json",
|
||||
"-show_entries",
|
||||
"format=size,bit_rate:frame=coded_picture_number,pkt_pts_time,pkt_pts,pkt_dts_time,pkt_dts,pkt_duration_time,pict_type,interlaced_frame,top_field_first,repeat_pict,width,height,sample_aspect_ratio,display_aspect_ratio,r_frame_rate,avg_frame_rate,time_base,pkt_size",
|
||||
"-select_streams",
|
||||
"v:0",
|
||||
input_path
|
||||
],
|
||||
stdout=output_file
|
||||
)
|
||||
process.wait()
|
||||
output_file.close()
|
||||
with open(path, "r") as file:
|
||||
data = json.load(file)
|
||||
return data["frames"]
|
||||
|
||||
|
||||
def create_nalu_entry(unit_id, index, probe_result, probe_result_index, offset_start, offset_end, nalu_header):
|
||||
forbidden_zero_bit = nalu_header >> 7
|
||||
nal_ref_idc = nalu_header >> 5 & 3
|
||||
nal_unit_type = nalu_header & 31
|
||||
size = offset_end - offset_start
|
||||
entry = {
|
||||
"id": unit_id,
|
||||
"offset_start": offset_start,
|
||||
"offset_end": offset_end,
|
||||
"size": size,
|
||||
"nalu_header": "0x%02X" % nalu_header,
|
||||
"forbidden_zero_bit": forbidden_zero_bit,
|
||||
"nal_ref_idc": nal_ref_idc,
|
||||
"nal_unit_type": nal_unit_type,
|
||||
"nal_unit_type_desc": NAL_UNIT_TYPES[nal_unit_type],
|
||||
}
|
||||
probe_result_index_mod = 0
|
||||
if nal_unit_type not in [6, 7, 8]:
|
||||
entry.update(probe_result[probe_result_index])
|
||||
probe_result_index_mod = 1
|
||||
index.append(entry)
|
||||
print(f"{unit_id}\t{size}\t{NAL_UNIT_TYPES[nal_unit_type]}")
|
||||
return unit_id + 1, probe_result_index + probe_result_index_mod
|
||||
|
||||
|
||||
def split_nalu(input_path, output_path):
|
||||
unit_id = 0
|
||||
index = []
|
||||
offset = 0
|
||||
offset_start = 0
|
||||
offset_end = None
|
||||
nalu_header = None
|
||||
buffer = b""
|
||||
probe_result = probe(input_path, output_path)
|
||||
probe_result_index = 0
|
||||
with open(input_path, "rb") as infile:
|
||||
while True:
|
||||
new_unit = False
|
||||
if len(buffer) == 3 and buffer == b'\x00\x00\x01':
|
||||
new_unit = True
|
||||
offset_end = offset - 3
|
||||
elif len(buffer) == 4 and buffer == b'\x00\x00\x00\x01':
|
||||
new_unit = True
|
||||
offset_end = offset - 4
|
||||
elif len(buffer) == 4 and buffer[1:] == b'\x00\x00\x01':
|
||||
new_unit = True
|
||||
offset_end = offset - 3
|
||||
if new_unit:
|
||||
if offset_end > offset_start:
|
||||
unit_id, probe_result_index = create_nalu_entry(unit_id, index, probe_result, probe_result_index, offset_start, offset_end, nalu_header)
|
||||
offset_start = offset_end
|
||||
next_byte = infile.read(1)
|
||||
offset += 1
|
||||
if new_unit > 0:
|
||||
nalu_header = next_byte[0]
|
||||
if next_byte == b"":
|
||||
create_nalu_entry(unit_id, index, probe_result, probe_result_index, offset_start, offset, nalu_header)
|
||||
break
|
||||
if len(buffer) < 4:
|
||||
buffer = buffer + next_byte
|
||||
else:
|
||||
buffer = buffer[1:] + next_byte
|
||||
with open(os.path.join(output_path, "nalu.csv"), "w", encoding="utf8", newline="") as file:
|
||||
writer = csv.DictWriter(file, fieldnames=get_fieldnames(index))
|
||||
writer.writeheader()
|
||||
writer.writerows(index)
|
||||
|
||||
|
||||
def get_fieldnames(items):
|
||||
fieldnames = []
|
||||
for item in items:
|
||||
for key in item:
|
||||
if key not in fieldnames:
|
||||
fieldnames.append(key)
|
||||
return fieldnames
|
||||
|
||||
|
||||
def preprocess(input_path, output_path, compression_options):
|
||||
setup_directory(output_path)
|
||||
encode_h264(input_path, os.path.join(output_path, "source.h264"), compression_options)
|
||||
split_nalu(os.path.join(output_path, "source.h264"), output_path)
|
||||
|
||||
|
||||
def rebuild(input_path, output_path):
|
||||
with open(os.path.join(input_path, "nalu.csv"), "r", encoding="utf8", newline="") as file:
|
||||
reader = csv.DictReader(file)
|
||||
index = list(reader)
|
||||
first = True
|
||||
with open(os.path.join(input_path, "output.h264"), "wb") as outfile:
|
||||
with open(os.path.join(input_path, "source.h264"), "rb") as infile:
|
||||
for nalu in tqdm.tqdm(index):
|
||||
data = infile.read(int(nalu["size"]))
|
||||
if nalu["pict_type"] == "I":
|
||||
if first:
|
||||
first = False
|
||||
else:
|
||||
continue
|
||||
outfile.write(data)
|
||||
subprocess.Popen(
|
||||
[
|
||||
"ffmpeg",
|
||||
"-loglevel",
|
||||
"quiet",
|
||||
"-stats",
|
||||
"-hide_banner",
|
||||
"-i",
|
||||
os.path.join(input_path, "output.h264"),
|
||||
output_path
|
||||
]
|
||||
).wait()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("action", type=str, choices=["preprocess", "rebuild", "full", "split", "probe"])
|
||||
parser.add_argument("input_path", type=str)
|
||||
parser.add_argument("output_path", type=str)
|
||||
parser.add_argument("-s", "--sc-threshold", type=int, default=40)
|
||||
parser.add_argument("-p", "--profile", type=str, choices=["high", "main", "baseline", "high10", "high422", "high444"], default="main")
|
||||
parser.add_argument("-l", "--level", type=float, default=4.0)
|
||||
parser.add_argument("-c", "--crf", type=int, default=23)
|
||||
parser.add_argument("-g", "--g", type=int, default=250)
|
||||
parser.add_argument("-k", "--keyint-min", type=int, default=25)
|
||||
parser.add_argument("-b", "--bf", type=int, default=0)
|
||||
args = parser.parse_args()
|
||||
compression_options = {
|
||||
"-crf": str(args.crf),
|
||||
"-sc_threshold": str(args.sc_threshold),
|
||||
"-profile:v": str(args.profile),
|
||||
"-level:v": str(args.level),
|
||||
"-g": str(args.g),
|
||||
"-keyint_min": str(args.keyint_min),
|
||||
"-bf": str(args.bf),
|
||||
}
|
||||
if args.action == "preprocess":
|
||||
preprocess(args.input_path, args.output_path, compression_options)
|
||||
elif args.action == "rebuild":
|
||||
rebuild(args.input_path, args.output_path)
|
||||
elif args.action == "full":
|
||||
tempdir = os.path.join(tempfile.gettempdir(), "foo")
|
||||
preprocess(args.input_path, tempdir, compression_options)
|
||||
rebuild(tempdir, args.output_path)
|
||||
elif args.action == "split":
|
||||
split_nalu(args.input_path, args.output_path)
|
||||
elif args.action == "probe":
|
||||
print(probe(args.input_path, args.output_path))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1 @@
|
||||
tqdm
|
||||
@@ -0,0 +1,38 @@
|
||||
# Optical Flow Transfer
|
||||
|
||||
Transfer optical flow from one video to an image.
|
||||
|
||||
## Getting Started
|
||||
|
||||
### Prerequisites
|
||||
|
||||
You'll need a working installation of [Python 3](https://www.python.org/), and [FFmpeg](https://ffmpeg.org/). Make sure they are in PATH.
|
||||
|
||||
### Installation
|
||||
|
||||
Download or clone this repository:
|
||||
|
||||
```console
|
||||
git clone https://github.com/ychalier/datamoshing.git
|
||||
cd datamoshing/optical-flow-transfer/
|
||||
```
|
||||
|
||||
Install the requirements:
|
||||
|
||||
```console
|
||||
pip -m install requirements.txt
|
||||
```
|
||||
|
||||
## Usage
|
||||
|
||||
Simply execute the main script:
|
||||
|
||||
```console
|
||||
python optical_flow_transfer.py <source-video> <source-image> <output-video>
|
||||
```
|
||||
|
||||
## Example
|
||||
|
||||
<video src="https://i.imgur.com/pt6Sq7A.mp4" autoplay loop mute controls></video>
|
||||
|
||||
I wrote some details [on my blog](https://chalier.fr/blog/datamoshing#opticalflowtransfer).
|
||||
@@ -0,0 +1,110 @@
|
||||
import os
|
||||
import shutil
|
||||
import argparse
|
||||
import subprocess
|
||||
import cv2
|
||||
import numpy
|
||||
import tqdm
|
||||
import PIL.Image
|
||||
|
||||
|
||||
def transfer_optical_flow(video_path, image_path, frame_folder=".frames"):
|
||||
if os.path.isdir(frame_folder):
|
||||
shutil.rmtree(frame_folder)
|
||||
os.makedirs(frame_folder)
|
||||
video_capture = cv2.VideoCapture(video_path)
|
||||
_, video_first_frame = video_capture.read()
|
||||
prev_gray = cv2.cvtColor(video_first_frame, cv2.COLOR_BGR2GRAY)
|
||||
reference_frame = PIL.Image.open(image_path)
|
||||
frame_index = 0
|
||||
reference_frame.save(os.path.join(
|
||||
frame_folder,
|
||||
"%06d.jpg" % frame_index
|
||||
))
|
||||
image_frame = numpy.array(reference_frame)
|
||||
height, width, depth = image_frame.shape
|
||||
framerate = video_capture.get(cv2.CAP_PROP_FPS)
|
||||
frame_count = int(video_capture.get(cv2.CAP_PROP_FRAME_COUNT))
|
||||
pbar = tqdm.tqdm(
|
||||
unit="frame",
|
||||
total=frame_count,
|
||||
desc="Computing optical flow"
|
||||
)
|
||||
base = numpy.zeros(height * width * depth, dtype=int)
|
||||
for l in range(height * width * depth):
|
||||
base[l] = l
|
||||
while video_capture.isOpened():
|
||||
pbar.update()
|
||||
frame_index += 1
|
||||
_, video_frame = video_capture.read()
|
||||
if video_frame is None:
|
||||
break
|
||||
gray = cv2.cvtColor(video_frame, cv2.COLOR_BGR2GRAY)
|
||||
flow = cv2.calcOpticalFlowFarneback(
|
||||
prev=prev_gray,
|
||||
next=gray,
|
||||
flow=None,
|
||||
pyr_scale=0.5,
|
||||
levels=3,
|
||||
winsize=15,
|
||||
iterations=3,
|
||||
poly_n=5,
|
||||
poly_sigma=1.2,
|
||||
flags=0
|
||||
).astype(int)
|
||||
flow_flat = numpy.repeat(
|
||||
flow[:, :, 1] * width * depth + flow[:, :, 0] * depth,
|
||||
depth
|
||||
)
|
||||
numpy.put(image_frame, base + flow_flat, image_frame.flat, mode="wrap")
|
||||
PIL.Image.fromarray(image_frame).save(os.path.join(
|
||||
frame_folder,
|
||||
"%06d.jpg" % frame_index
|
||||
))
|
||||
prev_gray = gray
|
||||
pbar.close()
|
||||
video_capture.release()
|
||||
return framerate
|
||||
|
||||
|
||||
def create_output_video(frame_folder, framerate, output_path=".frames"):
|
||||
subprocess.Popen([
|
||||
"ffmpeg",
|
||||
"-loglevel",
|
||||
"quiet",
|
||||
"-stats",
|
||||
"-hide_banner",
|
||||
"-framerate",
|
||||
"%.2f" % framerate,
|
||||
"-i",
|
||||
os.path.join(frame_folder, "%06d.jpg"),
|
||||
"-c:v",
|
||||
"libx264",
|
||||
"-pix_fmt",
|
||||
"yuv420p",
|
||||
output_path,
|
||||
"-y"
|
||||
]).wait()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("video_path", type=str)
|
||||
parser.add_argument("image_path", type=str)
|
||||
parser.add_argument("output_path", type=str)
|
||||
parser.add_argument("-f", "--frame-folder", type=str, default=".frames")
|
||||
args = parser.parse_args()
|
||||
framerate = transfer_optical_flow(
|
||||
args.video_path,
|
||||
args.image_path,
|
||||
args.frame_folder
|
||||
)
|
||||
create_output_video(
|
||||
args.frame_folder,
|
||||
framerate,
|
||||
args.output_path
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,4 @@
|
||||
tqdm
|
||||
Pillow
|
||||
python-opencv
|
||||
numpy
|
||||
+43
@@ -0,0 +1,43 @@
|
||||
# ffmpeg : to prores 422
|
||||
# DONNÉES
|
||||
|
||||
# video
|
||||
|
||||
wi = 3840
|
||||
he = 2160
|
||||
co = 1.5
|
||||
# coefficient
|
||||
|
||||
wi = wi * co
|
||||
he = he * co
|
||||
|
||||
scal = "neighbor" #sacle algo bilinear neighbor
|
||||
meth = "zero" #-me_method umh
|
||||
|
||||
# compression
|
||||
|
||||
br = "211277"
|
||||
ii = "20" #keyframes 1 - 100
|
||||
|
||||
#files
|
||||
|
||||
ext = "is#{co}K"
|
||||
|
||||
# DONNÉES
|
||||
|
||||
a = ARGV[0]
|
||||
aname = a.split(".")
|
||||
o = "#{aname[0]}_#{ext}_@#{br}k_M#{meth}_i#{ii}.avi"
|
||||
|
||||
puts "#{a} -> #{o}"
|
||||
|
||||
# FF SHELL
|
||||
|
||||
reverse = "" #",reverse"
|
||||
|
||||
value = %x( echo 'ffmpeg -y -i "#{a}" -vf scale=#{wi}:#{he}:flags=#{scal}#{reverse} -c:v libxvid -b:v #{br}k -sc_threshold 0 -g #{ii} -me_method #{meth} -c:a libmp3lame -b:a 256k "#{o}"' )
|
||||
value = %x[ #{value} ]
|
||||
|
||||
# -vf scale=3840:2160:flags=#{scal}
|
||||
|
||||
# ffplay -flags2 +export_mvs -i AVAFALL33_is1K_@222557k_Mumph_199999_ffo.avi -vf codecview=mv=pf+bf+bb
|
||||
Reference in New Issue
Block a user