-
Notifications
You must be signed in to change notification settings - Fork 0
/
dash_tool.py
142 lines (107 loc) · 4.23 KB
/
dash_tool.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ======================================================================================
# Copyright (C) 2023 by Data Specialist, Inc. All rights reserved.
# Released under the terms of the GNU General Public License version 2 or later.
# ======================================================================================
# Created by: [email protected]
# Created at: 30/05/2023
# ======================================================================================
# DASH: DASHBOARD TOOL (USING STREAMLIT)
# ======================================================================================
from timeit import default_timer as timer
import streamlit as st
from config import Config
from src.analysis.aggregates import GMLAggregates
from src.analysis.charts import GMLCharts
from src.analysis.filters import GMLFilters
from src.analysis.metrics import GMLMetrics
from src.analysis.tables import GMLTables
from src.utils.dw_operations import GMLDataWarehouseOperations
# Initializing objects
env = Config()
dw = GMLDataWarehouseOperations()
con = dw.connect_db()
agg = GMLAggregates(con)
chart = GMLCharts(con)
filter = GMLFilters(con)
metric = GMLMetrics(con)
table = GMLTables(con)
# Initializing variables
uf = ""
municipio = ""
porte = ""
start_timer = ""
end_timer = ""
def page_header():
st.title("Get More Leads (CNPJ)")
with st.expander("About the data"):
st.write(env.APP_FULL_DESC)
def page_footer(start_timer, end_timer):
st.write("Total running time: ", round(end_timer - start_timer, 3), " seconds")
with st.expander("Data Protection"):
st.write(env.APP_WARN_LGPD)
def export_data(df):
return df.to_csv(index=False).encode("utf-8")
def start():
page_header()
start_timer = timer()
st.subheader("Filter by")
col_filter1, col_filter2, col_filter3 = st.columns(3)
with col_filter1:
uf_df = filter.by_uf()
uf = st.selectbox("UF", uf_df)
with col_filter2:
municipio_df = filter.by_municipio(uf)
municipio = st.selectbox("Município", municipio_df)
with col_filter3:
porte_df = filter.by_porte(uf, municipio)
porte = st.selectbox("Porte", porte_df)
st.divider()
st.subheader("Data Indicators")
kpi1 = metric.get_value_total_empresas(uf, municipio, porte)
kpi2 = metric.get_value_total_empresas_individual(uf, municipio, porte)
kpi3 = metric.get_value_total_empresas_sociedade_anonima(uf, municipio, porte)
kpi4 = metric.get_value_total_empresas_sociedade_limitada(uf, municipio, porte)
col_kpi1, col_kpi2, col_kpi3, col_kpi4 = st.columns(4)
col_kpi1.metric("empresas", str(kpi1))
col_kpi2.metric("empresas individual", str(kpi2))
col_kpi3.metric("empresas soc anônima", str(kpi3))
col_kpi4.metric("empresas soc limitada", str(kpi4))
st.divider()
st.subheader("Data Aggregate")
col_agg1, col_agg2 = st.columns(2)
with col_agg1:
count_bairro = agg.by_bairro(uf, municipio, porte)
st.write("Top 5 Bairro", count_bairro)
with col_agg2:
count_cnae_fiscal = agg.by_cnae_fiscal(uf, municipio, porte)
st.write("Top 5 Cnae Fiscal", count_cnae_fiscal)
st.divider()
st.subheader("Data Distribution")
col_chart1, col_chart2 = st.columns(2)
with col_chart1:
with st.container():
st.write("Companies by matriz filial")
matriz_filial_df = chart.by_matriz_filial(uf, municipio, porte)
st.bar_chart(data=matriz_filial_df, x="matriz filial", y="qty")
with col_chart2:
with st.container():
st.write("Companies by opc mei")
opc_mei_df = chart.by_opc_mei(uf, municipio, porte)
st.bar_chart(data=opc_mei_df, x="opc mei", y="qty")
st.divider()
st.subheader("Data Preview")
table_head = table.list_rows(uf, municipio, porte)
table_full = table.export_rows(uf, municipio, porte)
st.write("Listing first 10 Companies", table_head)
st.download_button(
label="Export all data",
data=export_data(table_full),
file_name=f"empresas_{uf}_{municipio}.csv",
mime="text/csv",
key="download-csv",
)
st.divider()
end_timer = timer()
page_footer(start_timer, end_timer)