<jupyter_start><jupyter_text>Get reject list of profiles<jupyter_code>rList = glob('/storage/kakapo/*temp*')
wl = glob('/storage/kakapo/*.xml')
header = ['platform_number', 'cycle', 'lat', 'lon', 'c1', 'c2', 'c3', 'c4', 'c5']

def get_lines(filename):
    with open(filename, "r") as f:
        data = f.readlines()
        lines = []
        for line in data: 
            line = ' '.join(line.split())
            line = line.split(' ')
            line[-1] = line[-1].replace('/n', '')
            lines.append(line)
        print(filename)
    return lines

def parse_kakapo(filename, header):
    '''parses reject and white list in kakapo'''
    lines = get_lines(filename)
    df = pd.DataFrame(data=lines, columns=header)
    df['profile_id'] = df['platform_number'].apply(lambda x: str(x)) + '_'  + df['cycle'].apply(lambda x: str(x))
    return df

rejectDf = parse_kakapo(rList[0], header)
whiteList = parse_kakapo(wl[0], header)
print(rejectDf.shape)
rejectDf.head()
ok_ids = whiteList['profile_id'].tolist()
rejectDf = rejectDf[~rejectDf['profile_id'].isin(ok_ids)]
rejectList = rejectDf['profile_id'].tolist()
print(rejectDf.shape)
rejectDf.head()<jupyter_output>(1113, 10)<jupyter_text>Init database<jupyter_code>def create_collection(dbName, collectionName):
    dbUrl = 'mongodb://localhost:27017/'
    client = pymongo.MongoClient(dbUrl)
    db = client[dbName]
    coll = db[collectionName]
    coll = init_profiles_collection(coll)
    return coll    

def init_profiles_collection(coll):
    try:
        coll.create_index([('date', pymongo.DESCENDING)])
        coll.create_index([('dac', pymongo.DESCENDING)])
        coll.create_index([('lat', pymongo.DESCENDING)])
        coll.create_index([('lon', pymongo.DESCENDING)])
    except:
        logging.warning('not able to get collections or set indexes')
    return coll

coll = create_collection('JG', 'profiles')<jupyter_output><empty_output><jupyter_text>Parse data files<jupyter_code>files = glob('/storage/kakapo/*_padj.dat.gz')
header = ['latitude', 'longitude', 'day', 'month', 'year', 'hr', 'min',
          'sec', 'pressure', 'temperature', 'potential temperature',
          'salinity', 'DMODE status', 'WMO', 'cycle number', 'original netcdf file']

def get_gzip_lines(filename):
    with gzip.open(filename, "r") as f:
        data = f.readlines()
        lines = []
        for line in data:
            line = line.decode('utf-8')
            line = ' '.join(line.split())
            line = line.split(' ')
            line[-1] = line[-1].replace('/n', '')
            lines.append(line)
    return lines

def make_docs_from_lines(lines, presRange=[0,2000]):
    df = pd.DataFrame(data=lines, columns=header)
    df['profile_id'] = df['WMO'].apply(lambda x: str(x)) + '_'  + df['cycle number'].apply(lambda x: str(x))
    df = df[~df['profile_id'].isin(rejectList)]
    docs = []
    df = df.rename({'latitude': 'lat', 'longitude': 'lon', 'pressure':'pres',
                    'salinity':'psal', 'temperature': 'temp'}, axis=1)
    floatCols = ['lat', 'lon', 'pres', 'temp', 'psal']
    df[floatCols] = df[floatCols].astype(float)
    if presRange:
        df = df[(df['pres'].astype(float) >= presRange[0]) & (df['pres'].astype(float) <= presRange[1])]
    for profile_id, profDf in df.groupby('profile_id'):
        dl = profDf[['year', 'month', 'day']].astype(int).iloc[0]
        date = datetime(*dl)
        doc = {}
        doc['_id'] = profile_id
        doc['lat'] = profDf.lat.iloc[0]
        doc['lon'] = profDf.lon.iloc[0]
        doc['date'] = date
        meas = profDf[['pres', 'temp', 'psal']].to_dict(orient='records')
        if len(meas) == 0:
            continue
        
        doc['measurements'] = meas
        docs.append(doc)
        
    return docs
        
        
#coll.drop()
rejectProfiles = rejectDf.profile_id.tolist()
for fdx, filename in enumerate(files):
    lines = get_gzip_lines(filename)
    documents = make_docs_from_lines(lines, )
    print('index: {0}, filename: {1} lines: {2}'.format(fdx, filename, len(documents)))
    try:
        coll.insert_many(documents, ordered=False)
    except pymongo.errors.BulkWriteError as bwe:
        writeErrors = bwe.details['writeErrors']
        problem_idx = []
        for we in writeErrors:
            problem_idx.append(we['index'])
        trouble_list = [documents[i] for i in problem_idx]
        for doc in trouble_list:
            coll.replace_one({'_id': doc['_id']}, doc, upsert=True)<jupyter_output>index: 0, filename: /storage/kakapo/200401_padj.dat.gz lines: 1787
index: 1, filename: /storage/kakapo/200402_padj.dat.gz lines: 1707
index: 2, filename: /storage/kakapo/200403_padj.dat.gz lines: 1914
index: 3, filename: /storage/kakapo/200404_padj.dat.gz lines: 1875
index: 4, filename: /storage/kakapo/200405_padj.dat.gz lines: 2062
index: 5, filename: /storage/kakapo/200406_padj.dat.gz lines: 2009
index: 6, filename: /storage/kakapo/200407_padj.dat.gz lines: 2063
index: 7, filename: /storage/kakapo/200408_padj.dat.gz lines: 2179
index: 8, filename: /storage/kakapo/200409_padj.dat.gz lines: 2340
index: 9, filename: /storage/kakapo/200410_padj.dat.gz lines: 2527
index: 10, filename: /storage/kakapo/200411_padj.dat.gz lines: 2489
index: 11, filename: /storage/kakapo/200412_padj.dat.gz lines: 2701
index: 12, filename: /storage/kakapo/200501_padj.dat.gz lines: 2811
index: 13, filename: /storage/kakapo/200502_padj.dat.gz lines: 2660
index: 14, filename: /storage/kakapo/200503_padj.dat.gz li[...]<jupyter_text>test for rejects<jupyter_code>print(coll.find({'_id': rejectList[0]}).count())<jupyter_output>0<jupyter_text>Querying db into mat file<jupyter_code>startYear = 2007
endYear = 2016
JGDict = {}

profCycleNumberAggr = []
profFloatIDAggr = []
profJulDayAggr = []
profLatAggr = []
profLongAggr = []
profModeAggr = ''
profMonthAggr = []
profPresAggr = []
profPsalAggr = []
profTempAggr = []
profYearAggr = []
def repeat_array(val, nTimes):
    arr = [val for x in range(nTimes)]
    return arr

def records_to_arrays(meas):
    temp, pres, psal = [], [], []
    for row in meas:
        temp.append(row['temp'])
        pres.append(row['pres'])
        psal.append(row['psal'])
    return temp, pres, psal

def make_file_name(prefix='/storage/s/'):
    fileName = prefix + 'prof'
    fileName += str(obs).capitalize()
    fileName += '_at{}dbar'.format(str(presLevel))
    fileName += '_{0}_{1}'.format(minYear, maxYear)
    fileName += '.mat'
    return fileName
ed = datetime(endYear, 1, 1, 0)
dbLen = coll.count_documents({"date": {"$lte": ed} })
print(coll.count_documents({"date": {"$lte": ed} }))
from itertools import product
from calendar import monthrange
allIters = [[year for year in range(2004, 2016+1)], [month for month in range(1, 12+1)]]
yearMonth = list(product(*allIters))
year, month = yearMonth[1]
monthRange = monthrange(year, month)
print(monthRange)
datetime(year, 1, monthRange[1], 23, 59, 59)
12//12
for year, month in yearMonth:
    monthRange = monthrange(year, month)
    start = datetime(year, month, 1, 0)
    end = datetime(year + month//12, month%12 + 1, 1, 0, 0)
    ymQuery = {"date": {"$gte": start, "$lt": end}}
    cursor = coll.find(ymQuery).sort('date')
    print('year: {0}, month: {1}, nProf: {2}'.format(year, month, coll.count_documents(ymQuery)))
    
    profCycleNumberAggr = []
    profFloatIDAggr = []
    profJulDayAggr = []
    profLatAggr = []
    profLongAggr = []
    profModeAggr = ''
    profMonthAggr = []
    profPresAggr = []
    profPsalAggr = []
    profTempAggr = []
    profYearAggr = []
    for idx, doc in enumerate(cursor):

        date = doc['date']
        dateStr = date.strftime("%d-%b-%Y %H:%M:%S")
        profYear = date.year
        profMonth = date.month
        profLen = len(doc['measurements'])

        platCyc = doc['_id'].split('_')
        platform = platCyc[0]
        cycle = platCyc[1]

        temp, pres, psal = records_to_arrays(doc['measurements'])

        if len(temp) != profLen or len(pres) != profLen or len(psal) != profLen:
            pdb.set_trace()

        profCycleNumberAggr.append(cycle)
        profFloatIDAggr.append(platform)
        profJulDayAggr.append(dateStr)
        profLatAggr.append(doc['lat'])
        profLongAggr.append(doc['lon'])
        profYearAggr.append(profYear)
        profMonthAggr.append(profMonth)
        profModeAggr += ''
        profPresAggr.append(pres)
        profPsalAggr.append(psal)
        profTempAggr.append(temp)

    JGDict['profCycleNumberAggr'] = profCycleNumberAggr
    JGDict['profFloatIDAggr'] = profFloatIDAggr
    JGDict['profJulDayAggr'] = profJulDayAggr
    JGDict['profLatAggr'] = profLatAggr
    JGDict['profLongAggr'] = profLongAggr
    JGDict['profYearAggr'] = profYearAggr
    JGDict['profMonthAggr'] = profMonthAggr
    JGDict['profModeAggr'] = profModeAggr
    JGDict['profPresAggr'] = profPresAggr
    JGDict['profPsalAggr'] = profPsalAggr
    JGDict['profTempAggr'] = profTempAggr
    fileName = '/storage/forDonata/JG_{0}_{1}.mat'.format(year, month)
    savemat(fileName, JGDict)
JGDict['profCycleNumberAggr'] = profCycleNumberAggr
JGDict['profFloatIDAggr'] = profFloatIDAggr
JGDict['profJulDayAggr'] = profJulDayAggr
JGDict['profLatAggr'] = profLatAggr
JGDict['profLongAggr'] = profLongAggr
JGDict['profYearAggr'] = profYearAggr
JGDict['profMonthAggr'] = profMonthAggr
JGDict['profModeAggr'] = profModeAggr
JGDict['profPresAggr'] = profPresAggr
JGDict['profPsalAggr'] = profPsalAggr
JGDict['profTempAggr'] = profTempAggr
with open('/storage/JGtest.mat', 'ab') as f
    savemat(f, JGDict)
profJulDayAggr
doc
cursor = coll.find()
doc = cursor.next()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.15774735532047293, 'fraction_numerical': 0.027172785729101846, 'mean_word_length': 3.7039024390243904, 'pattern_counts': {'":': 7, '<': 19, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 1}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '365823', 'license': 'MIT', 'n_tokens_mistral': 3655, 'n_tokens_neox': 3423, 'n_words': 656, 'path': 'gridding/JG_pchip_interpolation/JG-data-to-db.ipynb', 'repo_name': 'BillMills/argo-database', 'size': '9642'}