Exercicis de realització i verificació de consultes amb tests a un fitxer CSV (Comma Separated Value) amb publicacions científiques (pel mòdul de bioinformàtica).

Com realitzar consultes a fitxers CSV en Python ?

Aquesta és una activitat per consolidar els aprenentatges de Python adquirits en estructures de control i de dades (i anteriors), en funcions, en tractament de fitxers i en testing per tal de realitzar consultes a un fitxer CSV amb informació útil per a realitzar investigació científica.

La Font de dades del CSV és el portal Scimago Journals

The SCImago Journal & Country Rank is a publicly available portal that includes the journals and country scientific indicators developed from the information contained in the Scopus® database. These indicators can be used to assess and analyze scientific domains. Journals can be compared or analysed separately. Country rankings may also be compared or analysed separately.

En altres paraules, és una organització que analitza mitjançant diverses mètriques la qualitat de les publicacions cientìfiques de diversos llocs del món: l'H-Index, el total absolut d'articles citats, documents publicats durant 1 i 3 anys…

I perquè volem extreure informació en un fitxer CSV (Comma Separated Values) en la informàtica ?

Doncs repassem els avantatges dels CSV:

  • Són fitxers de text que tenen informació organitzada i separada habitualment per simbols com: punt i coma, coma, tabulador, …
  • Això permet que siguin molt fàcils de llegir i escriure per a programadors, i que puguin ser tractats amb qualsevol llenguatge o fins i tot des del terminal de Linux.
  • Suporten tot el ventall de caràcters Unicode si és necessari (multiidioma).
  • Molt bon rendiment.
  • Es poden obrir fàcilment amb programes de fulls de càlcul (recomanable Libre Office) per si algú menys experimentat en informàtic hi ha de treballar.

Hi ha variacions dels fitxers CSV. Per exemple, si en comptes d'estar separats per comes estan separats per tabuladors, s'anomenen TSV (Tab Separated Value).

--

Extracció dades actualitzades de Scimago.

Per a extreure la versió més recent del fitxer accedir a la consulteu la web de Scimago i seguim els següents passos:

  1. Seleccionem els articles de "Medicine" de l'any 2022.
  2. La resta de camps deixem els que indica per defecte.
  3. Finalment, pitgem a "Download Data".
  4. Posem el fitxer csv dins la carpeta on crearem el projecte.

En aquesta captura teniu una mostra dels passos:

De tota manera, la versió que usarem per il·lustrar els tests és la del 2022 (més antiga). Si la voleu consultar n'hem guardat una còpia en un dels nostres projectes a Gitlab:

https://gitlab.com/xtec/bio/pandas/-/raw/main/data/scimago-medicine-2022.csv

Lectura del fitxer CSV.

Per llegir aquest fitxer en Python, utilitzarem una funció que li passarem la ruta i retornarà una llista amb totes les línies del fitxer.

Per facilitar la tasca usem la llibreria DictReader que crea el fitxer i el posa en un diccionari.

Cridarem aquesta funció i per provar que ha funcionat mostrem una línia. Només la primera perquè n’hi ha unes 7000 i pot tardar força en mostrar-les per pantalla!

import csv

# How to define a function in python with the word key
# the type date after the : is only documentation for Python
def read_csv_file(csv_file_path: str) -> list[dict]:
   
    with open(csv_file_path, newline='') as csv_file:
        csv_reader = csv.DictReader(csv_file, delimiter=';')
        result     = [row_dict for row_dict in csv_reader]

    return result

csv_file_path: str = "scimago-medicine.csv"
entries: list[dict] = read_csv_file(csv_file_path)
num: int = len(entries)
print("First entry") 
print(entries[0])
print(f"There are {num} entries.")

La resposta (pel fitxer de l'any 2022) és:

{'Rank': '1', 'Sourceid': '28773', 'Title': 'Ca-A Cancer Journal for Clinicians', 'Type': 'journal', 'Issn': '15424863, 00079235', 'SJR': '62,937', 'SJR Best Quartile': 'Q1', 'H index': '168', 'Total Docs. (2020)': '47', 'Total Docs. (3years)': '119', 'Total Refs.': '3452', 'Total Cites (3years)': '15499', 'Citable Docs. (3years)': '80', 'Cites / Doc. (2years)': '126,34', 'Ref. / Doc.': '73,45', 'Country': 'United States', 'Region': 'Northern America', 'Publisher': 'Wiley-Blackwell', 'Coverage': '1950-2020', 'Categories': 'Hematology (Q1); Oncology (Q1)'}
There are 7125 entries.

Crear i organitzar consultes.

Ens han demanat que realitzem diverses consultes al fitxer CSV. No només han de funcionar sinó que hem de separar el codi: el/s mètodes per la lectura de fitxers han d'estar en un mòdul apartat dels programes on fem consultes.

També hem de crear un programa per testejar que cada consulta funciona, que retorna els resultats que esperem en qualsevol cas.

Comencem amb aquestes 2 consultes; que ja tenim resoltes en el codi, i només hem d'organitzar-les.

  • Q1. How many entries are in scimago-medicine.csv?
  • Q2. Show the n first entries.

file_utils.py

# Imports
import csv

def read_csv_file(csv_file_path: str) -> list[dict]:
    '''Input:  The file contents as a single string.
      Output: A list of strings where each string is a row of the csv file.'''
    
    with open(csv_file_path, 'r', newline='', encoding='utf-8') as csv_file:
        # DictReader converts each row in csv_file in a dictionary.
        # The dictionary keys are the column names.
        csv_reader = csv.DictReader(csv_file, delimiter=';')
        result     = [row_dict for row_dict in csv_reader]

    return result

scimago_queries.py

# Our imports
import file_utils
# 3rd party imports
import pprint 

# File name and path.
csv_file_path: str = "scimago-medicine-2022.csv"   

# -----------------------------------------------------------------------------
# Q1. How many entries are in scimago-medicine.csv?
# -----------------------------------------------------------------------------
def q1_num_entries(entries: list[dict]) -> int:
    '''Input: List of entries
    Output: The number of entries.'''
    num: int = len(entries)
    #print("First entry:")
    #pprint.pp(entries[0])
    return num

# -----------------------------------------------------------------------------
# Q2. Show the n first entries.
# -----------------------------------------------------------------------------
def q2_first_entries(entries: list[dict], num_first_entries: int) -> list[dict]:
    '''Input: List of entries, number of first items.
    Output: The number of first entries defined in num_first_entries.'''
    return entries[0:num_first_entries]


# Main
# -----------------------------------------------------------------------------
if __name__ == "__main__":
    
    entries: list[dict] = file_utils.read_csv_file(csv_file_path)

    num_entries: list[dict] = q1_num_entries(entries)
    print(f"There are {num_entries} entries.")
    
    first_entries: list[dict] = q2_first_entries(entries,3)
    pprint.pp(first_entries)
# -----------------------------------------------------------------------------

Ja funcionen, i la lògica de les consultes està totalment separada de la de lectura del fitxer.

--

Testejar consultes.

Ja tenim les consultes creades. Ara, crearem funcions per verificar automàticament que funcionen, és a dir tests.

Si no ho has fet encara, instal·la Pytest dins del teu projecte.

pip install pytest

Si teniu problemes descobrint els tests amb el plugin de VSCode, esbborreu la carpeta .vscode i reinicieu el VSCode; tal i com explica la web oficial de VSCode:

[https://code.visualstudio.com/docs/python/testing]

Ara, ja podeu crear aquest programa amb els tests a les 2 queries. Segurament caldria afegir més tests per a la consulta 2 (més casos) però per ara servirà per executar correctament Pytest.

Recomanem usar aquesta estructura de fitxers/mòduls per tal de separar adequadament el codi:

- scimago-medicine-2022.csv
- file_utils.py
- scimago_queries.py
- scimago_tests.py

Donarem per suposat que el fitxer CSV del 2022 ja l'hem descarregat, els 2 anteriors fitxers Python ja els hem creat anteriorment i a continuació crearem el fitxer que usarem per executar els tests:

scimago_tests.py

import scimago_queries
import file_utils

entries: list[dict] = file_utils.read_csv_file('scimago-medicine.csv')

expected_entries_q1: int = 7125

def test_q1_2022_7125():
    assert scimago_queries.q1_num_entries(entries) == expected_entries_q1

expected_entry_q2_entry0Rank = '1'
expected_entry_q2_entry0Title = 'Ca-A Cancer Journal for Clinicians'

def test_q2_2022_entry0():
    first_entries = scimago_queries.q2_first_entries(entries,2)
    assert first_entries[0]['Rank'] == expected_entry_q2_entry0Rank
    assert first_entries[0]['Title'] == expected_entry_q2_entry0Title

Com podeu veure, els mètodes de test passen les proves perquè retorna els valors que esperavem tenint en compte el fitxer CSV del 2022:

Exercici previ. Per assegurar-te que et funcionen els tests, crea un altre test per provar la segona consulta (q2). Aquest test ha de provar que el contingut del segon registre és el que esperem, contrastant el rank, el títol, l'H-Index, i el tipus de publicació

expected_entry_q2_entry1Rank = '2'
expected_entry_q2_entry1Title = 'MMWR Recommendations and Reports'
expected_entry_q2_entry1Type = 'journal'
expected_entry_q2_entry1HIndex = '143'

def test_q2_2022_entry1():
    first_entries = scimago_queries.q2_first_entries(entries,2)
    assert first_entries[1]['Rank'] == expected_entry_q2_entry1Rank
    assert first_entries[1]['Title'] == expected_entry_q2_entry1Title
    assert first_entries[1]['Type'] == expected_entry_q2_entry1Type
    assert first_entries[1]['H index'] == expected_entry_q2_entry1HIndex

Ara que ja sabem com crear i provar consultes; en realitzarem de més interessants.

--

EXERCICIS: Crear i provar consultes.

Prova de crear les consultes que plantegem a continuació (dins del scimago_queries.py) i de provar-les (dins del scimago_tests.py).

Valorarem positivament que insereixis mètodes comuns en diverses consultes a file_utils.py.

Moltes d'aquestes consultes tenen diverses solucions i és important que provis de resoldre-les pel teu compte i només mirar les solucions quan et quedis encallat.

I un cop contrastis la solució amb la que has obtingut tu, que entenguis com has arribat a la solució, i plantejar-te pel teu compte consultes similars.

Així és com es dominen les tècniques de Big Data que cada cop s'aplicaquen més freqüentment en el món laboral.

Q3 - How many entries are from Spain? (Country = Spain)

Q4 - Show all the journals (Type = journal) published in UK (Country = United Kingdom) with an H-Index greater than 200, sorted by H-index (biggest H-Index first)

Q5 - What types of scientific publications are in the file (paràmetre Type)?
Resultat esperat:
['journal', 'book series', 'conference and proceedings', 'trade journal']

Q6 - Count the number of types of each scientific publication.

Q7 - Show all regions covered by all entries.
Resultat esperat:
{'Northern America', 'Middle East', 'Western Europe', 'Asiatic Region', 'Pacific Region', 'Latin America', 'Eastern Europe', 'Africa/Middle East', 'Africa'}

Q8 - Mean of H-index by region (difficult query)

Q9 - Count the number of journals from each country in the first quartile (SJR Best Quartile)='Q1'.

Possibles solucions consulta 3.

Codi consulta 3.

#Solució 31, iterativa.
def q3_spanish_entries(entries: list[dict]) -> int:
    '''Input: List of entries, number of first items.
    Output: The number of first entries defined in num_first_entries.'''
    numEntriesSpain: int = 0
    for entry in entries:
        if(entry['Country'] == 'Spain'):
            numEntriesSpain+=1
    return numEntriesSpain

#Solució 32, funcional.
def filterEntrySpain (entry:dict) -> bool:      
    return entry['Country'] == 'Spain'

def q3_spanish_entries_v2(entries: list[dict]) -> int:
    '''Input: List of entries, number of first items.
    Output: The number of first entries defined in num_first_entries.'''
    return len(list(filter(filterEntrySpain,entries)))

Test consulta 3.

expected_entries_q3: int = 137
def test_q3_2022_spanish_entries():
    assert scimago_main_queries.q3_spanish_entries_v2(entries) == expected_entries_q3

Possibles solucions consulta 4.

Codi consulta 4.

# -----------------------------------------------------------------------------
# Q4 - Show all the journals (Type = journal) published in UK 
# (Country = United Kingdom) with an H-Index greater than 200, 
# sorted by H-index (biggest H-Index first)
# -----------------------------------------------------------------------------
def q4(entries: list[dict]) -> list[dict]:
    '''Input: List of entries.
    Output: The number of .'''
    filtered_entries: list[dict] = list(filter(filterUKJournalHIndex200,entries))
    filtered_sorted_entries = sorted(filtered_entries, key=itemgetter('H index'),reverse=True)
    return filtered_sorted_entries

def filterUKJournalHIndex200(entry:dict) -> bool:
    '''Input: List of entries.
    Output: True if the country is UK, type journal, H-Index greater than 200.'''            
    return entry['Country'] == 'United Kingdom' and entry['Type'] == 'journal' \
        and int(entry['H index']) > 200

Test consulta 4.

expected_entry_q4_entry0Rank = '14'
expected_entry_q4_entry0HIndex = '762'

def test_q4_2022_entry0():
    first_entries_q4 = scimago_queries.q4(entries)
    assert first_entries_q4[0]['Rank'] == expected_entry_q4_entry0Rank
    assert first_entries_q4[0]['H index'] == expected_entry_q4_entry0HIndex

expected_numentries_q4 = 55

def test_q4_2022_num_entries():
    first_entries_q4 = scimago_queries.q4(entries)
    assert len(first_entries_q4)== expected_numentries_q4

Possibles solucions consultes 5 i 6.

Codi consultes 5 i 6.

# -----------------------------------------------------------------------------
# Q5 - What types of scientific publications are in the file (paràmetre Type)? 
# -----------------------------------------------------------------------------
def q5_set_types_each_pub(entries: list[dict]) -> set[str]:
    '''Input: List of entries.
    Output: A set of types of scientific publications'''
    types_set: set = set()
    for entry in entries:
        types_set.add(entry['Type'])

    return types_set

# -----------------------------------------------------------------------------
# Q6 - Count the number of types of each scientific publication.
# -----------------------------------------------------------------------------
def q6_number_types_each_pub(entries: list[dict]) -> dict[str,int]:
    '''Input: List of entries.
    Output: A dict with the type (key) and number (value) of each scientific publication.'''
    num_entries_type = {}
    for entry in entries:
        # if Type don't exist, we add it in the dict.
        if (not (entry['Type'] in num_entries_type)):
            num_entries_type[entry['Type']]=1
        # if Type exist, we sum 1 more publication.
        else:
            num_entries_type[entry['Type']] = num_entries_type[entry['Type']] + 1

    return num_entries_type

Test consultes 5 i 6.

expected_result_q5: set[str] = \
    {'journal', 'book series', 'conference and proceedings', 'trade journal'}
def test_q5_2022_types():
    assert expected_result_q5 == set(scimago_main_queries. \
        q5_set_types_each_pub(entries))

# expected_result_q6: dict[str,int] = {'journal': 7082, 'book series': 27, 'conference and proceedings': 5, 'trade journal': 4}

expected_result_q6: dict[str,int] = {'journal': 7216, 'book series': 28, 'conference and proceedings': 5, 'trade journal': 4}
def test_q6_2022_num_types():
    assert expected_result_q6 == scimago_main_queries. \
        q6_number_types_each_pub(entries)

Podem trobar més solucions dins dels fitxers:

Referències.