<commit_before>#
# <NAME>, CU Denver
#

from __future__ import absolute_import
import logging
import re
import os.path as osp
from datetime import datetime, timedelta
import subprocess
from utils import Dict, split_path, utc_to_utcf
from ingest.sat_source import SatSource

def parse_filename(file_name):
    info = {}
    pattern = 'FDC([CF]{1})-M([0-9]{1})_G([0-9]{2})_s([0-9]{14})_e([0-9]{14})'
    tfrmt = '%Y%j%H%M%S'
    match = re.search(pattern,file_name)
    if len(match.groups()) == 5:
        info['domain'] = match.group(1)
        info['mode'] = match.group(2)
        info['satellite'] = match.group(3)
        info['start_date'] = datetime.strptime(match.group(4)[0:13],tfrmt)
        info['end_date'] = datetime.strptime(match.group(5)[0:13],tfrmt)
    return info

def aws_search(awspaths):
    ls_cmd = 'aws s3 ls {} --recursive --no-sign-request'
    result = []
    for awspath in awspaths:
        cmd = ls_cmd.format(awspath).split()
        logging.debug('aws_search - {}'.format(' '.join(cmd)))
        max_retries = 5
        for tries in range(max_retries):
            try:
                r = subprocess.check_output(cmd)
            except:
                if tries == max_retries-1:
                    logging.error('error when requesting "{}", no more retries left'.format(' '.join(cmd)))
                else:
                    logging.warning('error when requesting "{}", retry {}/{}...'.format(' '.join(cmd),tries+1,max_retries))
                continue
            for line in r.decode().split('\n'):
                if len(line):
                    file_date,file_time,file_size,file_name = list(map(str.strip,line.split()))
                    info = parse_filename(file_name)
                    base = split_path(file_name)[0]
                    url = 's3://{}'.format(osp.join(base,file_name)) 
                    result.append({'producer_granule_id': file_name, 
                        'time_start': utc_to_utcf(info['start_date']), 
                        'updated': datetime.now(), 'url': url, 
                        'time_end': utc_to_utcf(info['end_date']), 
                        'domain': info['domain'], 'satellite': 'G{}'.format(info['satellite']),
                        'mode': info['mode']})
            break
    return result

class SatSourceAWSError(Exception):
    """
    Raised when a SatSourceAWS cannot retrieve satellite data.
    """
    pass

class SatSourceAWS(SatSource):
    """
    The parent class of all satellite sources that implements common functionality, for example

    - local satellite validation (file size check)
    - Satellite retrieval with retries (smart check whether server implements http-range)
    """

    def __init__(self, js):
        super(SatSourceAWS, self).__init__(js)

    def search_api_sat(self, sname, bounds, time, collection=None):
        """
        API search of the different satellite granules and return metadata dictionary

        :param sname: short name satellite product, ex: ''
        :param bounds: bounding box as (lon_min,lon_max,lat_min,lat_max)
        :param time: time interval (init_time_iso,final_time_iso)
        :param collection: id of the collection to specify

        :return metas: a dictionary with all the metadata for the API search
        """
        metas = []
        # hours first day
        stime = time[0]
        etime = min(time[0].replace(hour=23),time[1])
        n_first_hours = int((etime-stime).total_seconds()/3600)+1
        first_hours = [(stime + timedelta(hours=x)).strftime('%Y/%j/%H/') for x in range(n_first_hours)]
        awspaths = [osp.join(self.base_url.format(self.platform), self.product.format(self.sector), hour) for hour in first_hours]
        metas += aws_search(awspaths)
        # complete_days
        stime = (time[0]+timedelta(days=1)).replace(hour=0)
        etime = (time[1]-timedelta(days=1)).replace(hour=23)
        if stime < etime:
            n_complete_days = int((etime-stime).days)+1
            complete_days = [(stime + timedelta(days=x)).strftime('%Y/%j/') for x in range(n_complete_days)]
            awspaths = [osp.join(self.base_url.format(self.platform), self.product.format(self.sector), hour) for hour in complete_days]
            metas += aws_search(awspaths)
        # hours last day
        stime = max(time[0],time[1].replace(hour=0))
        etime = time[1]
        n_last_hours = int((etime-stime).total_seconds()/3600)+1
        last_hours = [h for h in ((stime + timedelta(hours=x)).strftime('%Y/%j/%H/') for x in range(n_last_hours)) if h not in first_hours]
        awspaths = [osp.join(self.base_url.format(self.platform), self.product.format(self.sector), hour) for hour in last_hours]
        metas += aws_search(awspaths)
        logging.info('SatSourceAWS.search_api_sat - {} metas found'.format(len(metas)))
        return metas

    def get_metas_sat(self, bounds, time):
        """
        Get all the meta data for all the necessary products

        :param bounds: bounding box as (lon_min,lon_max,lat_min,lat_max)
        :param time: time interval (init_time_datetime,final_time_datetime)
        :param maxg: max number of granules to process
        :return granules: dictionary with the metadata of all the products
        """
        metas=Dict({})
        metas.geo=self.search_api_sat(self.geo_prefix,bounds,time,collection=self.geo_collection_id)
        metas.fire=self.search_api_sat(self.fire_prefix,bounds,time,collection=self.fire_collection_id)
        metas.geo_nrt=self.search_api_sat(self.geo_nrt_prefix,bounds,time,collection=self.geo_nrt_collection_id)
        metas.fire_nrt=self.search_api_sat(self.fire_nrt_prefix,bounds,time,collection=self.fire_nrt_collection_id)
        return metas

    def group_metas(self, metas):
        """
        Group all the satellite metas before downloading to minimize number of files downloaded

        :param metas: satellite metadatas from API search
        :return result: groupped and cleanned metadata dictionary
        """
        g_id = lambda m: '_'.join(m['producer_granule_id'].split('.')[1:3])
        gmetas = Dict({})
        gmetas.geo = Dict({})
        gmetas.fire = Dict({})
        for m in metas['geo']:<commit_msg>AWS search working<commit_after>#
# <NAME>, CU Denver
#

from __future__ import absolute_import
import logging
import re
import os.path as osp
from datetime import datetime, timedelta
import subprocess
from utils import Dict, split_path, utc_to_utcf
from ingest.sat_source import SatSource

def parse_filename(file_name):
    info = {}
    pattern = 'FDC([CF]{1})-M([0-9]{1})_G([0-9]{2})_s([0-9]{14})_e([0-9]{14})'
    tfrmt = '%Y%j%H%M%S'
    match = re.search(pattern,file_name)
    if len(match.groups()) == 5:
        info['domain'] = match.group(1)
        info['mode'] = match.group(2)
        info['satellite'] = match.group(3)
        info['start_date'] = datetime.strptime(match.group(4)[0:13],tfrmt)
        info['end_date'] = datetime.strptime(match.group(5)[0:13],tfrmt)
    return info

def aws_search(awspaths, time=(datetime(2000,1,1),datetime.now())):
    ls_cmd = 'aws s3 ls {} --recursive --no-sign-request'
    result = []
    for awspath in awspaths:
        cmd = ls_cmd.format(awspath).split()
        logging.debug('aws_search - {}'.format(' '.join(cmd)))
        max_retries = 5
        for tries in range(max_retries):
            try:
                r = subprocess.check_output(cmd)
            except:
                if tries == max_retries-1:
                    logging.error('error when requesting "{}", no more retries left'.format(' '.join(cmd)))
                else:
                    logging.warning('error when requesting "{}", retry {}/{}...'.format(' '.join(cmd),tries+1,max_retries))
                continue
            for line in r.decode().split('\n'):
                if len(line):
                    _,_,file_size,file_name = list(map(str.strip,line.split()))
                    info = parse_filename(file_name)
                    if info['start_date'] <= time[1] and info['start_date'] >= time[0]:
                        base = split_path(file_name)[0]
                        url = 's3://{}'.format(osp.join(base,file_name)) 
                        result.append({'producer_granule_id': file_name, 
                            'time_start': utc_to_utcf(info['start_date']), 
                            'time_end': utc_to_utcf(info['end_date']), 
                            'updated': datetime.now(), 'url': url, 'file_size': file_size,
                            'domain': info['domain'], 'satellite': 'G{}'.format(info['satellite']),
                            'mode': info['mode']})
            break
    return result

class SatSourceAWSError(Exception):
    """
    Raised when a SatSourceAWS cannot retrieve satellite data.
    """
    pass

class SatSourceAWS(SatSource):
    """
    The parent class of all satellite sources that implements common functionality, for example

    - local satellite validation (file size check)
    - Satellite retrieval with retries (smart check whether server implements http-range)
    """

    def __init__(self, js):
        super(SatSourceAWS, self).__init__(js)

    def search_api_sat(self, sname, bounds, time, collection=None):
        """
        API search of the different satellite granules and return metadata dictionary

        :param sname: short name satellite product, ex: ''
        :param bounds: bounding box as (lon_min,lon_max,lat_min,lat_max)
        :param time: time interval (init_time_iso,final_time_iso)
        :param collection: id of the collection to specify

        :return metas: a dictionary with all the metadata for the API search
        """
        # complete_days
        stime = time[0]
        etime = time[1]
        n_complete_days = int((etime-stime).days)+1
        complete_days = [(stime + timedelta(days=x)).strftime('%Y/%j/') for x in range(n_complete_days)]
        awspaths = [osp.join(self.base_url.format(self.platform), self.product.format(self.sector), hour) for hour in complete_days]
        metas = aws_search(awspaths,time)
        logging.info('SatSourceAWS.search_api_sat - {} metas found'.format(len(metas)))
        return metas

    def get_metas_sat(self, bounds, time):
        """
        Get all the meta data for all the necessary products

        :param bounds: bounding box as (lon_min,lon_max,lat_min,lat_max)
        :param time: time interval (init_time_datetime,final_time_datetime)
        :param maxg: max number of granules to process
        :return granules: dictionary with the metadata of all the products
        """
        metas=Dict({})
        metas.geo=self.search_api_sat(self.geo_prefix,bounds,time,collection=self.geo_collection_id)
        metas.fire=self.search_api_sat(self.fire_prefix,bounds,time,collection=self.fire_collection_id)
        metas.geo_nrt=self.search_api_sat(self.geo_nrt_prefix,bounds,time,collection=self.geo_nrt_collection_id)
        metas.fire_nrt=self.search_api_sat(self.fire_nrt_prefix,bounds,time,collection=self.fire_nrt_collection_id)
        return metas

    def group_metas(self, metas):
        """
        Group all the satellite metas before downloading to minimize number of files downloaded

        :param metas: satellite metadatas from API search
        :return result: groupped and cleanned metadata dictionary
        """
        g_id = lambda m: '_'.join(m['producer_granule_id'].split('.')[1:3])
        gmetas = Dict({})
        gmetas.geo = Dict({})
        gmetas.fire = Dict({})
        for m in metas['geo']:
{'fraction_non_alphanumeric': 0.12634849400189868, 'fraction_numerical': 0.009320790541123673, 'mean_word_length': 2.684578696343402, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '898554', 'n_tokens_mistral': 3707, 'n_tokens_neox': 3532, 'n_words': 801}