libs/file_processing/file_processing_core.py
def do_process_user_file_chunks(
        page_size: int, error_handler: ErrorHandler, position: int, participant: Participant
):
    """
    Run through the files to process, pull their data, put it into s3 bins. Run the file through
    the appropriate logic path based on file type.

    If a file is empty put its ftp object to the empty_files_list, we can't delete objects
    in-place while iterating over the db.

    All files except for the audio recording files are in the form of CSVs, most of those files
    can be separated by "time bin" (separated into one-hour chunks) and concatenated and sorted
    trivially. A few files, call log, identifier file, and wifi log, require some triage
    beforehand.  The debug log cannot be correctly sorted by time for all elements, because it
    was not actually expected to be used by researchers, but is apparently quite useful.

    Any errors are themselves concatenated using the passed in error handler.

    In a single call to this function, page_size files will be processed,at the position specified.
    This is expected to exclude files that have previously errored in file processing.
    (some conflicts can be most easily resolved by just delaying a file until the next processing
    period, and it solves )
    """
    # Declare a defaultdict of a tuple of 2 lists
    all_binified_data = defaultdict(lambda: ([], []))
    ftps_to_remove = set()
    # The ThreadPool enables downloading multiple files simultaneously from the network, and continuing
    # to download files as other files are being processed, making the code as a whole run faster.
    pool = ThreadPool(CONCURRENT_NETWORK_OPS)  # Todo: make pool more global to prevent OS memory hogging
    survey_id_dict = {}

    # A Django query with a slice (e.g. .all()[x:y]) makes a LIMIT query, so it
    # only gets from the database those FTPs that are in the slice.
    # print(participant.as_unpacked_native_python())
    print(len(participant.files_to_process.exclude(deleted=True).all()))
    print(page_size)
    print(position)

    # ordering by path results in files grouped by type and chronological order, which is perfect
    # for efficiency.
    # Fixme: does this order_by break the skipping of items that failed to process? solution: exclude ids?
    files_to_process = participant.files_to_process \
        .exclude(deleted=True)  #.order_by("s3_file_path", "created_on")

    # This pool pulls in data for each FileForProcessing on a background thread and instantiates it.
    # Instantiating a FileForProcessing object queries S3 for the File's data. (network request))
    files_for_processing = pool.map(
        FileForProcessing, files_to_process[position: position + page_size], chunksize=1
    )

    for file_for_processing in files_for_processing:
        with error_handler:
            process_one_file(
                file_for_processing, survey_id_dict, all_binified_data, ftps_to_remove
            )
    pool.close()
    pool.terminate()

    # there are several failure modes and success modes, information for what to do with different
    # files percolates back to here.  Delete various database objects accordingly.
    more_ftps_to_remove, number_bad_files, earliest_time_bin, latest_time_bin = upload_binified_data(
        all_binified_data, error_handler, survey_id_dict
    )
    ftps_to_remove.update(more_ftps_to_remove)

    # Update the data quantity stats, if it actually processed any files
    if len(files_to_process) > 0:
        calculate_data_quantity_stats(participant,
                                      earliest_time_bin_number=earliest_time_bin,
                                      latest_time_bin_number=latest_time_bin)

    # Actually delete the processed FTPs from the database
    FileToProcess.objects.filter(pk__in=ftps_to_remove).delete()
    return number_bad_files

    Run through the files to process, pull their data, put it into s3 bins. Run the file through
    the appropriate logic path based on file type.

    If a file is empty put its ftp object to the empty_files_list, we can't delete objects
    in-place while iterating over the db.

    All files except for the audio recording files are in the form of CSVs, most of those files
    can be separated by "time bin" (separated into one-hour chunks) and concatenated and sorted
    trivially. A few files, call log, identifier file, and wifi log, require some triage
    beforehand.  The debug log cannot be correctly sorted by time for all elements, because it
    was not actually expected to be used by researchers, but is apparently quite useful.

    Any errors are themselves concatenated using the passed in error handler.

    In a single call to this function, page_size files will be processed,at the position specified.
    This is expected to exclude files that have previously errored in file processing.
    (some conflicts can be most easily resolved by just delaying a file until the next processing
    period, and it solves )
    
Run through the files to process, pull their data, put it into s3 bins. Run the file through
the appropriate logic path based on file type.

If a file is empty put its ftp object to the empty_files_list, we can't delete objects
in-place while iterating over the db.

All files except for the audio recording files are in the form of CSVs, most of those files
can be separated by "time bin" (separated into one-hour chunks) and concatenated and sorted
trivially. A few files, call log, identifier file, and wifi log, require some triage
beforehand.  The debug log cannot be correctly sorted by time for all elements, because it
was not actually expected to be used by researchers, but is apparently quite useful.

Any errors are themselves concatenated using the passed in error handler.

In a single call to this function, page_size files will be processed,at the position specified.
This is expected to exclude files that have previously errored in file processing.
(some conflicts can be most easily resolved by just delaying a file until the next processing
period, and it solves )
Run through the files to process pull their data put it into s3 bins . Run the file through the appropriate logic path based on file type . If a file is empty put its ftp object to the empty_files_list we can ' t delete objects in - place while iterating over the db . All files except for the audio recording files are in the form of CSVs most of those files can be separated by " time bin " ( separated into one - hour chunks ) and concatenated and sorted trivially . A few files call log identifier file and wifi log require some triage beforehand . The debug log cannot be correctly sorted by time for all elements because it was not actually expected to be used by researchers but is apparently quite useful . Any errors are themselves concatenated using the passed in error handler . In a single call to this function page_size files will be processed at the position specified . This is expected to exclude files that have previously errored in file processing . ( some conflicts can be most easily resolved by just delaying a file until the next processing period and it solves )
def do_process_user_file_chunks(
        page_size: int, error_handler: ErrorHandler, position: int, participant: Participant
):
    all_binified_data = defaultdict(lambda: ([], []))
    ftps_to_remove = set()
    pool = ThreadPool(CONCURRENT_NETWORK_OPS)  
    survey_id_dict = {}
    print(len(participant.files_to_process.exclude(deleted=True).all()))
    print(page_size)
    print(position)
    files_to_process = participant.files_to_process \
        .exclude(deleted=True)  
    files_for_processing = pool.map(
        FileForProcessing, files_to_process[position: position + page_size], chunksize=1
    )
    for file_for_processing in files_for_processing:
        with error_handler:
            process_one_file(
                file_for_processing, survey_id_dict, all_binified_data, ftps_to_remove
            )
    pool.close()
    pool.terminate()
    more_ftps_to_remove, number_bad_files, earliest_time_bin, latest_time_bin = upload_binified_data(
        all_binified_data, error_handler, survey_id_dict
    )
    ftps_to_remove.update(more_ftps_to_remove)
    if len(files_to_process) > 0:
        calculate_data_quantity_stats(participant,
                                      earliest_time_bin_number=earliest_time_bin,
                                      latest_time_bin_number=latest_time_bin)
    FileToProcess.objects.filter(pk__in=ftps_to_remove).delete()
    return number_bad_files
def do_process_user_file_chunks ( page_size : int , error_handler : ErrorHandler , position : int , participant : Participant ) : all_binified_data = defaultdict ( lambda : ( [ ] , [ ] ) ) ftps_to_remove = set ( ) pool = ThreadPool ( CONCURRENT_NETWORK_OPS ) survey_id_dict = { } print ( len ( participant . files_to_process . exclude ( deleted = True ) . all ( ) ) ) print ( page_size ) print ( position ) files_to_process = participant . files_to_process . exclude ( deleted = True ) files_for_processing = pool . map ( FileForProcessing , files_to_process [ position : position + page_size ] , chunksize = 1 ) for file_for_processing in files_for_processing : with error_handler : process_one_file ( file_for_processing , survey_id_dict , all_binified_data , ftps_to_remove ) pool . close ( ) pool . terminate ( ) more_ftps_to_remove , number_bad_files , earliest_time_bin , latest_time_bin = upload_binified_data ( all_binified_data , error_handler , survey_id_dict ) ftps_to_remove . update ( more_ftps_to_remove ) if len ( files_to_process ) > 0 : calculate_data_quantity_stats ( participant , earliest_time_bin_number = earliest_time_bin , latest_time_bin_number = latest_time_bin ) FileToProcess . objects . filter ( pk__in = ftps_to_remove ) . delete ( ) return number_bad_files
Run through the files to process, pull their data, put it into s3 bins.
Run through the files to process pull their data put it into s3 bins .
"""
    Run through the files to process, pull their data, put it into s3 bins. Run the file through
    the appropriate logic path based on file type.

    If a file is empty put its ftp object to the empty_files_list, we can't delete objects
    in-place while iterating over the db.

    All files except for the audio recording files are in the form of CSVs, most of those files
    can be separated by "time bin" (separated into one-hour chunks) and concatenated and sorted
    trivially. A few files, call log, identifier file, and wifi log, require some triage
    beforehand.  The debug log cannot be correctly sorted by time for all elements, because it
    was not actually expected to be used by researchers, but is apparently quite useful.

    Any errors are themselves concatenated using the passed in error handler.

    In a single call to this function, page_size files will be processed,at the position specified.
    This is expected to exclude files that have previously errored in file processing.
    (some conflicts can be most easily resolved by just delaying a file until the next processing
    period, and it solves )
    """ # Declare a defaultdict of a tuple of 2 lists # The ThreadPool enables downloading multiple files simultaneously from the network, and continuing # to download files as other files are being processed, making the code as a whole run faster. # Todo: make pool more global to prevent OS memory hogging # A Django query with a slice (e.g. .all()[x:y]) makes a LIMIT query, so it # only gets from the database those FTPs that are in the slice. # print(participant.as_unpacked_native_python()) # ordering by path results in files grouped by type and chronological order, which is perfect # for efficiency. # Fixme: does this order_by break the skipping of items that failed to process? solution: exclude ids? #.order_by("s3_file_path", "created_on") # This pool pulls in data for each FileForProcessing on a background thread and instantiates it. # Instantiating a FileForProcessing object queries S3 for the File's data. (network request)) # there are several failure modes and success modes, information for what to do with different # files percolates back to here.  Delete various database objects accordingly. # Update the data quantity stats, if it actually processed any files # Actually delete the processed FTPs from the database
page_size
int
error_handler
ErrorHandler
position
int
participant
Participant