Materiál k cvičení DBM1 týdny #4-5, sezona 2024/2025

Řešený příklad - využití PSČ k určení obce a vyšších územně-správních celků

Problém

Ukázková úloha spočívá v tom, že v rámci pseudonymizovaného (nebo nevhodně sesbíraného) datasetu máme k dispozici pouze PSČ (poštovní směrovací čísla) ale je požadováno, aby informace byla vykreslena na mapu obcí a vyšších hierarchických celků. Informace navíc může být požadována v přepočtu per capita, nebo jako absolutní hodnota.

Podproblémů je několik:

Data

📄 data_dle_psč.csv - obsahuje jednotlivé případy s informací o PSČ, pohlaví a stavu záznamu

📄 obce_psč_předpočítané.csv - obsahuje přepočítané hodnoty počtu čísel popisných a čísel evidenčních v obcích dle příslušného PSČ

📄 vazby-cr.csv - obsahuje vazby kódem obce a kódy vyšších územně-správních celků, do kterých přísluší. Viz OpenData.cz a RÚIAN endpoint

※ Pro zjednodušení byl obce_psč_předpočítané.csv předpzpracovaný a společně s daty o populaci bude řešen později

Vypracování

Načtení dat

Q1 v jakém formátu jsou datové soubory? Jaké enkódování a oddělovací znak používají?

Pro načtení dat budeme využívat knihovnu pandas a funkci read_csv. K lepší čitelnosti kódu strukturujme načítání do funkcí a definujem si konstanty pro cesty k souborům.


  

import pandas as pd

primary_data = 'data/data_dle_psč.csv'
psc_obce_data = 'data/obce_psč_předpočítané.csv'
hierarchy_data = 'data/vazby-cr.csv'


def preprocess_primary():
    df = pd.read_csv(...)

    return df

def preprocess_psc_obce():
    df = pd.read_csv(...)

    return df

def preprocess_hierarchy():
    df = pd.read_csv(...)
    
    return df

if __name__ == '__main__':
    print(preprocess_primary())
    print(preprocess_hierarchy())
    print(preprocess_psc_obce())

Opticky ověřte, že při výpisu dat není problém s kódováním a jsou správně oddělené hodnoty do příslušných sloupců.

Úpravy dat

U primárních dat by se hodilo odfiltrovat záznamy značené jako nekompletní. Dále bychom mohli (v rámci procviční) převést české popisky biologického pohlaví na běžně používané zkratky (např. muž → M).


  

def preprocess_primary():
    df = pd.read_csv(primary_data, sep=';', encoding='utf-8')
    df = df.loc[df["Complete?"] == "Complete"]
    df = df.drop(columns=['Complete?'])
    df = df.replace({"muž": "M", "žena": "F"})

    return df

Všimněte si, že akce jsou vykonávány sekvnenčně, tzn. že sloupec Complete? je nejprve filtrován a následně může být odstraněn.

U dat o obcích s PSČ zatím nemusíme nic řešit.

U hierarchických dat je potřeba vyřešit situaci, že nejnižší hierarchický prvek je COBCE_KOD, tzn. kód části obce. V našem případě ale používáme obec OBEC_KOD jako nejnižší hierarchický prvek. Potřebujeme tedy odstranit sloupec COBCE_KOD a zajistit, aby se odstranily duplicitní záznamy (pokud má obec více částí).


  

def preprocess_hierarchy():
    df = pd.read_csv(hierarchy_data, sep=';', encoding='cp1250')
    df = df.drop(columns=['COBCE_KOD'])
    df = df.drop_duplicates()

    return df

Všimněte si, že nejprve zahazujeme sloupec s největším detailem, čímž nám vzniknou duplicitní řádky, které následně odstraníme.

Q2 Proč nám vadí duplicity? Jaké problémy by mohly způsobit?

Q3 Kolik je v ČR obcí? Odpovídá počet záznamů v dataframe?

Náš dataframe má speciální vlastnost převodníku na vyšší celky. Bylo by vhodné zkontrolovat, zda všechny obce mají přiřazený kódy vyššího celků. To lze například voláním df.info(), kreré obsahuje informace o počtu neprázdných hodnot v jednotlivých sloupcích.

Pozorujeme jednu chybějící hodnotu a tak ji vypišme:


  

    print(df.info())
    
    print(df[df["OKRES_KOD"].isnull()])

Q4 O jakou obec nebo situaci se jedná?

K vyřešení problému můžeme doplnit vhodným způsobem chybějící hodnotu. V našem případě se můžeme orientovat podle kódu používaného v příslušném Shapefile, který využijeme k vizualizaci - zde je jako kód pro území Prahy používáno 9999.

Zároveň při té příležitosti převedeme sloupec na celočíselný typ, což nám usnadní následné spojování dat. Automaticky se chytil float kvůli chybějící hodnotě.


  

def preprocess_hierarchy():
    df = pd.read_csv(hierarchy_data, sep=';', encoding='cp1250')
    df = df.drop(columns=['COBCE_KOD'])
    df = df.drop_duplicates()

    df['OKRES_KOD'] = df['OKRES_KOD'].fillna(9999).astype(int) # Praha

return df

Pokud chceme mít kontrolu, že v datasetu probíhá vše podle očekvání (vhodné hlavně u měnících se vstupních souborů), můžeme využít tzv. assert statement, který zkontroluje, zda je podmínka pravdivá. Pokud ne, vyhodí AssertionError a program se zastaví.

Tímto způsobem můžeme zkontrolovat, že do okresu 9999 spadá pouze obec Praha (554782) a že OBEC_KOD je unikátní.


  

def preprocess_hierarchy():
    df = pd.read_csv(hierarchy_data, sep=';', encoding='cp1250')
    df = df.drop(columns=['COBCE_KOD'])
    df = df.drop_duplicates()

    df['OKRES_KOD'] = df['OKRES_KOD'].fillna(9999).astype(int) # Praha
    
    # validate uniqueness of OBEC_KOD
    duplicate_obec_kod = df[df.duplicated(subset='OBEC_KOD', keep=False)]
    assert duplicate_obec_kod.empty, "Duplicate OBEC_KOD found"

    # validate correctly asssigning Praha to OKRES_KOD
    assert set(df[df['OKRES_KOD'] == 9999]['OBEC_KOD']) == {554782}, "Invalid OKRES_KOD for OBEC_KOD"

    return df

Ověření dobrého mapování

Nyní by také bylo vhodné ověřit, že máme informaci o hierarchii pro všechny situace, které mohou nastat. Tzn. můžeme vytáhnout všechny kódy obcí z preprocess_psc_obce() a zjistit, zda se vyskytují v dataframu získaném z preprocess_hierarchy().


  

df_obce = preprocess_psc_obce()
df_hierarchy = preprocess_hierarchy()

missing_kod_obce = set(df_obce['Kód obce']) - set(df_hierarchy['OBEC_KOD'])
assert len(missing_kod_obce) == 0, "Missing 'Kód obce' in 'df_hierarchy'"

Všimněte si, že zde využíváme množinové operace. Vytvoříme množinu z kódů obcí a odečteme množinu kódů obcí z hierarchického dataframe. Pokud je výsledkem prázdná množina, znamená to, že všechny kódy obcí jsou pokryté.

Spojení dat

Nejjednoduším krokem k vyzkoušení spojení, je propojit každý záznam z primárního datasetu s odpovídajícími obcemi dle PSČ

※ Zatím neřešíme, jak rozpočítat hodnoty mezi obce se stejným PSČ


  

df_primary = preprocess_primary()
df_obce = preprocess_psc_obce()

df_merged = df_primary.merge(df_obce, left_on='PSČ', right_on='PSČ', how='inner')
print(df_merged)

U volání merge je možné nastavit parametr how, který odpovídá SQL JOIN operacím. V našem případě používáme inner, což znamená, že se spojí pouze záznamy, které mají shodné hodnoty v obou dataframech.

Q5 Jaká je rozdíl mezi inner, left, right a outer joinem?

Na data můžeme dále navázar hierarchickou informaci.


  

df_primary = preprocess_primary()
df_obce = preprocess_psc_obce()
df_hierarchy = preprocess_hierarchy()

df_merged = df_primary.merge(df_obce, left_on='PSČ', right_on='PSČ', how='inner')
df_merged = df_merged.merge(df_hierarchy, left_on='Kód obce', right_on='OBEC_KOD', how='inner')
print(df_merged)

Seskupení dat (naivní)

Pokud budeme chtít reportovat počty na různých úrovních hierarchie, můžeme využít metodu groupby a následně agregovat hodnoty.


  

df_grouped = df_merged.groupby(['OKRES_KOD']).size().reset_index(name='Count')
print(df_grouped)

Výsledkem bude dataframe, kde každý řádek odpovídá jednomu okresu a počet záznamů v tomto okrese. Funkce reset_index slouží k převedení indexu z OKRES_KOD na běžný sloupec. Bez jejího použití by se jednalo o datový typ Series, který by měl informaci o kódech okresů jako index.

Všimněte si, že neřešíme různé agregační funkce, případně sloupce, nad kterými je aplikovat.

💾 Aktuální stav kódu

Problém s naivním řešením seskupení

V předchozím případě bylo 11 původních záznamů rozpočítáno jako celkem 73 položek ve 4 skupinách seskupení. To je způsobeno tím, že při propojení PSČ s obcemi se vytvořili duplicitní záznamy a každý z nich měl stejnou váhu při seskupení.

Abychom toto vyřešili, je potřeba si pomoci s proporčním rozpočítáním. To znamená, že pokud máme 2 obce se stejným PSČ, musíme rozpočítat hodnoty podle nějaké metriky daných obcí (v našem případě dle počtu adresních míst).

Proporční rozpočítání

K řádnému rozpočítání potřebujeme znát:

Vytvořme pomocnou funkci, která bude vracet rozpočítání pro konkrétní PSČ a úroveň hierarchie. Konkrétně ve formě seznamu dvojic (kód, pravděpodobnost).


  

def aggregate_probabilities_by_hierarchy(df_zip_codes, df_hierarchy, psc_input, target_level):

    # Filter items for the given PSČ
    filtered_df = df_zip_codes[df_zip_codes['PSČ'] == psc_input]

    if filtered_df.empty:
        return []  # Return an empty list if the PSČ is not found

    # Calculate probabilities based on total_count
    probs_df = filtered_df[['Kód obce', 'total_count']].copy()
    total_sum = filtered_df['total_count'].sum()
    probs_df['probability'] = probs_df['total_count'] / total_sum

    # Merge with hierarchy to get target_level mappings
    merged_df = probs_df.merge(df_hierarchy, left_on='Kód obce', right_on='OBEC_KOD', how='left')

    if target_level not in df_hierarchy.columns:
        raise ValueError(f"Invalid target_level '{target_level}', must be one of: {list(df_hierarchy.columns[1:])}")

    # Aggregate probabilities by target_level
    aggregated_probabilities = merged_df.groupby(target_level)['probability'].sum().round(5)

    return list(aggregated_probabilities.items())

Nejprve je potřeba vybrat pouze obce, které mají hledané PSČ. Zároveň vracíme prázdný výsledek, pokud by se stalo, že PSČ není nalezeno.

Následně vytvořím pomocný dataframe obsahující pouze relevantní sloupce (Kód obce, total_count) a je vypočten celkový počet adresních míst s daným PSČ. To je použito k výpočtu odvozeného sloupce probability, což je proporce počtu adresních míst v obci.

Aby bylo možné vrátit libovolnou vyšší úroveň hierarchie, je připojena převodní tabulka na vyšší územně-správní celky.

Kvůli obecnosti je použit vstupní atribut target_level, který určuje, podle jakého sloupce se má seskupovat. Pokud by byl zadaný neplatný sloupec, funkce vyhodí ValueError. Operace groupby seskupí hodnoty podle daného sloupce a sečte pravděpodobnosti (výběr sloupce ['probability'] určuje, že se nasčítávají hodnoty právě odtud).

Na závěr je výsledek převeden na seznam dvojic (kód, pravděpodobnost).


  

p = aggregate_probabilities_by_hierarchy(df_obce, df_hierarchy, 33011, 'POU_KOD')
print(p)

Kontrolní volání funkce k rozpočtení PSČ 33011 na úroveň obcí s pověřeným obecním úřadem by mělo vrátit výsledek [(1350, 0.9211), (1368, 0.0789)]

Použití funkce k odvozenému sloupci v dataframe

Nyní lze využít vytvořenou funkci k vytvoření nového sloupce v dataframe, který bude obsahovat rozpočítané hodnoty pro daný PSČ a úroveň hierarchie. Pro zjednodušení vytvořme pomocnou funkci compute_probabilities(psc), která bude obalovat volání vytvořené funkce.


  

def compute_probabilities_POU(psc):
    return aggregate_probabilities_by_hierarchy(df_obce, df_hierarchy, psc, "POU_KOD")

df_primary['probabilities'] = df_primary["PSČ"].apply(compute_probabilities_POU)

Volání výše nejprve vybere sérii PSČ z dataframe (df_primary["PSČ"]) a na ní pomocí volání apply(compute_probabilities_POU) aplikuje vytvořenou funkci. To znamená, že hodnota PSČ z každého řádku bude předána funkci jako vstupní hodnota a výsledek funkce bude zpět uložen do nového sloupce probabilities.

Alternativou je volání funkce apply s anonymní (lambda) funkcí. Levá část je stejná a změnil se pouze vstup do volání apply(). Lambda funkce má jeden parametr psc, který je předán do funkce aggregate_probabilities_by_hierarchy. Vstupy df_obce a df_hierarchy jsou načteny z prostoru výše. Hodnota "POU_KOD" je pevně zvolena.


  

df_primary['probabilities'] = df_primary["PSČ"].apply(
        lambda psc: aggregate_probabilities_by_hierarchy(df_obce, df_hierarchy, psc, "POU_KOD")
)

Ověřte, že nový sloupec obsahuje seznam dvojic (kód, pravděpodobnost).


  

print(df_primary.to_string())

Q6 Jak by se musely kódy upravit, pokud bychom chtěli rozpočítat hodnoty na úroveň okresů?

Rozbití seznamu dvojic

V současném stavu se nám nebude dobře pracovat s informací o pravděpodobnostech, pokud bude uložena jako seznam dvojic. Intuitivně (ze znalosti SQL/databází) bychom chtěli rozbít seznam do jednotlivých řádek a zároveň dvojici rozdělit do dvou sloupců atomických hodnot.

Pro tento účel můžeme využít metodu explode, která rozdělí seznam do jednotlivých řádků. Následně můžeme využít metodu apply s anonymní funkcí, která rozdělí dvojici na dva sloupce.


  

df_exploded = df_primary.explode("probabilities")
df_exploded["code"] = df_exploded["probabilities"].apply(lambda x: x[0])
df_exploded["prob"] = df_exploded["probabilities"].apply(lambda x: x[1])
df_exploded = df_exploded.drop(columns=["probabilities"])

První řádek rozdělí seznam do jednotlivých řádek (podle počtu prvků v seznamu). Další 2 řádky slouží k vyvroření sloupců opisujících hodnotu na nultém a prvním indexu dvojice za použití lambda funkcí lambda x: x[0] a lambda x: x[1]. Poslední řádek jen uklízí dataframe od zbytečného sloupce.

Seskupení dat V2

Nyní můžeme využít nově vytvořené sloupce k seskupení dat na úrovni obcí s pověřeným obecním úřadem (případně jiné úrovně podle volby výše). Základní volání groupby je stejné jako v předchozím případě, ale nyní máme k dispozici podílovou informaci prob, kterou lze nasčítat.


  

df_grouped = df_exploded.groupby("code")["prob"].sum()
print(df_grouped)

Opět seskupujeme podle kódu code hierarchické úrovně (v tomto případě se jedná o 'POU_KOD'). Následně je vybrán hodnotový sloupec prob a nad ním je aplikována agregační funkce sum. Výsledkem je dataframe, kde každý řádek odpovídá jednomu kódu a hodnota je součtem pravděpodobností.

Alternativou pro komplexnější seskupování je použití funkce agg, která umožňuje specifikovat více agregačních funkcí najednou. V ukázkovém příkladu je použita funkce min a max pro výpočet minimálního a maximálního PSČ (leč nesmyslný a nevýznamný výpočet) v rámci skupiny. Výpočet sumy pravděpodobností je stejný jako v předchozím případě, ale zapsán v jiném formátu.


  

df_grouped = df_exploded.groupby("code").agg(
    prob_sum=("prob", "sum"),  # Sum of probabilities
    min_psc=("PSČ", "min"),    # Minimum PSČ in the group
    max_psc=("PSČ", "max")     # Maximum PSČ in the group
).reset_index()
print(df_grouped)

Manuální spočtení obce_psč_předpočítané.csv

Na závěr se můžeme dostat oklikou úplně na začátek, kdy jako jeden z vstupních datasetů byl poskytnut soubor obce_psč_předpočítané.csv. V tomto souboru jsou počty adresních míst v obcích dle PSČ.

📄 Původní podkladová data jsou k dispozici na OpenData, resp. přímý odkaz na RÚIAN endpoint.

Rozbalte soubory z archivu do složky data/adresy/ v projektu.

Q7 Prozkoumejte strukturu dat. Jaké sloupce obsahují, jaký význam má jeden řádek? Jaký význam má jeden soubor?

Postup zpracování není složitý.


  

def preprocess_zip_codes():
    all_files = [f for f in os.listdir(address_folder) if f.endswith('.csv')]
    df_list = [pd.read_csv(os.path.join(address_folder, file), sep=';', encoding='cp1250') for file in all_files]

    # missing vojenské újezdy 545422-boletice, 503941-libava with 0 address points
    # this triggers - FutureWarning: The behavior of DataFrame concatenation with empty or all-NA entries is deprecated

    combined_df = pd.concat(df_list, ignore_index=True)
    combined_df = combined_df[['Kód obce', 'Typ SO', 'PSČ']]

    # Group by 'Kód obce' and calculate counts
    grouped_df = combined_df.groupby(['Kód obce', 'PSČ']).agg(
        cp_count=('Typ SO', lambda x: (x == 'č.p.').sum()),  # Count of 'č.p.'
        cev_count=('Typ SO', lambda x: (x == 'č.ev.').sum()),  # Count of 'č.ev.'
        total_count=('Typ SO', 'count')  # Total count of rows
    ).reset_index()

    return grouped_df

Na začátku jsou použity dva list comprehensions. První získá seznam všech souborů v adresáři, které mají koncovku .csv. Druhý list comprehension načte jednotlivé soubory do dataframů, tzn. aplikuje pd.read_csv na každý soubor v seznamu a výsledek uloží jako položku v seznamu.

Následně jsou data spojena do jednoho dataframu pomocí pd.concat. Výsledný dataframe je omezen na sloupce Kód obce, Typ SO a PSČ.

※ Jelikož jsou 2 soubory prázdné (neobsahující ani jedno adresní místo), objeví se varování konstatující, že chování programu při spojování dataframů, pokud je jeden z nich prázdný, je deprecated. Tzn. výhledově v novějších verzích by se mohlo změnit.

Seskupení probíhá podle kombinace Kód obce a PSČ. Pro každou skupinu jsou spočítány hodnoty cp_count, cev_count a total_count. První dvě hodnoty jsou spočítány pomocí lambda funkcí, které počítají počet výskytů řetězců č.p. a č.ev. v atributu 'Typ SO'. Třetí hodnota je počet řádků v dané skupině.

Ověřte, že při použití preprocess_zip_codes() k získání df_obce v našem kódu se nezmění výsledek a vše bude fungovat stejně.


  

df_obce = preprocess_zip_codes()

💾 Aktuální stav kódu