{ "cells": [ { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "\"\"\"VVB_data_upate is a script that updates the data from a chosen year in the main table VVB_vragenlijst\"\"\"\n", "\n", "# Import needed libraries\n", "import os\n", "import pandas as pd\n", "import numpy as np\n", "import sqlalchemy\n", "import pymysql\n", "from sqlalchemy import String, Float, Integer, DECIMAL, Text, types\n", "from sqlalchemy.exc import SQLAlchemyError\n", "from sqlalchemy import MetaData, Table, text\n", "print(f\"Pandas version: {pd.__version__} (pandas version used: 2.1.4)\")\n", "print(f\"Numpy version: {np.__version__} (numpy version used: 2.0.25)\")\n", "print(f\"sqlalchemy version: {sqlalchemy.__version__} (sqlalchemy version used: 1.4.6)\")\n", "print(f\"pymysql version: {pymysql.__version__} (pymysql version used: 1.4.6)\")\n", "# Containrization? " ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#############################################################################################\n", "# Ophalen van de database verbinding inlog gegevens en het maken van de connectie\n", "#############################################################################################\n", "from get_secrets import SecretsManager\n", "# Maak een instantie van de SecretsManager-klasse\n", "secrets_manager = SecretsManager()\n", "# Haal een secret op\n", "db_credentials = secrets_manager.get_secret(\"dev/venster/gemeenten_amazon_db\")\n", "\n", "def create_connection():\n", " \"\"\"Maakt een databaseverbinding met credentials uit AWS Secrets Manager.\"\"\"\n", " if not db_credentials:\n", " raise Exception(\"Kon databasecredentials niet ophalen.\")\n", "\n", " username = db_credentials[\"username\"]\n", " password = db_credentials[\"password\"]\n", " host = db_credentials[\"host\"]\n", " database = db_credentials[\"dbname\"]\n", "\n", " # Gebruik de credentials in de databaseverbinding\n", " engine = sqlalchemy.create_engine(\n", " f\"mysql+pymysql://{username}:{password}@{host}/{database}\"\n", " )\n", " return engine" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#############################################################################################\n", "# Ophalen van de inummers/oude kolomnamen en de bijbehorende nieuwe kolomnamen\n", "#############################################################################################\n", "def get_source_target_dict():\n", " engine = create_connection() # Gebruik je bestaande functie om de connectie te maken\n", " \n", " query = \"SELECT INUM, COLUMN_NAME FROM inum_mappings_vvb\"\n", " \n", " with engine.connect() as connection:\n", " df = pd.read_sql(query, connection) # Haal de data op als DataFrame\n", " \n", " # Zet de DataFrame om naar een dictionary\n", " source_target_dict = dict(zip(df[\"INUM\"], df[\"COLUMN_NAME\"]))\n", "\n", " return source_target_dict\n", "\n", "# Voorbeeldgebruik\n", "source_target_dict = get_source_target_dict()" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#############################################################################################\n", "# Function to convert MySQL datatype strings to SQLAlchemy type\n", "#############################################################################################\n", "def convert_to_sqlalchemy_type(data_type):\n", " data_type = data_type.lower() # Zorg ervoor dat alles lowercase is\n", "\n", " if \"varchar\" in data_type:\n", " # Haal de lengte op (bijv. varchar(40) -> 40)\n", " length = int(data_type.split(\"(\")[-1].strip(\")\")) if \"(\" in data_type else 255\n", " return String(length)\n", " elif \"text\" in data_type:\n", " return Text() # SQLAlchemy behandelt dit als TEXT\n", " elif \"decimal\" in data_type:\n", " # Haal precisie en schaal op (bijv. decimal(14,4) -> DECIMAL(14,4))\n", " precision, scale = map(int, data_type.split(\"(\")[-1].strip(\")\").split(\",\"))\n", " return DECIMAL(precision, scale)\n", " elif \"int\" in data_type:\n", " return Integer()\n", " elif \"float\" in data_type:\n", " return Float()\n", " else:\n", " raise ValueError(f\"❌ ERROR: Unsupported data type: {data_type}\")\n", " \n", "#############################################################################################\n", "# Ophalen van de datatypes van de kolomen die weg worden geschreven naar de bestemming tabel\n", "#############################################################################################\n", "def get_export_dt_dict():\n", " engine = create_connection() # Gebruik je bestaande databaseverbinding\n", " \n", " query = \"SELECT COLUMN_NAME, DATA_TYPE_VVB_VRAGENLIJST FROM inum_mappings_vvb WHERE actief = 1\"\n", " \n", " with engine.connect() as connection:\n", " df = pd.read_sql(query, connection) # Haal de data op als DataFrame\n", "\n", " df[\"converted_dt\"] = df[\"DATA_TYPE_VVB_VRAGENLIJST\"].apply(convert_to_sqlalchemy_type)\n", " \n", " # Zet de DataFrame om naar dictionaries\n", " export_dt_dict = dict(zip(df[\"COLUMN_NAME\"], df[\"DATA_TYPE_VVB_VRAGENLIJST\"])) # Ruwe MySQL-datatypes\n", " export_dt_dict_converted = dict(zip(df[\"COLUMN_NAME\"], df[\"converted_dt\"])) # SQLAlchemy-datatypes\n", "\n", " return export_dt_dict, export_dt_dict_converted\n", "\n", "# Voorbeeldgebruik\n", "export_dt_dict, export_dt_dict_converted = get_export_dt_dict()\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#############################################################################################\n", "# Verwijderd alle rijen van de bestemming tabel voor het jaar dat je wilt verversen\n", "#############################################################################################\n", "\n", "def delete_existing_rows(engine, destination_table, year):\n", " with engine.begin() as conn: # Use `begin()` to ensure the transaction is committed\n", " # Count rows with year = {year} before deletion\n", " count_query = sqlalchemy.text(f\"SELECT COUNT(*) FROM {destination_table} WHERE year = {year}\")\n", " initial_count = conn.execute(count_query).scalar()\n", " \n", " if initial_count > 0:\n", " # Delete rows with year = {year}\n", " delete_query = sqlalchemy.text(f\"DELETE FROM {destination_table} WHERE year = {year}\")\n", " result = conn.execute(delete_query)\n", " \n", " # Commit is automatically handled with `begin()` context manager\n", " print(f\"Deleted {result.rowcount} rows from {destination_table}.\")\n", " \n", " # Count rows again after deletion to confirm\n", " final_count = conn.execute(count_query).scalar()\n", " print(f\"Rows with year {year} before deletion: {initial_count}\")\n", " print(f\"Rows with year {year} after deletion: {final_count}\")\n", " else:\n", " print(f\"No rows with year {year} found.\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#############################################################################################\n", "# Importing all specified columns from data_upd_{year} into a pandas dataframe. \n", "#############################################################################################\n", "\n", "def extract_data(engine, source_table, year):\n", " query = f\"SELECT * FROM {source_table}\"\n", " df = pd.read_sql(query, engine)\n", "\n", " # Stap 1: Verwijder de jaartal-prefix (bv. \"i2024.\") uit de kolomnamen\n", " prefix = f\"i{year}.\"\n", " df = df.rename(columns=lambda x: x.replace(prefix, \"\") if isinstance(x, str) else x)\n", "\n", " # Stap 2: Vervang kolomnamen op basis van de mapping in source_target_dict\n", " df = df.rename(columns=source_target_dict)\n", " return df" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#############################################################################################\n", "# Ophalen van een lijst met alle kolommen die gemapped moeten worden (het omzetten van een waarde zoals bijv. 'Ongeveer evenveel zelf als uitbesteed' naar 0.5)\n", "# De waarde correspondeert met het getal achter de functie map_values1() die zich bevind in transform data functie\n", "#############################################################################################\n", "def get_mapped_columns(serie):\n", " engine = create_connection() # Gebruik je bestaande databaseverbinding\n", " \n", " query = f\"SELECT COLUMN_NAME FROM inum_mappings_vvb WHERE mapping = {serie} and actief = 1\"\n", " \n", " with engine.connect() as connection:\n", " df = pd.read_sql(query, connection) # Haal de data op als DataFrame\n", "\n", " return df[\"COLUMN_NAME\"].tolist() # Zet de kolomnamen om in een lijst" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#############################################################################################\n", "# Ophalen van een lijst met alle kolommen die berekeningen zijn, deze worden gedropt aa het einde van transform functie omdat deze later worden opnieuw berekend\n", "#############################################################################################\n", "\n", "def get_calculated_columns():\n", " engine = create_connection() # Gebruik je bestaande databaseverbinding\n", " \n", " query = \"SELECT COLUMN_NAME FROM inum_mappings_vvb WHERE berekening IS NOT NULL AND berekening <> '' and actief = 1\"\n", " \n", " with engine.connect() as connection:\n", " df = pd.read_sql(query, connection) # Haal de data op als DataFrame\n", "\n", " return df[\"COLUMN_NAME\"].tolist() # Zet de kolomnamen om in een lijst" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#############################################################################################\n", "# Ophalen van een lijst met alle kolommen die niet meegneomen (meer) hoeven worden\n", "#############################################################################################\n", "def get_inactive_columns():\n", " engine = create_connection() # Gebruik je bestaande databaseverbinding\n", " \n", " query = \"SELECT COLUMN_NAME FROM inum_mappings_vvb WHERE actief = 0\"\n", " \n", " with engine.connect() as connection:\n", " df = pd.read_sql(query, connection) # Haal de data op als DataFrame\n", "\n", " return df[\"COLUMN_NAME\"].tolist() # Zet de kolomnamen om in een lijst" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#############################################################################################\n", "# Ophalen van een lijst met data type voor alle kolomen, we verdelen ze onder in decimaal en int kolom lijsten\n", "#############################################################################################\n", "def get_column_types():\n", " engine = create_connection() # Gebruik je bestaande databaseverbinding\n", " \n", " query = \"SELECT COLUMN_NAME, DATA_TYPE_VVB_VRAGENLIJST FROM inum_mappings_vvb where actief = 1 and (berekening IS NULL or berekening = '')\"\n", " \n", " with engine.connect() as connection:\n", " df = pd.read_sql(query, connection) # Haal de data op als DataFrame\n", "\n", " # Filter de kolommen per datatype\n", " decimal_columns = df[df[\"DATA_TYPE_VVB_VRAGENLIJST\"].str.contains(\"decimal\", case=False, na=False)][\"COLUMN_NAME\"].tolist()\n", " integer_columns = df[df[\"DATA_TYPE_VVB_VRAGENLIJST\"].str.contains(\"int\", case=False, na=False)][\"COLUMN_NAME\"].tolist()\n", "\n", " return decimal_columns, integer_columns" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#############################################################################################\n", "# Transforming the values suitable for export to VVB_vragenlijst\n", "#############################################################################################\n", "\n", "def transform_data(df, year):\n", " \n", " # Defining specific functions for transformations\n", " def map_values1(series):\n", " series = series.replace({\n", " 'Ongeveer evenveel zelf als uitbesteed': 0.5,\n", " 'Volledig of grotendeels uitbesteed': 1.0,\n", " 'Volledig of grotendeels in eigen beheer': 0.0,\n", " 'nan': np.nan,\n", " '2': np.nan,\n", " '1': 1.0,\n", " '0': 0.0,\n", " '0.5': 0.5,\n", " '' : np.nan\n", " })\n", " return series\n", " \n", " def map_values2(series):\n", " series = series.replace({\n", " 'nan': pd.NA,\n", " '2': 2,\n", " '4': 4,\n", " '1': 1,\n", " '0': 0,\n", " '3': 3,\n", " ' 25-50%': 2,\n", " ' >75%': 4,\n", " ' 50-75%': 3,\n", " '<25%': 1,\n", " 'Niet van toepassing': 0\n", " })\n", " return series\n", "\n", " # Check of 'naam_organisatie' bestaat voordat we de transformatie uitvoeren\n", " if 'naam_organisatie' in df.columns:\n", " mask = df['naam_organisatie'].str.startswith(('Gemeente ', 'gemeente ', 'Gemeenten '), na=False)\n", "\n", " df.loc[mask, 'naam_organisatie'] = df.loc[mask, 'naam_organisatie']\\\n", " .str.replace('Gemeente ', '', regex=False)\\\n", " .str.replace('gemeente ', '', regex=False)\\\n", " .str.replace('Gemeenten ', '', regex=False)\n", "\n", " # Controleer of 'aantal_inwoners' bestaat voordat we de transformatie uitvoeren\n", " if 'aantal_inwoners' in df.columns:\n", " df['aantal_inwoners'] = pd.to_numeric(df['aantal_inwoners'], errors='coerce').astype(pd.Int64Dtype())\n", "\n", " # Categorizing the values of multiple columns with floats using a function\n", " columnmap1 = get_mapped_columns(1)\n", " for column in columnmap1:\n", " if column in df.columns: # Controleer of de kolom bestaat\n", " df[column] = map_values1(df[column])\n", "\n", " # Adds a column to the dataframe where all values are 'year'\n", " df['year'] = year\n", "\n", " # Ophalen van de lijsten\n", " decimal_columns, integer_columns = get_column_types()\n", "\n", " for column in decimal_columns: \n", " if column in df.columns and df[column].dtype != 'float64': \n", " df[column] = df[column].astype(str).str.replace(',', '.', regex=False) \n", " df[column] = pd.to_numeric(df[column], errors='coerce').astype('float64') \n", "\n", " for column in integer_columns:\n", " if column in df.columns and df[column].dtype != 'Int64':\n", " df[column] = pd.to_numeric(df[column], errors='coerce').astype('Int64')\n", "\n", " # Making sure that missing values in object type columns are 'None'\n", " object_columns = df.select_dtypes(include=['object']).columns\n", " for col in object_columns:\n", " df[col] = df[col].where(pd.notna(df[col]), None)\n", "\n", " # Ophalen van kolommen die we moeten droppen\n", " columns_precalculated = get_calculated_columns()\n", " columns_inactive = get_inactive_columns()\n", " \n", " # Drop alleen kolommen als ze in de DataFrame staan\n", " df = df.drop(columns=[col for col in columns_precalculated if col in df.columns])\n", " df = df.drop(columns=[col for col in columns_inactive if col in df.columns])\n", "\n", " return df\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#############################################################################################\n", "# Het wegschrijven van de data, als er nieuwe kolomen zijn bij gekomen worden deze aan de bestemming tabel eerst toegevoegd.\n", "# Kolomen die niet in de inum_mappings_vvb tabel staan worden niet meegenomen.\n", "#############################################################################################\n", "def load_data(df, table_name, export_dt_dict):\n", " engine = create_connection() # Gebruik je bestaande databaseverbinding\n", " metadata = MetaData()\n", " \n", " with engine.connect() as connection:\n", " # Haal bestaande kolommen op uit de target tabel\n", " existing_columns = []\n", " if engine.dialect.has_table(connection, table_name): # Controleer of de tabel bestaat\n", " table = Table(table_name, metadata, autoload_with=engine)\n", " existing_columns = table.columns.keys()\n", " \n", " # **Verwijder kolommen uit df die niet in export_dt_dict staan**\n", " columns_to_keep = set(export_dt_dict.keys()).union({'year'}) \n", " columns_to_drop = [col for col in df.columns if col not in columns_to_keep]\n", " \n", " if columns_to_drop:\n", " print(f\"🗑️ Kolommen die niet in export_dt_dict staan, worden verwijderd uit df: {columns_to_drop}\")\n", " df.drop(columns=columns_to_drop, inplace=True, errors='ignore')\n", "\n", " # **Vind kolommen die nog niet in de database staan**\n", " new_columns = [col for col in df.columns if col not in existing_columns]\n", "\n", " if new_columns:\n", " print(f\"Nieuwe kolommen gevonden: {new_columns}. Deze worden toegevoegd aan de database.\")\n", "\n", " # Lijst om ongeldige kolommen uit df te verwijderen\n", " columns_to_drop = []\n", "\n", " # Voer ALTER TABLE statements uit om nieuwe kolommen toe te voegen\n", " for col in new_columns:\n", " col_type = export_dt_dict[col] # Haal het datatype op uit export_dt_dict\n", "\n", " if col in export_dt_dict:\n", "\n", " col_type = str(export_dt_dict[col]) # Converteer naar string\n", " alter_query = text(f\"ALTER TABLE `{table_name}` ADD COLUMN `{col}` {col_type};\")\n", " \n", " try:\n", " connection.execute(alter_query)\n", " print(f\"Kolom toegevoegd: {col} ({col_type})\")\n", " except SQLAlchemyError as err:\n", " print(f\"Error bij toevoegen van kolom {col}: {err}\")\n", " \n", " # Voeg de data toe aan de database\n", " try:\n", " df.to_sql(name=table_name, con=engine, \n", " if_exists='append', # Voeg toe aan de bestaande tabel\n", " index=False, \n", " dtype=export_dt_dict_converted # Gebruik de juiste datatypes\n", " )\n", " print(\"Data inserted successfully.\")\n", " except SQLAlchemyError as err:\n", " print(f\"Error inserting data: {err}\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#######################################################\n", "# Haalt data voor een specifiek jaar op uit vragenlijst_vvb (behalve als anders aangegeven)\n", "########################################################\n", "def extract_data_for_calc(table_name, year): \n", " engine = create_connection()\n", " # Loading data from MySQL table into Pandas DataFrame\n", " query = f\"SELECT * FROM {table_name} where year = {year}\"\n", " df = pd.read_sql(query, engine)\n", " return df" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#######################################################\n", "# Haalt de inummers, kolomnamen en berkeningen op uit inum_mappings_vvb \n", "########################################################\n", "def extract_calculations(): \n", " engine = create_connection()\n", " # Loading data from MySQL table into Pandas DataFrame\n", " query = f\"SELECT * FROM inum_mappings_vvb\"\n", " df = pd.read_sql(query, engine)\n", " return df" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#######################################################\n", "# Aanmaken van de berekende kolomen in de vragenlijst (worden opnieuw berekend)\n", "# Maakt gebruik van de berekening kolom in inum_mappings_vvb \n", "########################################################\n", "def add_vragenlijst_calculated_columns(df):\n", " # Haal de berekeningen op\n", " mappings = extract_calculations()\n", "\n", " # Maak een mapping van INUM naar kolomnaam\n", " inum_to_col = {row[\"INUM\"]: row[\"COLUMN_NAME\"] for _, row in mappings.iterrows()}\n", "\n", " for _, row in mappings.iterrows():\n", " if row[\"berekening\"]:\n", " formula = row[\"berekening\"]\n", "\n", " # Vervang inummers door kolomnamen in de berekening\n", " for inum, col_name in inum_to_col.items():\n", " formula = formula.replace(str(inum), f\"df['{col_name}'].fillna(0)\")\n", "\n", " # Voer de berekening uit en zorg dat deling door 0 wordt behandeld als None\n", " try:\n", " result = eval(formula)\n", " result.replace([np.inf, -np.inf], np.nan, inplace=True) # Vervang oneindige waarden door NaN\n", " result[result == 0] = np.nan # Optioneel: als je null wilt voor 0-resultaten\n", " df[row[\"COLUMN_NAME\"]] = result\n", " except ZeroDivisionError:\n", " df[row[\"COLUMN_NAME\"]] = None # Zet None bij deling door 0\n", " except Exception as e:\n", " print(f\"Fout bij berekening voor {row['COLUMN_NAME']}: {e}\")\n", "\n", " return df" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#######################################################\n", "# DASHBOARD VOORBEREKENINGEN\n", "#######################################################\n", "def add_dashboard_calculated_columns(df):\n", " df['d_short_long_ratio'] = (\n", " (df['totaal_medewerkers'].fillna(0) - (df['dienstverband_korter_dan_3_jaar'].fillna(0) + df['dienstverband_10_jaar_of_langer'].fillna(0))) / df['totaal_medewerkers'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_long_ratio'] = (\n", " df['dienstverband_10_jaar_of_langer'].fillna(0) / df['totaal_medewerkers'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_short_ratio'] = (\n", " df['dienstverband_korter_dan_3_jaar'].fillna(0) / df['totaal_medewerkers'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_safety_cost_ratio'] = (\n", " df['totale_kosten_veiligheidsregio_verbonden_partij'].fillna(0) / df['totaal_exploitatierekening'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_other_party_cost_ratio'] = (\n", " (df['totale_kosten_verbonden_partijen'].fillna(0) - df['totale_kosten_omgevingsdienst_verbonden_partij'].fillna(0) - df['totale_kosten_veiligheidsregio_verbonden_partij'].fillna(0) - df['totale_kosten_GGD_verbonden_partij'].fillna(0)) / df['totaal_exploitatierekening'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_env_service_cost_ratio'] = (\n", " df['totale_kosten_omgevingsdienst_verbonden_partij'].fillna(0) / df['totaal_exploitatierekening'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_health_service_cost_ratio'] = (\n", " df['totale_kosten_GGD_verbonden_partij'].fillna(0) / df['totaal_exploitatierekening'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_ext_hire_ratio'] = (\n", " df['aantal_externe_inhuur_personen'].fillna(0) / df['v_te_servicen_medewerkers'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_internal_abs'] = (\n", " df['v_te_servicen_medewerkers'].fillna(0) - df['aantal_externe_inhuur_personen'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_non_ext_hire_ratio'] = (\n", " 1 - df['aantal_externe_inhuur_personen'].fillna(0) / df['v_te_servicen_medewerkers'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_software_cost_ratio'] = (\n", " df['ict_kosten_software'].fillna(0) / df['v_te_servicen_medewerkers'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_hardware_cost_ratio'] = (\n", " df['ict_kosten_hardware'].fillna(0) / df['v_te_servicen_medewerkers'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_deprec_cost_ratio'] = (\n", " df['ict_kosten_afschrijvingen'].fillna(0) / df['v_te_servicen_medewerkers'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_outsource_cost_ratio'] = (\n", " df['ict_kosten_inhuur_uitbesteding'].fillna(0) / df['v_te_servicen_medewerkers'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_other_ict_cost_ratio'] = (\n", " df['v_overige_ict_kosten'].fillna(0) / df['v_te_servicen_medewerkers'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_ict_staff_cost_ratio'] = (\n", " df['ict_kosten_personeelskosten'].fillna(0) / df['v_te_servicen_medewerkers'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_telecom_cost_ratio'] = (\n", " df['ict_kosten_telefonie_datacommunicatie'].fillna(0) / df['v_te_servicen_medewerkers'].fillna(0)\n", " ).round(4)\n", "\n", " # Department ratios\n", " df['d_comm_ratio'] = (\n", " df['formatieve_omvang_communicatiefunctie'].fillna(0) / df['formatieve_omvang'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_div_ratio'] = (\n", " df['formatieve_omvang_div'].fillna(0) / df['formatieve_omvang'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_facilities_ratio'] = (\n", " df['formatieve_omvang_facilitaire_zaken'].fillna(0) / df['formatieve_omvang'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_finance_ratio'] = (\n", " df['formatieve_omvang_financien_toezicht_controle'].fillna(0) / df['formatieve_omvang'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_procure_ratio'] = (\n", " df['formatieve_omvang_inkoopfunctie'].fillna(0) / df['formatieve_omvang'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_legal_ratio'] = (\n", " df['formatieve_omvang_juridische_zaken'].fillna(0) / df['formatieve_omvang'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_hrm_ratio'] = (\n", " df['formatieve_omvang_p_o_hrm'].fillna(0) / df['formatieve_omvang'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_ict_ratio'] = (\n", " df['formatieve_omvang_ict'].fillna(0) / df['formatieve_omvang'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_gov_support_ratio'] = (\n", " df['formatieve_omvang_bestuurszaken'].fillna(0) / df['formatieve_omvang'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_line_manager_ratio'] = (\n", " (df['formatieve_omvang_leidinggevenden_hele_organisatie'].fillna(0) - df['formatieve_omvang_leidinggevenden_bedrijfsvoering'].fillna(0)) / df['formatieve_omvang'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_mgmt_support_ratio'] = (\n", " (df['formatieve_omvang_managementondersteuning_hele_organisatie'].fillna(0) - df['formatieve_omvang_managementondersteuning_bedrijfsvoering'].fillna(0)) / df['formatieve_omvang'].fillna(0)\n", " ).round(4)\n", "\n", " # Salary scale ratios\n", " df['d_salary_1_6_ratio'] = (\n", " df['aantal_fte_salarisschaal_1_6'].fillna(0) / df['v_optelsom_fte_salarisschalen'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_salary_10_12_ratio'] = (\n", " df['aantal_fte_salarisschaal_10_12'].fillna(0) / df['v_optelsom_fte_salarisschalen'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_salary_13_up_ratio'] = (\n", " df['aantal_fte_salarisschaal_13_hoger'].fillna(0) / df['v_optelsom_fte_salarisschalen'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_salary_7_9_ratio'] = (\n", " df['aantal_fte_salarisschaal_7_9'].fillna(0) / df['v_optelsom_fte_salarisschalen'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_unknown_salary_ratio'] = (\n", " df['aantal_fte_salarisschaal_onbekend'].fillna(0) / df['v_optelsom_fte_salarisschalen'].fillna(0)\n", " ).round(4)\n", "\n", " # Age category ratios\n", " df['d_under_35_ratio'] = (\n", " df['medewerkers_tot_35_jaar'].fillna(0) / df['v_som_medewerkers_leeftijdscategorie'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_over_55_ratio'] = (\n", " df['medewerkers_55_ouder'].fillna(0) / df['v_som_medewerkers_leeftijdscategorie'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_age_35_45_ratio'] = (\n", " df['medewerkers_35_45_jaar'].fillna(0) / df['v_som_medewerkers_leeftijdscategorie'].fillna(0)\n", " ).round(4)\n", "\n", " df['d_age_45_55_ratio'] = (\n", " df['medewerkers_45_55_jaar'].fillna(0) / df['v_som_medewerkers_leeftijdscategorie'].fillna(0)\n", " ).round(4)\n", "\n", " return df" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#######################################################\n", "# Voor deze berekeningen is er data van het jaar ervoor nodig daarom is het een aparte functie (Alleen mutatie van de formatie wordt zo berekend)\n", "#######################################################\n", "\n", "def add_calculations_using_last_year(df, year):\n", " last_year = year - 1\n", " df_lastyear = extract_data_for_calc(\"VVB_vragenlijst\", last_year)\n", "\n", " # Concateneer alleen de gemeenschappelijke kolommen\n", " df_union = pd.concat([df, df_lastyear])\n", " \n", " # Sort the DataFrame by VVB_id and year to ensure proper alignment\n", " df_union = df_union.sort_values(by=['VVB_id', 'year'])\n", "\n", " # Calculate last year's formatieve_omvang\n", " df_union['lastyear_formatieve_omvang'] = df_union.groupby('VVB_id')['formatieve_omvang'].shift(1)\n", "\n", " # Calculate mutatie_formatie\n", " df_union['d_mutatie_formatie'] = (\n", " (df_union['formatieve_omvang'] - df_union['lastyear_formatieve_omvang']) / df_union['lastyear_formatieve_omvang']\n", " )\n", "\n", " # Replace infinite values with NaN (this can occur if lastyear_formatieve_omvang is 0)\n", " df_union['d_mutatie_formatie'].replace([float('inf'), -float('inf')], float('nan'), inplace=True)\n", "\n", " # ✅ Drop alle rijen waar `year == last_year`\n", " df_union = df_union[df_union['year'] != last_year]\n", "\n", " return df_union" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#######################################################\n", "# Functie voor het opschonen van het te importeren dataframe, niet berekende kolomen worden verwijder, df wordt gesorteerd en infity wordt eruit gehaald\n", "#######################################################\n", "def clean_columns(df, original_df):\n", " \n", " # Sla de originele kolommen op voordat we berekeningen uitvoeren\n", " original_columns = set(original_df.columns)\n", "\n", " # Zorg dat 'year' en 'VVB_id' niet worden verwijderd\n", " keep_columns = {\"year\", \"VVB_id\"}\n", "\n", " # Zorg dat kollomen die gebruikt zijn voor berekeningen worden verwijderd\n", " calc_columns = {\"lastyear_formatieve_omvang\"}\n", "\n", " # Bepaal de kolommen die verwijderd moeten worden (alle originele, behalve 'year' en 'VVB_id')\n", " drop_columns = (original_columns - keep_columns).union(calc_columns)\n", " df = df.drop(columns=drop_columns, errors=\"ignore\")\n", "\n", " df.replace([np.inf, -np.inf], np.nan, inplace=True) # Zet inf en -inf om naar NaN\n", " df = df.where(pd.notna(df), None) # Zet NaN om naar None (NULL in MySQL)\n", "\n", " # Sorteer de kolommen op alfabetische volgorde\n", " df = df[sorted(df.columns)]\n", " return df\n", " " ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#######################################################\n", "# Creer een lijst met datatype van de berekende kolomen behalve vvb_id en year. Standaard op decimal(14,4)\n", "#######################################################\n", "def get_dtype_mapping(df):\n", " # Haal alle kolomnamen op\n", " column_names = list(df.columns)\n", "\n", " # Standaard alle kolommen als DECIMAL(14,4)\n", " dtype_mapping = {col: types.DECIMAL(14, 4) for col in column_names}\n", "\n", " # Specifieke kolommen als INTEGER\n", " if 'VVB_id' in dtype_mapping:\n", " dtype_mapping['VVB_id'] = types.INTEGER\n", " if 'year' in dtype_mapping:\n", " dtype_mapping['year'] = types.INTEGER\n", "\n", " return dtype_mapping\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#############################################################################################\n", "# Functie die alle andere functies in de juiste volgorde aanroept. Input is het jaar dat je wil verversen en de table waar je naar toe wilt schirjven\n", "# Hij haalt data op uit data_upd_{year} dus de tabel van dat jaar moet wel bestaan (bijv. 2023 moet de tabel data_upd_2023 wel bestaan in de venster_voor_bedrijfsvoering database)\n", "#############################################################################################\n", "def update_year(year, destination_table, destination_table_calc):\n", " print(f\"\\n--- Start update process for year {year} ---\")\n", "\n", " source_table = f'data_upd_{year}'\n", " engine = create_connection()\n", "\n", " try:\n", " print(f\"Extracting data from {source_table}...\")\n", " table = extract_data(engine, source_table, year)\n", " print(f\"✅ Data extracted successfully from {source_table}. Rows: {len(table)}\")\n", " except Exception as e:\n", " print(f\"❌ ERROR: Failed to extract data from {source_table}. Reason: {e}\")\n", " return\n", " \n", " try:\n", " print(\"Transforming data...\")\n", " table_transformed = transform_data(table, year)\n", " print(f\"✅ Data transformation completed successfully. Rows: {len(table_transformed)}\")\n", " except Exception as e:\n", " print(f\"❌ ERROR: Failed to transform data. Reason: {e}\")\n", " return\n", " \n", " try:\n", " print(f\"Deleting existing rows from {destination_table} for year {year}...\")\n", " delete_existing_rows(engine, destination_table, year)\n", " print(f\"✅ Old data deleted successfully from {destination_table}.\")\n", " except SQLAlchemyError as e:\n", " print(f\"❌ SQLAlchemy ERROR: Failed to delete data from {destination_table}. Reason: {e}\")\n", " return\n", " except Exception as e:\n", " print(f\"❌ ERROR: Unexpected issue during deletion from {destination_table}. Reason: {e}\")\n", " return \n", "\n", " try:\n", " print(f\"Loading transformed data into {destination_table}...\")\n", " load_data(table_transformed, destination_table, export_dt_dict)\n", " print(f\"✅ Data successfully loaded into {destination_table}.\")\n", " except SQLAlchemyError as e:\n", " print(f\"❌ SQLAlchemy ERROR: Failed to load data into {destination_table}. Reason: {e}\")\n", " return\n", " except Exception as e:\n", " print(f\"❌ ERROR: Unexpected issue during data load. Reason: {e}\")\n", " return\n", " \n", " try:\n", " print(\"Extracting newly updated data for calculations...\")\n", " df = extract_data_for_calc(destination_table, year)\n", " print(f\"✅ Data extracted successfully for calculations. Rows: {len(df)}\")\n", " except Exception as e:\n", " print(f\"❌ ERROR: Failed to extract data for calculations. Reason: {e}\")\n", " return\n", "\n", " try:\n", " print(\"Adding calculated columns...\")\n", " df = add_vragenlijst_calculated_columns(df)\n", " df = add_dashboard_calculated_columns(df)\n", " df = add_calculations_using_last_year(df, year)\n", " print(f\"✅ Calculated columns added successfully.\")\n", " except Exception as e:\n", " print(f\"❌ ERROR: Failed to add calculated columns. Reason: {e}\")\n", " return\n", "\n", " try:\n", " print(\"Cleaning columns...\")\n", " original_df = extract_data_for_calc(destination_table, year)\n", " df = clean_columns(df, original_df)\n", " print(f\"✅ Columns cleaned successfully.\")\n", " except Exception as e:\n", " print(f\"❌ ERROR: Failed to clean columns. Reason: {e}\")\n", " return\n", "\n", " try:\n", " print(\"Generating dtype mapping...\")\n", " dtype_mapping = get_dtype_mapping(df)\n", " print(f\"✅ Dtype mapping generated successfully.\")\n", " except Exception as e:\n", " print(f\"❌ ERROR: Failed to generate dtype mapping. Reason: {e}\")\n", " return\n", "\n", " try:\n", " print(f\"Deleting existing rows from {destination_table_calc} for year {year}...\")\n", " delete_existing_rows(engine, destination_table_calc, year)\n", " print(f\"✅ Old calculated data deleted successfully from {destination_table_calc}.\")\n", " except SQLAlchemyError as e:\n", " print(f\"❌ SQLAlchemy ERROR: Failed to delete data from {destination_table_calc}. Reason: {e}\")\n", " return\n", " except Exception as e:\n", " print(f\"❌ ERROR: Unexpected issue during deletion from {destination_table_calc}. Reason: {e}\")\n", " return \n", "\n", " try:\n", " print(f\"Loading calculated data into {destination_table_calc}...\")\n", " load_data(df, destination_table_calc, dtype_mapping)\n", " print(f\"✅ Data successfully loaded into {destination_table_calc}.\")\n", " except SQLAlchemyError as e:\n", " print(f\"❌ SQLAlchemy ERROR: Failed to load data into {destination_table_calc}. Reason: {e}\")\n", " return\n", " except Exception as e:\n", " print(f\"❌ ERROR: Unexpected issue during data load into {destination_table_calc}. Reason: {e}\")\n", " return\n", "\n", " print(f\"--- ✅ Update process for year {year} completed successfully ---\\n\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "if __name__ == '__main__':\n", " #\n", " # Als er in het platform data van 2023 wordt aangepast en data_upd_2023 wordt bijgewerkt kan je deze aanzetten om dat jaar ook bij te werken\n", " #\n", " # update_year(2023, 'VVB_vragenlijst', 'VVB_calculations')\n", "\n", " #Het huidige jaar dat we data ophalen (halen in 2025 data op voor 2024)\n", " update_year(2024, 'VVB_vragenlijst', 'VVB_calculations')" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#\n", "# Met jupyter nbconvert --to script vvb_data_update.ipynb \n", "# Kan het .ipynb omgezet worden naar .py \n", "# Dit kan van pas komen al je aanpassingen aan het script wilt doen, dan kan je de ipynb als ontwikkel script blijven gebruiken\n", "#" ] } ], "metadata": { "kernelspec": { "display_name": "Venster (venv)", "language": "python", "name": "myenv" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.10.12" } }, "nbformat": 4, "nbformat_minor": 2 }