day84/code-server

1from itertools import islice2import os3import sys4import base645import math6import json7import qrcode8import numpy as np9from multiprocessing import Pool, cpu_count10import av11from PIL import Image12from tqdm import tqdm13 14from checksum import checksum15 16# Constants17chunk_size = 500  # Adjust this as per the data capacity of QR codes and your specific requirements18frame_rate = 20.019width = 108020height = 108021 22def read_in_chunks(file_object, chunk_size=1024):23    """Generator to read a file piece by piece."""24    while True:25        data = file_object.read(chunk_size)26        if not data:27            break28        yield data29 30def create_qr(data_str, box_size=10, error_correction_level=qrcode.constants.ERROR_CORRECT_L):31    """Generate a QR code as a NumPy array from data string with specified box size and error correction level."""32    qr = qrcode.QRCode(33        version=1,34        error_correction=error_correction_level,35        box_size=box_size,36        border=4,37    )38    qr.add_data(data_str)39    img = qr.make_image(fill_color="black", back_color="white").convert('RGB')40    pil_img = img.resize((width, height), Image.Resampling.NEAREST)  # Use NEAREST for less interpolation blur41    return np.array(pil_img)42 43def process_chunk(data):44    """Encode data chunk into QR and return as image."""45    return create_qr(base64.b64encode(data).decode('ascii'))46 47def encode_and_write_frames(frames, stream, container):48    """Encode frames and write to video container."""49    for frame in frames:50        video_frame = av.VideoFrame.from_ndarray(frame, format='rgb24')51        for packet in stream.encode(video_frame):52            container.mux(packet)53 54def create_video(src, dest, read_file_lazy = False):55    """Create video from source file using PyAV."""56    md5_checksum = checksum(src)57    file_stats = os.stat(src)58    file_size = file_stats.st_size59    chunk_count = math.ceil(file_size / chunk_size)60 61    pbar = tqdm(total=chunk_count, desc="Generating Frames")62 63    meta_data = {64        "Filename": os.path.basename(src),65        "ChunkCount": chunk_count,66        "Filehash": md5_checksum,67        "ConverterUrl": "https://github.com/karaketir16/file2video",68        "ConverterVersion": "python_v1"69    }70 71    first_frame_data = json.dumps(meta_data, indent=4)72    first_frame = create_qr(first_frame_data)73 74    # Open output file75    container = av.open(dest, mode='w')76    stream = container.add_stream('h264', rate=frame_rate)77    stream.width = width78    stream.height = height79    stream.pix_fmt = 'yuv420p'80    stream.options = {'crf': '40'}81 82    # Write the first frame83    video_frame = av.VideoFrame.from_ndarray(first_frame, format='rgb24')84    for packet in stream.encode(video_frame):85        container.mux(packet)86 87    # Process chunks in batches using multiprocessing88    with open(src, 'rb') as f, Pool(cpu_count()) as pool:89        entire_file = []90        if not read_file_lazy:91            entire_file = f.read()92        i = 093        while True:94 95            chunks = []96 97            if read_file_lazy:98                chunks = list(islice(read_in_chunks(f, chunk_size), cpu_count()))99            else:100                for _ in range(cpu_count()):101                    chunks.append(entire_file[i: i + chunk_size])102                    if not chunks[-1]: #empty103                        chunks.pop()104                        break105                    i = i + chunk_size106            if not chunks:107                break108            frames = pool.map(process_chunk, chunks)109            encode_and_write_frames(frames, stream, container)110            pbar.update(len(frames))111 112    pbar.close()113 114    # Finalize the video file115    for packet in stream.encode():116        container.mux(packet)117    container.close()118 119if __name__ == '__main__':120    if len(sys.argv) < 3:121        print("Usage: python file2video.py source_file output_file.mp4")122        sys.exit(1)123    src = sys.argv[1]124    dest = sys.argv[2]125    create_video(src, dest)126