Philipp/aki_prj23_transparenzregister
1
0
Fork 0
mirror of https://github.com/fhswf/aki_prj23_transparenzregister.git synced 2025-06-22 00:23:55 +02:00
Code Issues Packages Projects Releases Wiki Activity

Feature/visualize verflechtungen (#324)

Browse Source
This commit is contained in:
Tim Ronneburg
2023-11-10 19:33:19 +01:00
committed by GitHub
parent d3158c4592 da72c3d0a8
commit 247719c76f
31 changed files with 158978 additions and 473 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2126
Jupyter/NetworkX/archive/Dev.ipynb Normal file
View File

File diff suppressed because one or more lines are too long

919
Jupyter/NetworkX/archive/networkX_with_sql.ipynb Normal file
View File

@ -0,0 +1,919 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": 18,
"metadata": {},
"outputs": [],
"source": [
"import os.path\n",
"\n",
"import pandas as pd\n",
"\n",
"# if not os.path.exists(\"src\"):\n",
"# %cd \"../\"\n",
"# os.path.abspath(\".\")"
]
},
{
"cell_type": "code",
"execution_count": 19,
"metadata": {},
"outputs": [],
"source": [
"from aki_prj23_transparenzregister.utils.sql import entities\n",
"from sqlalchemy.orm import aliased\n",
"from sqlalchemy import func, text\n",
"\n",
"# Alias for Company table for the base company\n",
"base_company = aliased(entities.Company, name=\"base_company\")\n",
"\n",
"# Alias for Company table for the head company\n",
"head_company = aliased(entities.Company, name=\"head_company\")"
]
},
{
"cell_type": "code",
"execution_count": 20,
"metadata": {},
"outputs": [],
"source": [
"from aki_prj23_transparenzregister.config.config_providers import JsonFileConfigProvider\n",
"from aki_prj23_transparenzregister.utils.sql.connector import get_session\n",
"\n",
"session = get_session(JsonFileConfigProvider(\"../secrets.json\"))"
]
},
{
"cell_type": "code",
"execution_count": 21,
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"'SELECT base_company.name AS name_company_base, relation.relation AS relation_type, head_company.name AS name_company_head \\nFROM company AS base_company JOIN (relation JOIN company_relation ON relation.id = company_relation.id) ON relation.company_id = base_company.id JOIN company AS head_company ON company_relation.company2_id = head_company.id'"
]
},
"execution_count": 21,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"# Query to fetch relations between companies\n",
"relations_query = (\n",
" session.query(\n",
" base_company.name.label(\"name_company_base\"),\n",
" entities.CompanyRelation.relation.label(\"relation_type\"),\n",
" head_company.name.label(\"name_company_head\"),\n",
" )\n",
" .join(\n",
" entities.CompanyRelation,\n",
" entities.CompanyRelation.company_id == base_company.id,\n",
" )\n",
" .join(\n",
" head_company,\n",
" entities.CompanyRelation.company2_id == head_company.id,\n",
" )\n",
")\n",
"str(relations_query)"
]
},
{
"cell_type": "code",
"execution_count": 22,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"121 ms ± 9.27 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)\n"
]
}
],
"source": [
"%timeit pd.read_sql_query(str(relations_query), session.bind)"
]
},
{
"cell_type": "code",
"execution_count": 23,
"metadata": {},
"outputs": [
{
"data": {
"text/html": [
"<div>\n",
"<style scoped>\n",
" .dataframe tbody tr th:only-of-type {\n",
" vertical-align: middle;\n",
" }\n",
"\n",
" .dataframe tbody tr th {\n",
" vertical-align: top;\n",
" }\n",
"\n",
" .dataframe thead th {\n",
" text-align: right;\n",
" }\n",
"</style>\n",
"<table border=\"1\" class=\"dataframe\">\n",
" <thead>\n",
" <tr style=\"text-align: right;\">\n",
" <th></th>\n",
" <th>name_company_base</th>\n",
" <th>relation_type</th>\n",
" <th>name_company_head</th>\n",
" </tr>\n",
" </thead>\n",
" <tbody>\n",
" <tr>\n",
" <th>0</th>\n",
" <td>2. Schaper Objekt GmbH &amp; Co. Kiel KG</td>\n",
" <td>KOMMANDITIST</td>\n",
" <td>Multi-Center Warenvertriebs GmbH</td>\n",
" </tr>\n",
" <tr>\n",
" <th>1</th>\n",
" <td>Alb-Windkraft GmbH &amp; Co. KG</td>\n",
" <td>KOMMANDITIST</td>\n",
" <td>EnBW Windkraftprojekte GmbH</td>\n",
" </tr>\n",
" <tr>\n",
" <th>2</th>\n",
" <td>Anneliese Köster GmbH &amp; Co. KG</td>\n",
" <td>KOMMANDITIST</td>\n",
" <td>INDUS Holding Aktiengesellschaft</td>\n",
" </tr>\n",
" <tr>\n",
" <th>3</th>\n",
" <td>AURELIUS Equity Opportunities SE &amp; Co. KGaA</td>\n",
" <td>HAFTENDER_GESELLSCHAFTER</td>\n",
" <td>AURELIUS Management SE</td>\n",
" </tr>\n",
" <tr>\n",
" <th>4</th>\n",
" <td>Aurelius KG</td>\n",
" <td>HAFTENDER_GESELLSCHAFTER</td>\n",
" <td>Aurelius Verwaltungs GmbH</td>\n",
" </tr>\n",
" <tr>\n",
" <th>...</th>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" </tr>\n",
" <tr>\n",
" <th>573</th>\n",
" <td>Zalando BTD 011 SE &amp; Co. KG</td>\n",
" <td>HAFTENDER_GESELLSCHAFTER</td>\n",
" <td>Zalando SE</td>\n",
" </tr>\n",
" <tr>\n",
" <th>574</th>\n",
" <td>Zalando BTD 011 SE &amp; Co. KG</td>\n",
" <td>KOMMANDITIST</td>\n",
" <td>Zalando Operations GmbH</td>\n",
" </tr>\n",
" <tr>\n",
" <th>575</th>\n",
" <td>zLabels Creation &amp; Sales GmbH &amp; Co. KG</td>\n",
" <td>HAFTENDER_GESELLSCHAFTER</td>\n",
" <td>zLabels GmbH</td>\n",
" </tr>\n",
" <tr>\n",
" <th>576</th>\n",
" <td>Zalando Customer Care International SE &amp; Co. KG</td>\n",
" <td>HAFTENDER_GESELLSCHAFTER</td>\n",
" <td>Zalando SE</td>\n",
" </tr>\n",
" <tr>\n",
" <th>577</th>\n",
" <td>Zalando Customer Care International SE &amp; Co. KG</td>\n",
" <td>KOMMANDITIST</td>\n",
" <td>Zalando Operations GmbH</td>\n",
" </tr>\n",
" </tbody>\n",
"</table>\n",
"<p>578 rows × 3 columns</p>\n",
"</div>"
],
"text/plain": [
" name_company_base \\\n",
"0 2. Schaper Objekt GmbH & Co. Kiel KG \n",
"1 Alb-Windkraft GmbH & Co. KG \n",
"2 Anneliese Köster GmbH & Co. KG \n",
"3 AURELIUS Equity Opportunities SE & Co. KGaA \n",
"4 Aurelius KG \n",
".. ... \n",
"573 Zalando BTD 011 SE & Co. KG \n",
"574 Zalando BTD 011 SE & Co. KG \n",
"575 zLabels Creation & Sales GmbH & Co. KG \n",
"576 Zalando Customer Care International SE & Co. KG \n",
"577 Zalando Customer Care International SE & Co. KG \n",
"\n",
" relation_type name_company_head \n",
"0 KOMMANDITIST Multi-Center Warenvertriebs GmbH \n",
"1 KOMMANDITIST EnBW Windkraftprojekte GmbH \n",
"2 KOMMANDITIST INDUS Holding Aktiengesellschaft \n",
"3 HAFTENDER_GESELLSCHAFTER AURELIUS Management SE \n",
"4 HAFTENDER_GESELLSCHAFTER Aurelius Verwaltungs GmbH \n",
".. ... ... \n",
"573 HAFTENDER_GESELLSCHAFTER Zalando SE \n",
"574 KOMMANDITIST Zalando Operations GmbH \n",
"575 HAFTENDER_GESELLSCHAFTER zLabels GmbH \n",
"576 HAFTENDER_GESELLSCHAFTER Zalando SE \n",
"577 KOMMANDITIST Zalando Operations GmbH \n",
"\n",
"[578 rows x 3 columns]"
]
},
"execution_count": 23,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"company_relations = pd.read_sql_query(str(relations_query), session.bind)\n",
"company_relations"
]
},
{
"cell_type": "code",
"execution_count": 24,
"metadata": {},
"outputs": [],
"source": [
"relations_query = (\n",
" session.query(\n",
" entities.Company.name.label(\"name_company\"),\n",
" entities.PersonRelation.relation.label(\"relation_type\"),\n",
" entities.Person.lastname.label(\"lastname\"),\n",
" entities.Person.firstname.label(\"firstname\"),\n",
" entities.Person.date_of_birth.label(\"date_of_birth\"),\n",
" )\n",
" .join(\n",
" entities.PersonRelation,\n",
" entities.PersonRelation.company_id == entities.Company.id,\n",
" )\n",
" .join(\n",
" entities.Person,\n",
" entities.PersonRelation.person_id == entities.Person.id,\n",
" )\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 25,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"373 ms ± 25.6 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)\n"
]
}
],
"source": [
"%timeit pd.read_sql_query(str(relations_query), session.bind)"
]
},
{
"cell_type": "code",
"execution_count": 26,
"metadata": {},
"outputs": [
{
"data": {
"text/html": [
"<div>\n",
"<style scoped>\n",
" .dataframe tbody tr th:only-of-type {\n",
" vertical-align: middle;\n",
" }\n",
"\n",
" .dataframe tbody tr th {\n",
" vertical-align: top;\n",
" }\n",
"\n",
" .dataframe thead th {\n",
" text-align: right;\n",
" }\n",
"</style>\n",
"<table border=\"1\" class=\"dataframe\">\n",
" <thead>\n",
" <tr style=\"text-align: right;\">\n",
" <th></th>\n",
" <th>name_company</th>\n",
" <th>relation_type</th>\n",
" <th>lastname</th>\n",
" <th>firstname</th>\n",
" <th>date_of_birth</th>\n",
" </tr>\n",
" </thead>\n",
" <tbody>\n",
" <tr>\n",
" <th>0</th>\n",
" <td>0 10 24 Telefondienste GmbH</td>\n",
" <td>GESCHAEFTSFUEHRER</td>\n",
" <td>Tetau</td>\n",
" <td>Nicolas</td>\n",
" <td>1971-01-02</td>\n",
" </tr>\n",
" <tr>\n",
" <th>1</th>\n",
" <td>0 10 24 Telefondienste GmbH</td>\n",
" <td>PROKURIST</td>\n",
" <td>Dammast</td>\n",
" <td>Lutz</td>\n",
" <td>1966-12-06</td>\n",
" </tr>\n",
" <tr>\n",
" <th>2</th>\n",
" <td>1. Staiger Grundstücksverwaltung GmbH &amp; Co. KG</td>\n",
" <td>KOMMANDITIST</td>\n",
" <td>Tutsch</td>\n",
" <td>Rosemarie</td>\n",
" <td>1941-10-09</td>\n",
" </tr>\n",
" <tr>\n",
" <th>3</th>\n",
" <td>1. Staiger Grundstücksverwaltung GmbH &amp; Co. KG</td>\n",
" <td>KOMMANDITIST</td>\n",
" <td>Staiger</td>\n",
" <td>Marc</td>\n",
" <td>1969-10-22</td>\n",
" </tr>\n",
" <tr>\n",
" <th>4</th>\n",
" <td>1. Staiger Grundstücksverwaltung GmbH &amp; Co. KG</td>\n",
" <td>KOMMANDITIST</td>\n",
" <td>Staiger</td>\n",
" <td>Michaela</td>\n",
" <td>1971-03-03</td>\n",
" </tr>\n",
" <tr>\n",
" <th>...</th>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" </tr>\n",
" <tr>\n",
" <th>14891</th>\n",
" <td>Wohnungsbaugesellschaft mit beschränkter Haftu...</td>\n",
" <td>GESCHAEFTSFUEHRER</td>\n",
" <td>Weirich</td>\n",
" <td>Torsten</td>\n",
" <td>1975-07-21</td>\n",
" </tr>\n",
" <tr>\n",
" <th>14892</th>\n",
" <td>Wohnungsbaugesellschaft mit beschränkter Haftu...</td>\n",
" <td>GESCHAEFTSFUEHRER</td>\n",
" <td>Brusinski</td>\n",
" <td>Bastian</td>\n",
" <td>1980-10-29</td>\n",
" </tr>\n",
" <tr>\n",
" <th>14893</th>\n",
" <td>Zalando Customer Care International SE &amp; Co. KG</td>\n",
" <td>PROKURIST</td>\n",
" <td>Pape</td>\n",
" <td>Ute</td>\n",
" <td>1978-12-13</td>\n",
" </tr>\n",
" <tr>\n",
" <th>14894</th>\n",
" <td>zebotec GmbH</td>\n",
" <td>GESCHAEFTSFUEHRER</td>\n",
" <td>Neff</td>\n",
" <td>Werner</td>\n",
" <td>1981-11-24</td>\n",
" </tr>\n",
" <tr>\n",
" <th>14895</th>\n",
" <td>zebotec GmbH</td>\n",
" <td>GESCHAEFTSFUEHRER</td>\n",
" <td>Morris</td>\n",
" <td>Richard</td>\n",
" <td>1971-01-02</td>\n",
" </tr>\n",
" </tbody>\n",
"</table>\n",
"<p>14896 rows × 5 columns</p>\n",
"</div>"
],
"text/plain": [
" name_company relation_type \\\n",
"0 0 10 24 Telefondienste GmbH GESCHAEFTSFUEHRER \n",
"1 0 10 24 Telefondienste GmbH PROKURIST \n",
"2 1. Staiger Grundstücksverwaltung GmbH & Co. KG KOMMANDITIST \n",
"3 1. Staiger Grundstücksverwaltung GmbH & Co. KG KOMMANDITIST \n",
"4 1. Staiger Grundstücksverwaltung GmbH & Co. KG KOMMANDITIST \n",
"... ... ... \n",
"14891 Wohnungsbaugesellschaft mit beschränkter Haftu... GESCHAEFTSFUEHRER \n",
"14892 Wohnungsbaugesellschaft mit beschränkter Haftu... GESCHAEFTSFUEHRER \n",
"14893 Zalando Customer Care International SE & Co. KG PROKURIST \n",
"14894 zebotec GmbH GESCHAEFTSFUEHRER \n",
"14895 zebotec GmbH GESCHAEFTSFUEHRER \n",
"\n",
" lastname firstname date_of_birth \n",
"0 Tetau Nicolas 1971-01-02 \n",
"1 Dammast Lutz 1966-12-06 \n",
"2 Tutsch Rosemarie 1941-10-09 \n",
"3 Staiger Marc 1969-10-22 \n",
"4 Staiger Michaela 1971-03-03 \n",
"... ... ... ... \n",
"14891 Weirich Torsten 1975-07-21 \n",
"14892 Brusinski Bastian 1980-10-29 \n",
"14893 Pape Ute 1978-12-13 \n",
"14894 Neff Werner 1981-11-24 \n",
"14895 Morris Richard 1971-01-02 \n",
"\n",
"[14896 rows x 5 columns]"
]
},
"execution_count": 26,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"df = pd.read_sql_query(str(relations_query), session.bind)\n",
"df"
]
},
{
"cell_type": "code",
"execution_count": 27,
"metadata": {},
"outputs": [
{
"data": {
"text/html": [
"<div>\n",
"<style scoped>\n",
" .dataframe tbody tr th:only-of-type {\n",
" vertical-align: middle;\n",
" }\n",
"\n",
" .dataframe tbody tr th {\n",
" vertical-align: top;\n",
" }\n",
"\n",
" .dataframe thead th {\n",
" text-align: right;\n",
" }\n",
"</style>\n",
"<table border=\"1\" class=\"dataframe\">\n",
" <thead>\n",
" <tr style=\"text-align: right;\">\n",
" <th></th>\n",
" <th>person_id</th>\n",
" </tr>\n",
" </thead>\n",
" <tbody>\n",
" <tr>\n",
" <th>0</th>\n",
" <td>2520</td>\n",
" </tr>\n",
" <tr>\n",
" <th>1</th>\n",
" <td>4993</td>\n",
" </tr>\n",
" <tr>\n",
" <th>2</th>\n",
" <td>3202</td>\n",
" </tr>\n",
" <tr>\n",
" <th>3</th>\n",
" <td>4611</td>\n",
" </tr>\n",
" <tr>\n",
" <th>4</th>\n",
" <td>4095</td>\n",
" </tr>\n",
" <tr>\n",
" <th>...</th>\n",
" <td>...</td>\n",
" </tr>\n",
" <tr>\n",
" <th>1804</th>\n",
" <td>3565</td>\n",
" </tr>\n",
" <tr>\n",
" <th>1805</th>\n",
" <td>3510</td>\n",
" </tr>\n",
" <tr>\n",
" <th>1806</th>\n",
" <td>530</td>\n",
" </tr>\n",
" <tr>\n",
" <th>1807</th>\n",
" <td>536</td>\n",
" </tr>\n",
" <tr>\n",
" <th>1808</th>\n",
" <td>4617</td>\n",
" </tr>\n",
" </tbody>\n",
"</table>\n",
"<p>1809 rows × 1 columns</p>\n",
"</div>"
],
"text/plain": [
" person_id\n",
"0 2520\n",
"1 4993\n",
"2 3202\n",
"3 4611\n",
"4 4095\n",
"... ...\n",
"1804 3565\n",
"1805 3510\n",
"1806 530\n",
"1807 536\n",
"1808 4617\n",
"\n",
"[1809 rows x 1 columns]"
]
},
"execution_count": 27,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"from sqlalchemy import func, text\n",
"\n",
"# Subquery to group and count the relations without joins\n",
"grouped_relations_subquery = (\n",
" session.query(\n",
" entities.PersonRelation.person_id,\n",
" )\n",
" .group_by(entities.PersonRelation.person_id)\n",
" .having(func.count() > 1)\n",
")\n",
"pd.DataFrame(grouped_relations_subquery.all())"
]
},
{
"cell_type": "code",
"execution_count": 28,
"metadata": {},
"outputs": [],
"source": [
"relations_query = (\n",
" session.query(\n",
" entities.Company.name.label(\"name_company\"),\n",
" entities.PersonRelation.relation.label(\"relation_type\"),\n",
" entities.Person.lastname.label(\"lastname\"),\n",
" entities.Person.firstname.label(\"firstname\"),\n",
" entities.Person.date_of_birth.label(\"date_of_birth\"),\n",
" )\n",
" .join(\n",
" entities.PersonRelation,\n",
" entities.PersonRelation.company_id == entities.Company.id,\n",
" )\n",
" .join(\n",
" entities.Person,\n",
" entities.PersonRelation.person_id == entities.Person.id,\n",
" )\n",
" .filter(entities.PersonRelation.person_id.in_(grouped_relations_subquery))\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 30,
"metadata": {},
"outputs": [
{
"data": {
"text/html": [
"<div>\n",
"<style scoped>\n",
" .dataframe tbody tr th:only-of-type {\n",
" vertical-align: middle;\n",
" }\n",
"\n",
" .dataframe tbody tr th {\n",
" vertical-align: top;\n",
" }\n",
"\n",
" .dataframe thead th {\n",
" text-align: right;\n",
" }\n",
"</style>\n",
"<table border=\"1\" class=\"dataframe\">\n",
" <thead>\n",
" <tr style=\"text-align: right;\">\n",
" <th></th>\n",
" <th>name_company</th>\n",
" <th>relation_type</th>\n",
" <th>lastname</th>\n",
" <th>firstname</th>\n",
" <th>date_of_birth</th>\n",
" <th>person_name</th>\n",
" </tr>\n",
" </thead>\n",
" <tbody>\n",
" <tr>\n",
" <th>0</th>\n",
" <td>0 10 24 Telefondienste GmbH</td>\n",
" <td>RelationshipRoleEnum.GESCHAEFTSFUEHRER</td>\n",
" <td>Tetau</td>\n",
" <td>Nicolas</td>\n",
" <td>1971-01-02</td>\n",
" <td>TetauNicolas</td>\n",
" </tr>\n",
" <tr>\n",
" <th>1</th>\n",
" <td>0 10 24 Telefondienste GmbH</td>\n",
" <td>RelationshipRoleEnum.PROKURIST</td>\n",
" <td>Dammast</td>\n",
" <td>Lutz</td>\n",
" <td>1966-12-06</td>\n",
" <td>DammastLutz</td>\n",
" </tr>\n",
" <tr>\n",
" <th>2</th>\n",
" <td>01050.com GmbH</td>\n",
" <td>RelationshipRoleEnum.GESCHAEFTSFUEHRER</td>\n",
" <td>Tetau</td>\n",
" <td>Nicolas</td>\n",
" <td>1971-01-02</td>\n",
" <td>TetauNicolas</td>\n",
" </tr>\n",
" <tr>\n",
" <th>3</th>\n",
" <td>01050.com GmbH</td>\n",
" <td>RelationshipRoleEnum.PROKURIST</td>\n",
" <td>Dammast</td>\n",
" <td>Lutz</td>\n",
" <td>1966-12-06</td>\n",
" <td>DammastLutz</td>\n",
" </tr>\n",
" <tr>\n",
" <th>4</th>\n",
" <td>AASP Filmproduktionsgesellschaft mbH &amp; Co. Leo...</td>\n",
" <td>RelationshipRoleEnum.KOMMANDITIST</td>\n",
" <td>Dellhofen</td>\n",
" <td>Jens</td>\n",
" <td>1977-04-19</td>\n",
" <td>DellhofenJens</td>\n",
" </tr>\n",
" <tr>\n",
" <th>...</th>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" <td>...</td>\n",
" </tr>\n",
" <tr>\n",
" <th>7071</th>\n",
" <td>Wohnungsbaugesellschaft mit beschränkter Haftu...</td>\n",
" <td>RelationshipRoleEnum.GESCHAEFTSFUEHRER</td>\n",
" <td>Karounos</td>\n",
" <td>Marita</td>\n",
" <td>1971-03-30</td>\n",
" <td>KarounosMarita</td>\n",
" </tr>\n",
" <tr>\n",
" <th>7072</th>\n",
" <td>Wohnungsbaugesellschaft mit beschränkter Haftu...</td>\n",
" <td>RelationshipRoleEnum.PROKURIST</td>\n",
" <td>Groll</td>\n",
" <td>Michael</td>\n",
" <td>1967-12-24</td>\n",
" <td>GrollMichael</td>\n",
" </tr>\n",
" <tr>\n",
" <th>7073</th>\n",
" <td>Wohnungsbaugesellschaft mit beschränkter Haftu...</td>\n",
" <td>RelationshipRoleEnum.GESCHAEFTSFUEHRER</td>\n",
" <td>Weirich</td>\n",
" <td>Torsten</td>\n",
" <td>1975-07-21</td>\n",
" <td>WeirichTorsten</td>\n",
" </tr>\n",
" <tr>\n",
" <th>7074</th>\n",
" <td>Wohnungsbaugesellschaft mit beschränkter Haftu...</td>\n",
" <td>RelationshipRoleEnum.GESCHAEFTSFUEHRER</td>\n",
" <td>Brusinski</td>\n",
" <td>Bastian</td>\n",
" <td>1980-10-29</td>\n",
" <td>BrusinskiBastian</td>\n",
" </tr>\n",
" <tr>\n",
" <th>7075</th>\n",
" <td>Zalando Customer Care International SE &amp; Co. KG</td>\n",
" <td>RelationshipRoleEnum.PROKURIST</td>\n",
" <td>Pape</td>\n",
" <td>Ute</td>\n",
" <td>1978-12-13</td>\n",
" <td>PapeUte</td>\n",
" </tr>\n",
" </tbody>\n",
"</table>\n",
"<p>7076 rows × 6 columns</p>\n",
"</div>"
],
"text/plain": [
" name_company \\\n",
"0 0 10 24 Telefondienste GmbH \n",
"1 0 10 24 Telefondienste GmbH \n",
"2 01050.com GmbH \n",
"3 01050.com GmbH \n",
"4 AASP Filmproduktionsgesellschaft mbH & Co. Leo... \n",
"... ... \n",
"7071 Wohnungsbaugesellschaft mit beschränkter Haftu... \n",
"7072 Wohnungsbaugesellschaft mit beschränkter Haftu... \n",
"7073 Wohnungsbaugesellschaft mit beschränkter Haftu... \n",
"7074 Wohnungsbaugesellschaft mit beschränkter Haftu... \n",
"7075 Zalando Customer Care International SE & Co. KG \n",
"\n",
" relation_type lastname firstname \\\n",
"0 RelationshipRoleEnum.GESCHAEFTSFUEHRER Tetau Nicolas \n",
"1 RelationshipRoleEnum.PROKURIST Dammast Lutz \n",
"2 RelationshipRoleEnum.GESCHAEFTSFUEHRER Tetau Nicolas \n",
"3 RelationshipRoleEnum.PROKURIST Dammast Lutz \n",
"4 RelationshipRoleEnum.KOMMANDITIST Dellhofen Jens \n",
"... ... ... ... \n",
"7071 RelationshipRoleEnum.GESCHAEFTSFUEHRER Karounos Marita \n",
"7072 RelationshipRoleEnum.PROKURIST Groll Michael \n",
"7073 RelationshipRoleEnum.GESCHAEFTSFUEHRER Weirich Torsten \n",
"7074 RelationshipRoleEnum.GESCHAEFTSFUEHRER Brusinski Bastian \n",
"7075 RelationshipRoleEnum.PROKURIST Pape Ute \n",
"\n",
" date_of_birth person_name \n",
"0 1971-01-02 TetauNicolas \n",
"1 1966-12-06 DammastLutz \n",
"2 1971-01-02 TetauNicolas \n",
"3 1966-12-06 DammastLutz \n",
"4 1977-04-19 DellhofenJens \n",
"... ... ... \n",
"7071 1971-03-30 KarounosMarita \n",
"7072 1967-12-24 GrollMichael \n",
"7073 1975-07-21 WeirichTorsten \n",
"7074 1980-10-29 BrusinskiBastian \n",
"7075 1978-12-13 PapeUte \n",
"\n",
"[7076 rows x 6 columns]"
]
},
"execution_count": 30,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"relations_df = pd.DataFrame(relations_query.all())\n",
"relations_df[\"person_name\"] = relations_df[\"lastname\"] + relations_df[\"firstname\"]\n",
"relations_df.rename(\n",
" columns={\"oldName1\": \"newName1\", \"oldName2\": \"newName2\"}, inplace=True\n",
")\n",
"relations_df"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"node_template = {\n",
" \"id\": \"(company|person)_\\d\",\n",
" \"label\": \"Name from entries\",\n",
" \"type\": \"Company|Person\",\n",
" \"shape\": \"dot\",\n",
" \"color\": \"#729b79ff\",\n",
" # TODO add title for hover effect in graph \"title\": \"\"\n",
"}\n",
"nodes = relations_df\n",
"for index in relations_df.index:\n",
" nodes[\"index\"] = {\n",
" \"label\": company_2.name,\n",
" \"type\": \"Company\",\n",
" \"shape\": \"dot\",\n",
" \"color\": \"#729b79ff\",\n",
" }"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import networkx as nx\n",
"import matplotlib.pyplot as plt\n",
"# relations_df[\"person_name\"] = relations_df[\"lastname\"] + relations_df[\"firstname\"]\n",
"\n",
"nodes = \n",
"# create edges from dataframe\n",
"graph = nx.from_pandas_edgelist(relations_df, source=\"name_company\", target=\"person_name\", edge_attr=\"relation_type\")\n",
"\n",
"# update node attributes from dataframe\n",
"nodes_attr = nodes.set_index(\"index\").to_dict(orient=\"index\")\n",
"nx.set_node_attributes(graph, nodes_attr)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from pyvis.network import Network\n",
"\n",
"net = Network(\n",
" directed=False, neighborhood_highlight=True, bgcolor=\"white\", font_color=\"black\"\n",
")\n",
"\n",
"# pass networkx graph to pyvis\n",
"net.from_nx(graph)\n",
"\n",
"net.inherit_edge_colors(False)\n",
"net.set_edge_smooth(\"dynamic\")\n",
"adj_list = net.get_adj_list()\n",
"\n",
"measure_type = \"degree\"\n",
"measure_vector = {}\n",
"\n",
"if measure_type == \"eigenvector\":\n",
" measure_vector = nx.eigenvector_centrality(graph)\n",
" df[\"eigenvector\"] = measure_vector.values()\n",
"if measure_type == \"degree\":\n",
" measure_vector = nx.degree_centrality(graph)\n",
" df[\"degree\"] = measure_vector.values()\n",
"if measure_type == \"betweeness\":\n",
" measure_vector = nx.betweenness_centrality(graph)\n",
" df[\"betweeness\"] = measure_vector.values()\n",
"if measure_type == \"closeness\":\n",
" measure_vector = nx.closeness_centrality(graph)\n",
" df[\"closeness\"] = measure_vector.values()\n",
"if measure_type == \"pagerank\":\n",
" measure_vector = nx.pagerank(graph)\n",
" df[\"pagerank\"] = measure_vector.values()\n",
"if measure_type == \"average_degree\":\n",
" measure_vector = nx.average_degree_connectivity(graph)\n",
" # df[\"average_degree\"] = measure_vector.values()\n",
" print(measure_vector.values())\n",
"\n",
"# calculate and update size of the nodes depending on their number of edges\n",
"for node_id, neighbors in adj_list.items():\n",
" # df[\"edges\"] = measure_vector.values()\n",
"\n",
" if measure_type == \"edges\":\n",
" size = 10 # len(neighbors)*5\n",
" else:\n",
" size = measure_vector[node_id] * 50\n",
" next(\n",
" (node.update({\"size\": size}) for node in net.nodes if node[\"id\"] == node_id),\n",
" None,\n",
" )\n",
"\n",
"\n",
"net.repulsion()\n",
"net.show_buttons(filter_=[\"physics\"])\n",
"\n",
"# net.show_buttons()\n",
"\n",
"# save graph as HTML\n",
"net.save_graph(\"./tmp.html\")"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "aki-prj23-transparenzregister-IY2hcXvW-py3.11",
"language": "python",
"name": "python3"
},
"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.11.4"
}
},
"nbformat": 4,
"nbformat_minor": 2
}

272
Jupyter/NetworkX/archive/network_graph.html Normal file
View File

File diff suppressed because one or more lines are too long

129790
Jupyter/NetworkX/archive/sql_alchemy_to_networkx.ipynb Normal file
View File

File diff suppressed because it is too large Load Diff

21843
Jupyter/NetworkX/archive/sql_alchemy_to_networkx_v2.ipynb Normal file
View File

File diff suppressed because one or more lines are too long

275
Jupyter/NetworkX/archive/tmp.html Normal file
View File

File diff suppressed because one or more lines are too long

272
Jupyter/NetworkX/archive_prod/network_graph.html Normal file
View File

File diff suppressed because one or more lines are too long

151
Jupyter/NetworkX/archive_prod/networkx_dash.py Normal file
View File

@ -0,0 +1,151 @@
"""Old NetworkX Graph which needs to be discarded in the next commits."""
import networkx as nx
import pandas as pd
import plotly.graph_objects as go
from dash import dcc, html
from aki_prj23_transparenzregister.config.config_providers import JsonFileConfigProvider
from aki_prj23_transparenzregister.utils.sql import connector, entities
test_company = 13 # 2213 # 13
def find_company_relations(company_id: int) -> pd.DataFrame:
"""_summary_.
Args:
company_id (int): _description_
Returns:
pd.DataFrame: _description_
"""
session = connector.get_session(JsonFileConfigProvider("./secrets.json"))
query_companies = session.query(entities.Company)
query_relations = session.query(entities.CompanyRelation)
companies_df: pd.DataFrame = pd.read_sql(str(query_companies), session.bind) # type: ignore
companies_relations_df: pd.DataFrame = pd.read_sql(str(query_relations), session.bind) # type: ignore
companies_relations_df = companies_relations_df.loc[
companies_relations_df["relation_id"] == company_id, :
][["relation_id", "company_relation_company2_id"]]
company_name = []
connected_company_name = []
for _, row in companies_relations_df.iterrows():
company_name.append(
companies_df.loc[companies_df["company_id"] == row["relation_id"]][
"company_name"
].iloc[0]
)
connected_company_name.append(
companies_df.loc[
companies_df["company_id"] == row["company_relation_company2_id"]
]["company_name"].iloc[0]
)
# print(company_name)
companies_relations_df["company_name"] = company_name
companies_relations_df["connected_company_name"] = connected_company_name
# print(companies_relations_df)
return companies_relations_df
# Plotly figure
def network_graph(company_id: int) -> go.Figure:
"""_summary_.
Args:
company_id (int): _description_
Returns:
go.Figure: _description_
"""
edges = []
for _, row in find_company_relations(company_id).iterrows():
edges.append([row["company_name"], row["connected_company_name"]])
network_graph = nx.Graph()
network_graph.add_edges_from(edges)
pos = nx.spring_layout(network_graph)
# edges trace
edge_x = []
edge_y = []
for edge in network_graph.edges():
x0, y0 = pos[edge[0]]
x1, y1 = pos[edge[1]]
edge_x.append(x0)
edge_x.append(x1)
edge_x.append(None)
edge_y.append(y0)
edge_y.append(y1)
edge_y.append(None)
edge_trace = go.Scatter(
x=edge_x,
y=edge_y,
line={"color": "black", "width": 1},
hoverinfo="none",
showlegend=False,
mode="lines",
)
# nodes trace
node_x = []
node_y = []
text = []
for node in network_graph.nodes():
x, y = pos[node]
node_x.append(x)
node_y.append(y)
text.append(node)
node_trace = go.Scatter(
x=node_x,
y=node_y,
text=text,
mode="markers+text",
showlegend=False,
hoverinfo="none",
marker={"color": "pink", "size": 50, "line": {"color": "black", "width": 1}},
)
# layout
layout = {
"plot_bgcolor": "white",
"paper_bgcolor": "white",
"margin": {"t": 10, "b": 10, "l": 10, "r": 10, "pad": 0},
"xaxis": {
"linecolor": "black",
"showgrid": False,
"showticklabels": False,
"mirror": True,
},
"yaxis": {
"linecolor": "black",
"showgrid": False,
"showticklabels": False,
"mirror": True,
},
}
# figure
return go.Figure(data=[edge_trace, node_trace], layout=layout)
def networkx_component(company_id: int) -> html.Div:
"""Retruns the Layout with a Graph.
Args:
company_id (int): _description_
Returns:
any: _description_
"""
return html.Div(
[
dcc.Graph(id="my-graph", figure=network_graph(company_id)),
]
)

186
Jupyter/NetworkX/archive_prod/networkx_dash_overall.py Normal file
View File

@ -0,0 +1,186 @@
"""Old Module for NetworkX Graphs."""
import networkx as nx
import pandas as pd
import plotly.graph_objects as go
from dash import Dash, Input, Output, dcc, html
from aki_prj23_transparenzregister.config.config_providers import JsonFileConfigProvider
from aki_prj23_transparenzregister.utils.sql import connector, entities
test_company = 13 # 2213 # 13
def find_all_company_relations() -> pd.DataFrame:
"""Searches for all companies and their relation in the DB.
Returns:
pd.DataFrame: _description_
"""
session = connector.get_session(JsonFileConfigProvider("./secrets.json"))
query_companies = session.query(entities.Company) # .all()
query_relations = session.query(entities.CompanyRelation) # .all()
companies_df: pd.DataFrame = pd.read_sql(str(query_companies), session.bind) # type: ignore
companies_relations_df: pd.DataFrame = pd.read_sql(str(query_relations), session.bind) # type: ignore
# print(companies_relations_df)
companies_relations_df = companies_relations_df[
["relation_id", "company_relation_company2_id"]
]
# print(companies_relations_df)
company_name = []
connected_company_name = []
companies_relations_df = companies_relations_df.head()
# print(companies_relations_df)
for _, row in companies_relations_df.iterrows():
company_name.append(
companies_df.loc[companies_df["company_id"] == row["relation_id"]][
"company_name"
].iloc[0]
)
connected_company_name.append(
companies_df.loc[
companies_df["company_id"] == row["company_relation_company2_id"]
]["company_name"].iloc[0]
)
# print(connected_company_name)
# print(company_name)
companies_relations_df["company_name"] = company_name
companies_relations_df["connected_company_name"] = connected_company_name
# print("Test")
# print(companies_relations_df)
return companies_relations_df
# Plotly figure
def create_network_graph() -> go.Figure:
"""Create a NetworkX Graph.
Returns:
go.Figure: _description_
"""
edges = []
for _, row in find_all_company_relations().iterrows():
edges.append([row["company_name"], row["connected_company_name"]])
network_graph = nx.Graph()
network_graph.add_edges_from(edges)
pos = nx.spring_layout(network_graph)
# edges trace
edge_x = []
edge_y = []
for edge in network_graph.edges():
x0, y0 = pos[edge[0]]
x1, y1 = pos[edge[1]]
edge_x.append(x0)
edge_x.append(x1)
edge_x.append(None)
edge_y.append(y0)
edge_y.append(y1)
edge_y.append(None)
edge_trace = go.Scatter(
x=edge_x,
y=edge_y,
line={"color": "black", "width": 1},
hoverinfo="none",
showlegend=False,
mode="lines",
)
# nodes trace
node_x = []
node_y = []
text = []
for node in network_graph.nodes():
x, y = pos[node]
node_x.append(x)
node_y.append(y)
text.append(node)
node_trace = go.Scatter(
x=node_x,
y=node_y,
text=text,
mode="markers+text",
showlegend=False,
hoverinfo="none",
marker={"color": "pink", "size": 50, "line": {"color": "black", "width": 1}},
)
# layout
layout = {
"plot_bgcolor": "white",
"paper_bgcolor": "white",
"margin": {"t": 10, "b": 10, "l": 10, "r": 10, "pad": 0},
"xaxis": {
"linecolor": "black",
"showgrid": False,
"showticklabels": False,
"mirror": True,
},
"yaxis": {
"linecolor": "black",
"showgrid": False,
"showticklabels": False,
"mirror": True,
},
}
measure_vector = {}
network_metrics_df = pd.DataFrame()
measure_vector = nx.eigenvector_centrality(network_graph)
network_metrics_df["eigenvector"] = measure_vector.values()
measure_vector = nx.degree_centrality(network_graph)
network_metrics_df["degree"] = measure_vector.values()
measure_vector = nx.betweenness_centrality(network_graph)
network_metrics_df["betweeness"] = measure_vector.values()
measure_vector = nx.closeness_centrality(network_graph)
network_metrics_df["closeness"] = measure_vector.values()
# figure
return go.Figure(data=[edge_trace, node_trace], layout=layout)
# Dash App
app = Dash(__name__)
app.title = "Dash Networkx"
# className="networkx_style"
app.layout = html.Div(
style={"width": "49%"},
children=[
html.I("Write your EDGE_VAR"),
html.Br(),
# dcc.Dropdown(['eigenvector', 'degree', 'betweeness', 'closeness'], 'eigenvector', id='metric-dropdown'),
dcc.Input(id="EGDE_VAR", type="text", value="K", debounce=True),
dcc.Graph(id="my-graph", style={"width": "49%"}),
],
)
@app.callback(
Output("my-graph", "figure"),
# Input('metric-dropdown', 'value'),
[Input("EGDE_VAR", "value")],
)
def update_output() -> go.Figure:
"""Just Returns the go Figure of Plotly.
Returns:
go.Figure: Returns a HTML Figure for Plotly.
"""
return create_network_graph()
if __name__ == "__main__":
"""Main Method to test this page."""
app.run(debug=True)

366
Jupyter/NetworkX/archive_prod/ui_elements.py Normal file
View File

@ -0,0 +1,366 @@
"""Dash elements."""
import pandas as pd
import plotly.graph_objs as go
from cachetools import TTLCache, cached
from dash import dash_table, dcc, html
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session
from aki_prj23_transparenzregister.ui.archive.networkx_dash import networkx_component
from aki_prj23_transparenzregister.utils.sql import entities
COLORS = {
"light": "#edefef",
"lavender-blush": "#f3e8ee",
"ash-gray": "#bacdb0",
"cambridge-blue": "#729b79",
"paynes-gray": "#475b63",
"raisin-black": "#2e2c2f",
}
def get_company_data(session: Session) -> pd.DataFrame:
"""Creates a session to the database and get's all available company data.
Args:
session: A session connecting to the database.
Returns:
A dataframe containing all available company data including the corresponding district court.
"""
query_company = session.query(entities.Company, entities.DistrictCourt.name).join(
entities.DistrictCourt
)
engine = session.bind
if not isinstance(engine, Engine):
raise TypeError
return pd.read_sql(str(query_company), engine, index_col="company_id")
def get_finance_data(session: Session) -> pd.DataFrame:
"""Collects all available company data.
Args:
session: A session connecting to the database.
Returns:
A dataframe containing all financial data of all companies.
"""
query_finance = session.query(
entities.AnnualFinanceStatement, entities.Company.name, entities.Company.id
).join(entities.Company)
engine = session.bind
if not isinstance(engine, Engine):
raise TypeError
return pd.read_sql(str(query_finance), engine)
@cached( # type: ignore
cache=TTLCache(maxsize=1, ttl=300),
key=lambda session: 0 if session is None else str(session.bind),
)
def get_options(session: Session | None) -> dict[int, str]:
"""Collects the search options for the companies.
Args:
session: A session connecting to the database.
Returns:
A dict containing the company id as key and its name.
"""
if not session:
return {}
return get_company_data(session)["company_name"].to_dict()
def create_header(options: dict) -> html:
"""Creates header for dashboard.
Args:
options: A dictionary with company names and ids for the dropdown.
Returns:
The html div to create the page's header including the name of the page and the search for companies.
"""
return html.Div(
className="header-wrapper",
children=[
html.Div(
className="header-title",
children=[
html.I(
id="home-button",
n_clicks=0,
className="bi-house-door-fill",
),
html.H1(
className="header-title-text",
children="Transparenzregister für Kapitalgesellschaften",
),
],
),
html.Div(
className="header-search",
children=[
html.Div(
className="header-search-dropdown",
children=[
dcc.Dropdown(
id="select_company",
options=[
{"label": o, "value": key}
for key, o in options.items()
],
placeholder="Suche nach Unternehmen oder Person",
),
],
),
],
),
],
)
def create_company_header(selected_company_name: str) -> html:
"""Create company header based on selected company.
Args:
selected_company_name: The company name that has been chosen in the dropdown.
Returns:
The html div to create the company header.
"""
return html.Div(
className="company-header",
children=[
html.H1(
className="company-header-title",
id="id-company-header-title",
children=selected_company_name,
),
],
)
def create_company_stats(selected_company_data: pd.Series) -> html:
"""Create company stats.
Args:
selected_company_data: A series containing all company information of the selected company.
Returns:
The html div to create the company stats table and the three small widgets.
"""
company_data = {
"col1": ["Unternehmen", "Straße", "Stadt"],
"col2": [
selected_company_data["company_name"],
selected_company_data["company_street"],
str(
selected_company_data["company_zip_code"]
+ " "
+ selected_company_data["company_city"]
),
],
"col3": ["Branche", "Amtsgericht", "Gründungsjahr"],
"col4": [
selected_company_data["company_sector"],
selected_company_data["district_court_name"],
"xxx",
],
}
df_company_data = pd.DataFrame(data=company_data)
return html.Div(
className="stats-wrapper",
children=[
html.Div(
className="widget-large",
children=[
html.H3(
className="widget-title",
children="Stammdaten",
),
dash_table.DataTable(
df_company_data.to_dict("records"),
[{"name": i, "id": i} for i in df_company_data.columns],
style_table={
"width": "90%",
"marginLeft": "auto",
"marginRight": "auto",
"paddingBottom": "20px",
"color": COLORS["raisin-black"],
},
# hide header of table
css=[
{
"selector": "tr:first-child",
"rule": "display: none",
},
],
style_cell={"textAlign": "center"},
style_cell_conditional=[
{"if": {"column_id": c}, "fontWeight": "bold"}
for c in ["col1", "col3"]
],
style_data={
"whiteSpace": "normal",
"height": "auto",
},
),
],
),
html.Div(
className="widget-small",
children=[
html.H3(
className="widget-title",
children="Stimmung",
),
],
),
html.Div(
className="widget-small",
children=[
html.H3(
className="widget-title",
children="Aktienkurs",
),
html.H1(
className="widget-content",
children="123",
),
],
),
html.Div(
className="widget-small",
children=[
html.H3(
className="widget-title",
children="Umsatz",
),
html.H1(
className="widget-content",
children="1234",
),
],
),
],
)
def create_tabs(selected_company_id: int, selected_finance_df: pd.DataFrame) -> html:
"""Create tabs for more company information.
Args:
selected_company_id: Id of the chosen company in the dropdown.
selected_finance_df: A dataframe containing all available finance information of the companies.
Returns:
The html div to create the tabs of the company page.
"""
return html.Div(
className="tabs",
children=[
dcc.Tabs(
id="tabs",
value="tab-1",
children=[
dcc.Tab(
label="Kennzahlen",
value="tab-1",
className="tab-style",
selected_className="selected-tab-style",
children=[kennzahlen_layout(selected_finance_df)],
),
dcc.Tab(
label="Beteiligte Personen",
value="tab-2",
className="tab-style",
selected_className="selected-tab-style",
),
dcc.Tab(
label="Stimmung",
value="tab-3",
className="tab-style",
selected_className="selected-tab-style",
),
dcc.Tab(
label="Verflechtungen",
value="tab-4",
className="tab-style",
selected_className="selected-tab-style",
children=[network_layout(selected_company_id)],
),
],
),
html.Div(id="tabs-example-content-1"),
],
)
def kennzahlen_layout(selected_finance_df: pd.DataFrame) -> html:
"""Create metrics tab.
Args:
selected_company_id: Id of the chosen company in the dropdown.
selected_finance_df: A dataframe containing all available finance information of the companies.
Returns:
The html div to create the metrics tab of the company page.
"""
return html.Div(
[
dcc.Graph(
figure=financials_figure(
selected_finance_df, "annual_finance_statement_ebit"
)
)
]
)
def financials_figure(selected_finance_df: pd.DataFrame, metric: str) -> go.Figure:
"""Creates plotly line chart for a specific company and a metric.
Args:
selected_finance_df: A dataframe containing all finance information of the selected company.
metric: The metric that should be visualized.
Returns:
A plotly figure showing the available metric data of the company.
"""
# create figure
fig_line = go.Figure()
# add trace for company 1
fig_line.add_trace(
go.Scatter(
x=selected_finance_df["annual_finance_statement_date"],
y=selected_finance_df[metric],
line_color=COLORS["raisin-black"],
marker_color=COLORS["raisin-black"],
)
)
# set title and labels
fig_line.update_layout(
title=metric,
xaxis_title="Jahr",
yaxis_title="in Mio.€",
plot_bgcolor=COLORS["light"],
)
return fig_line
def network_layout(selected_company_id: int) -> html:
"""Create network tab.
Args:
selected_company_id: Id of the chosen company in the dropdown.
Returns:
The html div to create the network tab of the company page.
"""
return networkx_component(selected_company_id)

22
documentations/meeting-notes/Meeting_2023-10-12.md Normal file
View File

@ -0,0 +1,22 @@
# Weekly *12*: 12.10.2023
## Teilnehmer
- Prof. Arinir
- Tristan Nolde
- Tim Ronneburg
- Philipp Horstenkamp
- Kim Mesewinkel-Risse
- Sascha Zhu
- Sebastian Zeleny
## Themen
- ABC:
- ...
## Abgeleitete Action Items
| Action Item | Verantwortlicher | Deadline |
|-------------|------------------|-----------------|
| Finanzdaten optimieren | Kim, Tristan | nächstes Weekly |
| Rebasen vom Branch für Sascha | Sascha | nächstes Weekly |

BIN
documentations/plots/first_complete_force_graph.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 406 KiB

BIN
documentations/seminararbeiten/Verflechtungsanalyse/Graph.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 1.2 KiB

After

Width:  |  Height:  |  Size: 2.3 KiB

180
documentations/seminararbeiten/Verflechtungsanalyse/metrics/test.html Normal file
View File

@ -0,0 +1,180 @@
<html>
<head>
<meta charset="utf-8">
<script src="lib/bindings/utils.js"></script>
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/vis-network/9.1.2/dist/dist/vis-network.min.css" integrity="sha512-WgxfT5LWjfszlPHXRmBWHkV2eceiWTOBvrKCNbdgDYTHrT2AeLCGbF4sZlZw3UMN3WtL0tGUoIAKsu8mllg/XA==" crossorigin="anonymous" referrerpolicy="no-referrer" />
<script src="https://cdnjs.cloudflare.com/ajax/libs/vis-network/9.1.2/dist/vis-network.min.js" integrity="sha512-LnvoEWDFrqGHlHmDD2101OrLcbsfkrzoSpvtSQtxK3RMnRV0eOkhhBN2dXHKRrUU8p2DGRTk35n4O8nWSVe1mQ==" crossorigin="anonymous" referrerpolicy="no-referrer"></script>
<center>
<h1></h1>
</center>
<!-- <link rel="stylesheet" href="../node_modules/vis/dist/vis.min.css" type="text/css" />
<script type="text/javascript" src="../node_modules/vis/dist/vis.js"> </script>-->
<link
href="https://cdn.jsdelivr.net/npm/bootstrap@5.0.0-beta3/dist/css/bootstrap.min.css"
rel="stylesheet"
integrity="sha384-eOJMYsd53ii+scO/bJGFsiCZc+5NDVN2yr8+0RDqr0Ql0h+rP48ckxlpbzKgwra6"
crossorigin="anonymous"
/>
<script
src="https://cdn.jsdelivr.net/npm/bootstrap@5.0.0-beta3/dist/js/bootstrap.bundle.min.js"
integrity="sha384-JEW9xMcG8R+pH31jmWH6WWP0WintQrMb4s7ZOdauHnUtxwoG2vI5DkLtS3qm9Ekf"
crossorigin="anonymous"
></script>
<center>
<h1></h1>
</center>
<style type="text/css">
#mynetwork {
width: 100%;
height: 600px;
background-color: white;
border: 1px solid lightgray;
position: relative;
float: left;
}
#config {
float: left;
width: 400px;
height: 600px;
}
</style>
</head>
<body>
<div class="card" style="width: 100%">
<div id="mynetwork" class="card-body"></div>
</div>
<div id="config"></div>
<script type="text/javascript">
// initialize global variables.
var edges;
var nodes;
var allNodes;
var allEdges;
var nodeColors;
var originalNodes;
var network;
var container;
var options, data;
var filter = {
item : '',
property : '',
value : []
};
// This method is responsible for drawing the graph, returns the drawn network
function drawGraph() {
var container = document.getElementById('mynetwork');
// parsing and collecting nodes and edges from the python
nodes = new vis.DataSet([{"branche": "Automobilhersteller", "color": " #729b79ff", "font": {"color": "black"}, "id": 2, "label": "Volkswagen AG", "shape": "dot", "size": 10, "title": "Volkswagen AG\nAutomobilhersteller", "type": "Company"}, {"branche": "Automobilhersteller", "color": " #729b79ff", "font": {"color": "black"}, "id": 1, "label": "Porsche Automobil Holding", "shape": "dot", "size": 10, "title": "Porsche Automobil Holding\nAutomobilhersteller", "type": "Company"}, {"branche": "Automobilhersteller", "color": " #729b79ff", "font": {"color": "black"}, "id": 3, "label": "Volkswagen", "shape": "dot", "size": 10, "title": "Volkswagen\nAutomobilhersteller", "type": "Company"}, {"branche": "Automobilhersteller", "color": " #729b79ff", "font": {"color": "black"}, "id": 4, "label": "Audi", "shape": "dot", "size": 10, "title": "Audi\nAutomobilhersteller", "type": "Company"}, {"branche": "Automobilhersteller", "color": " #729b79ff", "font": {"color": "black"}, "id": 5, "label": "Seat", "shape": "dot", "size": 10, "title": "Seat\nAutomobilhersteller", "type": "Company"}, {"branche": "Automobilhersteller", "color": " #729b79ff", "font": {"color": "black"}, "id": 6, "label": "Skoda Auto", "shape": "dot", "size": 10, "title": "Skoda Auto\nAutomobilhersteller", "type": "Company"}, {"branche": "Automobilhersteller", "color": " #729b79ff", "font": {"color": "black"}, "id": 7, "label": "Porsche AG", "shape": "dot", "size": 10, "title": "Porsche AG\nAutomobilhersteller", "type": "Company"}, {"branche": "Automobilhersteller", "color": " #729b79ff", "font": {"color": "black"}, "id": 8, "label": "Lamborghini", "shape": "dot", "size": 10, "title": "Lamborghini\nAutomobilhersteller", "type": "Company"}, {"branche": "Automobilhersteller", "color": " #729b79ff", "font": {"color": "black"}, "id": 9, "label": "Bentley", "shape": "dot", "size": 10, "title": "Bentley\nAutomobilhersteller", "type": "Company"}, {"branche": "Automobilzulieferer", "color": "#475b63ff", "font": {"color": "black"}, "id": 10, "label": "Forvia", "shape": "dot", "size": 10, "title": "Forvia\nAutomobilzulieferer", "type": "Company"}, {"branche": "Automobilzulieferer", "color": "#475b63ff", "font": {"color": "black"}, "id": 11, "label": "Hella", "shape": "dot", "size": 10, "title": "Hella\nAutomobilzulieferer", "type": "Company"}]);
edges = new vis.DataSet([{"from": 2, "label": "part_of", "to": 1, "width": 1}, {"from": 1, "label": "part_of", "to": 3, "width": 1}, {"from": 1, "label": "part_of", "to": 4, "width": 1}, {"from": 1, "label": "part_of", "to": 5, "width": 1}, {"from": 1, "label": "part_of", "to": 6, "width": 1}, {"from": 1, "label": "part_of", "to": 7, "width": 1}, {"from": 4, "label": "part_of", "to": 8, "width": 1}, {"from": 4, "label": "part_of", "to": 9, "width": 1}, {"from": 4, "label": "supplierer", "to": 10, "width": 1}, {"from": 11, "label": "part_of", "to": 10, "width": 1}]);
nodeColors = {};
allNodes = nodes.get({ returnType: "Object" });
for (nodeId in allNodes) {
nodeColors[nodeId] = allNodes[nodeId].color;
}
allEdges = edges.get({ returnType: "Object" });
// adding nodes and edges to the graph
data = {nodes: nodes, edges: edges};
var options = {
"configure": {
"enabled": true,
"filter": [
"physics"
]
},
"edges": {
"color": {
"inherit": false
},
"smooth": {
"enabled": true,
"type": "dynamic"
}
},
"interaction": {
"dragNodes": true,
"hideEdgesOnDrag": false,
"hideNodesOnDrag": false
},
"physics": {
"enabled": true,
"repulsion": {
"centralGravity": 0.2,
"damping": 0.09,
"nodeDistance": 150,
"springConstant": 0.05,
"springLength": 50
},
"solver": "repulsion",
"stabilization": {
"enabled": true,
"fit": true,
"iterations": 1000,
"onlyDynamicEdges": false,
"updateInterval": 50
}
}
};
// if this network requires displaying the configure window,
// put it in its div
options.configure["container"] = document.getElementById("config");
network = new vis.Network(container, data, options);
network.on("click", neighbourhoodHighlight);
return network;
}
drawGraph();
</script>
</body>
</html>

38
documentations/seminararbeiten/Verflechtungsanalyse/mockup_verflechtungsanalyse_with_networkx.ipynb
View File

File diff suppressed because one or more lines are too long

823
poetry.lock generated
View File

File diff suppressed because it is too large Load Diff

8
pyproject.toml
View File

@ -55,6 +55,9 @@ cachetools = "^5.3.1"
dash = "^2.14.1"
dash-auth = "^2.0.0"
dash-bootstrap-components = "^1.5.0"
dash-daq = "^0.5.0"
dash_cytoscape = "^0.2.0"
dashvis = "^0.1.3"
datetime = "^5.2"
deutschland = {git = "https://github.com/TrisNol/deutschland.git", branch = "hotfix/python-3.11-support"}
frozendict = "^2.3.8"
@ -63,9 +66,10 @@ matplotlib = "^3.8.1"
pgeocode = "^0.4.1"
psycopg2-binary = "^2.9.7"
pymongo = "^4.6.0"
python = "^3.11"
python = ">=3.11,<3.13"
python-dotenv = "^1.0.0"
rapidfuzz = "^3.5.2"
scipy = "^1.11.3"
seaborn = "^0.13.0"
selenium = "^4.15.2"
spacy = "^3.6.1"
@ -109,6 +113,7 @@ SQLAlchemy = {version = "*", extras = ["mypy"]}
black = "*"
loguru-mypy = "*"
mypy = "*"
networkx-stubs = "*"
pandas-stubs = "*"
pip-audit = "*"
pip-licenses = "*"
@ -121,6 +126,7 @@ types-setuptools = "*"
types-six = "*"
types-tabulate = "*"
types-tqdm = "*"
types-urllib3 = "*"
[tool.poetry.group.test.dependencies]
pytest = "^7.4.2"

112
src/aki_prj23_transparenzregister/ui/assets/networkx_style.css Normal file
View File

@ -0,0 +1,112 @@
.networkx_style {
float: right;
margin-top: 20px;
margin-left: 10px;
margin-right: 20px;
border: 1px solid;
border-color: var(--raisin-black);
border-radius: 20px;
width: 57%;
height: 100%;
}
.top_companytable_style {
float: left;
margin-top: 20px;
margin-left: 20px;
margin-right: 10px;
border: 1px solid;
border-color: var(--raisin-black);
border-radius: 20px;
width: 37%;
height: 100%;
}
.networkx_style .filter-wrapper {
float: left;
width: 100%;
}
.networkx_style .filter-wrapper .filter-wrapper-item {
display: inline-block;
padding: 10px;
width: 31%;
}
.networkx_style .filter-wrapper .filter-wrapper-item .dropdown_style {
padding-bottom: 10px;
margin-top: 1px;
padding-left: 10px;
width: 100%;
font-size: '30%'
/* background-color: var(--raisin-black); */
}
.networkx_style .header {
text-align: center;
align-items: center;
align-content: center;
vertical-align: middle;
padding: 20px;
margin: 0px;
/* margin: 5px; */
color: var(--raisin-black);
/* background-color: var(--raisin-black); */
}
.networkx_style .filter-wrapper .filter-wrapper-item .filter-description {
padding: 5px;
padding-left: 10px;
margin: 0px;
/* margin: 5px; */
color: var(--raisin-black);
/* background-color: var(--raisin-black); */
}
.networkx_style .switch-style {
align-self: left;
float: none;
padding-bottom: 10px;
margin-top: 1px;
padding-left: 10px;
width: 80%;
display: inline-block;
/* background-color: var(--raisin-black); */
}
.networkx_style .switch-style .jAjsxw {
align-self: left;
float: none;
display: inline-block;
padding: 0px;
margin: 0px;
/* background-color: var(--raisin-black); */
}
.networkx_style .graph-style {
margin-top: 10px;
padding-bottom: 0px;
display: inline-block;
margin: 0px;
width: 100%;
height: 100%;
/* float: inline-end, */
/* background-color: var(--raisin-black); */
}
.networkx_style .graph-style .canvas {
padding: 0px;
margin: 0px;
width: 100%
/* background-color: var(--raisin-black); */
}

60
src/aki_prj23_transparenzregister/ui/company_elements.py
View File

@ -10,6 +10,12 @@ from dash import dash_table, dcc, html
from sqlalchemy.orm import Session
from aki_prj23_transparenzregister.ui import data_elements, finance_elements
from aki_prj23_transparenzregister.utils.networkx.network_2d import create_2d_graph
from aki_prj23_transparenzregister.utils.networkx.network_base import initialize_network
from aki_prj23_transparenzregister.utils.networkx.networkx_data import (
create_edge_and_node_list_for_company,
find_company_relations,
)
COLORS = {
"light": "#edefef",
@ -344,7 +350,27 @@ def person_relations_layout(selected_company_id: int, session: Session) -> html:
)
def network_layout(selected_company_id: int) -> html:
def kennzahlen_layout(selected_finance_df: pd.DataFrame) -> html:
"""Create metrics tab.
Args:
selected_finance_df: A dataframe containing all available finance information of the companies.
Returns:
The html div to create the metrics tab of the company page.
"""
return html.Div(
[
dcc.Graph(
figure=finance_elements.financials_figure(
selected_finance_df, "annual_finance_statement_ebit"
)
)
]
)
def network_layout(selected_company_id: int) -> html.Div:
"""Create network tab.
Args:
@ -353,4 +379,34 @@ def network_layout(selected_company_id: int) -> html:
Returns:
The html div to create the network tab of the company page.
"""
return html.Div([f"Netzwerk von Unternehmen mit ID: {selected_company_id}"])
person_relations, company_relations = find_company_relations(selected_company_id)
# Create Edge and Node List from data
nodes, edges = create_edge_and_node_list_for_company(company_relations)
# Initialize the Network and receive the Graph and a DataFrame with Metrics
if nodes != {}:
graph, metrics = initialize_network(nodes=nodes, edges=edges)
metric = "None"
figure = create_2d_graph(
graph,
nodes,
edges,
metrics,
metric,
layout="Spring",
edge_annotation=True,
edge_thickness=1,
)
return html.Div(
children=[
dcc.Graph(figure=figure, id="company-graph", className="graph-style")
]
)
return html.Div(
[
html.H3(
f"Leider gibt es keine Verbindungen vom Unternehmen mit ID: {selected_company_id}"
)
]
)

409
src/aki_prj23_transparenzregister/ui/pages/home.py
View File

@ -1,22 +1,409 @@
"""Content of home page."""
from functools import lru_cache
import dash
from dash import html
import dash_daq as daq
import networkx as nx
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from cachetools import TTLCache, cached
from dash import Input, Output, callback, dash_table, dcc, html
from aki_prj23_transparenzregister.utils.networkx.network_2d import (
create_2d_graph,
)
from aki_prj23_transparenzregister.utils.networkx.network_3d import (
# initialize_network,
create_3d_graph,
)
from aki_prj23_transparenzregister.utils.networkx.network_base import (
initialize_network,
)
from aki_prj23_transparenzregister.utils.networkx.networkx_data import (
create_edge_and_node_list,
filter_relation_type,
get_all_company_relations,
get_all_person_relations,
)
dash.register_page(
__name__,
path="/",
title="Transparenzregister: Home",
redirect_from=[
"/unternehmensdetails",
"/personendetails",
"/unternehmensdetails/",
"/personendetails/",
],
# TODO Fix redirect in unit tests
# redirect_from=[
# "/unternehmensdetails",
# "/personendetails",
# "/unternehmensdetails/<value>",
# "/personendetails/<value>",
# ],
)
layout = html.Div(
metric = "None"
switch_edge_annotaion_value = False
egde_thickness = 1
network = None
def person_relation_type_filter() -> np.ndarray:
"""Returns an Numpy Array of String with Person telation types."""
return get_all_person_relations()["relation_type"].unique()
def company_relation_type_filter() -> np.ndarray:
"""Returns an Numpy Array of String with Company relation types."""
return get_all_company_relations()["relation_type"].unique()
def update_table(
metric_dropdown_value: str, metrics: pd.DataFrame
) -> tuple[dict, list]:
"""_summary_.
Args:
metric_dropdown_value (str): _description_
metrics (pd.DataFrame): _description_
Returns:
tuple[dict, list]: _description_
"""
table_df = metrics.sort_values(metric_dropdown_value, ascending=False).head(10)
table_df = table_df[["designation", "category", metric_dropdown_value]]
columns = [{"name": i, "id": i} for i in table_df.columns]
return table_df.to_dict("records"), columns # type: ignore
def layout() -> list[html.Div]:
"""Generates the Layout of the Homepage."""
person_relation_types = person_relation_type_filter()
company_relation_types = company_relation_type_filter()
top_companies_dict, top_companies_columns, figure = update_figure(
"None",
False,
False,
company_relation_types[0],
person_relation_types[0],
"Spring",
1,
"degree",
)
return html.Div(
children=html.Div(
children=[
html.Div(
className="top_companytable_style",
children=[
html.H1(
title="Top Ten Nodes in Graph by Metric",
style={"align": "mid"},
),
html.Div(
className="filter-wrapper-item",
children=[
html.H5(
className="filter-description",
children=["Filter Metric:"],
),
dcc.Dropdown(
[
"eigenvector",
"degree",
"betweenness",
"closeness",
],
"closeness",
id="dropdown_table_metric",
className="dropdown_style",
),
],
),
dash_table.DataTable(
top_companies_dict,
top_companies_columns,
id="metric_table",
),
],
),
html.Div(
className="networkx_style",
children=[
html.H1(className="header", children=["Social Graph"]),
html.Div(
className="filter-wrapper",
id="company_dropdown",
# style="visibility: hidden;",
children=[
html.Div(
className="filter-wrapper-item",
children=[
html.H5(
className="filter-description",
children=["Company Relation Type Filter:"],
),
dcc.Dropdown(
company_relation_types,
company_relation_types[0],
id="dropdown_company_relation_filter",
className="dropdown_style",
),
],
),
html.Div(
className="filter-wrapper-item",
# style="visibility: visible;",
children=[
html.H5(
className="filter-description",
children=["Person Relation Type Filter:"],
),
dcc.Dropdown(
person_relation_types,
person_relation_types[0],
id="dropdown_person_relation_filter",
className="dropdown_style",
),
],
),
html.Div(
className="filter-wrapper-item",
children=[
html.H5(
className="filter-description",
children=["Choose Graph Metric:"],
),
dcc.Dropdown(
[
"None",
"eigenvector",
"degree",
"betweenness",
"closeness",
],
"None",
id="dropdown",
className="dropdown_style",
),
],
),
html.Div(
className="filter-wrapper-item",
children=[
html.H5(
className="filter-description",
children=["Choose Layout:"],
),
dcc.Dropdown(
[
"Spring",
# "Bipartite",
"Circular",
"Kamada Kawai",
# "Planar",
"Random",
"Shell (only 2D)",
# "Spectral",
"Spiral (only 2D)",
# "Multipartite"
],
"Spring",
id="dropdown_layout",
className="dropdown_style",
),
],
),
html.Div(
className="filter-wrapper-item",
children=[
html.H5(
className="filter-description",
children=["Adjust Edge Thickness"],
),
dcc.Slider(
1,
4,
1,
value=1,
id="slider",
),
],
),
],
),
html.Div(
className="filter-wrapper",
children=[
html.Div(
className="filter-wrapper-item",
children=[
html.H5(
className="filter-description",
children=["Switch to 2D Diagramm"],
),
html.Div(
className="switch-style",
children=[
daq.BooleanSwitch(id="switch", on=False)
],
),
],
),
# html.Div(
# className="filter-wrapper-item",
# children=[
# html.H5(
# className="filter-description",
# children=["Enable Node Annotation"],
# ),
# html.Div(
# className="switch-style",
# children=[daq.BooleanSwitch(id="switch_node_annotation", on=False)],
# ),
# ],
# ),
html.Div(
className="filter-wrapper-item",
children=[
html.H5(
className="filter-description",
children=["Enable Edge Annotation"],
),
html.Div(
className="switch-style",
children=[
daq.BooleanSwitch(
id="switch_edge_annotation",
on=False,
)
],
),
],
),
],
),
dcc.Graph(
figure=figure, id="my-graph", className="graph-style"
),
],
),
]
)
)
@lru_cache(200)
def update_graph_data(
person_relation_type: str = "HAFTENDER_GESELLSCHAFTER",
company_relation_type: str = "GESCHAEFTSFUEHRER",
) -> tuple[nx.Graph, pd.DataFrame, dict, list]:
"""_summary_.
Args:
person_relation_type (str, optional): _description_. Defaults to "HAFTENDER_GESELLSCHAFTER".
company_relation_type (str, optional): _description_. Defaults to "GESCHAEFTSFUEHRER".
Returns:
tuple[nx.Graph, pd.DataFrame, dict, list]: _description_
"""
# Get Data
person_df = get_all_person_relations()
company_df = get_all_company_relations()
person_relation = filter_relation_type(person_df, person_relation_type)
company_relation = filter_relation_type(company_df, company_relation_type)
# Create Edge and Node List from data
nodes_tmp, edges_tmp = create_edge_and_node_list(person_relation, company_relation)
graph, metrics = initialize_network(nodes=nodes_tmp, edges=edges_tmp)
return graph, metrics, nodes_tmp, edges_tmp
@callback(
[
html.H1("This is our Home page", style={"margin": 0}),
html.Div("This is our Home page content."),
]
Output("metric_table", "data"),
Output("metric_table", "columns"),
Output("my-graph", "figure"),
],
[
Input("dropdown", "value"),
Input("switch", "on"),
# Input("switch_node_annotation", "on"),
Input("switch_edge_annotation", "on"),
Input("dropdown_company_relation_filter", "value"),
Input("dropdown_person_relation_filter", "value"),
Input("dropdown_layout", "value"),
Input("slider", "value"),
Input("dropdown_table_metric", "value"),
],
prevent_initial_call=True,
allow_duplicate=True,
)
# @lru_cache(20)
@cached(cache=TTLCache(maxsize=100, ttl=500))
def update_figure( # noqa: PLR0913
selected_metric: str,
switch_value: bool,
# switch_node_annotaion_value: bool,
switch_edge_annotaion_value: bool,
c_relation_filter_value: str,
p_relation_filter_value: str,
layout: str,
slider_value: float,
metric_dropdown_value: str,
) -> go.Figure:
"""In this Callback the Value of the Dropdown is used to filter the Data. In Addition it takes the filter for the Graph metrics and creates a new graph, or switches between 3D and 2D.
Args:
selected_metric (_type_): Selected Value
switch_value (bool): True if 2D, False if 3D
switch_edge_annotaion_value: True if Edge should have a description, Flase = No Descritpion
c_relation_filter_value (_type_): Variable with String value of Relation Type for Companies
p_relation_filter_value (_type_): Variable with String value of Relation Type for Persons
layout: String of the Layout Dropdown
metric_dropdown_value: String of the Metric Dropdown
slider_value: Sets the size of the Edge Connections
Returns:
Network Graph(Plotly Figure): Plotly Figure in 3 or 2D
"""
_ = c_relation_filter_value, p_relation_filter_value
graph, metrics, nodes, edges = update_graph_data(
person_relation_type=p_relation_filter_value,
company_relation_type=c_relation_filter_value,
)
table_dict, table_columns = update_table(metric_dropdown_value, metrics)
if switch_value:
return (
table_dict,
table_columns,
create_2d_graph(
graph,
nodes,
edges,
metrics,
selected_metric,
layout,
switch_edge_annotaion_value,
slider_value, # type: ignore
),
)
return (
table_dict,
table_columns,
create_3d_graph(
graph,
nodes,
edges,
metrics,
selected_metric,
layout,
switch_edge_annotaion_value,
slider_value, # type: ignore
),
)

1
src/aki_prj23_transparenzregister/utils/networkx/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Everything regarding data extraction for NetworkX Graphs."""

184
src/aki_prj23_transparenzregister/utils/networkx/network_2d.py Normal file
View File

@ -0,0 +1,184 @@
"""This Module contains the Method to create a 2D Network Graph with NetworkX."""
import networkx as nx
import numpy as np
import pandas as pd
import plotly.graph_objects as go
def create_2d_graph( # noqa PLR0913
graph: nx.Graph,
nodes: dict,
edges: list,
metrics: pd.DataFrame,
metric: str | None,
layout: str,
edge_annotation: bool,
edge_thickness: int,
) -> go.Figure:
"""This Method creates a 2d Network in Plotly with a Scatter Graph and retuns it.
Args:
graph (_type_): NetworkX Graph.
nodes (_type_): List of Nodes
edges (_type_): List of Edges
metrics (_type_): DataFrame with the MEtrics
metric (_type_): Selected Metric
edge_annotation: Enables the Description of Edges
edge_thickness: Int Value of the Edge thickness
layout: String which defines the Graph Layout
Returns:
_type_: Plotly Figure
"""
# Set 2D Layout
pos = nx.spring_layout(graph)
match layout:
case "Spring":
pos = nx.spring_layout(graph)
# case "Bipartite":
# pos = nx.bipartite_layout(graph)
case "Circular":
pos = nx.circular_layout(graph)
case "Kamada Kawai":
pos = nx.kamada_kawai_layout(graph)
# case "Planar":
# pos = nx.planar_layout(graph)
case "Random":
pos = nx.random_layout(graph)
case "(Shell only 2D)":
pos = nx.shell_layout(graph)
# case "Spectral":
# pos = nx.spectral_layout(graph)
case "(Spiral only 2D)":
pos = nx.spiral_layout(graph)
# case "Multipartite":
# pos = nx.multipartite_layout(graph)
# Initialize Variables to set the Position of the Edges.
edge_x = []
edge_y = []
# Initialize Node Position Variables.
node_x = []
node_y = []
# Initialize Position Variables for the Description Text of the edges.
edge_weight_x = []
edge_weight_y = []
# Getting the Positions from NetworkX and assign them to the variables.
for edge in graph.edges():
x0, y0 = pos[edge[0]]
x1, y1 = pos[edge[1]]
edge_x.append(x0)
edge_x.append(x1)
# edge_x.append(None)
edge_y.append(y0)
edge_y.append(y1)
# edge_y.append(None)
edge_weight_x.append((x1 + x0) / 2)
edge_weight_y.append((y1 + y0) / 2)
# edge_weight_y.append(None)
# Add the Edges to the scatter plot according to their Positions.
edge_trace = go.Scatter(
x=edge_x,
y=edge_y,
line={"width": 0.5, "color": "#888"},
hoverinfo="none",
mode="lines",
)
# Add the Edgedescriptiontext to the scatter plot according to its Position.
edge_weights_trace = go.Scatter(
x=edge_weight_x,
y=edge_weight_y,
mode="text",
marker_size=0.5,
text=[0.45, 0.7, 0.34],
textposition="top center",
hovertemplate="Relation: %{text}<extra></extra>",
)
# Getting the Positions from NetworkX and assign it to the variables.
for node in graph.nodes():
x, y = pos[node]
node_x.append(x)
node_y.append(y)
# Add the Nodes to the scatter plot according to their Positions.
node_trace = go.Scatter(
x=node_x,
y=node_y,
mode="markers",
hoverinfo="text",
marker={
"showscale": True,
"colorscale": "YlGnBu",
"reversescale": True,
"color": [],
"size": 10,
"colorbar": {
"thickness": 15,
"title": "Node Connections",
"xanchor": "left",
"titleside": "right",
},
"line_width": edge_thickness,
},
)
# # Set Color by using the nodes DataFrame with its Color Attribute. The sequence matters!
colors = list(nx.get_node_attributes(graph, "color").values())
node_names = list(nx.get_node_attributes(graph, "name").values())
# ids = list(nx.get_node_attributes(graph, "id").values())
# print(ids)
# # Get the Node Text
# node_names = []
# for key, value in nodes.items():
# if "name" in value:
# node_names.append(value["name"])
# else:
# node_names.append(value["firstname"] + " " + value["lastname"])
# # Add Color and Names to the Scatter Plot.
# print(colors)
# print(node_names)
node_trace.marker.color = colors
node_trace.text = node_names
# Highlight the Node Size in regards to the selected Metric.
if metric != "None":
node_trace.marker.size = list(np.sqrt(metrics[metric]) * 200)
# Add Relation_Type as a Description for the edges.
if edge_annotation:
edge_type_list = []
for row in edges:
edge_type_list.append(row["type"])
edge_weights_trace.text = edge_type_list
# Return the Plotly Figure
return go.Figure(
data=[edge_trace, edge_weights_trace, node_trace],
layout=go.Layout(
title="<br>Network graph made with Python",
titlefont_size=16,
showlegend=False,
hovermode="closest",
margin={"b": 20, "l": 5, "r": 5, "t": 20},
annotations=[
{
"showarrow": False,
"text": f"Companies (Red) & Person (Blue) Relation \n node_count: {len(nodes)}, edge_count: {len(edges)}",
"xref": "paper",
"yref": "paper",
"x": 0,
"y": 0.1,
"xanchor": "left",
"yanchor": "bottom",
"font": {"size": 14},
}
],
xaxis={"showgrid": False, "zeroline": False, "showticklabels": False},
yaxis={"showgrid": False, "zeroline": False, "showticklabels": False},
),
)

210
src/aki_prj23_transparenzregister/utils/networkx/network_3d.py Normal file
View File

@ -0,0 +1,210 @@
"""This Module contains the Method to create a 3D Network Graph with NetworkX."""
import networkx as nx
import numpy as np
import pandas as pd
import plotly.graph_objects as go
def create_3d_graph( # noqa : PLR0913
graph: nx.Graph,
nodes: dict,
edges: list,
metrics: pd.DataFrame,
metric: str | None,
layout: str,
edge_annotation: bool,
edge_thickness: int,
) -> go.Figure:
"""This Method creates a 3D Network in Plotly with a Scatter Graph and retuns it.
Args:
graph (_type_): NetworkX Graph.
nodes (_type_): List of Nodes
edges (_type_): List of Edges
metrics (_type_): DataFrame with the MEtrics
metric (_type_): Selected Metric
Returns:
_type_: Plotly Figure
"""
# 3d spring layout
pos = nx.spring_layout(graph, dim=3)
match layout:
case "Spring":
pos = nx.spring_layout(graph, dim=3)
# case "Bipartite":
# pos = nx.bipartite_layout(graph, dim=3)
case "Circular":
pos = nx.circular_layout(graph, dim=3)
case "Kamada Kawai":
pos = nx.kamada_kawai_layout(graph, dim=3)
# case "Planar":
# pos = nx.planar_layout(graph, dim=3)
case "Random":
pos = nx.random_layout(graph, dim=3)
# case "Shell":
# pos = nx.shell_layout(graph, dim=3)
# case "Spectral":
# pos = nx.spectral_layout(graph, dim=3)
# case "Spiral":
# pos = nx.spiral_layout(graph, dim=3)
# case "Multipartite":
# pos = nx.multipartite_layout(graph, dim=3)
# Initialize Variables to set the Position of the Edges.
edge_x = []
edge_y = []
edge_z = []
# Initialize Node Position Variables.
node_x = []
node_y = []
node_z = []
# Initialize Position Variables for the Description Text of the edges.
edge_weight_x = []
edge_weight_y = []
edge_weight_z = []
# Getting the Positions from NetworkX and assign them to the variables.
for edge in graph.edges():
x0, y0, z0 = pos[edge[0]]
x1, y1, z1 = pos[edge[1]]
edge_x.append(x0)
edge_x.append(x1)
edge_y.append(y0)
edge_y.append(y1)
edge_z.append(z0)
edge_z.append(z1)
# Calculate edge mid
edge_weight_x.append((x0 + x1) / 2)
edge_weight_y.append((y0 + y1) / 2)
edge_weight_z.append((z0 + z1) / 2)
# Add the Edges to the scatter plot according to their Positions.
edge_trace = go.Scatter3d(
x=edge_x,
y=edge_y,
z=edge_z,
mode="lines",
line={"color": "rgb(125,125,125)", "width": 1},
hoverinfo="none",
)
# Add the Edgedescriptiontext to the scatter plot according to its Position.
edge_weights_trace = go.Scatter3d(
x=edge_weight_x,
y=edge_weight_y,
z=edge_weight_z,
mode="text",
marker_size=1,
text=[0.45, 0.7, 0.34],
textposition="top center",
hovertemplate="weight: %{text}<extra></extra>",
)
# Getting the Positions from NetworkX and assign it to the variables.
for node in graph.nodes():
x, y, z = pos[node]
node_x.append(x)
node_y.append(y)
node_z.append(z)
# Add the Nodes to the scatter plot according to their Positions.
node_trace = go.Scatter3d(
x=node_x,
y=node_y,
z=node_z,
mode="markers",
name="actors",
marker={
"symbol": "circle",
"size": 6,
"color": "blue",
"colorscale": "Viridis",
"line": {"color": "rgb(50,50,50)", "width": 0.5},
},
# text=labels,
hoverinfo="text",
)
axis = {
"showbackground": False,
"showline": False,
"zeroline": False,
"showgrid": False,
"showticklabels": False,
"title": "",
}
layout = go.Layout(
title="Social Graph",
showlegend=False,
scene={
"xaxis": dict(axis),
"yaxis": dict(axis),
"zaxis": dict(axis),
},
margin={"t": 10},
hovermode="closest",
annotations=[
{
"showarrow": False,
"text": f"Companies (Red) & Person (Blue) Relation \n node_count: {len(nodes)}, edge_count: {len(edges)}",
"xref": "paper",
"yref": "paper",
"x": 0,
"y": 0.1,
"xanchor": "left",
"yanchor": "bottom",
"font": {"size": 14},
}
],
)
# Set Color by using the nodes DataFrame with its Color Attribute. The sequence matters!
colors = list(nx.get_node_attributes(graph, "color").values())
node_names = list(nx.get_node_attributes(graph, "name").values())
# node_names = []
# for key, value in nodes.items():
# if "name" in value:
# node_names.append(value["name"])
# else:
# node_names.append(value["firstname"] + " " + value["lastname"])
# Add Color and Names to the Scatter Plot.
# Add Color and Names to the Scatter Plot.
node_trace.marker.color = colors
node_trace.text = node_names
# Highlight the Node Size in regards to the selected Metric.
if metric != "None":
node_trace.marker.size = list(np.cbrt(metrics[metric]) * 200)
# Set the Color and width of Edges for better highlighting.
edge_colors = []
for row in edges:
if row["type"] == "HAFTENDER_GESELLSCHAFTER":
edge_colors.append("rgb(255,0,0)")
else:
edge_colors.append("rgb(255,105,180)")
edge_trace.line = {"color": edge_colors, "width": edge_thickness}
# Add Relation_Type as a Description for the edges.
if edge_annotation:
edge_type_list = []
for row in edges:
edge_type_list.append(row["type"])
edge_weights_trace.text = edge_type_list
# Set Color by using the nodes DataFrame with its Color Attribute. The sequence matters!
colors = list(nx.get_node_attributes(graph, "color").values())
node_trace.marker.color = colors
# Return the Plotly Figure
data = [edge_trace, edge_weights_trace, node_trace]
return go.Figure(data=data, layout=layout)

108
src/aki_prj23_transparenzregister/utils/networkx/network_base.py Normal file
View File

@ -0,0 +1,108 @@
"""This Module has a Method to initialize a Network with NetworkX which can be used for 2D and 3D Graphs."""
import networkx as nx
import pandas as pd
def initialize_network(edges: list, nodes: dict) -> tuple[nx.Graph, pd.DataFrame]:
"""This Method creates a Network from the Framework NetworkX with the help of a Node and Edge List. Furthemore it creates a DataFrame with the most important Metrics.
Args:
edges (list): List with the connections between Nodes.
nodes (dict): Dict with all Nodes.
Returns:
Graph: Plotly Figure
Metrices: DataFrame with Metrics
"""
# create edge dataframe
df_edges = pd.DataFrame(edges, columns=["from", "to", "type"])
graph = nx.from_pandas_edgelist(
df_edges, source="from", target="to", edge_attr="type"
)
# update node attributes from dataframe
nx.set_node_attributes(graph, nodes)
# Create a DataFrame with all Metrics
# Create a DataFrame with all Metrics
metrics = pd.DataFrame(
columns=[
"eigenvector",
"degree",
"betweenness",
"closeness",
"pagerank",
"category",
"designation",
"id",
]
)
metrics["eigenvector"] = nx.eigenvector_centrality(graph).values()
metrics["degree"] = nx.degree_centrality(graph).values()
metrics["betweenness"] = nx.betweenness_centrality(graph).values()
metrics["closeness"] = nx.closeness_centrality(graph).values()
metrics["pagerank"] = nx.pagerank(graph).values()
metrics["category"] = nx.get_node_attributes(graph, "type").values()
metrics["designation"] = nx.get_node_attributes(graph, "name").values()
metrics["id"] = nx.get_node_attributes(graph, "id").values()
return graph, metrics
def initialize_network_with_reduced_metrics(
edges: list, nodes: dict
) -> tuple[nx.Graph, pd.DataFrame]:
"""This Method creates a Network from the Framework NetworkX with the help of a Node and Edge List. Furthemore it creates a DataFrame with the most important Metrics.
Args:
edges (list): List with the connections between Nodes.
nodes (dict): Dict with all Nodes.
Returns:
Graph: Plotly Figure
Metrices: DataFrame with Metrics
"""
# create edge dataframe
df_edges = pd.DataFrame(edges, columns=["from", "to", "type"])
graph = nx.from_pandas_edgelist(
df_edges, source="from", target="to", edge_attr="type"
)
# update node attributes from dataframe
nx.set_node_attributes(graph, nodes)
# Create a DataFrame with all Metrics
metrics = pd.DataFrame(
columns=["degree", "betweenness", "closeness", "category", "designation", "id"]
)
# metrics["eigenvector"] = nx.eigenvector_centrality(graph).values()
metrics["degree"] = nx.degree_centrality(graph).values()
metrics["betweenness"] = nx.betweenness_centrality(graph).values()
metrics["closeness"] = nx.closeness_centrality(graph).values()
# metrics["pagerank"] = nx.pagerank(graph).values()
metrics["category"] = nx.get_node_attributes(graph, "type").values()
metrics["designation"] = nx.get_node_attributes(graph, "name").values()
metrics["id"] = nx.get_node_attributes(graph, "id").values()
return graph, metrics
def initialize_network_without_metrics(edges: list, nodes: dict) -> nx.Graph:
"""This Method creates a Network from the Framework NetworkX with the help of a Node and Edge List. Furthemore it creates a DataFrame with the most important Metrics.
Args:
edges (list): List with the connections between Nodes.
nodes (dict): Dict with all Nodes.
Returns:
Graph: Plotly Figure
"""
# create edge dataframe
df_edges = pd.DataFrame(edges, columns=["from", "to", "type"])
graph = nx.from_pandas_edgelist(
df_edges, source="from", target="to", edge_attr="type"
)
# update node attributes from dataframe
nx.set_node_attributes(graph, nodes)
return graph

382
src/aki_prj23_transparenzregister/utils/networkx/networkx_data.py Normal file
View File

@ -0,0 +1,382 @@
"""Module to receive and filter Data for working with NetworkX."""
from functools import lru_cache
import networkx as nx
import pandas as pd
from sqlalchemy.orm import aliased
from aki_prj23_transparenzregister.ui.session_handler import SessionHandler
from aki_prj23_transparenzregister.utils.networkx.network_base import (
initialize_network_with_reduced_metrics,
initialize_network_without_metrics,
)
from aki_prj23_transparenzregister.utils.sql import entities
# Gets the Session Key for the DB Connection.
# Alias for Company table for the base company
to_company = aliased(entities.Company, name="to_company")
# Alias for Company table for the head company
from_company = aliased(entities.Company, name="from_company")
COLOR_COMPANY = "blue"
COLOR_PERSON = "red"
def find_all_company_relations() -> pd.DataFrame:
"""_summary_.
Returns:
pd.DataFrame: _description_
"""
session = SessionHandler.session
assert session # noqa: S101
query_companies = session.query(entities.Company) # .all()
query_relations = session.query(entities.CompanyRelation) # .all()
companies_df: pd.DataFrame = pd.read_sql(str(query_companies), session.bind) # type: ignore
companies_relations_df: pd.DataFrame = pd.read_sql(str(query_relations), session.bind) # type: ignore
companies_relations_df = companies_relations_df[
["relation_id", "company_relation_company2_id"]
]
company_name = []
connected_company_name = []
companies_relations_df = companies_relations_df.head()
for _, row in companies_relations_df.iterrows():
company_name.append(
companies_df.loc[companies_df["company_id"] == row["relation_id"]][
"company_name"
].iloc[0]
)
connected_company_name.append(
companies_df.loc[
companies_df["company_id"] == row["company_relation_company2_id"]
]["company_name"].iloc[0]
)
companies_relations_df["company_name"] = company_name
companies_relations_df["connected_company_name"] = connected_company_name
return companies_relations_df
def get_all_company_relations() -> pd.DataFrame:
"""This Methods makes a Database Request for all Companies and their relations, modifies the ID Column and returns the Result as an DataFrame.
Returns:
DataFrame: DataFrame with all Relations between Companies.
"""
session = SessionHandler.session
assert session # noqa: S101
# Query to fetch relations between companies
relations_company_query = (
session.query(
to_company.id.label("id_company_to"),
to_company.name.label("name_company_to"),
entities.CompanyRelation.relation.label("relation_type"),
from_company.name.label("name_company_from"),
from_company.id.label("id_company_from"),
)
.join(
entities.CompanyRelation,
entities.CompanyRelation.company_id == to_company.id,
)
.join(
from_company,
entities.CompanyRelation.company2_id == from_company.id,
)
)
str(relations_company_query)
company_relations = pd.read_sql_query(str(relations_company_query), session.bind) # type: ignore
company_relations["id_company_from"] = company_relations["id_company_from"].apply(
lambda x: f"c_{x}"
)
company_relations["id_company_to"] = company_relations["id_company_to"].apply(
lambda x: f"c_{x}"
)
return company_relations
def get_all_person_relations() -> pd.DataFrame:
"""This Methods makes a Database Request for all Persons and their relations, modifies the ID Column and returns the Result as an DataFrame.
Returns:
DataFrame: DataFrame with all Relations between Persons and Companies.
"""
session = SessionHandler.session
assert session # noqa: S101
relations_person_query = (
session.query(
entities.Company.id.label("id_company"),
entities.Company.name.label("name_company"),
entities.PersonRelation.relation.label("relation_type"),
entities.Person.id.label("id_person"),
entities.Person.lastname.label("lastname"),
entities.Person.firstname.label("firstname"),
entities.Person.date_of_birth.label("date_of_birth"),
)
.join(
entities.PersonRelation,
entities.PersonRelation.company_id == entities.Company.id,
)
.join(
entities.Person,
entities.PersonRelation.person_id == entities.Person.id,
)
)
person_relations = pd.read_sql_query(str(relations_person_query), session.bind) # type: ignore
person_relations["id_company"] = person_relations["id_company"].apply(
lambda x: f"c_{x}"
)
person_relations["id_person"] = person_relations["id_person"].apply(
lambda x: f"p_{x}"
)
return person_relations
def filter_relation_type(
relation_dataframe: pd.DataFrame, selected_relation_type: str
) -> pd.DataFrame:
"""This Method filters the given DataFrame based on the selected Relation Type and returns it.
Args:
relation_dataframe (pd.DataFrame): DataFrame must contain a column named "relation_type".
selected_relation_type (str): String with the filter value.
Returns:
relation_dataframe (pd.DataFrame): The filtered DataFrame which now only contains entries with the selected Relation Type.
"""
return relation_dataframe.loc[
relation_dataframe["relation_type"] == selected_relation_type
]
def create_edge_and_node_list(
person_relations: pd.DataFrame, company_relations: pd.DataFrame
) -> tuple[dict, list]:
"""In this Method the given DataFrames with the relation will be morphed to Edge and Node lists and enhanced by a coloring for companies and person Nodes.
Args:
person_relations (pd.DataFrame): _description_
company_relations (pd.DataFrame): _description_
Returns:
_type_: _description_
"""
nodes: dict = {}
edges: list = []
# Iterate over person relations
for _, row in person_relations.iterrows():
if nodes.get(row["id_company"]) is None:
nodes[row["id_company"]] = {
"id": row["id_company"],
"name": row["name_company"],
"color": COLOR_COMPANY,
"type": "company",
}
if nodes.get(row["id_person"]) is None:
nodes[row["id_person"]] = {
"id": row["id_person"],
"name": str(row["firstname"]) + " " + str(row["lastname"]),
"date_of_birth": row["date_of_birth"],
"color": COLOR_PERSON,
"type": "person",
}
edges.append(
{
"from": row["id_person"],
"to": row["id_company"],
"type": row["relation_type"],
}
)
for _, row in company_relations.iterrows():
if nodes.get(row["id_company_from"]) is None: # noqa
nodes[row["id_company_from"]] = {
"id": row["id_company_from"],
"name": row["name_company_from"],
"color": COLOR_COMPANY,
"type": "company",
}
if nodes.get(row["id_company_to"]) is None:
nodes[row["id_company_to"]] = {
"id": row["id_company_to"],
"name": row["name_company_to"],
"color": COLOR_COMPANY,
"type": "company",
}
edges.append(
{
"from": row["id_company_from"],
"to": row["id_company_to"],
"type": row["relation_type"],
}
)
return nodes, edges
def find_company_relations(
selected_company_id: int,
) -> tuple[pd.DataFrame, pd.DataFrame]:
"""Finds all Relations for the given Company id.
Args:
selected_company_id: Id of the Company which Relations should be returned.
Returns:
Two Dataframes
"""
session = SessionHandler.session
assert session # noqa: S101
relations_company = (
session.query(
to_company.id.label("id_company_to"),
to_company.name.label("name_company_to"),
entities.CompanyRelation.relation.label("relation_type"),
from_company.name.label("name_company_from"),
from_company.id.label("id_company_from"),
)
.join(
entities.CompanyRelation,
entities.CompanyRelation.company_id == to_company.id,
)
.join(
from_company,
entities.CompanyRelation.company2_id == from_company.id,
)
.filter(
(from_company.id == selected_company_id)
| (to_company.id == selected_company_id)
)
.all()
)
company_relations = pd.DataFrame(
relations_company,
columns=[
"id_company_to",
"name_company_to",
"relation_type",
"name_company_from",
"id_company_from",
],
)
company_relations["id_company_from"] = company_relations["id_company_from"].apply(
lambda x: f"c_{x}"
)
company_relations["id_company_to"] = company_relations["id_company_to"].apply(
lambda x: f"c_{x}"
)
return pd.DataFrame(), company_relations # company_relations
def create_edge_and_node_list_for_company(
company_relations: pd.DataFrame,
) -> tuple[dict, list]:
"""In this Method the given DataFrames with the relation will be morphed to Edge and Node lists and enhanced by a coloring for companies and person Nodes.
Args:
person_relations (pd.DataFrame): _description_
company_relations (pd.DataFrame): _description_
Returns:
_type_: _description_
"""
nodes: dict = {}
edges: list = []
for _, row in company_relations.iterrows():
if nodes.get(row["id_company_from"]) is None:
nodes[row["id_company_from"]] = {
"id": row["id_company_from"],
"name": row["name_company_from"],
"color": COLOR_COMPANY,
"type": "company",
}
if nodes.get(row["id_company_to"]) is None:
nodes[row["id_company_to"]] = {
"id": row["id_company_to"],
"name": row["name_company_to"],
"color": COLOR_COMPANY,
"type": "company",
}
edges.append(
{
"from": row["id_company_from"],
"to": row["id_company_to"],
"type": row["relation_type"],
}
)
return nodes, edges
def get_all_metrics_from_id(company_id: int) -> pd.Series:
"""Returns all Metric for the given ID.
Args:
company_id (int): Id of the Company.
Returns:
pd.DataFrame: _description_
"""
# Get Data
person_df = get_all_person_relations()
company_df = get_all_company_relations()
# Create Edge and Node List from data
nodes_tmp, edges_tmp = create_edge_and_node_list(person_df, company_df)
graph, metrics = initialize_network_with_reduced_metrics(
nodes=nodes_tmp, edges=edges_tmp
)
filtered_metrics = metrics.loc[metrics["id"] == company_id]
if len(filtered_metrics) == 0:
return pd.Series([])
return filtered_metrics.iloc[0]
@lru_cache
def get_relations_number_from_id(id: int) -> tuple[int, int, int]:
"""Returns all Relation in 1, 2 and 3 lvl of one Node.
Args:
id (int): String of the Company or Person Id.
Returns:
tuple[int,int,int]: _description_
"""
# Get Data
person_df = get_all_person_relations()
company_df = get_all_company_relations()
# Create Edge and Node List from data
nodes_tmp, edges_tmp = create_edge_and_node_list(person_df, company_df)
graph = initialize_network_without_metrics(nodes=nodes_tmp, edges=edges_tmp)
neighbors = nx.all_neighbors(graph, id)
relations_lv1 = set(neighbors)
relations_lv2 = set()
relations_lv3 = set()
for node in relations_lv1:
relations_lv2 |= set(nx.all_neighbors(graph, node))
relations_lv2.discard(id)
for sub_node in relations_lv2:
relations_lv3 |= set(nx.all_neighbors(graph, sub_node))
relations_lv2.difference(relations_lv3)
return (len(relations_lv1), len(relations_lv2), len(relations_lv3))

12
tests/ui/company_elements_test.py
View File

@ -1,8 +1,20 @@
"""Tests for company elements."""
from collections.abc import Generator
import pytest
from sqlalchemy.orm import Session
from aki_prj23_transparenzregister.ui import company_elements, data_elements
from aki_prj23_transparenzregister.ui.session_handler import SessionHandler
@pytest.fixture(autouse=True)
def _set_session(full_db: Session) -> Generator[None, None, None]:
"""Sets a session for the dash application to be used."""
SessionHandler.session = full_db
yield
SessionHandler.session = None
def test_import() -> None:

195
tests/ui/home_page_test.py Normal file
View File

@ -0,0 +1,195 @@
"""Test for the Home Page."""
import datetime
from collections.abc import Generator
from unittest.mock import patch
import pandas as pd
import pytest
from aki_prj23_transparenzregister.ui.pages import home
def test_import() -> None:
"""Checks if an import co company_stats_dash can be made."""
assert home is not None
@pytest.mark.tim()
def test_person_relation_type_filter() -> None:
with patch(
"aki_prj23_transparenzregister.ui.pages.home.get_all_person_relations"
) as mock_filter:
data = [
{"relation_type": "Eigentümer"},
{"relation_type": "Inhaber"},
{"relation_type": "Eigentümer"},
]
mock_filter.return_value = pd.DataFrame(data)
assert list(home.person_relation_type_filter()) == ["Eigentümer", "Inhaber"]
@pytest.mark.tim()
def test_company_relation_type_filter() -> None:
with patch(
"aki_prj23_transparenzregister.ui.pages.home.get_all_company_relations"
) as mock_filter:
data = [
{"relation_type": "Eigentümer"},
{"relation_type": "Inhaber"},
{"relation_type": "Eigentümer"},
]
mock_filter.return_value = pd.DataFrame(data)
assert list(home.company_relation_type_filter()) == ["Eigentümer", "Inhaber"]
@pytest.mark.tim()
def test_update_table() -> None:
metrics = pd.DataFrame(
[
{
"designation": "Mustermann, Max",
"category": "Person",
"centrality": 3.14,
"betweenness": 42,
},
{
"designation": "Musterfrau, Martina",
"category": "Person",
"centrality": 42,
"betweenness": 3.14,
},
]
)
selected_metric = "centrality"
expected_result_df = [
{
"designation": "Musterfrau, Martina",
"category": "Person",
"centrality": 42.0,
},
{
"designation": "Mustermann, Max",
"category": "Person",
"centrality": 3.14,
},
]
expected_result_columns = [
{"name": "designation", "id": "designation"},
{"name": "category", "id": "category"},
{"name": "centrality", "id": "centrality"},
]
result_df, result_columns = home.update_table(selected_metric, metrics)
assert result_df == expected_result_df
assert result_columns == expected_result_columns
@pytest.fixture(scope="session", autouse=True)
def _get_person_relations() -> Generator:
data = [
{
"id_company": 1,
"name_company": "0 10 24 Telefondienste GmbH",
"relation_type": "GESCHAEFTSFUEHRER",
"id_person": 1,
"lastname": "Tetau",
"firstname": "Nicolas",
"date_of_birth": datetime.date(1971, 1, 2),
},
{
"id_company": 1,
"name_company": "0 10 24 Telefondienste GmbH",
"relation_type": "PROKURIST",
"id_person": 2,
"lastname": "Dammast",
"firstname": "Lutz",
"date_of_birth": datetime.date(1966, 12, 6),
},
{
"id_company": 2,
"name_company": "1. Staiger Grundstücksverwaltung GmbH & Co. KG",
"relation_type": "KOMMANDITIST",
"id_person": 3,
"lastname": "Tutsch",
"firstname": "Rosemarie",
"date_of_birth": datetime.date(1941, 10, 9),
},
{
"id_company": 2,
"name_company": "1. Staiger Grundstücksverwaltung GmbH & Co. KG",
"relation_type": "HAFTENDER_GESELLSCHAFTER",
"id_person": 4,
"lastname": "Staiger",
"firstname": "Marc",
"date_of_birth": datetime.date(1969, 10, 22),
},
{
"id_company": 2,
"name_company": "1. Staiger Grundstücksverwaltung GmbH & Co. KG",
"relation_type": "HAFTENDER_GESELLSCHAFTER",
"id_person": 5,
"lastname": "Staiger",
"firstname": "Michaela",
"date_of_birth": datetime.date(1971, 3, 3),
},
]
with patch(
"aki_prj23_transparenzregister.ui.pages.home.get_all_person_relations"
) as mock_get_person_relations:
mock_get_person_relations.return_value = pd.DataFrame(data)
yield
@pytest.fixture(scope="session", autouse=True)
def _get_company_relations() -> Generator:
data = [
{
"id_company_to": 2,
"name_company_to": "1. Staiger Grundstücksverwaltung GmbH & Co. KG",
"relation_type": "GESCHAEFTSFUEHRER",
"name_company_from": "Staiger I. Verwaltung-GmbH",
"id_company_from": 3226,
},
{
"id_company_to": 3,
"name_company_to": "1 A Autenrieth Kunststofftechnik GmbH & Co. KG",
"relation_type": "GESCHAEFTSFUEHRER",
"name_company_from": "Autenrieth Verwaltungs-GmbH",
"id_company_from": 3324,
},
{
"id_company_to": 5,
"name_company_to": "2. Schaper Objekt GmbH & Co. Kiel KG",
"relation_type": "KOMMANDITIST",
"name_company_from": "Multi-Center Warenvertriebs GmbH",
"id_company_from": 2213,
},
{
"id_company_to": 6,
"name_company_to": "AASP Filmproduktionsgesellschaft mbH & Co. Leonie KG",
"relation_type": "INHABER",
"name_company_from": "ABN AMRO Structured Products Gesellschaft für Fondsbeteiligungen mbH",
"id_company_from": 3332,
},
{
"id_company_to": 6,
"name_company_to": "AASP Filmproduktionsgesellschaft mbH & Co. Leonie KG",
"relation_type": "KOMMANDITIST",
"name_company_from": "Kallang GmbH",
"id_company_from": 3316,
},
]
with patch(
"aki_prj23_transparenzregister.ui.pages.home.get_all_company_relations"
) as mock_get_person_relations:
mock_get_person_relations.return_value = pd.DataFrame(data)
yield
@pytest.mark.tim()
def test_update_graph_data() -> None:
graph_result, metrics_result, nodes_result, edges_result = home.update_graph_data(
"HAFTENDER_GESELLSCHAFTER", "GESCHAEFTSFUEHRER"
)
assert graph_result is not None

103
tests/utils/networkx/network_2d_test.py Normal file
View File

@ -0,0 +1,103 @@
"""Test the initialize Network function."""
import datetime
from collections.abc import Generator
import plotly.graph_objects as go
import pytest
from sqlalchemy.orm import Session
from aki_prj23_transparenzregister.ui.session_handler import SessionHandler
from aki_prj23_transparenzregister.utils.networkx import network_2d
from aki_prj23_transparenzregister.utils.networkx.network_2d import create_2d_graph
from aki_prj23_transparenzregister.utils.networkx.network_base import initialize_network
def test_import() -> None:
"""Checks if an import co company_stats_dash can be made."""
assert network_2d is not None
@pytest.fixture(autouse=True)
def _set_session(full_db: Session) -> Generator[None, None, None]:
"""Sets a session for the dash application to be used."""
SessionHandler.session = full_db
yield
SessionHandler.session = None
def test_create_2d_graph() -> None:
"""Tests the creation of a 2D Graph."""
edges: list = [
{"from": "p_545", "to": "c_53", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "c_53", "to": "p_545", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "c_1", "to": "c_2", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "c_53", "to": "c_1", "type": "HAFTENDER_GESELLSCHAFTER"},
]
nodes: dict = {
"c_53": {
"id": "c_53",
"name": "1. Freiburger Solarfonds Beteiligungs-KG",
"type": "company",
"color": "blue",
},
"p_545": {
"id": "p_545",
"name": "Wetzel, Jürgen",
"type": "person",
"date_of_birth": datetime.date(1962, 11, 15),
"color": "red",
},
"c_1": {
"id": "c_1",
"name": "Musterfirma",
"type": "company",
"color": "blue",
},
"c_2": {
"id": "c_2",
"name": "Firma 1",
"type": "company",
"color": "blue",
},
}
graph, metrics = initialize_network(edges=edges, nodes=nodes)
metric = "None"
layout = "Spring"
edge_annotation = False
edge_thickness = 1
figure = create_2d_graph(
graph, nodes, edges, metrics, metric, layout, edge_annotation, edge_thickness
)
assert type(figure) is go.Figure
metric = "degree"
layout = "Circular"
figure = create_2d_graph(
graph, nodes, edges, metrics, metric, layout, edge_annotation, edge_thickness
)
assert type(figure) is go.Figure
edge_annotation = True
layout = "Kamada Kawai"
figure = create_2d_graph(
graph, nodes, edges, metrics, metric, layout, edge_annotation, edge_thickness
)
assert type(figure) is go.Figure
layout = "Random"
figure = create_2d_graph(
graph, nodes, edges, metrics, metric, layout, edge_annotation, edge_thickness
)
assert type(figure) is go.Figure
layout = "Shell (only 2D)"
figure = create_2d_graph(
graph, nodes, edges, metrics, metric, layout, edge_annotation, edge_thickness
)
assert type(figure) is go.Figure
layout = "Spiral (only 2D)"
figure = create_2d_graph(
graph, nodes, edges, metrics, metric, layout, edge_annotation, edge_thickness
)
assert type(figure) is go.Figure

79
tests/utils/networkx/network_3d_test.py Normal file
View File

@ -0,0 +1,79 @@
"""Test the initialize Network function."""
import datetime
import plotly.graph_objects as go
from aki_prj23_transparenzregister.utils.networkx import network_3d
from aki_prj23_transparenzregister.utils.networkx.network_3d import create_3d_graph
from aki_prj23_transparenzregister.utils.networkx.network_base import initialize_network
def test_import() -> None:
"""Checks if an import co company_stats_dash can be made."""
assert network_3d is not None
def test_create_3d_graph() -> None:
"""Tests the creation of a 3D Graph."""
edges: list = [
{"from": "p_545", "to": "c_53", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "c_53", "to": "p_545", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "c_1", "to": "c_2", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "c_53", "to": "c_1", "type": "HAFTENDER_GESELLSCHAFTER"},
]
nodes: dict = {
"c_53": {
"id": "c_53",
"name": "1. Freiburger Solarfonds Beteiligungs-KG",
"type": "company",
"color": "blue",
},
"p_545": {
"id": "p_545",
"name": "Wetzel, Jürgen",
"type": "person",
"date_of_birth": datetime.date(1962, 11, 15),
"color": "red",
},
"c_1": {
"id": "c_1",
"name": "Musterfirma",
"type": "company",
"color": "blue",
},
"c_2": {
"id": "c_2",
"name": "Firma 1",
"type": "company",
"color": "blue",
},
}
graph, metrics = initialize_network(edges=edges, nodes=nodes)
metric = "None"
layout = "Spring"
edge_annotation = False
edge_thickness = 1
figure = create_3d_graph(
graph, nodes, edges, metrics, metric, layout, edge_annotation, edge_thickness
)
assert type(figure) is go.Figure
metric = "degree"
layout = "Circular"
figure = create_3d_graph(
graph, nodes, edges, metrics, metric, layout, edge_annotation, edge_thickness
)
assert type(figure) is go.Figure
edge_annotation = True
layout = "Kamada Kawai"
figure = create_3d_graph(
graph, nodes, edges, metrics, metric, layout, edge_annotation, edge_thickness
)
assert type(figure) is go.Figure
layout = "Random"
figure = create_3d_graph(
graph, nodes, edges, metrics, metric, layout, edge_annotation, edge_thickness
)
assert type(figure) is go.Figure

85
tests/utils/networkx/network_base_test.py Normal file
View File

@ -0,0 +1,85 @@
"""Test the initialize Network function."""
import datetime
import networkx as nx
import pandas as pd
from aki_prj23_transparenzregister.utils.networkx import network_base
from aki_prj23_transparenzregister.utils.networkx.network_base import (
initialize_network,
initialize_network_with_reduced_metrics,
initialize_network_without_metrics,
)
def test_import() -> None:
"""Checks if an import co company_stats_dash can be made."""
assert network_base is not None
def test_initialize_network() -> None:
edges: list = [
{"from": "p_545", "to": "c_53", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "c_53", "to": "p_545", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "c_1", "to": "c_2", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "c_53", "to": "c_1", "type": "HAFTENDER_GESELLSCHAFTER"},
]
nodes: dict = {
"c_53": {
"id": "c_53",
"name": "1. Freiburger Solarfonds Beteiligungs-KG",
"type": "company",
"color": "blue",
},
"p_545": {
"id": "p_545",
"name": "Wetzel, Jürgen",
"type": "person",
"date_of_birth": datetime.date(1962, 11, 15),
"color": "red",
},
"c_1": {
"id": "c_1",
"name": "Musterfirma",
"type": "company",
"color": "blue",
},
"c_2": {
"id": "c_2",
"name": "Firma 1",
"type": "company",
"color": "blur",
},
}
# print(len(edges))
# print(len(nodes))
graph, metrics = initialize_network(edges=edges, nodes=nodes)
assert isinstance(graph, nx.Graph)
assert isinstance(metrics, pd.DataFrame)
assert list(metrics.columns) == [
"eigenvector",
"degree",
"betweenness",
"closeness",
"pagerank",
"category",
"designation",
"id",
]
graph_reduced, metrics_reduced = initialize_network_with_reduced_metrics(
edges=edges, nodes=nodes
)
assert isinstance(graph_reduced, nx.Graph)
assert isinstance(metrics_reduced, pd.DataFrame)
assert list(metrics_reduced.columns) == [
"degree",
"betweenness",
"closeness",
"category",
"designation",
"id",
]
graph = initialize_network_without_metrics(edges=edges, nodes=nodes)
assert isinstance(graph_reduced, nx.Graph)

240
tests/utils/networkx/networkx_data_test.py Normal file
View File

@ -0,0 +1,240 @@
"""Test the initialize Network function."""
import datetime
from collections.abc import Generator
from unittest.mock import patch
import pandas as pd
import pytest
from sqlalchemy.orm import Session
from aki_prj23_transparenzregister.ui.session_handler import SessionHandler
from aki_prj23_transparenzregister.utils.networkx import networkx_data
@pytest.fixture(scope="session", autouse=True)
def _get_person_relations() -> Generator:
data = [
{
"id_company": 1,
"name_company": "0 10 24 Telefondienste GmbH",
"relation_type": "GESCHAEFTSFUEHRER",
"id_person": 1,
"lastname": "Tetau",
"firstname": "Nicolas",
"date_of_birth": datetime.date(1971, 1, 2),
},
{
"id_company": 1,
"name_company": "0 10 24 Telefondienste GmbH",
"relation_type": "PROKURIST",
"id_person": 2,
"lastname": "Dammast",
"firstname": "Lutz",
"date_of_birth": datetime.date(1966, 12, 6),
},
{
"id_company": 2,
"name_company": "1. Staiger Grundstücksverwaltung GmbH & Co. KG",
"relation_type": "KOMMANDITIST",
"id_person": 3,
"lastname": "Tutsch",
"firstname": "Rosemarie",
"date_of_birth": datetime.date(1941, 10, 9),
},
{
"id_company": 2,
"name_company": "1. Staiger Grundstücksverwaltung GmbH & Co. KG",
"relation_type": "KOMMANDITIST",
"id_person": 4,
"lastname": "Staiger",
"firstname": "Marc",
"date_of_birth": datetime.date(1969, 10, 22),
},
{
"id_company": 2,
"name_company": "1. Staiger Grundstücksverwaltung GmbH & Co. KG",
"relation_type": "KOMMANDITIST",
"id_person": 5,
"lastname": "Staiger",
"firstname": "Michaela",
"date_of_birth": datetime.date(1971, 3, 3),
},
]
with patch(
"aki_prj23_transparenzregister.utils.networkx.networkx_data.get_all_person_relations"
) as mock_get_person_relations:
mock_get_person_relations.return_value = pd.DataFrame(data)
yield
@pytest.fixture(scope="session", autouse=True)
def _get_company_relations() -> Generator:
data = [
{
"id_company_to": 2,
"name_company_to": "1. Staiger Grundstücksverwaltung GmbH & Co. KG",
"relation_type": "HAFTENDER_GESELLSCHAFTER",
"name_company_from": "Staiger I. Verwaltung-GmbH",
"id_company_from": 3226,
},
{
"id_company_to": 3,
"name_company_to": "1 A Autenrieth Kunststofftechnik GmbH & Co. KG",
"relation_type": "HAFTENDER_GESELLSCHAFTER",
"name_company_from": "Autenrieth Verwaltungs-GmbH",
"id_company_from": 3324,
},
{
"id_company_to": 5,
"name_company_to": "2. Schaper Objekt GmbH & Co. Kiel KG",
"relation_type": "KOMMANDITIST",
"name_company_from": "Multi-Center Warenvertriebs GmbH",
"id_company_from": 2213,
},
{
"id_company_to": 6,
"name_company_to": "AASP Filmproduktionsgesellschaft mbH & Co. Leonie KG",
"relation_type": "INHABER",
"name_company_from": "ABN AMRO Structured Products Gesellschaft für Fondsbeteiligungen mbH",
"id_company_from": 3332,
},
{
"id_company_to": 6,
"name_company_to": "AASP Filmproduktionsgesellschaft mbH & Co. Leonie KG",
"relation_type": "KOMMANDITIST",
"name_company_from": "Kallang GmbH",
"id_company_from": 3316,
},
]
with patch(
"aki_prj23_transparenzregister.utils.networkx.networkx_data.get_all_company_relations"
) as mock_get_person_relations:
mock_get_person_relations.return_value = pd.DataFrame(data)
yield
@pytest.fixture(autouse=True)
def _set_session(full_db: Session) -> Generator[None, None, None]:
"""Sets a session for the dash application to be used."""
SessionHandler.session = full_db
yield
SessionHandler.session = None
def test_import() -> None:
"""Checks if an import co company_stats_dash can be made."""
assert networkx_data is not None
# def test_initialize_network() -> None:
# edges: list = [
# {"from": "p_545", "to": "c_53", "type": "HAFTENDER_GESELLSCHAFTER"},
# {"from": "c_53", "to": "p_545", "type": "HAFTENDER_GESELLSCHAFTER"},
# {"from": "c_1", "to": "c_2", "type": "HAFTENDER_GESELLSCHAFTER"},
# {"from": "c_53", "to": "c_1", "type": "HAFTENDER_GESELLSCHAFTER"},
# ]
# nodes: dict = {
# "c_53": {
# "id": "c_53",
# "name": "1. Freiburger Solarfonds Beteiligungs-KG",
# "type": "company",
# "color": "blue",
# },
# "p_545": {
# "id": "p_545",
# "name": "Wetzel, Jürgen",
# "type": "person",
# "date_of_birth": datetime.date(1962, 11, 15),
# "color": "red",
# },
# "c_1": {
# "id": "c_1",
# "name": "Musterfirma",
# "type": "company",
# "color": "blue",
# },
# "c_2": {
# "id": "c_2",
# "name": "Firma 1",
# "type": "company",
# "color": "blur",
# },
# }
def test_find_all_company_relations() -> None:
"""This Test methods tests if the correct type is returned for the corresponding Function."""
company_relations_df = networkx_data.find_all_company_relations()
assert type(company_relations_df) is pd.DataFrame
def test_get_all_company_relations() -> None:
"""This Test methods tests if the correct type is returned for the corresponding Function."""
company_relations_df = networkx_data.get_all_company_relations()
assert type(company_relations_df) is pd.DataFrame
def test_get_all_person_relations() -> None:
"""This Test methods tests if the correct type is returned for the corresponding Function."""
company_relations_df = networkx_data.get_all_person_relations()
assert type(company_relations_df) is pd.DataFrame
def test_filter_relation_type() -> None:
"""This Test methods tests if the correct type is returned for the corresponding Function."""
relation_dataframe = networkx_data.get_all_company_relations()
selected_relation_type = "HAFTENDER_GESELLSCHAFTER"
company_relations_df = networkx_data.filter_relation_type(
relation_dataframe, selected_relation_type
)
assert type(company_relations_df) is pd.DataFrame
def test_create_edge_and_node_list() -> None:
"""This Test methods tests if the correct type is returned for the corresponding Function."""
person_df = networkx_data.get_all_person_relations()
company_df = networkx_data.get_all_company_relations()
nodes, edges = networkx_data.create_edge_and_node_list(person_df, company_df)
assert isinstance(nodes, dict)
assert isinstance(edges, list)
def test_find_company_relations() -> None:
"""This Test methods tests if the correct type is returned for the corresponding Function."""
selected_company_id = 1
company_relations_df, person_df = networkx_data.find_company_relations(
selected_company_id
)
assert type(company_relations_df) is pd.DataFrame
assert type(person_df) is pd.DataFrame
def test_create_edge_and_node_list_for_company() -> None:
"""This Test methods tests if the correct type is returned for the corresponding Function."""
company_relations = networkx_data.get_all_company_relations()
nodes, edges = networkx_data.create_edge_and_node_list_for_company(
company_relations
)
assert isinstance(nodes, dict)
assert isinstance(edges, list)
def test_get_all_metrics_from_id() -> None:
"""This Test methods tests if the correct type is returned for the corresponding Function."""
company_id = 2
metrics = networkx_data.get_all_metrics_from_id(company_id)
assert type(metrics) is pd.Series
def test_get_relations_number_from_id() -> None:
"""This Test methods tests if the correct type and number of relations is received."""
# id = "c_2549"
id = 2
(
relations_lvl_1,
relations_lvl_2,
relations_lvl_3,
) = networkx_data.get_relations_number_from_id(id)
assert isinstance(relations_lvl_1, int)
assert isinstance(relations_lvl_2, int)
assert isinstance(relations_lvl_3, int)