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

Test Version

Browse Source
This commit is contained in:
Tim
2023-10-25 19:05:14 +02:00
parent 891ed277b6
commit 7e8adfafd5
12 changed files with 307 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -11,6 +11,9 @@ from sqlalchemy.orm import Session
from aki_prj23_transparenzregister.ui import data_elements, finance_elements
from aki_prj23_transparenzregister.ui.networkx_dash import networkx_component
from aki_prj23_transparenzregister.utils.networkx.networkx_data import (
find_company_relations,
)
COLORS = {
"light": "#edefef",
@ -344,8 +347,27 @@ def person_relations_layout(selected_company_id: int, session: Session) -> html:
style_filter={"text-align": "center", "backgroundColor": COLORS["light"]},
)
def kennzahlen_layout(selected_finance_df: pd.DataFrame) -> html:
"""Create metrics tab.
def network_layout(selected_company_id: int) -> html:
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:
@ -354,6 +376,16 @@ def network_layout(selected_company_id: int) -> html:
Returns:
The html div to create the network tab of the company page.
"""
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
# print(nodes)
# print(pd.DataFrame(edges))
# graph, metrics = initialize_network(nodes=nodes, edges=edges)
# metric = None
# figure = create_2d_graph(graph, nodes, edges, metrics, metric)
# selected_company_id
# print(company_relations)
return networkx_component(selected_company_id)
# return html.Div([f"Netzwerk von Unternehmen mit ID: {selected_company_id}"])

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

@ -94,7 +94,7 @@ layout = html.Div(
dcc.Dropdown(
company_relation_type_filter,
company_relation_type_filter[0],
id="dropdown_companyrelation_filter",
id="dropdown_company_relation_filter",
className="dropdown_style",
),
],
@ -109,7 +109,7 @@ layout = html.Div(
dcc.Dropdown(
person_relation_type_filter,
person_relation_type_filter[0],
id="dropdown_personrelation_filter",
id="dropdown_person_relation_filter",
className="dropdown_style",
),
],
@ -163,8 +163,8 @@ layout = html.Div(
[
Input("dropdown", "value"),
Input("switch", "on"),
Input("dropdown_companyrelation_filter", "value"),
Input("dropdown_personrelation_filter", "value"),
Input("dropdown_company_relation_filter", "value"),
Input("dropdown_person_relation_filter", "value"),
],
prevent_initial_call=True,
allow_duplicate=True,

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

@ -8,14 +8,14 @@ def initialize_network(edges: list, nodes: dict) -> tuple[nx.Graph, pd.DataFrame
Args:
edges (list): List with the connections between Nodes.
nodes (list): List with all Nodes.
nodes (dict): Dict with all Nodes.
Returns:
Graph: Plotly Figure
Metrices: DataFrame with Metrics
"""
# create edge dataframe
df_edges = pd.DataFrame(edges)
df_edges = pd.DataFrame(edges, columns=["from", "to", "type"])
graph = nx.from_pandas_edgelist(
df_edges, source="from", target="to", edge_attr="type"
)
@ -24,13 +24,13 @@ def initialize_network(edges: list, nodes: dict) -> tuple[nx.Graph, pd.DataFrame
nx.set_node_attributes(graph, nodes)
# Create a DataFrame with all Metrics
metrices = pd.DataFrame(
metrics = pd.DataFrame(
columns=["degree", "eigenvector", "betweeness", "closeness", "pagerank"]
)
metrices["eigenvector"] = nx.eigenvector_centrality(graph).values()
metrices["degree"] = nx.degree_centrality(graph).values()
metrices["betweeness"] = nx.betweenness_centrality(graph).values()
metrices["closeness"] = nx.closeness_centrality(graph).values()
metrices["pagerank"] = nx.pagerank(graph).values()
metrics["eigenvector"] = nx.eigenvector_centrality(graph).values()
metrics["degree"] = nx.degree_centrality(graph).values()
metrics["betweeness"] = nx.betweenness_centrality(graph).values()
metrics["closeness"] = nx.closeness_centrality(graph).values()
metrics["pagerank"] = nx.pagerank(graph).values()
return graph, metrices
return graph, metrics

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

@ -1,5 +1,6 @@
"""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
@ -201,7 +202,7 @@ def filter_relation_with_more_than_one_connection(
def create_edge_and_node_list(
person_relations: pd.DataFrame, company_relations: pd.DataFrame
) -> tuple[dict, list]:
"""In this Method the given DataFrame with the relation will be morphed to Edge and Node lists and enhanced by a coloring for companies and person Nodes.
"""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_
@ -261,3 +262,100 @@ def create_edge_and_node_list(
}
)
return nodes, edges
def find_company_relations(
selected_company_id: int,
) -> tuple[pd.DataFrame, pd.DataFrame()]:
print("In Method:" + str(selected_company_id))
selected_company_id = 312
# print( entities.Company.id == selected_company_id)
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,
)
.where(to_company.id == selected_company_id)
)
logger.debug(str(relations_company_query))
# relations_company_query = relations_company_query.filter(entities.CompanyRelation.company2_id == selected_company_id) # or (entities.CompanyRelation.company2_id == selected_company_id))
# logger.debug(str(relations_company_query))
# tmp = str(relations_company_query) + " WHERE from_company.id = " + str(selected_company_id)
# print(tmp)
# query = str(relations_company_query.filter((entities.CompanyRelation.company_id == selected_company_id) or (entities.CompanyRelation.company2_id == selected_company_id)))
# print(query)
# company_relations = get_all_company_relations()
company_relations = pd.DataFrame(
relations_company_query.all(),
columns=[
"id_company_to",
"name_company_to",
"relation_type",
"name_company_from",
"id_company_from",
],
)
logger.debug(str(relations_company_query))
print("query ergebnis:")
print(company_relations)
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 = []
COLOR_COMPANY = "blue"
for _index, row in company_relations.iterrows():
if node := 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,
}
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,
}
edges.append(
{
"from": row["id_company_from"],
"to": row["id_company_to"],
"type": row["relation_type"],
}
)
return nodes, edges

3
src/aki_prj23_transparenzregister/utils/sql/connector.py
View File

@ -40,12 +40,13 @@ def get_engine(conn_args: SQLConnectionString) -> Engine:
database=conn_args.database,
port=conn_args.port,
)
return sa.create_engine(url)
return sa.create_engine(url, echo=True)
if isinstance(conn_args, SQLiteConnectionString):
return sa.create_engine(
str(conn_args),
connect_args={"check_same_thread": True},
poolclass=SingletonThreadPool,
echo=True,
)
raise TypeError("The type of the configuration is invalid.")

6
tests/ui/cytoscape_dash_test.py
View File

@ -1,6 +0,0 @@
"""Test for the NetworkX Component."""
# def networkGraph(Edges) -> None:
# """Checks if an import co company_stats_dash can be made."""
# assert networkx_dash is not None

6
tests/ui/dashvis_networkx_test.py
View File

@ -1,6 +0,0 @@
"""Test for the NetworkX Component."""
# def networkGraph(Edges) -> None:
# """Checks if an import co company_stats_dash can be made."""
# assert networkx_dash is not None

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

@ -0,0 +1,7 @@
"""Test for the Home Page."""
# from aki_prj23_transparenzregister.ui.pages.home import networkx_dash
# def networkGraph(Edges: None) -> None:
# """Checks if an import co company_stats_dash can be made."""
# assert networkx_dash is not None

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

@ -0,0 +1,38 @@
"""Test the initialize Network function."""
import datetime
from unittest import TestCase
import networkx as nx
import pandas as pd
from aki_prj23_transparenzregister.utils.networkx.network_base import initialize_network
tc = TestCase()
def test_initialize_network() -> None:
edges: list = [
{"from": "p_545", "to": "c_53", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "p_758", "to": "c_77", "type": "HAFTENDER_GESELLSCHAFTER"},
]
nodes: dict = {
"c_53": {
"id": "c_53",
"name": "1. Freiburger Solarfonds Beteiligungs-KG",
"color": "blue",
},
"p_545": {
"id": "p_545",
"firstname": "Jürgen",
"lastname": "Wetzel",
"date_of_birth": datetime.date(1962, 11, 15),
"color": "red",
},
}
graph, metrics = initialize_network(edges=edges, nodes=nodes)
assert type(graph) is nx.Graph
assert type(metrics) is pd.DataFrame
tc.assertListEqual(
list(metrics.columns),
["degree", "eigenvector", "betweeness", "closeness", "pagerank"],
)

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

@ -0,0 +1,38 @@
"""Test the initialize Network function."""
import datetime
from unittest import TestCase
import networkx as nx
import pandas as pd
from aki_prj23_transparenzregister.utils.networkx.network_base import initialize_network
tc = TestCase()
def test_initialize_network() -> None:
edges: list = [
{"from": "p_545", "to": "c_53", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "p_758", "to": "c_77", "type": "HAFTENDER_GESELLSCHAFTER"},
]
nodes: dict = {
"c_53": {
"id": "c_53",
"name": "1. Freiburger Solarfonds Beteiligungs-KG",
"color": "blue",
},
"p_545": {
"id": "p_545",
"firstname": "Jürgen",
"lastname": "Wetzel",
"date_of_birth": datetime.date(1962, 11, 15),
"color": "red",
},
}
graph, metrics = initialize_network(edges=edges, nodes=nodes)
assert type(graph) is nx.Graph
assert type(metrics) is pd.DataFrame
tc.assertListEqual(
list(metrics.columns),
["degree", "eigenvector", "betweeness", "closeness", "pagerank"],
)

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

@ -0,0 +1,38 @@
"""Test the initialize Network function."""
import datetime
from unittest import TestCase
import networkx as nx
import pandas as pd
from aki_prj23_transparenzregister.utils.networkx.network_base import initialize_network
tc = TestCase()
def test_initialize_network() -> None:
edges: list = [
{"from": "p_545", "to": "c_53", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "p_758", "to": "c_77", "type": "HAFTENDER_GESELLSCHAFTER"},
]
nodes: dict = {
"c_53": {
"id": "c_53",
"name": "1. Freiburger Solarfonds Beteiligungs-KG",
"color": "blue",
},
"p_545": {
"id": "p_545",
"firstname": "Jürgen",
"lastname": "Wetzel",
"date_of_birth": datetime.date(1962, 11, 15),
"color": "red",
},
}
graph, metrics = initialize_network(edges=edges, nodes=nodes)
assert type(graph) is nx.Graph
assert type(metrics) is pd.DataFrame
tc.assertListEqual(
list(metrics.columns),
["degree", "eigenvector", "betweeness", "closeness", "pagerank"],
)

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

@ -0,0 +1,38 @@
"""Test the initialize Network function."""
import datetime
from unittest import TestCase
import networkx as nx
import pandas as pd
from aki_prj23_transparenzregister.utils.networkx.network_base import initialize_network
tc = TestCase()
def test_initialize_network() -> None:
edges: list = [
{"from": "p_545", "to": "c_53", "type": "HAFTENDER_GESELLSCHAFTER"},
{"from": "p_758", "to": "c_77", "type": "HAFTENDER_GESELLSCHAFTER"},
]
nodes: dict = {
"c_53": {
"id": "c_53",
"name": "1. Freiburger Solarfonds Beteiligungs-KG",
"color": "blue",
},
"p_545": {
"id": "p_545",
"firstname": "Jürgen",
"lastname": "Wetzel",
"date_of_birth": datetime.date(1962, 11, 15),
"color": "red",
},
}
graph, metrics = initialize_network(edges=edges, nodes=nodes)
assert type(graph) is nx.Graph
assert type(metrics) is pd.DataFrame
tc.assertListEqual(
list(metrics.columns),
["degree", "eigenvector", "betweeness", "closeness", "pagerank"],
)