src/dirbs/reports/csv_reports.py
def write_classified_triplets(logger: callable, conditions: list, report_dir: str, conn: callable):
    """Helper method to write classified triplets reports.

    Arguments:
        logger: DIRBS logger object
        conditions: conditions list to write report about
        report_dir: reporting directory
        conn: DIRBS PostgreSQL connection
    Returns:
        Report Metadata
    """
    logger.info('Generating per-condition classified triplets list...')
    with contextlib.ExitStack() as stack:
        # push files to the exit stack, to close them properly
        condition_labels = [c.label for c in conditions]
        filename_cond_map = {'classified_triplets_{0}.csv'.format(c): c for c in condition_labels}
        cond_label_file_map = {c: stack.enter_context(open(os.path.join(report_dir, fn), 'w', encoding='utf-8'))
                               for fn, c in filename_cond_map.items()}

        # create mapping between condition label and csv writer
        cond_label_csvwriter_map = {c: csv.writer(cond_label_file_map[c]) for c in condition_labels}

        # write headers to the files
        for _, writer in cond_label_csvwriter_map.items():
            writer.writerow(['imei', 'imsi', 'msisdn', 'operator'])

        # run query to find all classified triplets for the given conditions
        with conn.cursor() as cursor:
            query = """SELECT cs.imei_norm AS imei, cs.cond_name, mno.imsi,
                              mno.msisdn, mno.operator_id AS operator
                         FROM classification_state AS cs
                   INNER JOIN monthly_network_triplets_per_mno_no_null_imeis AS mno
                              ON mno.imei_norm = cs.imei_norm
                        WHERE cs.cond_name IN %s
                          AND cs.virt_imei_shard = calc_virt_imei_shard(cs.imei_norm)
                          AND cs.end_date IS NULL
                          AND mno.virt_imei_shard = calc_virt_imei_shard(mno.imei_norm)"""  # noqa: Q440
            sql_bytes = cursor.mogrify(query, [tuple([c for c in condition_labels])])
            query = sql.SQL(str(sql_bytes, conn.encoding))
            cursor.execute(query)

            for res in cursor:
                cond_label_csvwriter_map[res.cond_name].writerow([res.imei, res.imsi, res.msisdn, res.operator])
        logger.info('Per-condition classified triplets list generated successfully.')

    return _gen_metadata_for_reports(list(filename_cond_map.keys()), report_dir)
Helper method to write classified triplets reports.

    Arguments:
        logger: DIRBS logger object
        conditions: conditions list to write report about
        report_dir: reporting directory
        conn: DIRBS PostgreSQL connection
    Returns:
        Report Metadata
    
Helper method to write classified triplets reports.
Helper method to write classified triplets reports .
def write_classified_triplets(logger: callable, conditions: list, report_dir: str, conn: callable):
    logger.info('Generating per-condition classified triplets list...')
    with contextlib.ExitStack() as stack:
        condition_labels = [c.label for c in conditions]
        filename_cond_map = {'classified_triplets_{0}.csv'.format(c): c for c in condition_labels}
        cond_label_file_map = {c: stack.enter_context(open(os.path.join(report_dir, fn), 'w', encoding='utf-8'))
                               for fn, c in filename_cond_map.items()}
        cond_label_csvwriter_map = {c: csv.writer(cond_label_file_map[c]) for c in condition_labels}
        for _, writer in cond_label_csvwriter_map.items():
            writer.writerow(['imei', 'imsi', 'msisdn', 'operator'])
        with conn.cursor() as cursor:
            query = """SELECT cs.imei_norm AS imei, cs.cond_name, mno.imsi,
                              mno.msisdn, mno.operator_id AS operator
                         FROM classification_state AS cs
                   INNER JOIN monthly_network_triplets_per_mno_no_null_imeis AS mno
                              ON mno.imei_norm = cs.imei_norm
                        WHERE cs.cond_name IN %s
                          AND cs.virt_imei_shard = calc_virt_imei_shard(cs.imei_norm)
                          AND cs.end_date IS NULL
                          AND mno.virt_imei_shard = calc_virt_imei_shard(mno.imei_norm)"""  
            sql_bytes = cursor.mogrify(query, [tuple([c for c in condition_labels])])
            query = sql.SQL(str(sql_bytes, conn.encoding))
            cursor.execute(query)
            for res in cursor:
                cond_label_csvwriter_map[res.cond_name].writerow([res.imei, res.imsi, res.msisdn, res.operator])
        logger.info('Per-condition classified triplets list generated successfully.')
    return _gen_metadata_for_reports(list(filename_cond_map.keys()), report_dir)
def write_classified_triplets ( logger : callable , conditions : list , report_dir : str , conn : callable ) : logger . info ( 'Generating per-condition classified triplets list...' ) with contextlib . ExitStack ( ) as stack : condition_labels = [ c . label for c in conditions ] filename_cond_map = { 'classified_triplets_{0}.csv' . format ( c ) : c for c in condition_labels } cond_label_file_map = { c : stack . enter_context ( open ( os . path . join ( report_dir , fn ) , 'w' , encoding = 'utf-8' ) ) for fn , c in filename_cond_map . items ( ) } cond_label_csvwriter_map = { c : csv . writer ( cond_label_file_map [ c ] ) for c in condition_labels } for _ , writer in cond_label_csvwriter_map . items ( ) : writer . writerow ( [ 'imei' , 'imsi' , 'msisdn' , 'operator' ] ) with conn . cursor ( ) as cursor : query = """SELECT cs.imei_norm AS imei, cs.cond_name, mno.imsi,
                              mno.msisdn, mno.operator_id AS operator
                         FROM classification_state AS cs
                   INNER JOIN monthly_network_triplets_per_mno_no_null_imeis AS mno
                              ON mno.imei_norm = cs.imei_norm
                        WHERE cs.cond_name IN %s
                          AND cs.virt_imei_shard = calc_virt_imei_shard(cs.imei_norm)
                          AND cs.end_date IS NULL
                          AND mno.virt_imei_shard = calc_virt_imei_shard(mno.imei_norm)""" sql_bytes = cursor . mogrify ( query , [ tuple ( [ c for c in condition_labels ] ) ] ) query = sql . SQL ( str ( sql_bytes , conn . encoding ) ) cursor . execute ( query ) for res in cursor : cond_label_csvwriter_map [ res . cond_name ] . writerow ( [ res . imei , res . imsi , res . msisdn , res . operator ] ) logger . info ( 'Per-condition classified triplets list generated successfully.' ) return _gen_metadata_for_reports ( list ( filename_cond_map . keys ( ) ) , report_dir )
Helper method to write classified triplets reports.
Helper method to write classified triplets reports .
"""Helper method to write classified triplets reports.

    Arguments:
        logger: DIRBS logger object
        conditions: conditions list to write report about
        report_dir: reporting directory
        conn: DIRBS PostgreSQL connection
    Returns:
        Report Metadata
    """ # push files to the exit stack, to close them properly # create mapping between condition label and csv writer # write headers to the files # run query to find all classified triplets for the given conditions # noqa: Q440
None
logger
callable
DIRBS logger object
DIRBS logger object
conditions
list
conditions list to write report about
conditions list to write report about
report_dir
str
None
conn
callable
DIRBS PostgreSQL connection
DIRBS PostgreSQL connection