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

Bug fixes

Browse Source
This commit is contained in:
Tim
2023-11-04 13:23:31 +01:00
parent 31d7098d48
commit 152743597e
5 changed files with 208 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,11 +1,10 @@
"""Content of home page."""
from functools import lru_cache
import dash
import dash_daq as daq
import networkx as nx
import pandas as pd
import plotly.graph_objects as go
from dash import Input, Output, callback, dash_table, dcc, html, ctx
from dash import Input, Output, callback, dash_table, dcc, html
from aki_prj23_transparenzregister.utils.networkx.network_2d import (
create_2d_graph,
@ -21,18 +20,14 @@ from aki_prj23_transparenzregister.utils.networkx.networkx_data import (
create_edge_and_node_list,
filter_relation_type,
filter_relation_with_more_than_one_connection,
find_top_companies,
get_all_company_relations,
get_all_person_relations,
return_metric_table_df
)
dash.register_page(__name__, path="/")
# Get Data
person_relation = filter_relation_type(
get_all_person_relations(), "HAFTENDER_GESELLSCHAFTER"
)
person_relation = filter_relation_type(get_all_person_relations(), "NACHFOLGER")
company_relation = filter_relation_with_more_than_one_connection(
get_all_company_relations(), "id_company_to", "id_company_from"
)
@ -67,13 +62,33 @@ layout = "Spring"
# switch_node_annotaion_value = False
switch_edge_annotaion_value = False
egde_thickness = 1
network = create_3d_graph(graph, nodes, edges, metrics, metric, layout, switch_edge_annotaion_value, egde_thickness)
network = create_3d_graph(
graph,
nodes,
edges,
metrics,
metric,
layout,
switch_edge_annotaion_value,
egde_thickness,
)
# Get the possible Filter values for the Dropdowns.
person_relation_type_filter = get_all_person_relations()["relation_type"].unique()
company_relation_type_filter = get_all_company_relations()["relation_type"].unique()
top_companies_df = return_metric_table_df(metrics, nodes, "closeness") #find_top_companies()
def update_table(
metric_dropdown_value: str, metrics: pd.DataFrame
) -> tuple[dict, list]:
table_df = metrics.sort_values(metric_dropdown_value, ascending=False).head(10)
table_df = table_df[["designation", "category", metric_dropdown_value]]
table_df.to_dict("records")
columns = [{"name": i, "id": i} for i in table_df.columns]
return table_df.to_dict("records"), columns # type: ignore
top_companies_dict, top_companies_columns = update_table("closeness", metrics)
layout = html.Div(
children=html.Div(
@ -81,30 +96,32 @@ layout = html.Div(
html.Div(
className="top_companytable_style",
children=[
html.H1(title="Top Ten Nodes in Graph by Metric", style={"align": "mid"}),
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",
"betweeness",
"closeness",
],
"closeness",
id="dropdown_table_metric",
className="dropdown_style",
),
],
className="filter-wrapper-item",
children=[
html.H5(
className="filter-description",
children=["Filter Metric:"],
),
dash_table.DataTable(
top_companies_df.to_dict("records"),
[{"name": i, "id": i} for i in top_companies_df.columns],
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",
),
],
@ -135,7 +152,7 @@ layout = html.Div(
# ),
html.Div(
className="filter-wrapper",
id= "company_dropdown",
id="company_dropdown",
# style="visibility: hidden;",
children=[
html.Div(
@ -181,7 +198,7 @@ layout = html.Div(
"None",
"eigenvector",
"degree",
"betweeness",
"betweenness",
"closeness",
],
"None",
@ -219,15 +236,19 @@ layout = html.Div(
html.Div(
className="filter-wrapper-item",
children=[
html.H5(
className="filter-description",
children=["Adjust Edge Thickness"],
),
dcc.Slider( 1, 4, 1,
html.H5(
className="filter-description",
children=["Adjust Edge Thickness"],
),
dcc.Slider(
1,
4,
1,
value=1,
id='slider',
),
],),
id="slider",
),
],
),
],
),
html.Div(
@ -242,7 +263,9 @@ layout = html.Div(
),
html.Div(
className="switch-style",
children=[daq.BooleanSwitch(id="switch", on=False)],
children=[
daq.BooleanSwitch(id="switch", on=False)
],
),
],
),
@ -268,13 +291,16 @@ layout = html.Div(
),
html.Div(
className="switch-style",
children=[daq.BooleanSwitch(id="switch_edge_annotation", on=False)],
children=[
daq.BooleanSwitch(
id="switch_edge_annotation", on=False
)
],
),
],
),
],
),
dcc.Graph(figure=network, id="my-graph", className="graph-style"),
],
),
@ -282,11 +308,12 @@ layout = html.Div(
)
)
# @lru_cache(200)
def update_graph_data(
person_relation_type: str = "HAFTENDER_GESELLSCHAFTER",
company_relation_type: str = "GESCHAEFTSFUEHRER",
) -> tuple[nx.Graph, pd.DataFrame]:
) -> tuple[nx.Graph, pd.DataFrame, dict, list]:
# Get Data
person_df = get_all_person_relations()
company_df = get_all_company_relations()
@ -304,7 +331,11 @@ def update_graph_data(
@callback(
Output("my-graph", "figure"),
[
Output("metric_table", "data"),
Output("metric_table", "columns"),
Output("my-graph", "figure"),
],
[
Input("dropdown", "value"),
Input("switch", "on"),
@ -313,7 +344,8 @@ def update_graph_data(
Input("dropdown_company_relation_filter", "value"),
Input("dropdown_person_relation_filter", "value"),
Input("dropdown_layout", "value"),
Input('slider', 'value')
Input("slider", "value"),
Input("dropdown_table_metric", "value"),
],
prevent_initial_call=True,
allow_duplicate=True,
@ -327,7 +359,8 @@ def update_figure(
c_relation_filter_value: str,
p_relation_filter_value: str,
layout: str,
slider_value: float
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.
@ -342,33 +375,51 @@ def update_figure(
"""
_ = 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)
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 create_2d_graph(graph, nodes, edges, metrics, selected_metric, layout, switch_edge_annotaion_value, slider_value)
return (
table_dict,
table_columns,
create_2d_graph(
graph,
nodes,
edges,
metrics,
selected_metric,
layout,
switch_edge_annotaion_value,
slider_value,# type: ignore
),
)
else:
return create_3d_graph(graph, nodes, edges, metrics, selected_metric, layout, switch_edge_annotaion_value, slider_value)
return (
table_dict,
table_columns,
create_3d_graph(
graph,
nodes,
edges,
metrics,
selected_metric,
layout,
switch_edge_annotaion_value,
slider_value, # type: ignore
),
)
@callback(
Output("metric_table", "data"),
Output("company_dropdown", "style"),
[
Input("dropdown_table_metric", "value"),
Input("dropdown_data_soruce_filter", "value"),
],
)
def update_table(metric_dropdown_value: str) -> dict:
table_df = return_metric_table_df(metrics, nodes, metric_dropdown_value)
table_df.to_dict("records")
columns =[{"name": i, "id": i} for i in table_df.columns]
return table_df.to_dict("records")
@callback(
Output("company_dropdown", "style"),
[
Input("dropdown_data_soruce_filter", "value"),
],
)
def update_Dropdown(datasource_value: str) -> str:
style = ""
match datasource_value:
@ -378,4 +429,4 @@ def update_Dropdown(datasource_value: str) -> str:
style = "visibility: hidden;"
case "Company & Person Data":
style = "visibility: visible;"
return style
return style

67
src/aki_prj23_transparenzregister/utils/networkx/network_2d.py
View File

@ -6,7 +6,14 @@ import plotly.graph_objects as go
def create_2d_graph(
graph: nx.Graph, nodes: dict, edges: list, metrics: pd.DataFrame, metric: str | None, layout: str, edge_annotation: bool, edge_thickness: int
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.
@ -20,7 +27,7 @@ def create_2d_graph(
Returns:
_type_: Plotly Figure
"""
# Set 2D Layout
# Set 2D Layout
pos = nx.spring_layout(graph)
match layout:
case "Spring":
@ -66,8 +73,8 @@ def create_2d_graph(
edge_y.append(y1)
# edge_y.append(None)
edge_weight_x.append(((x1 + x0) / 2))
edge_weight_y.append(((y1 + y0) / 2))
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(
@ -115,16 +122,22 @@ def create_2d_graph(
},
)
# Set Color by using the nodes DataFrame with its Color Attribute. The sequence matters!
# # Set Color by using the nodes DataFrame with its Color Attribute. The sequence matters!
colors = list(nx.get_node_attributes(graph, "color").values())
# 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.
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
@ -139,9 +152,6 @@ def create_2d_graph(
edge_type_list.append(row["type"])
edge_weights_trace.text = edge_type_list
# print(edge_type_list)
# Return the Plotly Figure
return go.Figure(
@ -152,18 +162,19 @@ def create_2d_graph(
showlegend=False,
hovermode="closest",
margin={"b": 20, "l": 5, "r": 5, "t": 20},
annotations=[{
"showarrow": False,
"text": f"Companies (Blue) & Person (Red) 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},
}
],
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},
),

42
src/aki_prj23_transparenzregister/utils/networkx/network_3d.py
View File

@ -6,7 +6,14 @@ import plotly.graph_objects as go
def create_3d_graph(
graph: nx.Graph, nodes: dict, edges: list, metrics: pd.DataFrame, metric: str | None, layout: str, edge_annotation: bool, edge_thickness: int
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.
@ -26,7 +33,7 @@ def create_3d_graph(
case "Spring":
pos = nx.spring_layout(graph, dim=3)
# case "Bipartite":
# pos = nx.bipartite_layout(graph, dim=3)
# pos = nx.bipartite_layout(graph, dim=3)
case "Circular":
pos = nx.circular_layout(graph, dim=3)
case "Kamada Kawai":
@ -72,11 +79,11 @@ def create_3d_graph(
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)
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(
@ -147,7 +154,7 @@ def create_3d_graph(
annotations=[
{
"showarrow": False,
"text": f"Companies (Blue) & Person (Red) Relation \n node_count: {len(nodes)}, edge_count: {len(edges)}",
"text": f"Companies (Red) & Person (Blue) Relation \n node_count: {len(nodes)}, edge_count: {len(edges)}",
"xref": "paper",
"yref": "paper",
"x": 0,
@ -161,13 +168,14 @@ def create_3d_graph(
# 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 = []
for key, value in nodes.items():
if "name" in value:
node_names.append(value["name"])
else:
node_names.append(value["firstname"] + " " + value["lastname"])
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
@ -193,6 +201,10 @@ def create_3d_graph(
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]
data = [edge_trace, edge_weights_trace, node_trace]
return go.Figure(data=data, layout=layout)

6
src/aki_prj23_transparenzregister/utils/networkx/network_base.py
View File

@ -25,12 +25,14 @@ def initialize_network(edges: list, nodes: dict) -> tuple[nx.Graph, pd.DataFrame
# Create a DataFrame with all Metrics
metrics = pd.DataFrame(
columns=["degree", "eigenvector", "betweeness", "closeness", "pagerank"]
columns=["degree", "eigenvector", "betweenness", "closeness", "pagerank"]
)
metrics["eigenvector"] = nx.eigenvector_centrality(graph).values()
metrics["degree"] = nx.degree_centrality(graph).values()
metrics["betweeness"] = nx.betweenness_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()
return graph, metrics

67
src/aki_prj23_transparenzregister/utils/networkx/networkx_data.py
View File

@ -1,6 +1,5 @@
"""Module to receive and filter Data for working with NetworkX."""
import pandas as pd
from loguru import logger
from sqlalchemy.orm import aliased
from aki_prj23_transparenzregister.config.config_providers import JsonFileConfigProvider
@ -97,7 +96,7 @@ def get_all_company_relations() -> pd.DataFrame:
)
)
str(relations_company_query)
company_relations = pd.read_sql_query(str(relations_company_query), session.bind) # type: ignore
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}"
@ -134,7 +133,7 @@ def get_all_person_relations() -> pd.DataFrame:
entities.PersonRelation.person_id == entities.Person.id,
)
)
person_relations = pd.read_sql_query(str(relations_person_query), session.bind) # type: ignore
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}"
@ -158,7 +157,6 @@ def filter_relation_type(
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
]
@ -191,7 +189,7 @@ def filter_relation_with_more_than_one_connection(
if count > 1:
# tmp_df = pd.concat([tmp_df, pd.DataFrame(row)])+
tmp_df.loc[len(tmp_df)] = row
tmp_df.loc[len(tmp_df)] = row # type: ignore
count = 0
else:
count = 0
@ -225,14 +223,15 @@ def create_edge_and_node_list(
"id": row["id_company"],
"name": row["name_company"],
"color": COLOR_COMPANY,
"type": "company",
}
if node := nodes.get(row["id_person"]) is None:
nodes[row["id_person"]] = {
"id": row["id_person"],
"firstname": row["firstname"],
"lastname": row["lastname"],
"name": str(row["firstname"]) + " " + str(row["lastname"]),
"date_of_birth": row["date_of_birth"],
"color": COLOR_PERSON,
"type": "person",
}
edges.append(
{
@ -248,12 +247,14 @@ def create_edge_and_node_list(
"id": row["id_company_from"],
"name": row["name_company_from"],
"color": COLOR_COMPANY,
"type": "company",
}
if node := 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(
{
@ -262,12 +263,13 @@ def create_edge_and_node_list(
"type": row["relation_type"],
}
)
return nodes, edges
def find_company_relations(
selected_company_id: int,
) -> tuple[pd.DataFrame, pd.DataFrame()]:
) -> tuple[pd.DataFrame, pd.DataFrame]:
relations_company_query = (
session.query(
to_company.id.label("id_company_to"),
@ -284,7 +286,10 @@ def find_company_relations(
from_company,
entities.CompanyRelation.company2_id == from_company.id,
)
.filter((from_company.id == selected_company_id) | (to_company.id == selected_company_id))
.filter(
(from_company.id == selected_company_id)
| (to_company.id == selected_company_id)
)
)
company_relations = pd.DataFrame(relations_company_query.all())
# logger.debug(str(relations_company_query))
@ -305,7 +310,7 @@ def find_company_relations(
"id_company_from",
],
)
company_relations["id_company_from"] = company_relations["id_company_from"].apply(
lambda x: f"c_{x}"
)
@ -356,40 +361,7 @@ def create_edge_and_node_list_for_company(
return nodes, edges
def return_metric_table_df(metrics: pd.DataFrame, nodes: dict, metric: str)-> pd.DataFrame:
"""_summary_
Args:
metrics (pd.DataFrame): _description_
nodes (dict): _description_
metric (str): _description_
Returns:
pd.DataFrame: _description_
"""
# tmp = pd.DataFrame(columns=["Platzierung", "company_name", "Umsatz M€"])
tmp_list = []
category = []
for key, values in nodes.items():
# print(values["id"])
if str(values["id"]).split("_")[0] == "c":
tmp_list.append(values["name"])
category.append("company")
if str(values["id"]).split("_")[0] == "p":
tmp_list.append(str(values["firstname"]) + " " + str(values["lastname"]))
category.append("person")
metrics["designation"] = tmp_list
metrics["category"] = category
tmp_df = metrics.sort_values(metric, ascending=False)
tmp_df = tmp_df[["designation", metric, "category"]].head(10)
tmp_df.rename(columns={metric: "Metric"}, inplace=True)
# print(tmp_df[["designation", metric, "category"]].head(10))
return tmp_df
def get_all_metrics_from_id(company_id: int)-> pd.DataFrame:
def get_all_metrics_from_id(company_id: int) -> pd.DataFrame:
"""_summary_
Args:
@ -398,9 +370,10 @@ def get_all_metrics_from_id(company_id: int)-> pd.DataFrame:
Returns:
pd.DataFrame: _description_
"""
return pd.DataFrame
return pd.DataFrame()
def get_relations_number_from_id(company_id: int)->tuple[int,int,int]:
def get_relations_number_from_id(company_id: int) -> tuple[int, int, int]:
"""_summary_
Args:
@ -409,4 +382,4 @@ def get_relations_number_from_id(company_id: int)->tuple[int,int,int]:
Returns:
tuple[int,int,int]: _description_
"""
return (1,2,3)
return (1, 2, 3)