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

Integrated NetworkX graphs into App

Browse Source
This commit is contained in:
Tim
2023-09-26 20:17:56 +02:00
parent 6a313f9803
commit e81b5fb518
4 changed files with 470 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
src/aki_prj23_transparenzregister/ui/networkx_dash.py
View File

@ -1,7 +1,7 @@
import pandas as pd
import networkx as nx
import plotly.graph_objects as go
from dash import Dash, Input, Output, dcc, html
from dash import Dash, Input, Output, dcc, html, callback
from aki_prj23_transparenzregister.config.config_providers import JsonFileConfigProvider
from aki_prj23_transparenzregister.utils.sql import connector, entities
@ -32,10 +32,10 @@ def find_company_relations(company_id: int) -> pd.DataFrame:
return companies_relations_df
# Plotly figure
def networkGraph(EGDE_VAR: None) -> go.Figure:
def networkGraph(company_id: int) -> go.Figure:
# df = find_company_relations(test_company)
edges = []
for index, row in find_company_relations(test_company).iterrows():
for index, row in find_company_relations(company_id).iterrows():
edges.append([row["company_name"], row["connected_company_name"]])
# print(row["company_name"], row["connected_company_name"])
# print(edges)
@ -113,27 +113,36 @@ def networkGraph(EGDE_VAR: None) -> go.Figure:
# Dash App
app = Dash(__name__)
# app = Dash(__name__)
app.title = "Dash Networkx"
# app.title = "Dash Networkx"
app.layout = html.Div(
# app.layout = html.Div(
# [
# html.I("Write your EDGE_VAR"),
# html.Br(),
# dcc.Input(id="EGDE_VAR", type="text", value="K", debounce=True),
# dcc.Graph(id="my-graph"),
# ]
# )
def networkx_component(company_id: int):
layout = html.Div(
[
html.I("Write your EDGE_VAR"),
html.Br(),
dcc.Input(id="EGDE_VAR", type="text", value="K", debounce=True),
dcc.Graph(id="my-graph"),
dcc.Graph(id="my-graph", figure=networkGraph(company_id)),
]
)
)
return layout
# callback(
# Output("my-graph", "figure", allow_duplicate=True),
# [Input("EGDE_VAR", "value")],
# prevent_initial_call=True,
# )
# def update_output() -> None:
# return networkGraph()
@app.callback(
Output("my-graph", "figure"),
[Input("EGDE_VAR", "value")],
)
def update_output(EGDE_VAR: None) -> None:
return networkGraph(EGDE_VAR)
if __name__ == "__main__":
app.run(debug=True)
# if __name__ == "__main__":
# app.run(debug=True)

8
src/aki_prj23_transparenzregister/ui/networkx_dash_overall.py
View File

@ -148,14 +148,16 @@ def networkGraph(EGDE_VAR: None) -> go.Figure:
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"),
dcc.Graph(id="my-graph", style={'width': '49%'}),
]
)

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

@ -1,14 +1,13 @@
"""Content of home page."""
import dash
import networkx as nx
from dash import html
import pandas as pd
import networkx as nx
import plotly.graph_objects as go
from dash import Input, Output, callback, html
from dash import Dash, Input, Output, dcc, html, callback
from aki_prj23_transparenzregister.config.config_providers import JsonFileConfigProvider
from aki_prj23_transparenzregister.utils.sql import connector, entities
from aki_prj23_transparenzregister.utils.networkx.networkx_data import (
find_all_company_relations,
find_top_companies,
)
dash.register_page(
__name__,
@ -21,7 +20,36 @@ dash.register_page(
"/personendetails/",
],
)
def find_all_company_relations() -> pd.DataFrame:
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():
# print(companies_df.loc[companies_df["company_id"] == row["relation_id"]]["company_name"].values[0])
# print("TEst")
company_name.append(companies_df.loc[companies_df["company_id"] == row["relation_id"]]["company_name"].values[0])
connected_company_name.append(companies_df.loc[companies_df["company_id"] == row["company_relation_company2_id"]]["company_name"].values[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 networkGraph(EGDE_VAR: None) -> go.Figure:
@ -30,6 +58,10 @@ def networkGraph(EGDE_VAR: None) -> go.Figure:
edges = []
for index, row in find_all_company_relations().iterrows():
edges.append([row["company_name"], row["connected_company_name"]])
# print(row["company_name"], row["connected_company_name"])
# print(edges)
# edges = df[["relation_id","company_relation_company2_id"]]
# edges = [[EGDE_VAR, "B"], ["B", "C"], ["B", "D"]]
network_graph = nx.Graph()
network_graph.add_edges_from(edges)
pos = nx.spring_layout(network_graph)
@ -99,6 +131,7 @@ def networkGraph(EGDE_VAR: None) -> go.Figure:
measure_vector = {}
network_metrics_df = pd.DataFrame()
measure_vector = nx.eigenvector_centrality(network_graph)
network_metrics_df["eigenvector"] = measure_vector.values()
@ -122,42 +155,35 @@ def networkGraph(EGDE_VAR: None) -> go.Figure:
return go.Figure(data=[edge_trace, node_trace], layout=layout)
df = find_top_companies()
with open("src/aki_prj23_transparenzregister/ui/assets/network_graph.html") as file:
html_content = file.read()
# Dash App
# app = Dash(__name__)
# app.title = "Dash Networkx"
layout = html.Div(
children = html.Div(
children=[
# NOTE lib dir created by NetworkX has to be placed in assets
html.Iframe(
src="assets/network_graph.html",
style={"height": "100vh", "width": "100vw"},
allow="*",
html.Div(
className="top_companytable_style",
children=[
html.I("Write your EDGE_VAR")
]
),
html.Div(
className="networkx_style",
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"),
]
)
]
# children = html.Div(
# children=[
# html.Div(
# className="top_companytable_style",
# children=[
# html.Title(title="Top Ten Unternehmen", style={"align": "mid"}),
# dash_table.DataTable(df.to_dict('records'), [{"name": i, "id": i} for i in df.columns])
# ]
# ),
# html.Div(
# className="networkx_style",
# children=[
# html.Header(title="Social Graph"),
# dcc.Dropdown(['eigenvector', 'degree', 'betweeness', 'closeness'], 'eigenvector', id='demo-dropdown'),
# "Text",
# dcc.Input(id="EGDE_VAR", type="text", value="K", debounce=True),
# # dcc.Dropdown(['eigenvector', 'degree', 'betweeness', 'closeness'], 'eigenvector', id='metric-dropdown'),
# dcc.Graph(id="my-graph"),
# ]
# )
# ]
# )
)
)
@ -167,5 +193,4 @@ layout = html.Div(
[Input("EGDE_VAR", "value")],
)
def update_output(EGDE_VAR: None) -> None:
find_top_companies()
return networkGraph(EGDE_VAR)

368
src/aki_prj23_transparenzregister/ui/ui_elements.py Normal file
View File

@ -0,0 +1,368 @@
"""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.utils.sql import entities
from aki_prj23_transparenzregister.ui.networkx_dash import networkx_component
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.
"""
selected_company_id
return networkx_component(selected_company_id)
# return html.Div([f"Netzwerk von Unternehmen mit ID: {selected_company_id}"])