cowidev.testing.batch

cowidev.testing.batch.andorra

class cowidev.testing.batch.andorra.Andorra

Bases: CountryTestBase

_df_builder(regex_key: str, text: str) → DataFrame

Builds Dataframe

_get_relevant_element(soup: BeautifulSoup) → Tag

Gets the relevant element.

_get_text_from_element(elem: Tag) → str

Extracts text from the element.

_parse_data(soup: BeautifulSoup) → DataFrame

Gets data from the source page.

_parse_metrics(text: str) → DataFrame

Get metrics from text.

export()

Exports data to CSV.

location: str = 'Andorra'

pipe_correct_dp(df: DataFrame)

Pipes the replacement data point.

pipe_date(df: DataFrame) → DataFrame

Pipes date column.

pipe_metrics(df: DataFrame) → DataFrame

Pipes metrics.

pipeline(df: DataFrame) → DataFrame

Pipeline for data.

read() → DataFrame

Reads data from the source page.

regex = {'pcr': "'n_pcr': { type: 'line', data: { labels:(.*?), datasets: .*? data: (.*?), fill:", 'script': "'n_serologics': {", 'tma': "'n_tma': { type: 'line', data: { labels:(.*?), datasets: .*? data: (.*?), fill:"}

source_label: str = "Tauler COVID-19, Govern d'Andorra"

source_url_ref: str = 'https://covid19.govern.ad'

units: str = 'tests performed'

cowidev.testing.batch.andorra.main()

cowidev.testing.batch.argentina

class cowidev.testing.batch.argentina.Argentina

Bases: CountryTestBase

export()

location: str = 'Argentina'

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'fecha': 'Date', 'positivos': 'positive', 'total': 'Daily change in cumulative total'}

source_label: str = 'Government of Argentina'

source_url: str = 'https://sisa.msal.gov.ar/datos/descargas/covid-19/files/Covid19Determinaciones.zip'

source_url_ref: str = 'https://datos.gob.ar/dataset/salud-covid-19-determinaciones-registradas-republica-argentina'

units: str = 'tests performed'

cowidev.testing.batch.argentina.main()

cowidev.testing.batch.armenia

class cowidev.testing.batch.armenia.Armenia

Bases: CountryTestBase

_build_df(data: dict) → DataFrame

Create df from raw data

_get_data_id_from_source(source_url: str) → str

Get Data ID from source

_load_data(data_id)

Load data from source

export()

Export data to csv

location: str = 'Armenia'

pipe_date(df: DataFrame) → DataFrame

Clean date

pipe_metrics(df: DataFrame) → DataFrame

Process metrics

pipe_pr(df: DataFrame) → DataFrame

Calculate Positive Rate

pipeline(df: DataFrame) → DataFrame

Pipeline for data processing

read() → DataFrame

Read data from source

regex: dict = {'element': 'window\\.infographicData=({.*})', 'entity': 'f5b6e83c-39b1-47c6-a84f-cd7ebaa3b7b1'}

rename_columns: dict = {'': 'Date', 'Բացասական թեստերի արդյունքներ': 'negative', 'Հաստատված դեպքեր': 'positive'}

source_label: str = 'National Center for Disease Control'

source_url: str = 'https://e.infogram.com/'

source_url_ref: str = 'https://ncdc.am/coronavirus/confirmed-cases-by-days/'

units: str = 'tests performed'

cowidev.testing.batch.armenia.main()

cowidev.testing.batch.australia

class cowidev.testing.batch.australia.Australia

Bases: CountryTestBase

export()

location: str = 'Australia'

notes: str = 'Made available by CovidbaseAU'

pipe_date(df: DataFrame) → DataFrame

pipe_metrics(df: DataFrame) → DataFrame

pipe_pr(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: str = {'Total': 'Cumulative total'}

source_label: str = 'Australian Government Department of Health'

source_url: str = 'https://covidbaseau.com/historical/COVID-19%20Tests%20Australia.csv'

source_url_ref: str = 'https://covidbaseau.com/tests/'

units: str = 'tests performed'

cowidev.testing.batch.australia.main()

cowidev.testing.batch.austria

class cowidev.testing.batch.austria.Austria

Bases: CountryTestBase

export()

location: str = 'Austria'

pipe_date(df: DataFrame)

pipe_exluce_dp(df: DataFrame)

pipe_filter(df: DataFrame)

pipeline(df: DataFrame)

read() → DataFrame

rename_columns: str = {'Meldedat': 'Date', 'TestGesamt': 'Cumulative total'}

source_label: str = 'Federal Ministry for Social Affairs, Health, Care and Consumer Protection'

source_url: str = 'https://covid19-dashboard.ages.at/data/CovidFallzahlen.csv'

source_url_ref: str = 'https://www.data.gv.at/katalog/dataset/846448a5-a26e-4297-ac08-ad7040af20f1'

units: str = 'tests performed'

cowidev.testing.batch.austria.main()

cowidev.testing.batch.belgium

class cowidev.testing.batch.belgium.Belgium

Bases: CountryTestBase

export()

location: str = 'Belgium'

pipeline(df: DataFrame) → DataFrame

read()

rename_columns: str = {'DATE': 'Date', 'PR': 'Positive rate', 'TESTS_ALL': 'Daily change in cumulative total'}

source_label: str = 'Sciensano (Belgian institute for health)'

source_url: str = 'https://epistat.sciensano.be/Data/COVID19BE_tests.csv'

source_url_ref: str = 'https://epistat.sciensano.be/Data/COVID19BE_tests.csv'

units: str = 'tests performed'

cowidev.testing.batch.belgium.main()

cowidev.testing.batch.bolivia

class cowidev.testing.batch.bolivia.Bolivia

Bases: CountryTestBase

export()

location: str = 'Bolivia'

notes: str = 'Made available by BoliGráfica on GitHub'

pipe_filter_rows(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'Bolivia': 'Cumulative total', 'Unnamed: 0': 'Date'}

source_label: str = 'Bolivia Ministry of Health'

source_url: str = 'https://raw.githubusercontent.com/dquintani/covid/main/pruebas_acum.csv'

source_url_ref: str = 'https://www.boligrafica.com/'

units: str = 'tests performed'

cowidev.testing.batch.bolivia.main()

cowidev.testing.batch.bosnia_herzegovina

class cowidev.testing.batch.bosnia_herzegovina.BosniaHerzegovina

Bases: CountryTestBase

_get_records(url: str) → dict

_load_data(url: str)

_parse_date(elem)

_parse_metric(elem)

_remove_typo(df: DataFrame) → DataFrame

export()

location: str = 'Bosnia and Herzegovina'

pipeline(df: DataFrame) → DataFrame

read()

source_label: str = 'Ministry of Civil Affairs'

source_url: str = ['http://mcp.gov.ba/publication/read/epidemioloska-slika-covid-19?pageId=3', 'http://mcp.gov.ba/publication/read/epidemioloska-slika-novo?pageId=97']

source_url_ref: str = 'http://mcp.gov.ba/publication/read/epidemioloska-slika-covid-19?pageId=3, http://mcp.gov.ba/publication/read/epidemioloska-slika-novo?pageId=97'

units: str = 'tests performed'

cowidev.testing.batch.bosnia_herzegovina.main()

cowidev.testing.batch.brazil

class cowidev.testing.batch.brazil.Brazil

Bases: CountryTestBase

export()

location: str = 'Brazil'

notes: str = 'Made available by Wesley Cota on GitHub'

pipe_metrics(df: DataFrame) → DataFrame

pipe_rename_columns(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'date': 'Date', 'tests': 'Cumulative total'}

source_label: str = 'Coronavírus Brasil'

source_url: str = 'https://raw.githubusercontent.com/wcota/covid19br/master/cases-brazil-states.csv'

source_url_ref: str = 'https://coronavirusbra1.github.io/'

units: str = 'tests performed'

cowidev.testing.batch.brazil.main()

cowidev.testing.batch.canada

class cowidev.testing.batch.canada.Canada

Bases: CountryTestBase

export()

location: str = 'Canada'

pipeline(df: DataFrame, units: str, metric_field: str) → DataFrame

pipeline_metric(df: DataFrame, units: str, metric_field: str) → DataFrame

read()

rename_columns: dict = {'date': 'Date', 'numtests': 'Cumulative total', 'prname': 'Country'}

source_label: str = 'Government of Canada'

source_url: str = 'https://health-infobase.canada.ca/src/data/covidLive/covid19-download.csv'

source_url_ref: str = 'https://health-infobase.canada.ca/src/data/covidLive/covid19-download.csv'

cowidev.testing.batch.canada.main()

cowidev.testing.batch.chile

class cowidev.testing.batch.chile.Chile

Bases: CountryTestBase

export()

location: str = 'Chile'

source_label: str = 'Ministry of Health, via Ministry of Science GitHub repository'

source_url_ref: str = 'https://github.com/MinCiencia/Datos-COVID19/tree/master/output/producto49'

units: str = 'tests performed'

cowidev.testing.batch.chile.main()

cowidev.testing.batch.colombia

class cowidev.testing.batch.colombia.Colombia

Bases: CountryTestBase

_read_antigens()

_read_pcr()

export()

location: str = 'Colombia'

pipe_cumulative_total(df: DataFrame)

pipe_positive_rate(df: DataFrame)

pipeline(df: DataFrame)

read()

source_label: str = 'National Institute of Health'

source_url_ref: str = 'https://www.ins.gov.co/Noticias/Paginas/coronavirus-pcr.aspx'

units: str = 'tests performed'

cowidev.testing.batch.colombia.main()

cowidev.testing.batch.costa_rica

class cowidev.testing.batch.costa_rica.CostaRica

Bases: CountryTestBase

export()

location: str = 'Costa Rica'

pipe_aggregate(df: DataFrame)

pipe_date(df: DataFrame)

pipe_metrics(df: DataFrame)

pipeline(df: DataFrame)

read()

source_label: str = 'Ministry of Health'

property source_url

source_url_ref: str = 'https://geovision.uned.ac.cr/oges/'

units: str = 'people tested'

cowidev.testing.batch.costa_rica.main()

cowidev.testing.batch.cuba

class cowidev.testing.batch.cuba.Cuba

Bases: CountryTestBase

_parse_data(data)

export()

location: str = 'Cuba'

notes: str = 'Made available on GitHub by covid19cubadata'

pipeline(df: DataFrame) → DataFrame

read()

source_label: str = 'Ministry of Public Health'

source_url: str = 'https://raw.githubusercontent.com/covid19cubadata/covid19cubadata.github.io/master/data/covid19-cuba.json'

source_url_ref: str = 'https://covid19cubadata.github.io/#cuba'

units: str = 'tests performed'

cowidev.testing.batch.cuba.main()

cowidev.testing.batch.cyprus

class cowidev.testing.batch.cyprus.Cyprus

Bases: CountryTestBase

export()

location: str = 'Cyprus'

pipeline(df: DataFrame) → DataFrame

read()

rename_columns: dict = {'date': 'Date', 'total tests': 'Cumulative total'}

source_label: str = 'Ministry of Health'

source_url: str = 'https://www.data.gov.cy/node/4617?language=en'

source_url_ref: str = 'https://www.data.gov.cy/node/4617?language=en'

units: str = 'tests performed'

cowidev.testing.batch.cyprus.main()

cowidev.testing.batch.czechia

class cowidev.testing.batch.czechia.Czechia

Bases: CountryTestBase

export()

location: str = 'Czechia'

pipe_metric(df: DataFrame) → DataFrame

pipe_pr(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read()

rename_columns: dict = {'datum': 'Date', 'incidence_pozitivni': 'positive', 'pocet_AG_testy': 'antigen', 'pocet_PCR_testy': 'pcr'}

source_label: str = 'Ministry of Health'

source_url: str = 'https://onemocneni-aktualne.mzcr.cz/api/v2/covid-19/testy-pcr-antigenni.csv'

source_url_ref: str = 'https://onemocneni-aktualne.mzcr.cz/api/v2/covid-19'

units: str = 'tests performed'

cowidev.testing.batch.czechia.main()

cowidev.testing.batch.denmark

class cowidev.testing.batch.denmark.Denmark

Bases: CountryTestBase

_parse_data_url()

export()

location: str = 'Denmark'

pipe_date(df: DataFrame)

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read()

source_label: str = 'Statens Serum Institut'

source_url_ref: str = 'https://covid19.ssi.dk/overvagningsdata/download-fil-med-overvaagningdata'

units: str = 'tests performed'

cowidev.testing.batch.denmark.main()

cowidev.testing.batch.ecdc

class cowidev.testing.batch.ecdc.ECDC

Bases: CountryTestBase

_yearweek_to_date(year_week: str) → str

Convert year_week(yyyy-Www) to date.

columns_use: list = ['year_week', 'region_name', 'tests_done']

export()

Export data to CSV.

export_countries(df: DataFrame)

Export data to the relevant csv and log the confirmation.

location: str = 'ECDC'

pipe_cumsum(df: DataFrame) → DataFrame

Calculate cumulative sum of tests.

pipe_date(df: DataFrame) → DataFrame

Pipe to convert year_week to date.

pipe_filter_entries(df: DataFrame) → DataFrame

Get valid entries:

• Discard subnational data.

• Countries not coming from OWID (avoid loop).

pipe_metadata(df: DataFrame) → DataFrame

Add metadata to DataFrame.

pipe_rename_countries(df: DataFrame) → DataFrame

Rename countries to match OWID naming convention.

pipeline(df: DataFrame)

Pipeline for data.

read()

Read data from source.

rename_columns: dict = {'region_name': 'location', 'tests_done': 'Cumulative total', 'year_week': 'Date'}

source_label: str = 'European Centre for Disease Prevention and Control (ECDC)'

source_url: str = 'https://opendata.ecdc.europa.eu/covid19/testing/csv/data.csv'

source_url_ref: str = 'https://www.ecdc.europa.eu/en/publications-data/covid-19-testing'

units: str = 'tests performed'

cowidev.testing.batch.ecdc.main()

cowidev.testing.batch.ecuador

class cowidev.testing.batch.ecuador.Ecuador

Bases: CountryTestBase

export()

Export data to CSV

location: str = 'Ecuador'

notes: str = 'Sum of confirmados and descartados'

pipe_date(df: DataFrame) → DataFrame

Convert date to datetime

pipe_metrics(df: DataFrame) → DataFrame

Calculate metrics

pipe_pr(df: DataFrame) → DataFrame

Calculate positive rate

pipeline(df: DataFrame) → DataFrame

Pipeline for data

read() → DataFrame

Read data from source

rename_columns: dict = {'created_at': 'Date', 'negativas_pcr': 'negative', 'positivas_pcr': 'positive'}

source_label: str = 'Ministerio de Salud Pública del Ecuador (via Ecuacovid)'

source_url: str = 'https://github.com/andrab/ecuacovid/raw/master/datos_crudos/ecuacovid.csv'

source_url_ref: str = 'https://github.com/andrab/ecuacovid'

units: str = 'people tested'

cowidev.testing.batch.ecuador.main()

cowidev.testing.batch.estonia

class cowidev.testing.batch.estonia.Estonia

Bases: CountryTestBase

export()

location: str = 'Estonia'

pipeline(df: DataFrame) → DataFrame

read()

rename_columns: dict = {'StatisticsDate': 'Date', 'TotalTests': 'Cumulative total'}

source_label: str = 'Estonian Health Board'

source_url: str = 'https://opendata.digilugu.ee/opendata_covid19_test_county_all.csv'

source_url_ref: str = 'https://www.terviseamet.ee/et/koroonaviirus/avaandmed'

units: str = 'tests performed'

cowidev.testing.batch.estonia.main()

cowidev.testing.batch.finland

class cowidev.testing.batch.finland.Finland

Bases: CountryTestBase

base_url = 'https://sampo.thl.fi/pivot/prod/en/epirapo/covid19case/fact_epirapo_covid19case.csv'

export()

location: str = 'Finland'

pipe_filter_columns(df: DataFrame) → DataFrame

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

source_label: str = 'Finnish Department of Health and Welfare'

source_url: str = 'https://sampo.thl.fi/pivot/prod/en/epirapo/covid19case/fact_epirapo_covid19case.csv?row=dateweek20200101-509093L&column=measure-444833.445356.492118.&&fo=1'

source_url_ref: str = 'https://sampo.thl.fi/pivot/prod/en/epirapo/covid19case/fact_epirapo_covid19case'

units: str = 'tests performed'

cowidev.testing.batch.finland.main()

cowidev.testing.batch.france

class cowidev.testing.batch.france.France

Bases: CountryTestBase

export()

location: str = 'France'

pipeline(df: DataFrame)

read()

source_label: str = 'National Public Health Agency'

source_url_ref: str = 'https://www.data.gouv.fr/fr/datasets/donnees-relatives-aux-resultats-des-tests-virologiques-covid-19/'

units: str = 'people tested'

cowidev.testing.batch.france.main()

cowidev.testing.batch.germany

class cowidev.testing.batch.germany.Germany

Bases: CountryTestBase

export()

location: str = 'Germany'

pipeline(df: DataFrame) → DataFrame

read()

source_url: str = 'https://www.rki.de/DE/Content/InfAZ/N/Neuartiges_Coronavirus/Daten/Testzahlen-gesamt.xlsx?__blob=publicationFile'

cowidev.testing.batch.germany.main()

cowidev.testing.batch.germany.read_xlsx_from_url(url, **kwargs)

cowidev.testing.batch.guam

class cowidev.testing.batch.guam.Guam

Bases: CountryTestBase

export()

location: str = 'Guam'

pipe_date(df: DataFrame) → DataFrame

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read()

rename_columns: dict = {'date': 'Date'}

source_label: str = 'Department of Health & Human Services'

source_url: str = 'https://healthdata.gov/api/views/j8mb-icvb/rows.csv'

source_url_ref: str = 'https://healthdata.gov/dataset/COVID-19-Diagnostic-Laboratory-Testing-PCR-Testing/j8mb-icvb'

units: str = 'tests performed'

cowidev.testing.batch.guam.main()

cowidev.testing.batch.guatemala

Constructs daily time series of COVID-19 testing data for Guatemala.

Official dashboard: https://tablerocovid.mspas.gob.gt/.

class cowidev.testing.batch.guatemala.Guatemala

Bases: CountryTestBase

export()

location: str = 'Guatemala'

source_label: str = 'Ministry of Health and Social Assistance'

source_url: str = 'https://gtmvigilanciacovid.shinyapps.io/3869aac0fb95d6baf2c80f19f2da5f98'

source_url_ref: str = 'https://gtmvigilanciacovid.shinyapps.io/3869aac0fb95d6baf2c80f19f2da5f98'

units: str = 'people tested'

cowidev.testing.batch.guatemala.main()

cowidev.testing.batch.hong_kong

class cowidev.testing.batch.hong_kong.HongKong

Bases: CountryTestBase

_load_cases()

export()

location: str = 'Hong Kong'

pipe_date(df)

pipe_metrics(df)

pipe_pr(df: DataFrame) → DataFrame

pipe_row_sum(df)

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'日期由 From Date': 'from', '日期至 To Date': 'Date', '普及社區檢測計劃下的檢測數目 Number of tests under Universal Community Testing Programme': 't3', '檢測數字 Number of tests': 't1', '特定群組檢測計劃下的檢測數目 Number of tests under Target Group Testing Scheme': 't2', '社區檢測中心的檢測數目 Number of tests in Community Testing Centres': 't5', '臨時檢測中心的檢測數目 Number of tests in Temporary Testing Centres': 't4'}

source_label: str = 'Department of Health'

source_url: str = 'http://www.chp.gov.hk/files/misc/statistics_on_covid_19_testing_cumulative.csv'

source_url_ref: str = 'http://www.chp.gov.hk/files/misc/statistics_on_covid_19_testing_cumulative.csv'

units: str = 'tests performed'

cowidev.testing.batch.hong_kong.main()

cowidev.testing.batch.hungary

class cowidev.testing.batch.hungary.Hungary

Bases: CountryTestBase

export()

location: str = 'Hungary'

notes: str = 'Made available by Atlo.team'

pipe_date(df)

pipe_filter(df)

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'Dátum': 'Date', 'Új mintavételek száma': 'Daily change in cumulative total'}

source_label: str = 'Government of Hungary'

source_url_ref: str = 'https://atlo.team/koronamonitor/'

units: str = 'tests performed'

cowidev.testing.batch.hungary.main()

cowidev.testing.batch.iceland

class cowidev.testing.batch.iceland.Iceland

Bases: CountryTestBase

_build_df(data: dict) → DataFrame

Create dfs from raw data

_get_data_id_from_source(source_url: str) → str

Get Data ID from source

_load_data(data_id)

Load data from source

export()

Export data to csv

location: str = 'Iceland'

pipe_date(df: DataFrame) → DataFrame

Clean date

pipe_pr(df: DataFrame) → DataFrame

Calculate Positive Rate

pipe_row_sum(df: DataFrame) → DataFrame

Sum rows

pipeline(df: DataFrame) → DataFrame

Pipeline for data processing

read() → DataFrame

Read data from source

regex: dict = {'element': 'window\\.infographicData=({.*})', 'title_positive': 'Fjöldi smita innanlands', 'title_test': 'Fjöldi sýna eftir dögum'}

rename_columns: dict = {'Antigen domestic tests': 't2', 'Border tests': 't4', 'Border tests 1 and 2': 't4', 'Domestic infections': 'p1', 'Domestic infections (PCR or antigen test)': 'p1', 'Domestic infections PCR': 'p2', 'Domestic tests (PCR or antigen)': 't1', 'PCR domestic tests': 't1', 'Quarantine- and random screening': 'p2', 'Quarantine- and random tests': 't2', 'Screening by deCODE Genetics': 'p3', 'Sympotmatic tests': 't1', 'Symptomatic screening': 'p1', 'Symptomatic tests': 't1', 'deCODE Genetics screening': 't3'}

source_label: str = 'Government of Iceland'

source_url: str = 'https://e.infogram.com/'

source_url_ref: str = 'https://www.covid.is/data'

units: str = 'tests performed'

cowidev.testing.batch.iceland.main()

cowidev.testing.batch.india

class cowidev.testing.batch.india.India

Bases: CountryTestBase

export()

location: str = 'India'

notes: str = 'Made available by DataMeet on GitHub'

pipe_filter_rows(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'report_time': 'Date', 'samples': 'Cumulative total'}

source_label: str = 'Indian Council of Medical Research'

source_url: str = 'https://raw.githubusercontent.com/datameet/covid19/master/data/icmr_testing_status.json'

source_url_ref: str = 'https://github.com/datameet/covid19'

units: str = 'samples tested'

cowidev.testing.batch.india.main()

cowidev.testing.batch.indonesia

class cowidev.testing.batch.indonesia.Indonesia

Bases: CountryTestBase

export()

location: str = 'Indonesia'

pipeline(df: DataFrame)

read()

source_label: str = 'Government of Indonesia'

source_url_ref: str = 'https://covid19.go.id/peta-sebaran'

units: str = 'people tested'

cowidev.testing.batch.indonesia.main()

cowidev.testing.batch.ireland

Constructs daily time series of COVID-19 testing data for Ireland.

Dashboard: https://covid19ireland-geohive.hub.arcgis.com/pages/hospitals-icu–testing

class cowidev.testing.batch.ireland.Ireland

Bases: CountryTestBase

TESTING_TYPE = 'PCR only'

export() → None

location: str = 'Ireland'

pipe_date(df: DataFrame)

pipeline(df: DataFrame)

read()

rename_columns: dict = {'Date_HPSC': 'Date', 'PosR7': 'Positive rate', 'Test24': 'Daily change in cumulative total', 'TotalLabs': 'Cumulative total'}

source_label: str = 'Government of Ireland'

source_url: str = 'https://services-eu1.arcgis.com/z6bHNio59iTqqSUY/arcgis/rest/services/LaboratoryLocalTimeSeriesHistoricView/FeatureServer/0/query'

source_url_ref: str = 'https://covid19ireland-geohive.hub.arcgis.com/pages/hospitals-icu--testing'

units: str = 'tests performed'

cowidev.testing.batch.ireland.main()

cowidev.testing.batch.ireland.sanity_checks(df: DataFrame) → None

checks that there are no obvious errors in the scraped data.

cowidev.testing.batch.israel

class cowidev.testing.batch.israel.Israel

Bases: CountryTestBase

export()

Export to CSV

location: str = 'Israel'

pipe_pr(df: DataFrame) → DataFrame

Calculates the positive rate

pipeline(df: DataFrame) → DataFrame

Pipeline for the data

read()

Reads data from the source

rename_columns: dict = {'amount': 'positive', 'amountPersonTested': 'Daily change in cumulative total', 'date': 'Date'}

source_label: str = 'Israel Ministry of Health'

source_url: str = {'positive': 'https://datadashboardapi.health.gov.il/api/queries/infectedPerDate', 'tests': 'https://datadashboardapi.health.gov.il/api/queries/testResultsPerDate'}

source_url_ref: str = 'https://datadashboard.health.gov.il/COVID-19/general'

units: str = 'people tested'

cowidev.testing.batch.israel.main()

cowidev.testing.batch.italy

class cowidev.testing.batch.italy.Italy

Bases: CountryTestBase

export()

location: str = 'Italy'

notes: str = 'Made available by the Department of Civil Protection on GitHub'

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'data': 'Date', 'tamponi': 'Cumulative total'}

source_label: str = 'Presidency of the Council of Ministers'

source_url: str = 'https://raw.githubusercontent.com/pcm-dpc/COVID-19/master/dati-andamento-nazionale/dpc-covid19-ita-andamento-nazionale.csv'

source_url_ref: str = 'https://github.com/pcm-dpc/COVID-19/tree/master/dati-andamento-nazionale/'

units: str = 'tests performed'

cowidev.testing.batch.italy.main()

cowidev.testing.batch.japan

class cowidev.testing.batch.japan.Japan

Bases: CountryTestBase

export()

location: str = 'Japan'

pipe_date(df: DataFrame) → DataFrame

pipe_filter_rows(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'PCR 検査実施人数(単日)': 'Daily change in cumulative total', '日付': 'Date'}

source_label: str = 'Ministry of Health, Labour and Welfare'

source_url: str = 'https://www.mhlw.go.jp/content/pcr_tested_daily.csv'

source_url_ref: str = 'https://www.mhlw.go.jp/content/pcr_tested_daily.csv'

units: str = 'people tested'

cowidev.testing.batch.japan.main()

cowidev.testing.batch.kazakhstan

Constructs daily time series of COVID-19 testing data for Kazakhstan.

Dashboard: https://hls.kz/

Notes:

• This module requires ChromeDriver to be installed

  (https://chromedriver.chromium.org/downloads) and in your executable $PATH.

class cowidev.testing.batch.kazakhstan.Kazakhstan

Bases: CountryTestBase

export() → None

location: str = 'Kazakhstan'

cowidev.testing.batch.kazakhstan.get_data() → DataFrame

cowidev.testing.batch.kazakhstan.main()

cowidev.testing.batch.kazakhstan.sanity_checks(df: DataFrame) → None

checks that there are no obvious errors in the scraped data.

cowidev.testing.batch.latvia

class cowidev.testing.batch.latvia.Latvia

Bases: CountryTestBase

export()

location: str = 'Latvia'

notes: str = 'Collected from the Latvian Open Data Portal'

pipe_date(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'Datums': 'Date', 'TestuSkaits': 'Daily change in cumulative total'}

source_label: str = 'Center for Disease Prevention and Control'

source_url: str = 'https://data.gov.lv/dati/dataset/f01ada0a-2e77-4a82-8ba2-09cf0cf90db3/resource/d499d2f0-b1ea-4ba2-9600-2c701b03bd4a/download/covid_19_izmeklejumi_rezultati.csv'

source_url_ref: str = 'https://data.gov.lv/dati/eng/dataset/covid-19'

units: str = 'tests performed'

cowidev.testing.batch.latvia.main()

cowidev.testing.batch.liechtenstein

class cowidev.testing.batch.liechtenstein.Liechtenstein

Bases: CountryTestBase

export()

location: str = 'Liechtenstein'

pipe_filter_rows(df: DataFrame) → DataFrame

pipe_groupby(df: DataFrame) → DataFrame

pipe_positive_rate(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'datum': 'Date', 'entries': 'Daily change in cumulative total'}

source_label: str = 'Federal Office of Public Health'

source_url: str = 'https://www.covid19.admin.ch/api/data/20220112-m4gbccen/sources/COVID19Test_geoRegion_PCR_Antigen.json'

source_url_ref: str = 'https://opendata.swiss/en/dataset/covid-19-schweiz'

units: str = 'tests performed'

cowidev.testing.batch.liechtenstein.main()

cowidev.testing.batch.lithuania

class cowidev.testing.batch.lithuania.Lithuania

Bases: CountryTestBase

export()

location: str = 'Lithuania'

pipe_date(df: DataFrame) → DataFrame

pipe_filter(df: DataFrame) → DataFrame

pipe_pr(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'date': 'Date', 'dgn_pos_day': 'daily_pos', 'dgn_tot_day': 'Daily change in cumulative total'}

source_label: str = 'Government of Lithuania'

source_url: str = 'https://opendata.arcgis.com/datasets/d49a63c934be4f65a93b6273785a8449_0.csv?where=municipality_code%20%3D%20%2700%27'

source_url_ref: str = 'https://open-data-ls-osp-sdg.hub.arcgis.com/datasets/d49a63c934be4f65a93b6273785a8449_0'

units: str = 'tests performed'

cowidev.testing.batch.lithuania.main()

cowidev.testing.batch.luxembourg

class cowidev.testing.batch.luxembourg.Luxembourg

Bases: CountryTestBase

_get_relevant_table(url: str) → Tag

Get the table with the relevant data

export()

Export data to CSV

location: str = 'Luxembourg'

pipe_date(df: DataFrame) → DataFrame

Convert date to datetime

pipe_pr(df: DataFrame) → DataFrame

Calculate positive rate

pipeline(df: DataFrame) → DataFrame

Pipeline for data

read() → DataFrame

Read data from source

rename_columns: dict = {'Nombre de personnes testées "positif"': 'positive', 'Nombre de tests PCR effectués': 'Cumulative total'}

source_label: str = 'Luxembourg Ministry of Health'

source_url_ref: str = 'https://msan.gouvernement.lu/fr/graphiques-evolution.html'

units: str = 'tests performed'

cowidev.testing.batch.luxembourg.main()

cowidev.testing.batch.malaysia

class cowidev.testing.batch.malaysia.Malaysia

Bases: CountryTestBase

export()

location: str = 'Malaysia'

notes: str = 'Made available by the Malaysia Ministry of Health on GitHub'

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'date': 'Date'}

source_label: str = 'Malaysia Ministry of Health'

source_url: str = 'https://raw.githubusercontent.com/MoH-Malaysia/covid19-public/main/epidemic/tests_malaysia.csv'

source_url_ref: str = 'https://github.com/MoH-Malaysia/covid19-public'

units: str = 'people tested'

cowidev.testing.batch.malaysia.main()

cowidev.testing.batch.malta

class cowidev.testing.batch.malta.Malta

Bases: CountryTestBase

export()

location: str = 'Malta'

notes: str = <NA>

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'Publication date': 'Date', 'Total NAA and rapid antigen tests': 'Cumulative total'}

source_label: str = 'COVID-19 Public Health Response Team (Ministry for Health)'

source_url: str = 'https://raw.githubusercontent.com/COVID19-Malta/COVID19-Data/master/COVID-19%20Malta%20-%20COVID%20Tests.csv'

source_url_ref: str = 'https://github.com/COVID19-Malta/COVID19-Data/'

units: str = 'tests performed'

cowidev.testing.batch.malta.main()

cowidev.testing.batch.marshall_islands

class cowidev.testing.batch.marshall_islands.MarshallIslands

Bases: CountryTestBase

export()

location: str = 'Marshall Islands'

pipe_date(df: DataFrame) → DataFrame

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read()

rename_columns: dict = {'date': 'Date'}

source_label: str = 'Department of Health & Human Services'

source_url: str = 'https://healthdata.gov/api/views/j8mb-icvb/rows.csv'

source_url_ref: str = 'https://healthdata.gov/dataset/COVID-19-Diagnostic-Laboratory-Testing-PCR-Testing/j8mb-icvb'

units: str = 'tests performed'

cowidev.testing.batch.marshall_islands.main()

cowidev.testing.batch.mexico

class cowidev.testing.batch.mexico.Mexico

Bases: CountryTestBase

export()

location: str = 'Mexico'

pipe_daily_change_in_cum_total(df: DataFrame) → DataFrame

pipe_filter_rows(df: DataFrame) → DataFrame

pipe_positive_rate(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

source_label: str = 'Health Secretary'

source_url: str = 'https://datos.covid-19.conacyt.mx/#DownZCSV'

source_url_ref: str = 'https://datos.covid-19.conacyt.mx/#DownZCSV'

units: str = 'people tested'

url_melt(url: str, name: str) → DataFrame

cowidev.testing.batch.mexico.main()

cowidev.testing.batch.netherlands

class cowidev.testing.batch.netherlands.Netherlands

Bases: CountryTestBase

export()

location: str = 'Netherlands'

notes: str = <NA>

pipe_positive_rate(df: DataFrame)

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'Date_of_statistics': 'Date', 'Tested_with_result': 'Daily change in cumulative total'}

source_label: str = 'Dutch National Institute for Public Health and the Environment'

source_url: str = 'https://data.rivm.nl/covid-19/COVID-19_uitgevoerde_testen.json'

source_url_ref: str = 'https://data.rivm.nl/covid-19/'

units: str = 'tests performed'

cowidev.testing.batch.netherlands.main()

cowidev.testing.batch.north_korea

class cowidev.testing.batch.north_korea.NorthKorea

Bases: CountryTestBase

_parse_table()

export()

location: str = 'North Korea'

pipe_filter(df: DataFrame)

pipeline(df: DataFrame)

read()

rename_columns: dict = {'Unnamed: 1': 'Daily change in cumulative total', 'Unnamed: 4': 'Source URL', 'Unnamed: 5': 'Date'}

source_label: str = 'NK News'

source_url_ref: str = 'https://www.nknews.org/pro/coronavirus-in-north-korea-tracker/'

units: str = 'people tested'

cowidev.testing.batch.north_korea.main()

cowidev.testing.batch.northern_mariana_islands

class cowidev.testing.batch.northern_mariana_islands.NorthernMarianaIslands

Bases: CountryTestBase

export()

location: str = 'Northern Mariana Islands'

pipe_date(df: DataFrame) → DataFrame

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read()

rename_columns: dict = {'date': 'Date'}

source_label: str = 'Department of Health & Human Services'

source_url: str = 'https://healthdata.gov/api/views/j8mb-icvb/rows.csv'

source_url_ref: str = 'https://healthdata.gov/dataset/COVID-19-Diagnostic-Laboratory-Testing-PCR-Testing/j8mb-icvb'

units: str = 'tests performed'

cowidev.testing.batch.northern_mariana_islands.main()

cowidev.testing.batch.norway

class cowidev.testing.batch.norway.Norway

Bases: CountryTestBase

export()

Exports data to csv

location: str = 'Norway'

pipe_date(df: DataFrame) → DataFrame

Cleans date column

pipe_metrics(df: DataFrame)

Pipes metrics

pipe_pr(df: DataFrame) → DataFrame

Pipes pr

pipeline(df: DataFrame) → DataFrame

pipeline for data

read() → DataFrame

Reads data from source.

source_label: str = 'Norwegian Institute of Public Health'

source_url: str = 'https://www.fhi.no/api/chartdata/api/90789'

source_url_ref: str = 'https://www.fhi.no/en/id/infectious-diseases/coronavirus/daily-reports/daily-reports-COVID19'

units: str = 'people tested'

cowidev.testing.batch.norway.main()

cowidev.testing.batch.peru

class cowidev.testing.batch.peru.Peru

Bases: CountryTestBase

date_start: str = '2020-04-08'

export()

Export data to CSV

location: str = 'Peru'

notes: str = 'Ministerio de Salud via https://github.com/jmcastagnetto/covid-19-peru-data'

pipe_filter(df: DataFrame) → DataFrame

Filter data

pipe_pr(df: DataFrame) → DataFrame

Calculate positive rate

pipeline(df: DataFrame) → DataFrame

Pipeline for data

read() → DataFrame

Read data from source

rename_columns: dict = {'confirmed': 'positive', 'date': 'Date', 'total_tests': 'Cumulative total'}

source_label: str = 'National Institute of Health'

source_url: str = 'https://raw.githubusercontent.com/jmcastagnetto/covid-19-peru-data/main/datos/covid-19-peru-data.csv'

source_url_ref: str = 'https://datos.ins.gob.pe/dataset/dataset-de-pruebas-moleculares-del-instituto-nacional-de-salud-ins'

units: str = 'tests performed'

cowidev.testing.batch.peru.main()

cowidev.testing.batch.philippines

class cowidev.testing.batch.philippines.Philippines

Bases: CountryTestBase

_base_url = 'https://drive.google.com/drive/folders/1ZPPcVU4M7T-dtRyUceb0pMAd8ickYf8o'

_get_file_id()

_get_id_folder()

_parse_drive_id_from_pdf(pdf_path)

export()

location: str = 'Philippines'

pipe_aggregate(df: DataFrame)

pipe_checks(df: DataFrame)

pipe_date(df: DataFrame)

pipeline(df: DataFrame)

read()

rename_columns: dict = {'cumulative_unique_individuals': 'Cumulative total', 'report_date': 'Date'}

source_label: str = 'Philippines Department of Health'

units: str = 'people tested'

cowidev.testing.batch.philippines.main()

cowidev.testing.batch.portugal

class cowidev.testing.batch.portugal.Portugal

Bases: CountryTestBase

export()

location: str = 'Portugal'

pipeline(df: DataFrame)

read()

source_label: str = 'Ministry of Health'

source_url: str = 'https://services5.arcgis.com/eoFbezv6KiXqcnKq/arcgis/rest/services/Covid19_Amostras/FeatureServer/0/query?f=json&where=1%3D1&returnGeometry=false&spatialRel=esriSpatialRelIntersects&outFields=*&orderByFields=Data_do_Relatorio%20asc&resultOffset=0&resultRecordCount=32000&resultType=standard&cacheHint=true'

source_url_ref: str = 'https://covid19.min-saude.pt/ponto-de-situacao-atual-em-portugal/'

units: str = 'tests performed'

cowidev.testing.batch.portugal.main()

cowidev.testing.batch.puerto_rico

class cowidev.testing.batch.puerto_rico.PuertoRico

Bases: CountryTestBase

export()

location: str = 'Puerto Rico'

pipe_date(df: DataFrame) → DataFrame

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read()

rename_columns: dict = {'date': 'Date'}

source_label: str = 'Department of Health & Human Services'

source_url: str = 'https://healthdata.gov/api/views/j8mb-icvb/rows.csv'

source_url_ref: str = 'https://healthdata.gov/dataset/COVID-19-Diagnostic-Laboratory-Testing-PCR-Testing/j8mb-icvb'

units: str = 'tests performed'

cowidev.testing.batch.puerto_rico.main()

cowidev.testing.batch.qatar

class cowidev.testing.batch.qatar.Qatar

Bases: CountryTestBase

export()

Exports data to csv

header: dict = {'Accept': 'application/json; odata=verbose', 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.16; rv:86.0) Gecko/20100101 Firefox/86.0'}

location: str = 'Qatar'

pipe_date(df: DataFrame) → DataFrame

Cleans date column

pipe_metrics(df: DataFrame)

Pipes metrics

pipeline(df: DataFrame) → DataFrame

pipeline for data

read() → DataFrame

Reads data from source.

rename_columns: dict = {'PublishingDate': 'Date', 'TotalTests24H': 'Daily change in cumulative total'}

source_label: str = 'Qatar Ministry of Public Health'

source_url: str = "https://covid19.moph.gov.qa/EN/_api/web/lists/getbytitle('Covid19DailyStatus')/items?$top=5000"

source_url_ref: str = 'https://covid19.moph.gov.qa/EN/Pages/default.aspx'

units: str = 'tests performed'

cowidev.testing.batch.qatar.main()

cowidev.testing.batch.rwanda

class cowidev.testing.batch.rwanda.Rwanda

Bases: CountryTestBase

date_start: str = '2021-11-08'

export()

Exports data to csv

location: str = 'Rwanda'

params: dict = {'cacheHint': True, 'f': 'json', 'orderByFields': 'created_date desc', 'outFields': '*', 'resultRecordCount': 32000, 'resultType': 'standard', 'returnGeometry': False, 'spatialRel': 'esriSpatialRelIntersects', 'where': '1=1'}

pipe_date(df: DataFrame) → DataFrame

Cleans date column

pipe_filter(df: DataFrame) → DataFrame

Filter data

pipeline(df: DataFrame) → DataFrame

pipeline for data

read() → DataFrame

Reads data from source.

rename_columns: dict = {'attributes.created_date': 'Date', 'attributes.cumulative_test': 'Cumulative total'}

source_label: str = 'Rwanda Ministry of Health'

source_url: str = 'https://gis.rbc.gov.rw/server/rest/services/Hosted/service_b580a3db9319449e82045881f1667b01/FeatureServer/0/query'

source_url_ref: str = 'https://rbc.gov.rw/index.php?id=707'

units: str = 'samples tested'

cowidev.testing.batch.rwanda.main()

cowidev.testing.batch.saudi_arabia

class cowidev.testing.batch.saudi_arabia.SaudiArabia

Bases: CountryTestBase

export()

Exports data to csv

location: str = 'Saudi Arabia'

params: dict = {'f': 'json', 'orderByFields': 'ReportDate asc', 'outFields': 'ReportDate,DailyTest', 'resultOffset': 0, 'resultRecordCount': 32000, 'resultType': 'standard', 'returnGeometry': False, 'spatialRel': 'esriSpatialRelIntersects', 'where': "ReportDate>'2020-01-01 00:00:00'"}

pipe_date(df: DataFrame) → DataFrame

Cleans date column

pipe_metrics(df: DataFrame)

Pipes metrics

pipeline(df: DataFrame) → DataFrame

pipeline for data

read() → DataFrame

Reads data from source.

rename_columns: dict = {'attributes.DailyTest': 'Daily change in cumulative total', 'attributes.ReportDate': 'Date'}

source_label: str = 'Ministry of Health'

source_url: str = 'https://services6.arcgis.com/bKYAIlQgwHslVRaK/arcgis/rest/services/DailyTestPerformance_ViewLayer/FeatureServer/0/query'

source_url_ref: str = 'https://covid19.moh.gov.sa/'

units: str = 'tests performed'

cowidev.testing.batch.saudi_arabia.main()

cowidev.testing.batch.senegal

class cowidev.testing.batch.senegal.Senegal

Bases: CountryTestBase

export()

Exports data to csv

location: str = 'Senegal'

params: dict = {'f': 'json', 'orderByFields': 'Date asc', 'outFields': 'Date,Nombre_de_tests_realises', 'resultOffset': 0, 'resultRecordCount': 32000, 'resultType': 'standard', 'returnGeometry': False, 'spatialRel': 'esriSpatialRelIntersects', 'where': '1=1'}

pipe_date(df: DataFrame) → DataFrame

Cleans date column

pipe_metrics(df: DataFrame)

Pipes metrics

pipeline(df: DataFrame) → DataFrame

pipeline for data

read() → DataFrame

Reads data from source.

rename_columns: dict = {'attributes.Date': 'Date', 'attributes.Nombre_de_tests_realises': 'Daily change in cumulative total'}

source_label: str = 'Ministry for Health and Social Action'

source_url: str = 'https://services7.arcgis.com/Z6qiqUaS6ImjYL5S/arcgis/rest/services/tendance_nationale/FeatureServer/0/query'

source_url_ref: str = 'http://www.sante.gouv.sn/'

units: str = 'tests performed'

cowidev.testing.batch.senegal.main()

cowidev.testing.batch.serbia

class cowidev.testing.batch.serbia.Serbia

Bases: CountryTestBase

base_url: str = 'https://github.com/aleksandar-jovicic/COVID19-Serbia/{}/master/timeseries.csv'

export()

location: str = 'Serbia'

notes: str = 'Made available by Aleksandar Jovičić on Github'

pipe_cumulative_total(df: DataFrame) → DataFrame

pipe_date(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'tested': 'Cumulative total', 'ts': 'Date'}

source_label: str = 'Ministry of Health'

property source_url

property source_url_ref

testing_type: str = 'PCR only'

units: str = 'people tested'

cowidev.testing.batch.serbia.main()

cowidev.testing.batch.slovakia

class cowidev.testing.batch.slovakia.Slovakia

Bases: CountryTestBase

export()

Export data to CSV

location: str = 'Slovakia'

notes: str = 'Ministry of Health via https://github.com/Institut-Zdravotnych-Analyz'

pipe_metrics(df: DataFrame) → DataFrame

Pipes metrics

pipe_pr(df: DataFrame) → DataFrame

Calculate positive rate

pipeline(df: DataFrame) → DataFrame

Pipeline for data

read() → DataFrame

Read data from source

rename_columns: dict = {'Datum': 'Date'}

source_label: str = 'Ministry of Health'

source_url: str = 'https://github.com/Institut-Zdravotnych-Analyz/covid19-data/raw/main/DailyStats/OpenData_Slovakia_Covid_DailyStats.csv'

source_url_ref: str = 'https://github.com/Institut-Zdravotnych-Analyz/covid19-data'

units: str = 'tests performed'

cowidev.testing.batch.slovakia.main()

cowidev.testing.batch.slovenia

class cowidev.testing.batch.slovenia.Slovenia

Bases: CountryTestBase

export()

Export data to CSV

location: str = 'Slovenia'

notes: str = 'National Institute of Public Health via Sledilnik'

pipe_date(df: DataFrame) → DataFrame

Pipes Date

pipe_filter(df: DataFrame) → DataFrame

Filter data

pipe_metrics(df: DataFrame) → DataFrame

Pipes metrics

pipe_pr(df: DataFrame) → DataFrame

Calculate positive rate

pipeline(df: DataFrame) → DataFrame

Pipeline for data

read() → DataFrame

Read data from source

rename_columns: dict = {'data.hagt.performed.today': 'ag', 'data.hagt.positive.today': 'positive_ag', 'total.performed.today': 'pcr', 'total.positive.today': 'positive_pcr'}

source_label: str = 'National Institute of Public Health'

source_url: str = 'https://api.sledilnik.org/api/lab-tests'

source_url_ref: str = 'https://covid-19.sledilnik.org/en/data'

units: str = 'tests performed'

cowidev.testing.batch.slovenia.main()

cowidev.testing.batch.south_africa

class cowidev.testing.batch.south_africa.SouthAfrica

Bases: CountryTestBase

export()

location: str = 'South Africa'

notes: str = 'Made available by the University of Pretoria on Github'

pipe_add_datapoint(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read()

rename_columns: dict = {'YYYYMMDD': 'Date', 'cumulative_tests': 'Cumulative total'}

source_label: str = 'National Institute for Communicable Diseases (NICD)'

source_url: str = 'https://raw.githubusercontent.com/dsfsi/covid19za/master/data/covid19za_timeline_testing.csv'

source_url_ref: str = 'https://github.com/dsfsi/covid19za'

units: str = 'people tested'

cowidev.testing.batch.south_africa.main()

cowidev.testing.batch.south_africa.pipe_date(df: DataFrame) → DataFrame

cowidev.testing.batch.south_africa.pipe_drop_nan(df: DataFrame)

cowidev.testing.batch.south_africa.pipe_metrics(df: DataFrame) → DataFrame

cowidev.testing.batch.south_korea

class cowidev.testing.batch.south_korea.SouthKorea

Bases: CountryTestBase

_read_new()

_read_old()

export()

location: str = 'South Korea'

notes: str = 'Data made available by Asia Regional Information Center at Seoul National University'

pipeline(df: DataFrame)

read()

source_label: str = 'Korea Disease Control and prevention Agency'

source_url_ref: str = 'https://sites.google.com/view/snuaric/data-service/covid-19/covid-19-data'

units: str = 'people tested'

cowidev.testing.batch.south_korea.main()

cowidev.testing.batch.spain

class cowidev.testing.batch.spain.Spain

Bases: CountryTestBase

base_url = 'https://www.sanidad.gob.es/profesionales/saludPublica/ccayes/alertasActual/nCov'

date = '03102022'

export()

location: str = 'Spain'

pipe_filter_columns(df: DataFrame) → DataFrame

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

source_label: str = 'Ministry of Health'

source_url: str = 'https://www.sanidad.gob.es/profesionales/saludPublica/ccayes/alertasActual/nCov/documentos/Datos_Pruebas_Realizadas_Historico_03102022.csv'

source_url_ref: str = 'https://www.sanidad.gob.es/profesionales/saludPublica/ccayes/alertasActual/nCov/pruebasRealizadas.htm'

test_url = 'https://www.sanidad.gob.es/profesionales/saludPublica/ccayes/alertasActual/nCov/documentos/Datos_Pruebas_Realizadas_Historico_03102022.csv'

units: str = 'tests performed'

cowidev.testing.batch.spain.main()

cowidev.testing.batch.sri_lanka

class cowidev.testing.batch.sri_lanka.SriLanka

Bases: CountryTestBase

_parse_data(data: dict) → DataFrame

Parses data from source.

export()

Exports data to csv

location: str = 'Sri Lanka'

pipe_date(df: DataFrame) → DataFrame

Cleans date column

pipe_metrics(df: DataFrame)

Pipes metrics

pipeline(df: DataFrame) → DataFrame

pipeline for data

read() → DataFrame

Reads data from source.

rename_columns: dict = {'date': 'Date'}

source_label: str = 'Sri Lanka Health Promotion Bureau'

source_url: str = 'https://www.hpb.health.gov.lk/api/get-current-statistical'

source_url_ref: str = 'https://www.hpb.health.gov.lk'

units: str = 'tests performed'

cowidev.testing.batch.sri_lanka.main()

cowidev.testing.batch.switzerland

class cowidev.testing.batch.switzerland.Switzerland

Bases: CountryTestBase

_get_related_url_from_source(source_url: str) → str

Get related url from source

export()

Export data to csv

location: str = 'Switzerland'

pipe_metrics(df: DataFrame) → DataFrame

Process metrics

pipeline(df: DataFrame) → DataFrame

Pipeline

read()

Read data from source

rename_columns: str = {'datum': 'Date', 'entries': 'Daily change in cumulative total'}

source_label: str = 'Federal Office of Public Health'

source_url: str = 'https://www.covid19.admin.ch/api/data/context'

source_url_ref: str = 'https://opendata.swiss/en/dataset/covid-19-schweiz'

units: str = 'tests performed'

cowidev.testing.batch.switzerland.main()

cowidev.testing.batch.taiwan

class cowidev.testing.batch.taiwan.Taiwan

Bases: CountryTestBase

export()

Export data to CSV

location: str = 'Taiwan'

pipe_date(df: DataFrame) → DataFrame

Convert date to datetime

pipe_filter(df: DataFrame) → DataFrame

Filter data

pipeline(df: DataFrame) → DataFrame

Pipeline for data

read() → DataFrame

Read data from source

rename_columns: dict = {'Total': 'Daily change in cumulative total', '通報日': 'Date'}

source_label: str = 'Taiwan CDC Open Data Portal'

source_url: str = 'https://od.cdc.gov.tw/eic/covid19/covid19_tw_specimen.csv'

source_url_ref: str = 'https://data.cdc.gov.tw/en/dataset/daily-cases-suspected-sars-cov-2-infection_tested'

units: str = 'people tested'

cowidev.testing.batch.taiwan.main()

cowidev.testing.batch.thailand

class cowidev.testing.batch.thailand.Thailand

Bases: CountryTestBase

export()

location: str = 'Thailand'

pipe_date(df: DataFrame)

pipe_filter(df: DataFrame)

pipeline(df: DataFrame)

read()

rename_columns: dict = {'Total Testing': 'Daily change in cumulative total'}

source_label: str = 'Department of Medical Sciences Ministry of Public Health'

source_url: str = 'https://data.go.th/dataset/9f6d900f-f648-451f-8df4-89c676fce1c4/resource/0092046c-db85-4608-b519-ce8af099315e/download'

source_url_ref: str = 'https://data.go.th/dataset/covid-19-testing-data'

units: str = 'tests performed'

cowidev.testing.batch.thailand.main()

cowidev.testing.batch.trinidad_and_tobago

class cowidev.testing.batch.trinidad_and_tobago.TrinidadAndTobago

Bases: CountryTestBase

export()

Exports data to csv

location: str = 'Trinidad and Tobago'

pipe_date(df: DataFrame) → DataFrame

Cleans date column

pipe_pr(df: DataFrame) → DataFrame

Calculate Positive Rate

pipeline(df: DataFrame) → DataFrame

pipeline for data

read() → DataFrame

Reads data from source.

rename_columns: dict = {'attributes.CreationDate': 'Date', 'attributes.positive_tests': 'positive', 'attributes.unique_public_private_test': 'Cumulative total'}

source_label: str = 'Ministry of Health'

source_url: str = 'https://services3.arcgis.com/x3I4DqUw3b3MfTwQ/arcgis/rest/services/service_7a519502598f492a9094fd0ad503cf80/FeatureServer/0/query?f=json&resultOffset=0&resultRecordCount=32000&where=report_dt%20IS%20NOT%20NULL%20AND%20report_dt%20BETWEEN%20timestamp%20%272020-12-20%2021%3A00%3A00%27%20AND%20CURRENT_TIMESTAMP&orderByFields=report_dt%20asc&outFields=unique_public_private_test,CreationDate,positive_tests&resultType=standard&returnGeometry=false&spatialRel=esriSpatialRelIntersects'

source_url_ref: str = 'https://www.covid19.gov.tt/'

units: str = 'people tested'

cowidev.testing.batch.trinidad_and_tobago.main()

cowidev.testing.batch.turkey

class cowidev.testing.batch.turkey.Turkey

Bases: CountryTestBase

_parse_table()

export()

location: str = 'Turkey'

pipe_date(df: DataFrame)

pipe_filter(df: DataFrame)

pipeline(df: DataFrame)

read()

rename_columns: dict = {'Bugünkü Test Sayısı': 'Daily change in cumulative total', 'Tarih': 'Date', 'Toplam Test Sayısı': 'Cumulative total'}

source_label: str = 'Turkish Ministry of Health'

source_url: str = 'https://covid19.saglik.gov.tr/TR-66935/genel-koronavirus-tablosu.html'

source_url_ref: str = 'https://covid19.saglik.gov.tr/TR-66935/genel-koronavirus-tablosu.html'

units: str = 'tests performed'

cowidev.testing.batch.turkey.main()

cowidev.testing.batch.united_arab_emirates

class cowidev.testing.batch.united_arab_emirates.UnitedArabEmirates

Bases: CountryTestBase

export()

Exports data to csv

location: str = 'United Arab Emirates'

params: dict = {'f': 'json', 'orderByFields': 'DATE_ asc', 'outFields': 'DATE_,TESTS', 'returnGeometry': False, 'spatialRel': 'esriSpatialRelIntersects', 'where': '1=1'}

pipe_date(df: DataFrame) → DataFrame

Cleans date column

pipe_metrics(df: DataFrame)

Pipes metrics

pipeline(df: DataFrame) → DataFrame

pipeline for data

read() → DataFrame

Reads data from source.

rename_columns: dict = {'attributes.DATE_': 'Date', 'attributes.TESTS': 'Daily change in cumulative total'}

source_label: str = 'UAE Federal Competitiveness and Statistics Authority'

source_url: str = 'https://geostat.fcsa.gov.ae/gisserver/rest/services/UAE_COVID19_Statistics_Rates_Layer/FeatureServer/0/query'

source_url_ref: str = 'https://fcsc.gov.ae/en-us/Pages/Covid19/UAE-Covid-19-Updates.aspx'

units: str = 'tests performed'

cowidev.testing.batch.united_arab_emirates.main()

cowidev.testing.batch.united_kingdom

class cowidev.testing.batch.united_kingdom.UnitedKingdom

Bases: CountryTestBase

export()

location: str = 'United Kingdom'

pipeline(df: DataFrame)

read() → DataFrame

source_label: str = 'Public Health England'

source_url_ref: str = 'https://coronavirus.data.gov.uk/details/testing'

units: str = 'tests performed'

cowidev.testing.batch.united_kingdom.main()

cowidev.testing.batch.united_states

class cowidev.testing.batch.united_states.UnitedStates

Bases: CountryTestBase

export()

location: str = 'United States'

pipe_date(df: DataFrame) → DataFrame

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read()

rename_columns: dict = {'date': 'Date'}

source_label: str = 'Department of Health & Human Services'

source_url: str = 'https://healthdata.gov/api/views/j8mb-icvb/rows.csv'

source_url_ref: str = 'https://healthdata.gov/dataset/COVID-19-Diagnostic-Laboratory-Testing-PCR-Testing/j8mb-icvb'

units: str = 'tests performed'

cowidev.testing.batch.united_states.main()

cowidev.testing.batch.united_states_virgin_islands

class cowidev.testing.batch.united_states_virgin_islands.UnitedStatesVirginIslands

Bases: CountryTestBase

export()

location: str = 'United States Virgin Islands'

pipe_date(df: DataFrame) → DataFrame

pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read()

rename_columns: dict = {'date': 'Date'}

source_label: str = 'Department of Health & Human Services'

source_url: str = 'https://healthdata.gov/api/views/j8mb-icvb/rows.csv'

source_url_ref: str = 'https://healthdata.gov/dataset/COVID-19-Diagnostic-Laboratory-Testing-PCR-Testing/j8mb-icvb'

units: str = 'tests performed'

cowidev.testing.batch.united_states_virgin_islands.main()

cowidev.testing.batch.uruguay

class cowidev.testing.batch.uruguay.Uruguay

Bases: CountryTestBase

export()

location: str = 'Uruguay'

static pipe_date(df: DataFrame) → DataFrame

static pipe_metrics(df: DataFrame) → DataFrame

pipeline(df: DataFrame) → DataFrame

read() → DataFrame

rename_columns: dict = {'date': 'Date', 'total': 'Cumulative total'}

source_label: str = 'Ministry of Public Health'

source_url: str = 'https://estadisticas.msp-uy.com/data.json'

source_url_ref: str = 'https://estadisticas.msp-uy.com/data.json'

testing_type = 'PCR only'

units: str = 'people tested'

cowidev.testing.batch.uruguay.main()

cowidev.testing.batch.zambia

Constructs daily time series of COVID-19 testing data for Zambia. ArcGIS Dashboard: https://zambia-open-data-nsdi-mlnr.hub.arcgis.com/pages/zambia-covid19

class cowidev.testing.batch.zambia.Zambia

Bases: CountryTestBase

export() → None

get_data() → DataFrame

location: str = 'Zambia'

cowidev.testing.batch.zambia.main()