操作指南：上传（高级）

1
math.ceiling(float(21136250) / float(4))
2
# 5284063 bytes

1
# Where we are going to queue new files.
2
FILE_QUEUE = Queue()
3
4
# Where we are going to queue new chunks.
5
CHUNK_QUEUE = Queue()
6
7
# The http session that will handle TCP 
8
# connection pooling for us.
9
HTTP_SESSION = http.Session()
10
11
def take_picture():
12
    """Snaps a picture for the user."""
13
14
    image = MY_DEVICE.capture()
15
    file_path = MY_DEVICE.write_image(image)
16
    FILE_QUEUE.add(file_path)
17
18
def task_register_assets():
19
    """
20
    Pulls snapped pictures from the FILE_QUEUE, registers
21
    with Frame.io, and adds the chunks to the CHUNK_QUEUE.
22
    """
23
    while True:
24
        # Get the latest file added to the queue and register 
25
        # a C2C Asset for it.
26
        new_file = FILE_QUEUE.get()
27
        asset = c2c.crete_asset_for_file(HTTP_SESSION, new_file)
28
29
        # Calculate the size for each chunk
30
        chunk_size = c2c.calculate_chunk_size(asset, new_file)
31
32
        # Create a message for each chunk with it's parameters 
33
        # and add it to the 
34
        # queue.
35
        chunk_start = 0
36
        for chunk_url in asset.upload_urls:
37
            message = {
38
                "file_path": new_file,
39
                "chunk_url": chunk_url,
40
                "chunk_start": chunk_start,
41
                "chunk_size": chunk_size,
42
            }
43
44
            # Put the message in the queue and 
45
            CHUNK_QUEUE.put(message)
46
            chunk_start += chunk_size
47
48
def task_upload_chunk():
49
    """Takes a chunks and uploads them."""
50
51
    while True:
52
        info = CHUNK_QUEUE.get()
53
        c2c.upload_chunk(HTTP_SESSION, info)
54
55
def launch_upload_tasks():
56
    """Lauches our Frame.io upload tasks."""
57
    # Create a list to hold all of our tasks.
58
    tasks = list()
59
60
    # Create one task for registering assets.
61
    asset_task = run_task_in_thread(task_register_assets)
62
    tasks.append(asset_task)
63
64
    # Create 2 tasks per CPU core for uploading chunks.
65
    for _ in range(0, GET_CPU_COUNT() * 2):
66
        chunk_task = run_task_in_thread(task_upload_chunk)
67
        tasks.append(chunk_task)
68
69
    # Run these tasks until shutdown
70
    run_forever(tasks)

1
# Our persistence layer for queuing uploads, potentially using SQLite 
2
# or another embedded database
3
C2C_UPLOAD_STORE = NewC2CUploadStore()
4
5
# HTTP session for connection pooling
6
HTTP_SESSION = http.Session()
7
8
def take_picture():
9
    """Captures an image and adds it to the upload queue."""
10
    image = MY_DEVICE.capture()
11
    file_path = MY_DEVICE.write_image(image)
12
13
    # Register the file with our persistent store
14
    C2C_UPLOAD_STORE.add_file(file_path)
15
16
def task_register_assets():
17
    """
18
    Processes files from persistent storage and registers
19
    them with Frame.io for upload.
20
    """
21
    while True:
22
        # Get the next available file from our store
23
        file_record = C2C_UPLOAD_STORE.get_file()
24
25
        try:
26
            # Register the asset with Frame.io
27
            asset = c2c.create_asset_for_file(HTTP_SESSION, file_record)
28
            chunk_size = c2c.calculate_chunk_size(asset, file_record)
29
30
            # Create entries for each chunk in our persistent store
31
            chunk_start = 0
32
            for chunk_url in asset.upload_urls:
33
                message = {
34
                    "file_path": file_record,
35
                    "chunk_url": chunk_url,
36
                    "chunk_start": chunk_start,
37
                    "chunk_size": chunk_size,
38
                }
39
                
40
                C2C_UPLOAD_STORE.new_chunk(message)
41
                chunk_start += chunk_size
42
                
43
        except BaseException as error:
44
            # Record the error in our persistent store
45
            C2C_UPLOAD_STORE.file_asset_create_error(file_record, error)
46
        else:
47
            # Mark the asset as successfully created
48
            C2C_UPLOAD_STORE.file_asset_created(file_record)
49
50
def task_upload_chunk():
51
    """Uploads individual file chunks from the persistent queue."""
52
    while True:
53
        # Get the next chunk, marking it as "in progress" to prevent 
54
        # other tasks from processing it simultaneously
55
        chunk_record = C2C_UPLOAD_STORE.get_chunk()
56
57
        try:
58
            c2c.upload_chunk(HTTP_SESSION, chunk_record)
59
        except BaseException as error:
60
            # Record the error for potential retry
61
            C2C_UPLOAD_STORE.chunk_error(chunk_record, error)
62
        else:
63
            # Mark successful completion
64
            C2C_UPLOAD_STORE.chunk_success(chunk_record)
65
66
def launch_upload_tasks():
67
    """Launches Frame.io upload processing tasks."""
68
    tasks = []
69
70
    # Asset registration task
71
    asset_task = run_task_in_thread(task_register_assets)
72
    tasks.append(asset_task)
73
74
    # Multiple parallel chunk upload tasks
75
    worker_count = GET_CPU_COUNT() * 2
76
    for _ in range(worker_count):
77
        chunk_task = run_task_in_thread(task_upload_chunk)
78
        tasks.append(chunk_task)
79
80
    # Run indefinitely
81
    run_forever(tasks)