amegianeg/gist:f471a74c307c650547de8b5917202c62

## gistfile1.txt
import io
import shutil
import timeit
import os
import uuid
from functools import partial


BYTES_PER_MB = 1048576
NUM_ROUNDS = 5
NUM_EXECUTIONS = 100
CHUNK_SIZE = (64 * 2 ** 10) * 2  # 64kb * 2
NUMPY_ENABLED = False


def write_dicom_instance(instance_path, dicom_data_bytes):
    dicom_data_bytes.seek(0)
    os.makedirs(os.path.dirname(instance_path), exist_ok=True)
    with open(instance_path, "wb") as outf:
        outf.write(dicom_data_bytes.getbuffer())


# TODO chunks not working with BytesIO
#def write_dicom_file_chunks(instance_path, dicom_data_bytes):
#    os.makedirs(os.path.dirname(instance_path), exist_ok=True)
#    with open(instance_path, "wb") as outf:
#        for chunk in dicom_data_bytes.chunks():
#            outf.write(chunk)


def write_dicom_instance_buffer(instance_path, dicom_data_bytes):
    dicom_data_bytes.seek(0)
    os.makedirs(os.path.dirname(instance_path), exist_ok=True)
    with open(instance_path, "wb", buffering=CHUNK_SIZE) as outf:
        outf.write(dicom_data_bytes.getbuffer())


def write_dicom_instance_buffer_4096(instance_path, dicom_data_bytes):
    dicom_data_bytes.seek(0)
    os.makedirs(os.path.dirname(instance_path), exist_ok=True)
    with open(instance_path, "wb", buffering=4096) as outf:
        outf.write(dicom_data_bytes.getbuffer())


# TODO chunks not working with BytesIO
#def write_dicom_instance_buffer_rw(instance_path, dicom_data_bytes):
#    os.makedirs(os.path.dirname(instance_path), exist_ok=True)
#    with open(instance_path, "wb", buffering=CHUNK_SIZE) as outf:
#        for chunk in dicom_data_bytes.chunks(chunk_size=CHUNK_SIZE):
#            outf.write(chunk.getbuffer())


def write_dicom_instance_numpy(instance_path, dicom_data_bytes):
    dicom_data_bytes.seek(0)
    os.makedirs(os.path.dirname(instance_path), exist_ok=True)
    array = np.array(dicom_data_bytes.getbuffer())
    with open('binary.file', 'wb') as f:
        array.tofile(instance_path)


def write_file_fast_copy(instance_path, dicom_data_bytes):
    dicom_data_bytes.seek(0)
    os.makedirs(os.path.dirname(instance_path), exist_ok=True)
    with open(instance_path, "wb") as outf:
        shutil.copyfileobj(dicom_data_bytes, outf)


def write_file_fast_copy_change_chunk_128(instance_path, dicom_data_bytes):
    dicom_data_bytes.seek(0)
    os.makedirs(os.path.dirname(instance_path), exist_ok=True)
    with open(instance_path, "wb") as outf:
        shutil.copyfileobj(dicom_data_bytes, outf, length=128 * 1024)


def write_file_fast_copy_change_chunk_256(instance_path, dicom_data_bytes):
    dicom_data_bytes.seek(0)
    os.makedirs(os.path.dirname(instance_path), exist_ok=True)
    with open(instance_path, "wb") as outf:
        shutil.copyfileobj(dicom_data_bytes, outf, length=256 * 1024)


def write_file_fast_copy_change_chunk_512(instance_path, dicom_data_bytes):
    dicom_data_bytes.seek(0)
    os.makedirs(os.path.dirname(instance_path), exist_ok=True)
    with open(instance_path, "wb") as outf:
        shutil.copyfileobj(dicom_data_bytes, outf, length=512 * 1024)


if __name__ == '__main__':
    file_plain_films = io.BytesIO(os.urandom(BYTES_PER_MB * 15))  # Plain films 10-20Mb per dicom
    file_ct = io.BytesIO(os.urandom(int(BYTES_PER_MB * 0.5)))  # CT 520kb per dicom
    file_mamo = io.BytesIO(os.urandom(int(BYTES_PER_MB * 30)))  # Mamo 30MB per dicom
    file_mr = io.BytesIO(os.urandom(int(BYTES_PER_MB * 0.25)))  # MR 250kb per dicom
    files = {
        "PLAIN FILMS": file_plain_films,
        "CT": file_ct,
        "MAMO": file_mamo,
        "MR": file_mr
    }

    for test, file in files.items():
        print(f"Testing {test}")

        output_path = f"./outputs/{uuid.uuid4()}/{test}"

        print("Test 'write_file_fast_copy' started...")
        result = timeit.repeat(
            partial(write_file_fast_copy, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
        print(min(result))
        print("Test 'write_file_fast_copy' finished.")

        print("Test 'write_file_fast_copy_change_chunk_128' started...")
        result = timeit.repeat(
            partial(write_file_fast_copy_change_chunk_128, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
        print(min(result))
        print("Test 'write_file_fast_copy_change_chunk_128' finished.")

        print("Test 'write_file_fast_copy_change_chunk_256' started...")
        result = timeit.repeat(
            partial(write_file_fast_copy_change_chunk_256, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
        print(min(result))
        print("Test 'write_file_fast_copy_change_chunk_256' finished.")

        print("Test 'write_file_fast_copy_change_chunk_512' started...")
        result = timeit.repeat(
            partial(write_file_fast_copy_change_chunk_512, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
        print(min(result))
        print("Test 'write_file_fast_copy_change_chunk_512' finished.")

        print("Test 'write_dicom_instance_buffer' started...")
        result = timeit.repeat(
            partial(write_dicom_instance_buffer, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
        print(min(result))
        print("Test 'write_dicom_instance_buffer' finished.")

        print("Test 'write_dicom_instance_buffer_4096' started...")
        result = timeit.repeat(
            partial(write_dicom_instance_buffer_4096, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
        print(min(result))
        print("Test 'write_dicom_instance_buffer_4096' finished.")

        print("Test 'write_dicom_instance' started...")
        result = timeit.repeat(
            partial(write_dicom_instance, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
        print(min(result))
        print("Test 'write_dicom_instance' finished.")

        #import numpy as np
        #print("Test 'write_dicom_instance_numpy' started...")
        #result = timeit.repeat(
        #    partial(write_dicom_instance_numpy, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
        #print(min(result))
        #print("Test 'write_dicom_instance_numpy' finished.")

        print(f"Tests {test} finished!")
        print("::::::::::::::::::::::::::::::::::::::::::::")
	import io
	import shutil
	import timeit
	import os
	import uuid
	from functools import partial


	BYTES_PER_MB = 1048576
	NUM_ROUNDS = 5
	NUM_EXECUTIONS = 100
	CHUNK_SIZE = (64 * 2 ** 10) * 2 # 64kb * 2
	NUMPY_ENABLED = False


	def write_dicom_instance(instance_path, dicom_data_bytes):
	dicom_data_bytes.seek(0)
	os.makedirs(os.path.dirname(instance_path), exist_ok=True)
	with open(instance_path, "wb") as outf:
	outf.write(dicom_data_bytes.getbuffer())


	# TODO chunks not working with BytesIO
	#def write_dicom_file_chunks(instance_path, dicom_data_bytes):
	# os.makedirs(os.path.dirname(instance_path), exist_ok=True)
	# with open(instance_path, "wb") as outf:
	# for chunk in dicom_data_bytes.chunks():
	# outf.write(chunk)


	def write_dicom_instance_buffer(instance_path, dicom_data_bytes):
	dicom_data_bytes.seek(0)
	os.makedirs(os.path.dirname(instance_path), exist_ok=True)
	with open(instance_path, "wb", buffering=CHUNK_SIZE) as outf:
	outf.write(dicom_data_bytes.getbuffer())


	def write_dicom_instance_buffer_4096(instance_path, dicom_data_bytes):
	dicom_data_bytes.seek(0)
	os.makedirs(os.path.dirname(instance_path), exist_ok=True)
	with open(instance_path, "wb", buffering=4096) as outf:
	outf.write(dicom_data_bytes.getbuffer())


	# TODO chunks not working with BytesIO
	#def write_dicom_instance_buffer_rw(instance_path, dicom_data_bytes):
	# os.makedirs(os.path.dirname(instance_path), exist_ok=True)
	# with open(instance_path, "wb", buffering=CHUNK_SIZE) as outf:
	# for chunk in dicom_data_bytes.chunks(chunk_size=CHUNK_SIZE):
	# outf.write(chunk.getbuffer())


	def write_dicom_instance_numpy(instance_path, dicom_data_bytes):
	dicom_data_bytes.seek(0)
	os.makedirs(os.path.dirname(instance_path), exist_ok=True)
	array = np.array(dicom_data_bytes.getbuffer())
	with open('binary.file', 'wb') as f:
	array.tofile(instance_path)


	def write_file_fast_copy(instance_path, dicom_data_bytes):
	dicom_data_bytes.seek(0)
	os.makedirs(os.path.dirname(instance_path), exist_ok=True)
	with open(instance_path, "wb") as outf:
	shutil.copyfileobj(dicom_data_bytes, outf)


	def write_file_fast_copy_change_chunk_128(instance_path, dicom_data_bytes):
	dicom_data_bytes.seek(0)
	os.makedirs(os.path.dirname(instance_path), exist_ok=True)
	with open(instance_path, "wb") as outf:
	shutil.copyfileobj(dicom_data_bytes, outf, length=128 * 1024)


	def write_file_fast_copy_change_chunk_256(instance_path, dicom_data_bytes):
	dicom_data_bytes.seek(0)
	os.makedirs(os.path.dirname(instance_path), exist_ok=True)
	with open(instance_path, "wb") as outf:
	shutil.copyfileobj(dicom_data_bytes, outf, length=256 * 1024)


	def write_file_fast_copy_change_chunk_512(instance_path, dicom_data_bytes):
	dicom_data_bytes.seek(0)
	os.makedirs(os.path.dirname(instance_path), exist_ok=True)
	with open(instance_path, "wb") as outf:
	shutil.copyfileobj(dicom_data_bytes, outf, length=512 * 1024)


	if __name__ == '__main__':
	file_plain_films = io.BytesIO(os.urandom(BYTES_PER_MB * 15)) # Plain films 10-20Mb per dicom
	file_ct = io.BytesIO(os.urandom(int(BYTES_PER_MB * 0.5))) # CT 520kb per dicom
	file_mamo = io.BytesIO(os.urandom(int(BYTES_PER_MB * 30))) # Mamo 30MB per dicom
	file_mr = io.BytesIO(os.urandom(int(BYTES_PER_MB * 0.25))) # MR 250kb per dicom
	files = {
	"PLAIN FILMS": file_plain_films,
	"CT": file_ct,
	"MAMO": file_mamo,
	"MR": file_mr
	}

	for test, file in files.items():
	print(f"Testing {test}")

	output_path = f"./outputs/{uuid.uuid4()}/{test}"

	print("Test 'write_file_fast_copy' started...")
	result = timeit.repeat(
	partial(write_file_fast_copy, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
	print(min(result))
	print("Test 'write_file_fast_copy' finished.")

	print("Test 'write_file_fast_copy_change_chunk_128' started...")
	result = timeit.repeat(
	partial(write_file_fast_copy_change_chunk_128, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
	print(min(result))
	print("Test 'write_file_fast_copy_change_chunk_128' finished.")

	print("Test 'write_file_fast_copy_change_chunk_256' started...")
	result = timeit.repeat(
	partial(write_file_fast_copy_change_chunk_256, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
	print(min(result))
	print("Test 'write_file_fast_copy_change_chunk_256' finished.")

	print("Test 'write_file_fast_copy_change_chunk_512' started...")
	result = timeit.repeat(
	partial(write_file_fast_copy_change_chunk_512, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
	print(min(result))
	print("Test 'write_file_fast_copy_change_chunk_512' finished.")

	print("Test 'write_dicom_instance_buffer' started...")
	result = timeit.repeat(
	partial(write_dicom_instance_buffer, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
	print(min(result))
	print("Test 'write_dicom_instance_buffer' finished.")

	print("Test 'write_dicom_instance_buffer_4096' started...")
	result = timeit.repeat(
	partial(write_dicom_instance_buffer_4096, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
	print(min(result))
	print("Test 'write_dicom_instance_buffer_4096' finished.")

	print("Test 'write_dicom_instance' started...")
	result = timeit.repeat(
	partial(write_dicom_instance, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
	print(min(result))
	print("Test 'write_dicom_instance' finished.")

	#import numpy as np
	#print("Test 'write_dicom_instance_numpy' started...")
	#result = timeit.repeat(
	# partial(write_dicom_instance_numpy, output_path, file), repeat=NUM_ROUNDS, number=NUM_EXECUTIONS)
	#print(min(result))
	#print("Test 'write_dicom_instance_numpy' finished.")

	print(f"Tests {test} finished!")
	print("::::::::::::::::::::::::::::::::::::::::::::")