<s> """<STR_LIT>""" <EOL> import logging <EOL> import os <EOL> import time <EOL> import shutil <EOL> from agdc . cube_util import DatasetError , create_directory <EOL> from tile_contents import TileContents <EOL> from acquisition_record import AcquisitionRecord <EOL> from ingest_db_wrapper import IngestDBWrapper <EOL> LOGGER = logging . getLogger ( __name__ ) <EOL> LOGGER . setLevel ( logging . INFO ) <EOL> class Collection ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , datacube ) : <EOL> """<STR_LIT>""" <EOL> self . datacube = datacube <EOL> self . db = IngestDBWrapper ( datacube . db_connection ) <EOL> self . new_bands = self . __reindex_bands ( datacube . bands ) <EOL> self . transaction_stack = [ ] <EOL> self . temp_tile_directory = os . path . join ( self . datacube . tile_root , <EOL> '<STR_LIT>' , <EOL> self . datacube . process_id ) <EOL> create_directory ( self . temp_tile_directory ) <EOL> def cleanup ( self ) : <EOL> """<STR_LIT>""" <EOL> shutil . rmtree ( self . temp_tile_directory , ignore_errors = True ) <EOL> @ staticmethod <EOL> def get_dataset_key ( dataset ) : <EOL> """<STR_LIT>""" <EOL> derived_levels = { '<STR_LIT>' , '<STR_LIT>' } <EOL> satellite = dataset . get_satellite_tag ( ) <EOL> sensor = dataset . get_sensor_name ( ) <EOL> level = dataset . get_processing_level ( ) <EOL> if level in derived_levels : <EOL> satellite = '<STR_LIT>' <EOL> sensor = level <EOL> return ( satellite , sensor , level ) <EOL> def get_temp_tile_directory ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . temp_tile_directory <EOL> def check_metadata ( self , dataset ) : <EOL> """<STR_LIT>""" <EOL> self . __check_satellite_and_sensor ( dataset ) <EOL> self . __check_processing_level ( dataset ) <EOL> self . __check_bands ( dataset ) <EOL> def transaction ( self , db = None ) : <EOL> """<STR_LIT>""" <EOL> return Transaction ( self . db if db is None else db , <EOL> self . transaction_stack ) <EOL> def lock_datasets ( self , dataset_list ) : <EOL> """<STR_LIT>""" <EOL> lock_list = [ '<STR_LIT>' + str ( dataset_id ) <EOL> for dataset_id in dataset_list ] <EOL> return Lock ( self . datacube , lock_list ) <EOL> def create_acquisition_record ( self , dataset ) : <EOL> """<STR_LIT>""" <EOL> return AcquisitionRecord ( self , dataset ) <EOL> def create_tile_contents ( self , tile_type_id , tile_footprint , <EOL> band_stack ) : <EOL> """<STR_LIT>""" <EOL> tile_type_info = self . datacube . tile_type_dict [ tile_type_id ] <EOL> tile_contents = TileContents ( self . datacube . tile_root , tile_type_info , <EOL> tile_footprint , band_stack ) <EOL> return tile_contents <EOL> def current_transaction ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . transaction_stack [ - <NUM_LIT:1> ] <EOL> def mark_tile_for_removal ( self , tile_pathname ) : <EOL> """<STR_LIT>""" <EOL> self . current_transaction ( ) . mark_tile_for_removal ( tile_pathname ) <EOL> def mark_tile_for_creation ( self , tile_contents ) : <EOL> """<STR_LIT>""" <EOL> self . current_transaction ( ) . mark_tile_for_creation ( tile_contents ) <EOL> @ staticmethod <EOL> def __reindex_bands ( bands ) : <EOL> """<STR_LIT>""" <EOL> new_bands = { } <EOL> for ( tile_type , band_dict ) in bands . items ( ) : <EOL> for ( ( satellite , sensor ) , sensor_dict ) in band_dict . items ( ) : <EOL> for ( file_number , band_info ) in sensor_dict . items ( ) : <EOL> dataset_key = ( satellite , sensor , band_info [ '<STR_LIT>' ] ) <EOL> new_bands . setdefault ( dataset_key , { } ) <EOL> new_bands [ dataset_key ] . setdefault ( tile_type , { } ) <EOL> new_bands [ dataset_key ] [ tile_type ] [ file_number ] = band_info <EOL> return new_bands <EOL> def __check_satellite_and_sensor ( self , dataset ) : <EOL> """<STR_LIT>""" <EOL> satellite_id = self . db . get_satellite_id ( dataset . get_satellite_tag ( ) ) <EOL> if satellite_id is None : <EOL> raise DatasetError ( "<STR_LIT>" % <EOL> dataset . get_satellite_tag ( ) ) <EOL> sensor_id = self . db . get_sensor_id ( satellite_id , <EOL> dataset . get_sensor_name ( ) ) <EOL> if sensor_id is None : <EOL> msg = ( "<STR_LIT>" % <EOL> ( dataset . get_satellite_tag ( ) , dataset . get_sensor_name ( ) ) ) <EOL> raise DatasetError ( msg ) <EOL> def __check_processing_level ( self , dataset ) : <EOL> """<STR_LIT>""" <EOL> level_id = self . db . get_level_id ( dataset . get_processing_level ( ) ) <EOL> if level_id is None : <EOL> raise DatasetError ( "<STR_LIT>" % <EOL> dataset . get_processing_level ( ) ) <EOL> def __check_bands ( self , dataset ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> dataset_bands = self . new_bands [ self . get_dataset_key ( dataset ) ] <EOL> except KeyError : <EOL> raise DatasetError ( '<STR_LIT>' ) <EOL> for tile_type_bands in dataset_bands . values ( ) : <EOL> for band_info in tile_type_bands . values ( ) : <EOL> dataset . find_band_file ( band_info [ '<STR_LIT>' ] ) <EOL> class Transaction ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , db , tr_stack = None ) : <EOL> """<STR_LIT>""" <EOL> self . db = db <EOL> self . tr_stack = tr_stack <EOL> self . tile_remove_list = None <EOL> self . tile_create_list = None <EOL> self . previous_commit_mode = None <EOL> def __enter__ ( self ) : <EOL> """<STR_LIT>""" <EOL> self . tile_remove_list = [ ] <EOL> self . tile_create_list = [ ] <EOL> self . previous_commit_mode = self . db . turn_off_autocommit ( ) <EOL> if self . tr_stack is not None : <EOL> self . tr_stack . append ( self ) <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_value , exc_traceback ) : <EOL> """<STR_LIT>""" <EOL> if exc_type is None : <EOL> self . __commit ( ) <EOL> else : <EOL> self . __rollback ( ) <EOL> self . tile_remove_list = None <EOL> self . tile_create_list = None <EOL> self . db . restore_commit_mode ( self . previous_commit_mode ) <EOL> if self . tr_stack is not None : <EOL> tr = self . tr_stack . pop ( ) <EOL> assert tr is self , "<STR_LIT>" <EOL> def __commit ( self ) : <EOL> """<STR_LIT>""" <EOL> for tile_contents in self . tile_create_list : <EOL> tile_contents . make_permanent ( ) <EOL> self . db . commit ( ) <EOL> tile_create_set = { t . get_output_path ( ) <EOL> for t in self . tile_create_list } <EOL> for tile_pathname in self . tile_remove_list : <EOL> if tile_pathname not in tile_create_set : <EOL> if os . path . isfile ( tile_pathname ) : <EOL> os . remove ( tile_pathname ) <EOL> def __rollback ( self ) : <EOL> """<STR_LIT>""" <EOL> for tile_contents in self . tile_create_list : <EOL> tile_contents . remove ( ) <EOL> self . db . rollback ( ) <EOL> def mark_tile_for_removal ( self , tile_pathname ) : <EOL> """<STR_LIT>""" <EOL> if tile_pathname not in self . tile_remove_list : <EOL> self . tile_remove_list . append ( tile_pathname ) <EOL> def mark_tile_for_creation ( self , tile_contents ) : <EOL> """<STR_LIT>""" <EOL> self . tile_create_list . append ( tile_contents ) <EOL> class Lock ( object ) : <EOL> """<STR_LIT>""" <EOL> DEFAULT_WAIT = <NUM_LIT:10> <EOL> DEFAULT_RETRIES = <NUM_LIT:6> <EOL> def __init__ ( self , <EOL> datacube , <EOL> lock_list , <EOL> wait = DEFAULT_WAIT , <EOL> retries = DEFAULT_RETRIES ) : <EOL> """<STR_LIT>""" <EOL> self . datacube = datacube <EOL> self . lock_list = sorted ( lock_list ) <EOL> self . wait = wait <EOL> self . retries = retries <EOL> def __enter__ ( self ) : <EOL> """<STR_LIT>""" <EOL> for dummy_tries in range ( self . retries + <NUM_LIT:1> ) : <EOL> try : <EOL> self . __acquire_locks ( self . lock_list ) <EOL> break <EOL> except LockError : <EOL> time . sleep ( self . wait ) <EOL> else : <EOL> raise LockError ( ( "<STR_LIT>" % <EOL> self . retries ) + <EOL> self . lock_list ) <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_value , exc_traceback ) : <EOL> """<STR_LIT>""" <EOL> for object_to_unlock in self . lock_list : <EOL> self . datacube . unlock_object ( object_to_unlock ) <EOL> def __acquire_locks ( self , lock_list ) : <EOL> """<STR_LIT>""" <EOL> if lock_list : <EOL> if self . datacube . lock_object ( lock_list [ <NUM_LIT:0> ] ) : <EOL> try : <EOL> self . __acquire_locks ( lock_list [ <NUM_LIT:1> : ] ) <EOL> except : <EOL> self . datacube . unlock_object ( lock_list [ <NUM_LIT:0> ] ) <EOL> raise <EOL> else : <EOL> raise LockError ( ) <EOL> class LockError ( DatasetError ) : <EOL> """<STR_LIT>""" <EOL> pass </s>
