Upload 12 files

genz/Dockerfile

@@ -0,0 +1,33 @@
+# Use an official Python runtime as a parent image
+FROM python:3.10-slim-buster
+
+# Set the working directory in the Docker image
+WORKDIR /app
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    libgl1 \
+    libglib2.0-0 \
+    libzbar0 \
+    ffmpeg \
+    git \
+ && rm -rf /var/lib/apt/lists/*
+
+RUN pip install uv
+
+# Clone the repository
+RUN git clone --depth 1 --branch master https://github.com/karaketir16/file2video.git /app
+
+# Copy the current directory contents into the container at /app
+#COPY . /app
+
+RUN uv venv
+
+# Install any needed packages specified in requirements.txt
+RUN uv pip install --no-cache-dir -r requirements.txt
+
+WORKDIR /data
+
+# Run file2video.py when the container launches
+ENTRYPOINT ["/app/.venv/bin/python", "/app/file2video.py", "--docker"]
+CMD []

genz/checksum.py

@@ -0,0 +1,12 @@
+import hashlib
+
+def checksum(large_file):
+    """Calculate MD5 checksum for file."""
+    md5_object = hashlib.md5()
+    block_size = 128 * md5_object.block_size
+    with open(large_file, 'rb') as a_file:
+        chunk = a_file.read(block_size)
+        while chunk:
+            md5_object.update(chunk)
+            chunk = a_file.read(block_size)
+    return md5_object.hexdigest()

genz/convert.py

@@ -0,0 +1,62 @@
+from encode import create_video
+from decode_video import decode
+from youtube_decode import youtube_decode
+import argparse
+import sys
+
+def enc_file(source_file, output_video):
+    print (f"Encoding {source_file} to {output_video}")
+    create_video(source_file, output_video)
+
+def dec_video(source_video, destination_folder, docker_mode):
+    print (f"Decoding {source_video} to {destination_folder}")
+    decode(source_video, destination_folder)
+
+def y_decode(video_url, destination_folder, docker_mode):
+    print (f"Decoding {video_url} to {destination_folder}")
+    youtube_decode(video_url, destination_folder)
+
+def main():
+
+    # First, check if '--docker' is in the command line arguments
+    docker_mode = '--docker' in sys.argv
+    if docker_mode:
+        sys.argv.remove('--docker')  # Remove it so it doesn't interfere with the main parser
+
+    docker_usage = """\
+docker run --rm -v $(pwd):/data karaketir16/file2video [-h]  [--encode source_file output_video] 
+                                                        [--decode source_video destination_folder] 
+                                                        [--youtube-decode youtube_url destination_folder]"""
+    
+    if docker_mode:
+        parser = argparse.ArgumentParser(description="Program to encode files into videos and decode videos back to files.", usage=docker_usage)
+    else:
+        parser = argparse.ArgumentParser(description="Program to encode files into videos and decode videos back to files.")
+
+    # Optional argument for encoding
+    parser.add_argument("--encode", nargs=2, metavar=('source_file', 'output_video'), 
+                        help="Encode a file into a video: source_file output_video.mp4")
+    
+    # Optional argument for decoding
+    parser.add_argument("--decode", nargs=2, metavar=('source_video', 'destination_folder'), 
+                        help="Decode a video to a file: source_video.mp4 destination_folder")
+    
+    # Optional argument for YouTube video decoding
+    parser.add_argument("--youtube-decode", nargs=2, metavar=('youtube_url', 'destination_folder'), 
+                        help="Decode a video from a YouTube URL to a file: 'youtube_url' destination_folder")
+
+    
+    args = parser.parse_args()
+
+    # Check which command is used and call the corresponding function
+    if args.encode:
+        enc_file(*args.encode)
+    elif args.decode:
+        dec_video(*args.decode, docker_mode)
+    elif args.youtube_decode:
+        y_decode(*args.youtube_decode, docker_mode)
+    else:
+        parser.print_help()
+
+if __name__ == "__main__":
+    main()

genz/decode/usage.md

@@ -0,0 +1,114 @@
+python file2video.py --encode go.txt out.mp4 
+python file2video.py --decode out.mp4 ./     
+```python
+import cv2
+import sys
+import yt_dlp
+import os
+from datetime import datetime
+
+from decode_video import decode_video
+
+def youtube_decode(src, dest_folder):
+    base_filename = "downloaded_video"
+    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+    output_file = f"{dest_folder}/{base_filename}_{timestamp}.mp4"
+
+    # Create a yt-dlp configuration to download the video
+    ydl_opts = {
+        'format': 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best',
+        'outtmpl': output_file,
+        'noplaylist': True,
+        'quiet': True,
+        'merge_output_format': 'mp4',  # Ensure the output is mp4 if separate streams are downloaded
+    }
+
+    # Use yt-dlp to download the video
+    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+        ydl.download([src])
+
+    # Check if the file is downloaded and exists
+    if os.path.exists(output_file):
+        # Open the downloaded video file using OpenCV
+        cap = cv2.VideoCapture(output_file)
+
+        # Read and process the video
+        decode_video(cap, dest_folder)
+    else:
+        print("Failed to download video.")
+        sys.exit(1)
+
+if __name__ == '__main__':
+    if len(sys.argv) < 3:
+        print("Usage: python video2file.py \"https://video_url\" destination_folder")
+        sys.exit(1)
+
+    src = sys.argv[1]
+    dest_folder = sys.argv[2]
+    youtube_decode(src, dest_folder)
+
+```
+```python
+from encode import create_video
+from decode_video import decode
+from youtube_decode import youtube_decode
+import argparse
+import sys
+
+def enc_file(source_file, output_video):
+    print (f"Encoding {source_file} to {output_video}")
+    create_video(source_file, output_video)
+
+def dec_video(source_video, destination_folder, docker_mode):
+    print (f"Decoding {source_video} to {destination_folder}")
+    decode(source_video, destination_folder)
+
+def y_decode(video_url, destination_folder, docker_mode):
+    print (f"Decoding {video_url} to {destination_folder}")
+    youtube_decode(video_url, destination_folder)
+
+def main():
+
+    # First, check if '--docker' is in the command line arguments
+    docker_mode = '--docker' in sys.argv
+    if docker_mode:
+        sys.argv.remove('--docker')  # Remove it so it doesn't interfere with the main parser
+
+    docker_usage = """\
+docker run --rm -v $(pwd):/data karaketir16/file2video [-h]  [--encode source_file output_video] 
+                                                        [--decode source_video destination_folder] 
+                                                        [--youtube-decode youtube_url destination_folder]"""
+    
+    if docker_mode:
+        parser = argparse.ArgumentParser(description="Program to encode files into videos and decode videos back to files.", usage=docker_usage)
+    else:
+        parser = argparse.ArgumentParser(description="Program to encode files into videos and decode videos back to files.")
+
+    # Optional argument for encoding
+    parser.add_argument("--encode", nargs=2, metavar=('source_file', 'output_video'), 
+                        help="Encode a file into a video: source_file output_video.mp4")
+    
+    # Optional argument for decoding
+    parser.add_argument("--decode", nargs=2, metavar=('source_video', 'destination_folder'), 
+                        help="Decode a video to a file: source_video.mp4 destination_folder")
+    
+    # Optional argument for YouTube video decoding
+    parser.add_argument("--youtube-decode", nargs=2, metavar=('youtube_url', 'destination_folder'), 
+                        help="Decode a video from a YouTube URL to a file: 'youtube_url' destination_folder")
+
+    
+    args = parser.parse_args()
+
+    # Check which command is used and call the corresponding function
+    if args.encode:
+        enc_file(*args.encode)
+    elif args.decode:
+        dec_video(*args.decode, docker_mode)
+    elif args.youtube_decode:
+        y_decode(*args.youtube_decode, docker_mode)
+    else:
+        parser.print_help()
+
+if __name__ == "__main__":
+    main()
+```

genz/decode_video.py

@@ -0,0 +1,103 @@
+import cv2
+import json
+import os
+import sys
+import base64
+import logging
+from pyzbar import pyzbar
+from tqdm import tqdm
+from multiprocessing import Pool, cpu_count
+from checksum import checksum
+
+# Setup basic logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+def apply_preprocessing(frame):
+    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+    _, threshold = cv2.threshold(gray, 128, 255, cv2.THRESH_BINARY)
+    return threshold
+
+def read_barcode(preprocessed_frame):
+    barcodes = pyzbar.decode(preprocessed_frame)
+    for barcode in barcodes:
+        barcode_info = barcode.data.decode('utf-8')
+        return True, barcode_info
+    return False, None
+
+def process_frame(frame):
+    preprocessed_frame = apply_preprocessing(frame)
+    success, data = read_barcode(preprocessed_frame)
+    if not success:
+        logging.warning("Failed to read QR code")
+        return None  # Return None if no barcode is found
+    return data
+
+def decode_video(cap, dest_folder):
+    if not os.path.exists(dest_folder):
+        os.makedirs(dest_folder)
+
+    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+    pbar = tqdm(total=total_frames, desc="Processing Frames")
+
+    ret, first_frame = cap.read()
+    if not ret:
+        logging.error("Cannot read first frame")
+        return
+
+    metadata = process_frame(first_frame)
+    if metadata is None:
+        logging.error("No QR code in first frame; cannot proceed")
+        return
+    meta_data = json.loads(metadata)
+    dest = os.path.join(dest_folder, meta_data["Filename"])
+    file = open(dest, "wb")
+
+    # Start worker processes
+    num_workers = cpu_count()
+
+    with Pool(num_workers) as pool:
+        while cap.isOpened():
+            frames = []
+            done = False
+            for _ in range(num_workers):
+                ret, frame = cap.read()
+                if ret:
+                    frames.append(frame)
+                else:
+                    done = True
+                    break
+
+            datas = pool.map(process_frame, frames)
+
+            pbar.update(len(frames))
+
+            for data in datas:
+                file.write(base64.b64decode(data))
+
+            if done:
+                break
+
+
+    file.close()
+    cap.release()
+    pbar.close()
+
+    logging.info("Verify file integrity")
+    md5_sum = checksum(dest)
+    assert md5_sum == meta_data["Filehash"], "Data corrupted"
+    logging.info("File integrity verified: ")
+    logging.info(dest)
+
+def decode(src, dest_folder):
+    cap = cv2.VideoCapture(src)
+    decode_video(cap, dest_folder)
+
+if __name__ == '__main__':
+    if len(sys.argv) < 3:
+        logging.error("Usage: python script.py source_file.mp4 destination_folder")
+        sys.exit(1)
+    src = sys.argv[1]
+    dest_folder = sys.argv[2]
+
+    decode(src, dest_folder)
+

genz/encode.py

@@ -0,0 +1,125 @@
+from itertools import islice
+import os
+import sys
+import base64
+import math
+import json
+import qrcode
+import numpy as np
+from multiprocessing import Pool, cpu_count
+import av
+from PIL import Image
+from tqdm import tqdm
+
+from checksum import checksum
+
+# Constants
+chunk_size = 500  # Adjust this as per the data capacity of QR codes and your specific requirements
+frame_rate = 20.0
+width = 1080
+height = 1080
+
+def read_in_chunks(file_object, chunk_size=1024):
+    """Generator to read a file piece by piece."""
+    while True:
+        data = file_object.read(chunk_size)
+        if not data:
+            break
+        yield data
+
+def create_qr(data_str, box_size=10, error_correction_level=qrcode.constants.ERROR_CORRECT_L):
+    """Generate a QR code as a NumPy array from data string with specified box size and error correction level."""
+    qr = qrcode.QRCode(
+        version=1,
+        error_correction=error_correction_level,
+        box_size=box_size,
+        border=4,
+    )
+    qr.add_data(data_str)
+    img = qr.make_image(fill_color="black", back_color="white").convert('RGB')
+    pil_img = img.resize((width, height), Image.Resampling.NEAREST)  # Use NEAREST for less interpolation blur
+    return np.array(pil_img)
+
+def process_chunk(data):
+    """Encode data chunk into QR and return as image."""
+    return create_qr(base64.b64encode(data).decode('ascii'))
+
+def encode_and_write_frames(frames, stream, container):
+    """Encode frames and write to video container."""
+    for frame in frames:
+        video_frame = av.VideoFrame.from_ndarray(frame, format='rgb24')
+        for packet in stream.encode(video_frame):
+            container.mux(packet)
+
+def create_video(src, dest, read_file_lazy = False):
+    """Create video from source file using PyAV."""
+    md5_checksum = checksum(src)
+    file_stats = os.stat(src)
+    file_size = file_stats.st_size
+    chunk_count = math.ceil(file_size / chunk_size)
+
+    pbar = tqdm(total=chunk_count, desc="Generating Frames")
+
+    meta_data = {
+        "Filename": os.path.basename(src),
+        "ChunkCount": chunk_count,
+        "Filehash": md5_checksum,
+        "ConverterUrl": "https://github.com/karaketir16/file2video",
+        "ConverterVersion": "python_v1"
+    }
+
+    first_frame_data = json.dumps(meta_data, indent=4)
+    first_frame = create_qr(first_frame_data)
+
+    # Open output file
+    container = av.open(dest, mode='w')
+    stream = container.add_stream('h264', rate=frame_rate)
+    stream.width = width
+    stream.height = height
+    stream.pix_fmt = 'yuv420p'
+    stream.options = {'crf': '40'}
+
+    # Write the first frame
+    video_frame = av.VideoFrame.from_ndarray(first_frame, format='rgb24')
+    for packet in stream.encode(video_frame):
+        container.mux(packet)
+
+    # Process chunks in batches using multiprocessing
+    with open(src, 'rb') as f, Pool(cpu_count()) as pool:
+        entire_file = []
+        if not read_file_lazy:
+            entire_file = f.read()
+        i = 0
+        while True:
+
+            chunks = []
+
+            if read_file_lazy:
+                chunks = list(islice(read_in_chunks(f, chunk_size), cpu_count()))
+            else:
+                for _ in range(cpu_count()):
+                    chunks.append(entire_file[i: i + chunk_size])
+                    if not chunks[-1]: #empty
+                        chunks.pop()
+                        break
+                    i = i + chunk_size
+            if not chunks:
+                break
+            frames = pool.map(process_chunk, chunks)
+            encode_and_write_frames(frames, stream, container)
+            pbar.update(len(frames))
+
+    pbar.close()
+
+    # Finalize the video file
+    for packet in stream.encode():
+        container.mux(packet)
+    container.close()
+
+if __name__ == '__main__':
+    if len(sys.argv) < 3:
+        print("Usage: python file2video.py source_file output_file.mp4")
+        sys.exit(1)
+    src = sys.argv[1]
+    dest = sys.argv[2]
+    create_video(src, dest)

genz/usage.md

@@ -0,0 +1,114 @@
+python file2video.py --encode go.txt out.mp4 
+python file2video.py --decode out.mp4 ./     
+```python
+import cv2
+import sys
+import yt_dlp
+import os
+from datetime import datetime
+
+from decode_video import decode_video
+
+def youtube_decode(src, dest_folder):
+    base_filename = "downloaded_video"
+    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+    output_file = f"{dest_folder}/{base_filename}_{timestamp}.mp4"
+
+    # Create a yt-dlp configuration to download the video
+    ydl_opts = {
+        'format': 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best',
+        'outtmpl': output_file,
+        'noplaylist': True,
+        'quiet': True,
+        'merge_output_format': 'mp4',  # Ensure the output is mp4 if separate streams are downloaded
+    }
+
+    # Use yt-dlp to download the video
+    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+        ydl.download([src])
+
+    # Check if the file is downloaded and exists
+    if os.path.exists(output_file):
+        # Open the downloaded video file using OpenCV
+        cap = cv2.VideoCapture(output_file)
+
+        # Read and process the video
+        decode_video(cap, dest_folder)
+    else:
+        print("Failed to download video.")
+        sys.exit(1)
+
+if __name__ == '__main__':
+    if len(sys.argv) < 3:
+        print("Usage: python video2file.py \"https://video_url\" destination_folder")
+        sys.exit(1)
+
+    src = sys.argv[1]
+    dest_folder = sys.argv[2]
+    youtube_decode(src, dest_folder)
+
+```
+```python
+from encode import create_video
+from decode_video import decode
+from youtube_decode import youtube_decode
+import argparse
+import sys
+
+def enc_file(source_file, output_video):
+    print (f"Encoding {source_file} to {output_video}")
+    create_video(source_file, output_video)
+
+def dec_video(source_video, destination_folder, docker_mode):
+    print (f"Decoding {source_video} to {destination_folder}")
+    decode(source_video, destination_folder)
+
+def y_decode(video_url, destination_folder, docker_mode):
+    print (f"Decoding {video_url} to {destination_folder}")
+    youtube_decode(video_url, destination_folder)
+
+def main():
+
+    # First, check if '--docker' is in the command line arguments
+    docker_mode = '--docker' in sys.argv
+    if docker_mode:
+        sys.argv.remove('--docker')  # Remove it so it doesn't interfere with the main parser
+
+    docker_usage = """\
+docker run --rm -v $(pwd):/data karaketir16/file2video [-h]  [--encode source_file output_video] 
+                                                        [--decode source_video destination_folder] 
+                                                        [--youtube-decode youtube_url destination_folder]"""
+    
+    if docker_mode:
+        parser = argparse.ArgumentParser(description="Program to encode files into videos and decode videos back to files.", usage=docker_usage)
+    else:
+        parser = argparse.ArgumentParser(description="Program to encode files into videos and decode videos back to files.")
+
+    # Optional argument for encoding
+    parser.add_argument("--encode", nargs=2, metavar=('source_file', 'output_video'), 
+                        help="Encode a file into a video: source_file output_video.mp4")
+    
+    # Optional argument for decoding
+    parser.add_argument("--decode", nargs=2, metavar=('source_video', 'destination_folder'), 
+                        help="Decode a video to a file: source_video.mp4 destination_folder")
+    
+    # Optional argument for YouTube video decoding
+    parser.add_argument("--youtube-decode", nargs=2, metavar=('youtube_url', 'destination_folder'), 
+                        help="Decode a video from a YouTube URL to a file: 'youtube_url' destination_folder")
+
+    
+    args = parser.parse_args()
+
+    # Check which command is used and call the corresponding function
+    if args.encode:
+        enc_file(*args.encode)
+    elif args.decode:
+        dec_video(*args.decode, docker_mode)
+    elif args.youtube_decode:
+        y_decode(*args.youtube_decode, docker_mode)
+    else:
+        parser.print_help()
+
+if __name__ == "__main__":
+    main()
+```

genz/youtube_decode.py

@@ -0,0 +1,46 @@
+import cv2
+import sys
+import yt_dlp
+import os
+from datetime import datetime
+
+from decode_video import decode_video
+
+def youtube_decode(src, dest_folder):
+    base_filename = "downloaded_video"
+    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+    output_file = f"{dest_folder}/{base_filename}_{timestamp}.mp4"
+
+    # Create a yt-dlp configuration to download the video
+    ydl_opts = {
+        'format': 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best',
+        'outtmpl': output_file,
+        'noplaylist': True,
+        'quiet': True,
+        'merge_output_format': 'mp4',  # Ensure the output is mp4 if separate streams are downloaded
+    }
+
+    # Use yt-dlp to download the video
+    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+        ydl.download([src])
+
+    # Check if the file is downloaded and exists
+    if os.path.exists(output_file):
+        # Open the downloaded video file using OpenCV
+        cap = cv2.VideoCapture(output_file)
+
+        # Read and process the video
+        decode_video(cap, dest_folder)
+    else:
+        print("Failed to download video.")
+        sys.exit(1)
+
+if __name__ == '__main__':
+    if len(sys.argv) < 3:
+        print("Usage: python video2file.py \"https://video_url\" destination_folder")
+        sys.exit(1)
+
+    src = sys.argv[1]
+    dest_folder = sys.argv[2]
+    youtube_decode(src, dest_folder)
+