FTTH / dashboard_v2.py
Geoeasy's picture
Update dashboard_v2.py
e885005 verified
Raw
History Blame Contribute Delete
92.5 kB
"""
Dashboard Analítico FTTH da França (V2) - Região PACA Sul
===========================================================
Interface moderna e rápida baseada em dados pré-processados da ARCEP e OpenStreetMap.
Focado nos departamentos: Bouches-du-Rhône (13), Var (83) e Vaucluse (84).
Suporta filtragem administrativa, download de relatórios Markdown e Q&A Inteligente.
"""
import re
import json
import flask
import unicodedata
import pandas as pd
import geopandas as gpd
from pathlib import Path
import dash
from dash import dcc, html, Input, Output, State
import dash_bootstrap_components as dbc
import plotly.express as px
import plotly.graph_objects as go
# ─── 1. Caminhos e Carregamento dos Dados Pré-Processados ───────────────────
# BASE_DIR: resolve sempre relativo ao script, funciona local e em Docker (/app)
BASE_DIR = Path(__file__).parent.resolve()
DATA_DIR = BASE_DIR / "data_processed"
REPORTS_DIR = BASE_DIR / "data_reports"
# Carregar dados tabulares
df_summary = pd.read_csv(DATA_DIR / "departments_summary.csv")
df_pmz = pd.read_csv(DATA_DIR / "pmz_map_data.csv")
df_national_hist = pd.read_csv(DATA_DIR / "national_history.csv")
df_dept_hist = pd.read_csv(DATA_DIR / "departments_history.csv")
# Carregar outros pontos físicos da infraestrutura de rede
df_nro = pd.read_csv(DATA_DIR / "nro_map_data.csv") if (DATA_DIR / "nro_map_data.csv").exists() else pd.DataFrame()
df_nra = pd.read_csv(DATA_DIR / "nra_map_data.csv") if (DATA_DIR / "nra_map_data.csv").exists() else pd.DataFrame()
df_copper = pd.read_csv(DATA_DIR / "copper_map_data.csv") if (DATA_DIR / "copper_map_data.csv").exists() else pd.DataFrame()
df_pt = pd.read_csv(DATA_DIR / "pt_map_data.csv") if (DATA_DIR / "pt_map_data.csv").exists() else pd.DataFrame()
# Carregar geojson de fronteiras de departamentos (ADM2)
with open(DATA_DIR / "departments_boundaries.geojson", "r", encoding="utf-8") as f:
geojson_depts = json.load(f)
# Carregar geojson de fronteiras de regiões (ADM1)
with open(DATA_DIR / "regions_boundaries.geojson", "r", encoding="utf-8") as f:
geojson_regions = json.load(f)
for i, feat in enumerate(geojson_regions['features']):
feat['id'] = feat['properties'].get('NAME_1', str(i))
# Carregar geojson de Arrondissements (ADM3 — 320 unidades)
_adm3_path = DATA_DIR / "adm3_arrondissements.geojson"
if _adm3_path.exists():
with open(_adm3_path, "r", encoding="utf-8") as f:
geojson_adm3 = json.load(f)
for i, feat in enumerate(geojson_adm3['features']):
feat['id'] = feat['properties'].get('shapeID', str(i))
else:
geojson_adm3 = None
# Carregar geojson de Cantons (ADM4 — 2054 unidades)
_adm4_path = DATA_DIR / "adm4_cantons.geojson"
if _adm4_path.exists():
with open(_adm4_path, "r", encoding="utf-8") as f:
geojson_adm4 = json.load(f)
for i, feat in enumerate(geojson_adm4['features']):
feat['id'] = feat['properties'].get('shapeID', str(i))
else:
geojson_adm4 = None
# Carregar centroides das comunas (ADM5 — 35.010 unidades) para zoom preciso
_centroids_path = DATA_DIR / "commune_centroids.csv"
if _centroids_path.exists():
df_commune_centroids = pd.read_csv(_centroids_path)
# Normalizar nomes para lookup
df_commune_centroids['name_norm'] = df_commune_centroids['shapeName'].str.strip().str.lower()
else:
df_commune_centroids = pd.DataFrame()
# Garantir tipos de dados corretos e consistentes
for df in [df_summary, df_pmz, df_dept_hist, df_nro, df_nra, df_copper, df_pt]:
if not df.empty and 'dep_code' in df.columns:
df['dep_code'] = df['dep_code'].astype(str).str.zfill(2)
# Ordenar departamentos numericamente (2A=20.1, 2B=20.2 para ficarem entre 19 e 21)
def _sort_key(code):
c = str(code).lower()
if c == '2a': return 20.1
if c == '2b': return 20.2
try: return float(c)
except: return 999
df_summary_sorted = df_summary.copy()
df_summary_sorted['_sort'] = df_summary_sorted['dep_code'].apply(_sort_key)
df_summary_sorted = df_summary_sorted.sort_values('_sort').drop(columns=['_sort'])
# Mapeamento amigável para o Dropdown de departamentos
depts_options = [{"label": "🇫🇷 Visão Geral Nacional", "value": "national"}]
for _, r in df_summary_sorted.iterrows():
depts_options.append({
"label": f"📍 {r['dep_code']} - {r['dep_name']} ({r['region_name']})",
"value": r['dep_code']
})
# Extrair regiões únicas ordenadas (ADM1)
regions_list = sorted(df_summary_sorted['region_name'].dropna().unique().tolist())
regions_options = [{"label": "🗺️ Todas as Regiões", "value": "all"}]
regions_options += [{"label": f"🏙️ {r}", "value": r} for r in regions_list]
# Centroides de cada região (média dos centroides dos departamentos)
region_centroids = df_summary_sorted.groupby('region_name').agg(
lat=("centroid_lat", "mean"),
lon=("centroid_lon", "mean")
).to_dict('index')
custom_styles = {
"bg": "#0a0c10",
"panel-bg": "#121620",
"border": "1px solid #1e2538",
"text-main": "#f0f4f8",
"text-muted": "#8a99ad",
"accent-blue": "#00d2ff",
"accent-green": "#39ff14",
"accent-orange": "#ff8700",
"accent-red": "#ff3838",
"accent-purple": "#bd00ff"
}
# ─── 2. Inicialização do App Dash ───────────────────────────────────────────
app = dash.Dash(
__name__,
external_stylesheets=[
dbc.themes.CYBORG,
"https://cdnjs.cloudflare.com/ajax/libs/font-awesome/6.4.0/css/all.min.css"
],
meta_tags=[{"name": "viewport", "content": "width=device-width, initial-scale=1"}]
)
app.title = "INFRAESTRUTURA FTTH FRANÇA"
# ─── 3. Endpoints Flask para Downloads Estáticos de Alta Performance ─────────
server = app.server
@server.route("/download/report/<dep_code>")
def download_report(dep_code):
if dep_code == "national":
return flask.send_from_directory(
directory=str(REPORTS_DIR),
path="national_summary.md",
as_attachment=True
)
else:
dep_code_clean = str(dep_code).zfill(2)
if dep_code_clean in df_summary['dep_code'].values:
return flask.send_from_directory(
directory=str(REPORTS_DIR / "departments"),
path=f"dep_{dep_code_clean}.md",
as_attachment=True
)
return "Departamento inválido", 400
@server.route("/download/communes/<dep_code>")
def download_communes(dep_code):
if dep_code == "national":
return flask.send_from_directory(
directory=str(DATA_DIR),
path="communes_summary.csv",
as_attachment=True
)
else:
dep_code_clean = str(dep_code).zfill(2)
if dep_code_clean in df_summary['dep_code'].values:
return flask.send_from_directory(
directory=str(DATA_DIR / "communes_by_dep"),
path=f"communes_{dep_code_clean}.csv",
as_attachment=True
)
return "Departamento inválido", 400
@server.route("/download/zapm/<dep_code>")
def download_zapm(dep_code):
dep_code_clean = str(dep_code).zfill(2)
if dep_code_clean in df_summary['dep_code'].values:
return flask.send_from_directory(
directory=str(DATA_DIR / "zapm_by_dep"),
path=f"zapm_{dep_code_clean}.geojson",
as_attachment=True
)
return "Departamento inválido ou escopo nacional (selecione um departamento)", 400
# ─── 4. Layout da Interface do Dashboard ────────────────────────────────────
app.layout = html.Div(
[
# Header / Barra Superior Fixa
html.Div(
[
html.Div(
[
html.I(className="fa-solid fa-network-wired me-3", style={"color": custom_styles["accent-blue"], "font-size": "24px"}),
html.Span("INFRAESTRUTURA FTTH FRANÇA", style={"font-family": "Outfit", "font-weight": "800", "color": "#ffffff", "font-size": "20px", "letter-spacing": "1px"}),
html.Span("OSINT MONITOR", style={"font-size": "11px", "color": custom_styles["accent-orange"], "border": f"1px solid {custom_styles['accent-orange']}", "padding": "2px 6px", "border-radius": "4px", "margin-left": "15px", "font-weight": "bold", "letter-spacing": "1px"})
],
style={"display": "flex", "align-items": "center"}
),
html.Div(
[
html.Span(["Escopo: ", html.Strong("França Metropolitana (Todos os Níveis)", style={"color": "#fff"})], className="me-4"),
html.Span(["Status da API: ", html.Strong("ONLINE (Offline Cache)", style={"color": custom_styles["accent-green"]})])
],
style={"font-size": "13px", "color": custom_styles["text-muted"]}
)
],
style={
"position": "fixed",
"top": "0",
"left": "0",
"right": "0",
"height": "75px",
"background-color": custom_styles["panel-bg"],
"border-bottom": "2px solid #1e2538",
"padding": "0 30px",
"z-index": "1030",
"display": "flex",
"justify-content": "space-between",
"align-items": "center"
}
),
# Barra Lateral de Filtros e Downloads
html.Div(
[
# Seção 1: Filtro Geográfico Principal
html.Div(
[
html.H5("Nível Administrativo", style={"font-size": "12px", "text-transform": "uppercase", "color": custom_styles["text-muted"], "letter-spacing": "1.5px", "border-left": f"3px solid {custom_styles['accent-blue']}", "padding-left": "8px", "font-weight": "bold"}),
# ADM1 — Região
html.Span("ADM1 — Região", style={"font-size": "10px", "color": custom_styles["accent-blue"], "display": "block", "margin-top": "10px", "letter-spacing": "0.5px"}),
dcc.Dropdown(
id="region-selector",
options=regions_options,
value=["all"],
multi=True,
clearable=False,
style={"background-color": "#0a0c10", "color": "#ffffff", "border": "1px solid #1e2538", "margin-top": "4px"}
),
# ADM2 — Departamento
html.Span("ADM2 — Departamento", style={"font-size": "10px", "color": custom_styles["accent-orange"], "display": "block", "margin-top": "8px", "letter-spacing": "0.5px"}),
dcc.Dropdown(
id="dep-selector",
options=depts_options,
value=["national"],
multi=True,
clearable=False,
style={"background-color": "#0a0c10", "color": "#ffffff", "border": "1px solid #1e2538", "margin-top": "4px"}
),
# ADM3 — Commune / Município
html.Span("ADM3 — Commune / Município", style={"font-size": "10px", "color": custom_styles["accent-green"], "display": "block", "margin-top": "8px", "letter-spacing": "0.5px"}),
dcc.Dropdown(
id="commune-selector",
options=[],
value=None,
placeholder="Selecione um departamento primeiro...",
clearable=True,
disabled=True,
style={"background-color": "#0a0c10", "color": "#ffffff", "border": "1px solid #1e2538", "margin-top": "4px"}
)
],
className="mb-4"
),
# Seção 2: Operadores Principais
html.Div(
[
html.H5("Operadores no Mapa", style={"font-size": "12px", "text-transform": "uppercase", "color": custom_styles["text-muted"], "letter-spacing": "1.5px", "border-left": f"3px solid {custom_styles['accent-blue']}", "padding-left": "8px", "font-weight": "bold"}),
dcc.Checklist(
id="operator-selector",
options=[
{"label": " Orange", "value": "Orange"},
{"label": " SFR", "value": "SFR"},
{"label": " Free", "value": "Free"},
{"label": " Bouygues Telecom", "value": "Bouygues Telecom"},
{"label": " Outros / RAPs", "value": "Outros"}
],
value=["Orange", "SFR", "Free", "Bouygues Telecom", "Outros"],
labelStyle={"display": "block", "margin-bottom": "8px", "cursor": "pointer", "font-size": "14px", "color": "#f0f4f8"},
style={"margin-top": "10px"}
)
],
className="mb-4"
),
# Seção 3: Camadas e Modos do Mapa
html.Div(
[
html.H5("Camadas & Exibição", style={"font-size": "12px", "text-transform": "uppercase", "color": custom_styles["text-muted"], "letter-spacing": "1.5px", "border-left": f"3px solid {custom_styles['accent-blue']}", "padding-left": "8px", "font-weight": "bold"}),
dcc.Checklist(
id="map-layers",
options=[
{"label": " ⚡ Nós de Rede Óptica (NROs)", "value": "nro"},
{"label": " ☎️ Centrais Telefônicas (NRAs)", "value": "nra"},
{"label": " 📦 Pontos de Mutualização (PMZs)", "value": "pmz"},
{"label": " 🪵 Redes de Cobre (COPPER)", "value": "copper"},
{"label": " 🚇 Pontos de Transporte (PTs)", "value": "pt"},
{"label": " 📐 Zonas de PM (ZAPMs)", "value": "zapm"},
{"label": " 🗺️ Fronteiras de Regiões (ADM1)", "value": "region_boundaries"},
{"label": " 📍 Divisas de Departamentos (ADM2)", "value": "boundaries"},
{"label": " 🏙️ Arrondissements (ADM3)", "value": "adm3"},
{"label": " 📌 Cantons (ADM4)", "value": "adm4"}
],
value=["nro", "nra", "pmz", "boundaries", "region_boundaries"],
labelStyle={"display": "block", "margin-bottom": "8px", "cursor": "pointer", "font-size": "14px", "color": "#f0f4f8"},
style={"margin-top": "10px"}
)
],
className="mb-4"
),
# Seção 4: Hub de Downloads OSINT
html.Div(
[
html.H5("Downloads Disponíveis", style={"font-size": "12px", "text-transform": "uppercase", "color": custom_styles["text-muted"], "letter-spacing": "1.5px", "border-left": f"3px solid {custom_styles['accent-blue']}", "padding-left": "8px", "font-weight": "bold"}),
html.P("Baixe relatórios, tabelas segmentadas por departamento e o mapeamento completo.", style={"font-size": "11.5px", "color": custom_styles["text-muted"], "margin-top": "10px"}),
html.A(
html.Button(
[html.I(className="fa-solid fa-file-arrow-down me-2"), "Baixar Relatório MD"],
className="btn btn-outline-info w-100 mb-2 btn-sm",
style={"font-weight": "bold"}
),
id="download-report-link",
href="/download/report/national",
target="_blank"
),
html.A(
html.Button(
[html.I(className="fa-solid fa-file-csv me-2"), "Baixar Comunas (CSV)"],
className="btn btn-outline-success w-100 mb-2 btn-sm",
style={"font-weight": "bold"}
),
id="download-communes-link",
href="/download/communes/national",
target="_blank"
),
html.A(
html.Button(
[html.I(className="fa-solid fa-draw-polygon me-2"), "Baixar ZAPM (GeoJSON)"],
className="btn btn-outline-primary w-100 mb-2 btn-sm",
style={"font-weight": "bold"}
),
id="download-zapm-link",
href="/download/zapm/national",
target="_blank",
style={"display": "none"} # Invisível por padrão no escopo nacional
),
],
className="mt-4"
)
],
style={
"position": "fixed",
"top": "75px",
"left": "0",
"bottom": "0",
"width": "320px",
"background-color": custom_styles["panel-bg"],
"border-right": "2px solid #1e2538",
"padding": "25px",
"overflow-y": "auto",
"z-index": "1020"
}
),
# Painel de Conteúdo Principal
html.Div(
[
# Fila Superior: Kpi Cards Dinâmicos
dbc.Row(
[
dbc.Col(
dbc.Card(
dbc.CardBody(
[
html.Div("Locais Conectáveis (Fibra Elegível)", style={"font-size": "11px", "color": custom_styles["text-muted"], "text-transform": "uppercase"}),
html.H3(id="kpi-raccordables", style={"color": custom_styles["accent-blue"], "font-weight": "800", "margin": "5px 0"}),
html.Div(id="kpi-raccordables-sub", style={"font-size": "12px"})
]
),
style={"background-color": custom_styles["panel-bg"], "border": custom_styles["border"]}
),
md=3
),
dbc.Col(
dbc.Card(
dbc.CardBody(
[
html.Div("Locais Estimados (Domicílios)", style={"font-size": "11px", "color": custom_styles["text-muted"], "text-transform": "uppercase"}),
html.H3(id="kpi-estimados", style={"color": "#ffffff", "font-weight": "800", "margin": "5px 0"}),
html.Div("Estimativa total oficial ARCEP", id="kpi-estimados-sub", style={"font-size": "12px", "color": custom_styles["text-muted"]})
]
),
style={"background-color": custom_styles["panel-bg"], "border": custom_styles["border"]}
),
md=3
),
dbc.Col(
dbc.Card(
dbc.CardBody(
[
html.Div("Taxa de Cobertura Ponderada", style={"font-size": "11px", "color": custom_styles["text-muted"], "text-transform": "uppercase"}),
html.H3(id="kpi-cobertura", style={"color": custom_styles["accent-green"], "font-weight": "800", "margin": "5px 0"}),
html.Div(id="kpi-cobertura-status", style={"font-size": "12px", "font-weight": "bold"})
]
),
style={"background-color": custom_styles["panel-bg"], "border": custom_styles["border"]}
),
md=3
),
dbc.Col(
dbc.Card(
dbc.CardBody(
[
html.Div("Operador Dominante e PMZs", style={"font-size": "11px", "color": custom_styles["text-muted"], "text-transform": "uppercase"}),
html.H3(id="kpi-operator", style={"color": custom_styles["accent-orange"], "font-weight": "800", "margin": "5px 0", "font-size": "20px"}),
html.Div(id="kpi-pmz-total", style={"font-size": "12px", "color": custom_styles["text-muted"]})
]
),
style={"background-color": custom_styles["panel-bg"], "border": custom_styles["border"]}
),
md=3
)
],
className="g-3 mb-3"
),
# Fila Central: Mapa Geográfico (Ocupa 42% da altura)
html.Div(
[
dcc.Graph(
id="mapbox-map-v2",
style={"height": "100%", "width": "100%", "border-radius": "8px", "overflow": "hidden", "border": custom_styles["border"]},
config={"displayModeBar": "hover", "scrollZoom": True}
)
],
style={"flex": "0 0 38vh", "margin-bottom": "15px", "height": "38vh"}
),
# Fila Inferior: Abas Analíticas Expandidas (Ocupa o resto da tela)
dbc.Card(
[
dbc.CardHeader(
dbc.Tabs(
[
dbc.Tab(label="📊 Histórico de Implantação", tab_id="tab-history", label_style={"cursor": "pointer", "font-weight": "bold"}),
dbc.Tab(label="🍕 Participação dos Operadores", tab_id="tab-market", label_style={"cursor": "pointer", "font-weight": "bold"}),
dbc.Tab(label="📑 Relatório Técnico OSINT (.md)", tab_id="tab-report-md", label_style={"cursor": "pointer", "font-weight": "bold", "color": custom_styles["accent-blue"]}),
dbc.Tab(label="🔍 Glossário Técnico", tab_id="tab-glossary", label_style={"cursor": "pointer", "font-weight": "bold", "color": custom_styles["accent-green"]}),
dbc.Tab(label="🤖 Assistente Q&A", tab_id="tab-qa", label_style={"cursor": "pointer", "font-weight": "bold", "color": custom_styles["accent-orange"]})
],
id="tabs-panel-v2",
active_tab="tab-history",
),
style={"background-color": "#121620", "border-bottom": "1px solid #1e2538"}
),
dbc.CardBody(
html.Div(
[
# Conteúdo Aba 1: Histórico de Implantação
html.Div(
[
dbc.Row(
[
dbc.Col(
[
html.H6("Evolução Temporal da Cobertura de Fibra (2017-2025)", className="text-muted text-uppercase mb-2", style={"font-size": "11px", "letter-spacing": "1px"}),
dcc.Graph(id="chart-history-cobertura", style={"height": "210px"})
],
md=6
),
dbc.Col(
[
html.H6("Crescimento de Locais Elegíveis (Conectáveis vs Estimados)", className="text-muted text-uppercase mb-2", style={"font-size": "11px", "letter-spacing": "1px"}),
dcc.Graph(id="chart-history-locaux", style={"height": "210px"})
],
md=6
)
]
)
],
id="tab-content-history",
style={"display": "block"}
),
# Conteúdo Aba 2: Participação de Mercado
html.Div(
[
dbc.Row(
[
dbc.Col(
[
html.H6("Representação de Mercado Baseada em PMZs Instalados", className="text-muted text-uppercase mb-2", style={"font-size": "11px", "letter-spacing": "1px"}),
dcc.Graph(id="chart-operator-pie", style={"height": "210px"})
],
md=6
),
dbc.Col(
[
html.H6("Tabela Comparativa de PMZs por Operador", className="text-muted text-uppercase mb-2", style={"font-size": "11px", "letter-spacing": "1px"}),
html.Div(id="table-operator-data", style={"height": "210px", "overflow-y": "auto"})
],
md=6
)
]
)
],
id="tab-content-market",
style={"display": "none"}
),
# Conteúdo Aba 3: Relatório Markdown
html.Div(
[
html.Div(
id="report-markdown-container",
style={
"background-color": "#0a0c10",
"border": "1px solid #1e2538",
"border-radius": "8px",
"padding": "20px",
"height": "220px",
"overflow-y": "auto",
"color": "#f0f4f8"
}
)
],
id="tab-content-report-md",
style={"display": "none"}
),
# Conteúdo Aba 4: Glossário Técnico
html.Div(
[
html.Div(
[
dbc.Row(
[
dbc.Col(
[
html.Div(
[
html.H5("NRA (Central Telefônica - Cobre)", style={"color": custom_styles["accent-blue"], "font-size": "13px", "font-weight": "bold"}),
html.P("Noeud de Raccordement d'Abonnés. A central histórica da rede de cobre da Orange onde as linhas telefônicas clássicas convergem. Está sob plano de encerramento definitivo na França até 2030, sendo substituída por fibra.", style={"font-size": "11.5px", "color": "#d0dceb"})
],
className="p-2 mb-2", style={"background-color": "#0a0c10", "border": "1px solid #1e2538", "border-radius": "6px"}
),
html.Div(
[
html.H5("NRO (Nó Óptico - Fibra)", style={"color": custom_styles["accent-orange"], "font-size": "13px", "font-weight": "bold"}),
html.P("Noeud de Raccordement Optique. O equivalente à central telefônica para a rede de fibra. Centraliza os equipamentos ativos de rede (OLTs) que modulam os sinais luminosos que vão para os clientes.", style={"font-size": "11.5px", "color": "#d0dceb"})
],
className="p-2 mb-2", style={"background-color": "#0a0c10", "border": "1px solid #1e2538", "border-radius": "6px"}
),
html.Div(
[
html.H5("PMZ (Ponto de Mutualização)", style={"color": custom_styles["accent-green"], "font-size": "13px", "font-weight": "bold"}),
html.P("Point de Mutualisation de Zone. Armário metálico que se encontra na rua. Ele concentra as conexões de fibra de até 1000 residências. Serve de ponto de interconexão comum onde qualquer operadora comercial pode ligar suas linhas.", style={"font-size": "11.5px", "color": "#d0dceb"})
],
className="p-2", style={"background-color": "#0a0c10", "border": "1px solid #1e2538", "border-radius": "6px"}
)
],
md=6
),
dbc.Col(
[
html.Div(
[
html.H5("COPPER (Rede de Cobre / Legada)", style={"color": custom_styles["accent-red"], "font-size": "13px", "font-weight": "bold"}),
html.P("A rede histórica francesa baseada em cabos de cobre de par trançado. Provê telefonia tradicional e internet ADSL/VDSL. Apresenta alta atenuação com a distância e está em desativação física e desinstalação.", style={"font-size": "11.5px", "color": "#d0dceb"})
],
className="p-2 mb-2", style={"background-color": "#0a0c10", "border": "1px solid #1e2538", "border-radius": "6px"}
),
html.Div(
[
html.H5("PT (Ponto de Transporte)", style={"color": custom_styles["accent-purple"], "font-size": "13px", "font-weight": "bold"}),
html.P("Point de Transport. Representa a rede física de tráfego que liga os nós locais da ARCEP (como PMZs) de volta até os grandes anéis de transmissão de tráfego de alta velocidade (backbone/backhaul).", style={"font-size": "11.5px", "color": "#d0dceb"})
],
className="p-2 mb-2", style={"background-color": "#0a0c10", "border": "1px solid #1e2538", "border-radius": "6px"}
),
html.Div(
[
html.H5("ZAPM (Zona d'Arrière de PM)", style={"color": "#ffffff", "font-size": "13px", "font-weight": "bold"}),
html.P("A delimitação poligonal que engloba todos os domicílios físicos cujos cabos ópticos de distribuição convergem fisicamente para o mesmo PMZ.", style={"font-size": "11.5px", "color": "#d0dceb"})
],
className="p-2", style={"background-color": "#0a0c10", "border": "1px solid #1e2538", "border-radius": "6px"}
)
],
md=6
)
]
)
],
style={"padding-right": "5px"}
)
],
id="tab-content-glossary",
style={"display": "none"}
),
# Conteúdo Aba 5: Assistente Q&A Inteligente
html.Div(
[
dbc.Row(
[
dbc.Col(
dbc.Input(
id="qa-input-v2",
placeholder="Ex: Qual é a cobertura de Bouches-du-Rhône? / Quem lidera em Var? / Comparação de depts",
type="text",
style={"background-color": "#0a0c10", "border": "1px solid #1e2538", "color": "#fff", "height": "38px"}
),
md=10
),
dbc.Col(
dbc.Button(
"Consultar Agente",
id="qa-btn-v2",
color="info",
className="w-100",
style={"height": "38px", "font-weight": "bold"}
),
md=2
)
],
className="g-2 mb-2"
),
html.Div(
id="qa-output-area-v2",
style={
"background-color": "#0a0c10",
"border": "1px solid #1e2538",
"border-radius": "8px",
"padding": "12px",
"height": "170px",
"overflow-y": "auto"
}
)
],
id="tab-content-qa",
style={"display": "none"}
)
],
style={"height": "100%"}
),
style={"background-color": "#121620", "height": "calc(100% - 42px)", "padding": "15px", "overflow-y": "auto"}
)
],
style={
"border": custom_styles["border"],
"border-radius": "8px",
"overflow": "hidden",
"flex": "1 1 auto",
"display": "flex",
"flex-direction": "column",
"min-height": "320px",
"height": "calc(62vh - 150px)"
}
)
],
style={
"margin-top": "75px",
"margin-left": "320px",
"padding": "20px",
"background-color": custom_styles["bg"],
"min-height": "calc(100vh - 75px)",
"display": "flex",
"flex-direction": "column",
"overflow-y": "auto"
}
)
],
style={"background-color": custom_styles["bg"], "min-height": "100vh"}
)
# ─── 5. Lógica de Callbacks do Dashboard ────────────────────────────────────
# 1. Chaveador de Abas
@app.callback(
[
Output("tab-content-history", "style"),
Output("tab-content-market", "style"),
Output("tab-content-report-md", "style"),
Output("tab-content-glossary", "style"),
Output("tab-content-qa", "style")
],
[Input("tabs-panel-v2", "active_tab")]
)
def toggle_tabs(active_tab):
styles = [{"display": "none"} for _ in range(5)]
tabs_map = {
"tab-history": 0,
"tab-market": 1,
"tab-report-md": 2,
"tab-glossary": 3,
"tab-qa": 4
}
if active_tab in tabs_map:
styles[tabs_map[active_tab]] = {"display": "block"}
return tuple(styles)
# 1b. Cascade ADM1 → ADM2: Região filtra Departamentos
@app.callback(
[Output("dep-selector", "options"),
Output("dep-selector", "value")],
[Input("region-selector", "value")]
)
def update_dept_by_region(region_value):
region_list = region_value if isinstance(region_value, list) else [region_value]
national_opt = [{"label": "🇫🇷 Visão Geral Nacional", "value": "national"}]
if not region_list or "all" in region_list:
return depts_options, ["national"]
filtered = df_summary_sorted[df_summary_sorted['region_name'].isin(region_list)]
opts = national_opt + [
{"label": f"📍 {r['dep_code']} - {r['dep_name']}", "value": r['dep_code']}
for _, r in filtered.iterrows()
]
return opts, ["national"]
def update_commune_by_dept(dep_value):
dep_list = dep_value if isinstance(dep_value, list) else [dep_value]
if not dep_list or "national" in dep_list:
return [{"label": "📊 Todo o Departamento", "value": "all"}], ["all"], True, "Selecione um departamento para ver comunas"
dep_val = dep_list[0]
communes_file = DATA_DIR / "communes_by_dep" / f"communes_{dep_val}.csv"
if not communes_file.exists():
return [], ["all"], True, "Sem dados de communes"
df_comm = pd.read_csv(communes_file).sort_values('NOM_COM')
opts = [{"label": "📊 Todo o Departamento", "value": "all"}]
opts += [{"label": f"🏘️ {row['NOM_COM']}", "value": str(row['INSEE_COM'])} for _, row in df_comm.iterrows()]
return opts, ["all"], False, "Selecione uma commune..."
# 2. KPI Cards e Links de Download Dinâmicos
@app.callback(
[
Output("kpi-raccordables", "children"),
Output("kpi-raccordables-sub", "children"),
Output("kpi-estimados", "children"),
Output("kpi-estimados-sub", "children"),
Output("kpi-cobertura", "children"),
Output("kpi-cobertura-status", "children"),
Output("kpi-cobertura-status", "style"),
Output("kpi-operator", "children"),
Output("kpi-pmz-total", "children"),
Output("download-report-link", "href"),
Output("download-communes-link", "href"),
Output("download-zapm-link", "href"),
Output("download-zapm-link", "style")
],
[Input("dep-selector", "value"),
Input("commune-selector", "value")]
)
def update_kpis(dep_value, commune_value):
dep_value = (dep_value[0] if dep_value else "national") if isinstance(dep_value, list) else dep_value
commune_value = (commune_value[0] if commune_value else "all") if isinstance(commune_value, list) else commune_value
# ── Nível ADM3: Commune específica ────────────────────────────────────
if dep_value and dep_value != "national" and commune_value and commune_value != "all":
communes_file = DATA_DIR / "communes_by_dep" / f"communes_{dep_value}.csv"
if communes_file.exists():
df_comm = pd.read_csv(communes_file)
df_comm['INSEE_COM'] = df_comm['INSEE_COM'].astype(str)
row_c = df_comm[df_comm['INSEE_COM'] == str(commune_value)]
if not row_c.empty:
row_c = row_c.iloc[0]
racc = int(row_c['locaux_raccordables'])
estim = int(row_c['locaux_estim'])
pmzs = int(row_c['pmz_count'])
cob = float(row_c['cobertura_pct'])
nom = row_c['NOM_COM']
cob_status = "🟢 Alta Cobertura" if cob >= 90 else ("🟡 Transição Ativa" if cob >= 75 else "🔴 Crítico / Expansão")
cob_color = custom_styles["accent-green"] if cob >= 90 else (custom_styles["accent-orange"] if cob >= 75 else custom_styles["accent-red"])
return (
f"{racc:,}", f"Commune: {nom}",
f"{estim:,}", "Estimativa local ARCEP",
f"{cob:.1f}%", cob_status, {"color": cob_color, "font-weight": "bold"},
"N/D", f"{pmzs} PMZs na commune",
f"/download/report/{dep_value}",
f"/download/communes/{dep_value}",
f"/download/zapm/{dep_value}", {"display": "block"}
)
if dep_value == "national":
# Somar dados de todos os departamentos com dados estimados válidos
df_valid = df_summary[df_summary['locaux_estim'] > 0]
racc = int(df_valid['locaux_raccordables'].sum())
estim = int(df_valid['locaux_estim'].sum())
cobertura = (racc / estim * 100) if estim > 0 else 0
total_pmzs = int(df_summary['pmz_total'].sum())
# Operador dominante geral nacional
dominant_op = "Orange"
sub_text = "Consolidado Nacional (França)"
op_text = f"{dominant_op}"
pmz_sub = f"{total_pmzs:,} PMZs mapeados"
download_url = "/download/report/national"
download_communes = "/download/communes/national"
download_zapm = "#"
zapm_style = {"display": "none"}
racc_str = f"{racc:,}"
estim_str = f"{estim:,}"
estim_sub_text = "Consolidado nacional ZAPM/ARCEP"
cobertura_str = f"{cobertura:.1f}%"
cobertura_status = "🟢 Cobertura Monitorada (França)"
cobertura_color = custom_styles["accent-green"]
else:
# Departamento específico
row = df_summary[df_summary['dep_code'] == dep_value].iloc[0]
racc = int(row['locaux_raccordables'])
estim = int(row['locaux_estim'])
cobertura = row['cobertura_pct']
total_pmzs = int(row['pmz_total'])
dominant_op = row['operador_dominante']
sub_text = f"Departamento {dep_value}"
op_text = dominant_op if pd.notna(dominant_op) and str(dominant_op).strip() != "" else "N/D"
pmz_sub = f"{total_pmzs:,} PMZs ativos" if total_pmzs > 0 else "Sem dados PMZ"
download_url = f"/download/report/{dep_value}"
download_communes = f"/download/communes/{dep_value}"
download_zapm = f"/download/zapm/{dep_value}"
zapm_style = {"display": "block"}
if estim > 0:
racc_str = f"{racc:,}"
estim_str = f"{estim:,}"
cobertura_str = f"{cobertura:.1f}%"
cobertura_status = "🟢 Alta Cobertura" if cobertura >= 90 else ("🟡 Transição Ativa" if cobertura >= 75 else "🔴 Crítico / Expansão")
cobertura_color = custom_styles["accent-green"] if cobertura >= 90 else (custom_styles["accent-orange"] if cobertura >= 75 else custom_styles["accent-red"])
if int(row['logement_insee_2016']) > 0:
estim_sub_text = f"Censo ARCEP (Insee: {int(row['logement_insee_2016']):,})"
else:
estim_sub_text = "Estimativa agregada via ZAPM"
else:
racc_str = "N/D (ARCEP)"
estim_str = "N/D"
estim_sub_text = "Dados ZAPM/ARCEP ausentes"
cobertura_str = "N/D"
cobertura_status = "⚪ Sem Polígonos ZAPM"
cobertura_color = custom_styles["text-muted"]
return (
racc_str,
sub_text,
estim_str,
estim_sub_text,
cobertura_str,
cobertura_status,
{"color": cobertura_color, "font-weight": "bold"},
op_text,
pmz_sub,
download_url,
download_communes,
download_zapm,
zapm_style
)
# 3. Gráficos de Histórico
@app.callback(
[
Output("chart-history-cobertura", "figure"),
Output("chart-history-locaux", "figure")
],
[Input("dep-selector", "value")]
)
def update_history_charts(dep_value):
dep_value = (dep_value[0] if dep_value else "national") if isinstance(dep_value, list) else dep_value
if dep_value == "national":
df_hist = df_national_hist.copy()
title_prefix = "Nacional (PACA)"
else:
df_hist = df_dept_hist[df_dept_hist['dep_code'] == dep_value].copy()
title_prefix = f"Dept {dep_value}"
if df_hist.empty or len(df_hist) == 0:
# Tentar carregar dados de comunas para este departamento como fallback
communes_file = DATA_DIR / "communes_by_dep" / f"communes_{dep_value}.csv"
if communes_file.exists():
df_comm = pd.read_csv(communes_file)
if not df_comm.empty:
# ── Gráfico 1: Distribuição de cobertura — Top 5 + Bottom 5 para mostrar contraste ──
# Ordenar e selecionar top 5 maiores + 5 menores (com dados válidos)
df_valid_pct = df_comm[df_comm["cobertura_pct"] > 0].sort_values("cobertura_pct", ascending=False)
n = len(df_valid_pct)
if n >= 10:
df_top = df_valid_pct.head(5)
df_bot = df_valid_pct.tail(5)
df_comm_cob = pd.concat([df_top, df_bot]).drop_duplicates().sort_values("cobertura_pct", ascending=True)
chart_title_cob = "Cobertura FTTH — Top 5 e Bottom 5 Comunas (%)"
else:
df_comm_cob = df_valid_pct.sort_values("cobertura_pct", ascending=True)
chart_title_cob = "Cobertura FTTH por Comunas (%)"
# Palette: gradiente do vermelho ao ciano pelo índice (mostra contraste real)
n_bars = len(df_comm_cob)
bar_colors = [
f"rgba({int(220 - 180*(i/(max(n_bars-1,1))))}, {int(80 + 160*(i/(max(n_bars-1,1))))}, {int(100 + 155*(i/(max(n_bars-1,1))))}, 0.9)"
for i in range(n_bars)
]
fig_cob = go.Figure()
fig_cob.add_trace(go.Bar(
x=df_comm_cob["cobertura_pct"],
y=df_comm_cob["NOM_COM"],
orientation="h",
marker=dict(
color=bar_colors,
line=dict(color="rgba(255,255,255,0.05)", width=0.5)
),
text=[f"{v:.1f}%" for v in df_comm_cob["cobertura_pct"]],
textposition="outside",
textfont=dict(size=9, color="#c0c8d8"),
hovertemplate="<b>%{y}</b><br>Cobertura: %{x:.1f}%<extra></extra>"
))
fig_cob.update_layout(
title=dict(text=chart_title_cob, font=dict(size=10, color="#7a8aaa"), x=0, pad=dict(l=0)),
margin={"r": 60, "t": 28, "l": 10, "b": 30},
paper_bgcolor="#121620",
plot_bgcolor="#0e1118",
font=dict(color="#7a8aaa", size=9),
xaxis=dict(
gridcolor="#1e2538", showline=False, ticksuffix="%",
range=[0, 115], zeroline=False, fixedrange=True
),
yaxis=dict(
gridcolor="#1e2538", showline=False, tickfont=dict(size=8.5),
automargin=True
),
bargap=0.25,
showlegend=False
)
# ── Gráfico 2: Top 10 por volume de locais raccordables ──
df_comm_racc = df_comm[df_comm["locaux_raccordables"] > 0].sort_values("locaux_raccordables", ascending=True).tail(10)
n_racc = len(df_comm_racc)
# Palette: azul escuro → ciano brilhante pelo volume
racc_colors = [
f"rgba({int(30 + 30*(i/(max(n_racc-1,1))))}, {int(130 + 100*(i/(max(n_racc-1,1))))}, {int(190 + 65*(i/(max(n_racc-1,1))))}, 0.85)"
for i in range(n_racc)
]
fig_loc = go.Figure()
fig_loc.add_trace(go.Bar(
x=df_comm_racc["locaux_raccordables"],
y=df_comm_racc["NOM_COM"],
orientation="h",
marker=dict(
color=racc_colors,
line=dict(color="rgba(255,255,255,0.05)", width=0.5)
),
text=[f"{int(v):,}" for v in df_comm_racc["locaux_raccordables"]],
textposition="outside",
textfont=dict(size=9, color="#c0c8d8"),
hovertemplate="<b>%{y}</b><br>Locais raccordables: %{x:,}<extra></extra>"
))
fig_loc.update_layout(
title=dict(text="Top 10 Comunas por Volume de Locais Raccordables", font=dict(size=10, color="#7a8aaa"), x=0, pad=dict(l=0)),
margin={"r": 70, "t": 28, "l": 10, "b": 30},
paper_bgcolor="#121620",
plot_bgcolor="#0e1118",
font=dict(color="#7a8aaa", size=9),
xaxis=dict(
gridcolor="#1e2538", showline=False, zeroline=False,
fixedrange=True
),
yaxis=dict(
gridcolor="#1e2538", showline=False, tickfont=dict(size=8.5),
automargin=True
),
bargap=0.25,
showlegend=False
)
return fig_cob, fig_loc
# Criar figuras de aviso se o histórico e comunas estiverem indisponíveis
fig_cob = go.Figure()
fig_cob.add_annotation(
text="Dados históricos de cobertura ARCEP indisponíveis para este depto.",
xref="paper", yref="paper", x=0.5, y=0.5, showarrow=False,
font=dict(size=12, color=custom_styles["text-muted"])
)
fig_cob.update_layout(
margin={"r": 10, "t": 10, "l": 10, "b": 10},
paper_bgcolor="#121620",
plot_bgcolor="#121620",
xaxis=dict(visible=False),
yaxis=dict(visible=False)
)
fig_loc = go.Figure()
fig_loc.add_annotation(
text="Consulte ativos físicos mapeados e ZAPMs na aba de Mapa e Relatórios.",
xref="paper", yref="paper", x=0.5, y=0.5, showarrow=False,
font=dict(size=11, color=custom_styles["text-muted"])
)
fig_loc.update_layout(
margin={"r": 10, "t": 10, "l": 10, "b": 10},
paper_bgcolor="#121620",
plot_bgcolor="#121620",
xaxis=dict(visible=False),
yaxis=dict(visible=False)
)
return fig_cob, fig_loc
df_hist['período'] = df_hist['annee'].astype(str) + " " + df_hist['trimestre'].astype(str)
df_hist = df_hist.sort_values(by=['annee', 'trimestre'])
# Gráfico 1: Cobertura %
fig_cob = go.Figure()
fig_cob.add_trace(go.Scatter(
x=df_hist['período'],
y=df_hist['cobertura_pct'],
mode='lines+markers',
line=dict(color=custom_styles["accent-green"], width=3),
marker=dict(size=6, color="#ffffff", line=dict(color=custom_styles["accent-green"], width=2)),
name='Cobertura %',
hovertemplate='<b>Período:</b> %{x}<br><b>Cobertura:</b> %{y:.1f}%<br><extra></extra>'
))
fig_cob.update_layout(
margin={"r": 10, "t": 10, "l": 40, "b": 30},
paper_bgcolor="#121620",
plot_bgcolor="#121620",
font=dict(color=custom_styles["text-muted"], size=9),
xaxis=dict(gridcolor="#1e2538", showline=True, linecolor="#1e2538"),
yaxis=dict(gridcolor="#1e2538", ticksuffix="%", range=[0, 100]),
showlegend=False
)
# Gráfico 2: Conectáveis vs Estimados
fig_loc = go.Figure()
fig_loc.add_trace(go.Bar(
x=df_hist['período'],
y=df_hist['locaux_raccordables'],
name='Conectáveis (Fibra)',
marker_color=custom_styles["accent-blue"]
))
fig_loc.add_trace(go.Scatter(
x=df_hist['período'],
y=df_hist['locaux_estim'],
name='Locais Estimados',
line=dict(color="#ffffff", dash='dash', width=2),
hovertemplate='<b>Locais Estimados:</b> %{y:,.0f}<br><extra></extra>'
))
fig_loc.update_layout(
margin={"r": 10, "t": 10, "l": 40, "b": 30},
paper_bgcolor="#121620",
plot_bgcolor="#121620",
font=dict(color=custom_styles["text-muted"], size=9),
xaxis=dict(gridcolor="#1e2538"),
yaxis=dict(gridcolor="#1e2538"),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1, bgcolor="rgba(0,0,0,0)", font=dict(color="#fff", size=8))
)
return fig_cob, fig_loc
# 4. Gráfico e Tabela de Operadores (Aba Participação)
@app.callback(
[
Output("chart-operator-pie", "figure"),
Output("table-operator-data", "children")
],
[
Input("dep-selector", "value"),
Input("operator-selector", "value")
]
)
def update_operator_charts(dep_value, selected_ops):
dep_value = (dep_value[0] if dep_value else "national") if isinstance(dep_value, list) else dep_value
# Filtrar PMZs pelo departamento
if dep_value == "national":
df_pmz_filtered = df_pmz.copy()
else:
df_pmz_filtered = df_pmz[df_pmz['dep_code'] == dep_value].copy()
if len(df_pmz_filtered) == 0:
return go.Figure(), "Sem dados disponíveis"
# Agrupar operadora
def normalize_op(op):
op_str = str(op).strip()
if "Orange" in op_str: return "Orange"
if "SFR" in op_str: return "SFR"
if "Free" in op_str: return "Free"
if "Bouygues" in op_str: return "Bouygues Telecom"
return "Outros"
df_pmz_filtered['operator_group'] = df_pmz_filtered['operator'].apply(normalize_op)
# Filtrar operadores selecionados na checklist
df_pmz_filtered = df_pmz_filtered[df_pmz_filtered['operator_group'].isin(selected_ops)]
if len(df_pmz_filtered) == 0:
return go.Figure(), "Nenhum operador selecionado"
counts = df_pmz_filtered['operator_group'].value_counts()
labels = counts.index.tolist()
values = counts.values.tolist()
# Cores fixas corporativas
op_colors = {
"Orange": "#ff7900",
"SFR": "#e2001a",
"Free": "#bd00ff",
"Bouygues Telecom": "#009ebd",
"Outros": "#39ff14"
}
colors = [op_colors.get(lbl, "#555") for lbl in labels]
fig = go.Figure(data=[go.Pie(labels=labels, values=values, hole=.4, marker=dict(colors=colors))])
fig.update_layout(
margin={"r": 10, "t": 10, "l": 10, "b": 10},
paper_bgcolor="#121620",
plot_bgcolor="#121620",
font=dict(color=custom_styles["text-muted"], size=9),
showlegend=True,
legend=dict(orientation="v", yanchor="middle", y=0.5, xanchor="right", x=1.1, font=dict(color="#fff", size=8))
)
# Criar tabela HTML
total = sum(values)
table_header = html.Thead(
html.Tr([html.Th("Operador"), html.Th("PMZs Ativos"), html.Th("% Share")]),
style={"background-color": "#1e2538", "color": "#fff", "font-size": "12px"}
)
rows = []
for lbl, val in zip(labels, values):
pct = (val / total * 100) if total > 0 else 0
rows.append(html.Tr([
html.Td([
html.Span(style={"display": "inline-block", "width": "10px", "height": "10px", "backgroundColor": op_colors.get(lbl, "#555"), "marginRight": "8px", "borderRadius": "50%"}),
lbl
]),
html.Td(f"{val:,}"),
html.Td(f"{pct:.1f}%", style={"font-weight": "bold", "color": custom_styles["accent-blue"]})
]))
table_body = html.Tbody(rows, style={"font-size": "11.5px", "color": "#f0f4f8"})
table_html = dbc.Table(
[table_header, table_body],
bordered=True,
hover=True,
responsive=True,
striped=True,
style={"border-color": "#1e2538"}
)
return fig, table_html
# 5. Renderização do Relatório Markdown Original na Aba (.md)
@app.callback(
Output("report-markdown-container", "children"),
[Input("dep-selector", "value")]
)
def load_report_markdown(dep_value):
dep_value = (dep_value[0] if dep_value else "national") if isinstance(dep_value, list) else dep_value
try:
# ── Carregar conteúdo base do ficheiro .md ──────────────────────────
if dep_value == "national":
file_path = REPORTS_DIR / "national_summary.md"
else:
file_path = REPORTS_DIR / "departments" / f"dep_{dep_value}.md"
base_content = ""
if file_path.exists():
with open(file_path, "r", encoding="utf-8") as f:
base_content = f.read()
else:
base_content = "_Relatório base não encontrado no servidor._"
# ── Gerar Secção de Métricas Complementares ─────────────────────────
extra_md = "\n\n---\n\n## 📐 Métricas Técnicas Complementares\n\n"
if dep_value == "national":
# Dados nacionais agregados
df_valid = df_summary[df_summary['locaux_estim'] > 0]
total_racc = int(df_valid['locaux_raccordables'].sum())
total_estim = int(df_valid['locaux_estim'].sum())
total_pmz = int(df_summary['pmz_total'].sum())
cobertura = (total_racc / total_estim * 100) if total_estim > 0 else 0
# 1. Densidade média por PMZ (nacional)
dens_pmz = round(total_racc / total_pmz, 1) if total_pmz > 0 else 0
# 2. % dependente de cobre (estimado)
pct_cobre = round(100.0 - cobertura, 1)
# 3. PBOs estimados (1 PBO / 10 locais raccordables — norma ARCEP)
pbos_est = total_racc // 10
# 4. Taxa urbana/rural — heurística nacional via média de coberturas
# Comunas com cobertura >= 75% e >= 500 locais = "urbana"
df_comm_all = pd.read_csv(DATA_DIR / "communes_summary.csv") \
if (DATA_DIR / "communes_summary.csv").exists() else pd.DataFrame()
if not df_comm_all.empty:
urban = df_comm_all[(df_comm_all['cobertura_pct'] >= 75) & (df_comm_all['locaux_estim'] >= 500)]
pct_urban = round(len(urban) / len(df_comm_all) * 100, 1) if len(df_comm_all) > 0 else 0
pct_rural = round(100.0 - pct_urban, 1)
taxa_urb_str = f"~{pct_urban}% urbana / ~{pct_rural}% rural (heurística ZAPM)"
else:
taxa_urb_str = "N/D (dados de comunas não disponíveis)"
# 5. Classificação de zona (heurística nacional)
if cobertura >= 90:
zona_class = "**Zone Dense** (cobertura ≥ 90% — área metropolitana predominante)"
elif cobertura >= 70:
zona_class = "**AMII** (Appel à Manifestations d'Intentions d'Investissements — 70-90%)"
else:
zona_class = "**RIP** (Réseau d'Initiative Publique — cobertura < 70%, área de interesse público)"
extra_md += f"""| Métrica | Valor |
| :--- | :--- |
| **Densidade média por PMZ** | {dens_pmz:,.0f} locais raccordables / PMZ |
| **Taxa urbana vs rural (estimada)** | {taxa_urb_str} |
| **% estimado ainda em cobre** | ~{pct_cobre}% dos locais sem fibra elegível |
| **PBOs estimados** | ~{pbos_est:,} PBOs (norma: 1 PBO / 10 locais) |
| **Classificação de zona predominante** | {zona_class} |
> ⚠️ *Taxa urbana/rural e classificação de zona são estimativas heurísticas baseadas em dados ZAPM/ARCEP. Para dados oficiais, consulte o zonage ARCEP/DINUM.*
"""
else:
# Departamento específico
rows_dep = df_summary[df_summary['dep_code'] == dep_value]
if rows_dep.empty:
return [dcc.Markdown(base_content), dcc.Markdown("_Métricas complementares: departamento não encontrado._")]
row = rows_dep.iloc[0]
racc = int(row['locaux_raccordables'])
estim = int(row['locaux_estim'])
pmz_total = int(row['pmz_total'])
cobertura = float(row['cobertura_pct'])
# 1. Densidade média por PMZ
dens_pmz = round(racc / pmz_total, 1) if pmz_total > 0 else 0
dens_str = f"{dens_pmz:,.0f} locais raccordables / PMZ" if pmz_total > 0 else "N/D (sem PMZs mapeados)"
# 2. % dependente de cobre (estimado)
pct_cobre = round(100.0 - cobertura, 1) if estim > 0 else None
cobre_str = f"~{pct_cobre}% dos locais sem fibra elegível" if pct_cobre is not None else "N/D (dados de cobertura ausentes)"
# 3. PBOs estimados
pbos_est = racc // 10 if racc > 0 else 0
pbos_str = f"~{pbos_est:,} PBOs (norma: 1 PBO / 10 locais)" if racc > 0 else "N/D"
# 4. Taxa urbana/rural — via communes deste departamento
communes_path = DATA_DIR / "communes_by_dep" / f"communes_{dep_value}.csv"
if communes_path.exists():
df_comm = pd.read_csv(communes_path)
urban = df_comm[(df_comm['cobertura_pct'] >= 75) & (df_comm['locaux_estim'] >= 500)]
pct_urban = round(len(urban) / len(df_comm) * 100, 1) if len(df_comm) > 0 else 0
pct_rural = round(100.0 - pct_urban, 1)
n_urban = len(urban)
n_rural = len(df_comm) - n_urban
taxa_urb_str = f"~{pct_urban}% urbana ({n_urban} comunas) / ~{pct_rural}% rural ({n_rural} comunas)"
else:
taxa_urb_str = "N/D (dados de comunas não disponíveis)"
# 5. Classificação de zona (heurística por cobertura + densidade PMZ)
if cobertura >= 90 and dens_pmz >= 5000:
zona_class = "**Zone Dense** — cobertura ≥ 90% e alta densidade urbana"
elif cobertura >= 90:
zona_class = "**Zone Dense / AMII** — alta cobertura com densidade moderada"
elif cobertura >= 70:
zona_class = "**AMII** (Appel à Manifestations d'Intentions d'Investissements)"
elif cobertura > 0:
zona_class = "**RIP** (Réseau d'Initiative Publique) — território prioritário"
else:
zona_class = "**Indeterminada** — dados de cobertura insuficientes"
extra_md += f"""| Métrica | Valor |
| :--- | :--- |
| **Densidade média por PMZ** | {dens_str} |
| **Taxa urbana vs rural (estimada)** | {taxa_urb_str} |
| **% estimado ainda em cobre** | {cobre_str} |
| **PBOs estimados** | {pbos_str} |
| **Classificação de zona** | {zona_class} |
> ⚠️ *Taxa urbana/rural e classificação de zona são estimativas heurísticas baseadas em dados ZAPM. Para o zonage oficial, consulte o ficheiro de zoneamento ARCEP/DINUM.*
"""
# ── Renderizar ambos os blocos como componentes separados ───────────
return [
dcc.Markdown(base_content, dangerously_allow_html=False),
dcc.Markdown(extra_md, dangerously_allow_html=False)
]
except Exception as e:
return f"Erro ao gerar relatório: {str(e)}"
# 6. Mapa Mapbox Interativo e Responsivo
@app.callback(
Output("mapbox-map-v2", "figure"),
[
Input("dep-selector", "value"),
Input("operator-selector", "value"),
Input("map-layers", "value"),
Input("region-selector", "value"),
Input("commune-selector", "value")
]
)
def update_map(dep_value, selected_ops, layers, region_value, commune_value):
region_list = region_value if isinstance(region_value, list) else [region_value]
dep_list = dep_value if isinstance(dep_value, list) else [dep_value]
comm_list = commune_value if isinstance(commune_value, list) else [commune_value]
# Extract first items for zoom logic and compatibility
region_value = region_list[0] if region_list else "all"
if "all" in region_list: region_value = "all"
dep_value = dep_list[0] if dep_list else "national"
if "national" in dep_list: dep_value = "national"
commune_value = comm_list[0] if comm_list else "all"
if "all" in comm_list: commune_value = "all"
fig = go.Figure()
# 1. Obter Centróide e Zoom
if dep_value != "national" and commune_value and commune_value not in ("all", None):
# Nível ADM3: zoom na commune
row = df_summary[df_summary['dep_code'] == dep_value].iloc[0]
# Verificar se existe centroide exato da commune
if not df_commune_centroids.empty:
match = df_commune_centroids[df_commune_centroids['name_norm'] == str(commune_value).lower()]
if not match.empty:
lat_center = float(match.iloc[0]['lat'])
lon_center = float(match.iloc[0]['lon'])
zoom_level = 12.0
else:
lat_center = row['centroid_lat']
lon_center = row['centroid_lon']
zoom_level = 11.0
else:
# Fallback
df_pmz_dep = df_pmz[df_pmz['dep_code'] == dep_value]
if not df_pmz_dep.empty:
lat_center = float(df_pmz_dep['lat'].mean())
lon_center = float(df_pmz_dep['lon'].mean())
else:
lat_center = row['centroid_lat']
lon_center = row['centroid_lon']
zoom_level = 11.0
elif dep_value != "national":
# Nível ADM2: zoom no departamento
row = df_summary[df_summary['dep_code'] == dep_value].iloc[0]
lat_center = row['centroid_lat']
lon_center = row['centroid_lon']
zoom_level = 9.0
elif region_value and region_value != "all" and region_value in region_centroids:
# Nível ADM1: zoom na região
lat_center = region_centroids[region_value]['lat']
lon_center = region_centroids[region_value]['lon']
zoom_level = 7.0
else:
# Nível nacional
lat_center, lon_center = 46.5, 2.5
zoom_level = 5.0
# 2a. Fronteiras de Regiões (ADM1) — linha amarela acima dos departamentos
if "region_boundaries" in layers:
region_ids = [f['id'] for f in geojson_regions['features']]
region_names = [f['properties'].get('NAME_1', 'Região') for f in geojson_regions['features']]
if region_value and region_value != "all":
z_vals = [1 if name in region_list else 0 for name in region_names] if "all" not in region_list else [0] * len(region_ids)
else:
z_vals = [0] * len(region_ids)
fig.add_trace(go.Choroplethmap(
geojson=geojson_regions,
locations=region_ids,
z=z_vals,
featureidkey="id",
colorscale=[[0, "rgba(255, 200, 0, 0.02)"], [1, "rgba(255, 200, 0, 0.25)"]],
marker_line_color="#ffc800",
marker_line_width=2.5,
showscale=False,
text=region_names,
hovertemplate="<b>%{text}</b><extra></extra>"
))
# 2b. Fronteiras de Departamentos (ADM2)
if "boundaries" in layers:
locations = [f['properties']['dep_code'] for f in geojson_depts['features']]
text_vals = [f['properties'].get('NAME_2', '') for f in geojson_depts['features']]
if "national" not in dep_list:
z_vals = [1 if loc in dep_list else 0 for loc in locations]
elif "all" not in region_list:
deps_regiao = set(df_summary_sorted[df_summary_sorted['region_name'].isin(region_list)]['dep_code'])
z_vals = [1 if loc in deps_regiao else 0 for loc in locations]
else:
z_vals = [0] * len(locations)
fig.add_trace(go.Choroplethmap(
geojson=geojson_depts,
locations=locations,
z=z_vals,
featureidkey="properties.dep_code",
colorscale=[[0, "rgba(0, 210, 255, 0.02)"], [1, "rgba(0, 210, 255, 0.25)"]],
marker_line_color=custom_styles["accent-blue"],
marker_line_width=2.0,
showscale=False,
text=text_vals,
hovertemplate="<b>%{text}</b><extra></extra>"
))
# 2c. Fronteiras de Arrondissements (ADM3)
if "adm3" in layers and geojson_adm3:
if "national" not in dep_list:
features_adm3 = [f for f in geojson_adm3['features'] if f['properties'].get('dep_code') in dep_list]
geojson_adm3_filtered = {"type": "FeatureCollection", "features": features_adm3}
else:
geojson_adm3_filtered = geojson_adm3
features_adm3 = geojson_adm3['features']
adm3_ids = [f['id'] for f in features_adm3]
adm3_names = [f['properties'].get('shapeName', 'Arrondissement') for f in features_adm3]
fig.add_trace(go.Choroplethmap(
geojson=geojson_adm3_filtered,
locations=adm3_ids,
z=[1] * len(adm3_ids),
featureidkey="id",
colorscale=[[0, "rgba(255, 100, 100, 0.01)"], [1, "rgba(255, 100, 100, 0.01)"]],
marker_line_color="#ff6b6b",
marker_line_width=1.5,
showscale=False,
text=adm3_names,
hovertemplate="<b>Arrondissement: %{text}</b><extra></extra>"
))
# 2d. Fronteiras de Cantons (ADM4)
if "adm4" in layers and geojson_adm4:
if "national" not in dep_list:
features_adm4 = [f for f in geojson_adm4['features'] if f['properties'].get('dep_code') in dep_list]
geojson_adm4_filtered = {"type": "FeatureCollection", "features": features_adm4}
else:
geojson_adm4_filtered = geojson_adm4
features_adm4 = geojson_adm4['features']
adm4_ids = [f['id'] for f in features_adm4]
adm4_names = [f['properties'].get('shapeName', 'Canton') for f in features_adm4]
fig.add_trace(go.Choroplethmap(
geojson=geojson_adm4_filtered,
locations=adm4_ids,
z=[1] * len(adm4_ids),
featureidkey="id",
colorscale=[[0, "rgba(100, 255, 100, 0.01)"], [1, "rgba(100, 255, 100, 0.01)"]],
marker_line_color="#4cd137",
marker_line_width=1.0,
showscale=False,
text=adm4_names,
hovertemplate="<b>Canton: %{text}</b><extra></extra>"
))
# 3. Adicionar camada de Polígonos de ZAPM (apenas no nível do departamento)
if "zapm" in layers and dep_value != "national":
zapm_geojson_path = Path("data_processed") / "zapm_by_dep" / f"zapm_{dep_value}.geojson"
if zapm_geojson_path.exists():
try:
with open(zapm_geojson_path, "r", encoding="utf-8") as f:
geojson_zapm = json.load(f)
# Para colorir, criamos um ID e um z fictício para cada feature
z_vals_zapm = [1 for _ in range(len(geojson_zapm['features']))]
locations_zapm = [f['properties']['id'] if 'id' in f['properties'] else str(i) for i, f in enumerate(geojson_zapm['features'])]
# Garantir que as features tenham o ID correspondente no GeoJSON
for i, f in enumerate(geojson_zapm['features']):
if 'id' not in f['properties']:
f['properties']['id'] = str(i)
f['id'] = f['properties']['id']
fig.add_trace(go.Choroplethmap(
geojson=geojson_zapm,
locations=locations_zapm,
z=z_vals_zapm,
featureidkey="id",
colorscale=[[0, "rgba(57, 255, 20, 0.06)"], [1, "rgba(57, 255, 20, 0.06)"]],
marker_line_color="#39ff14",
marker_line_width=0.8,
showscale=False,
name="Zona ZAPM",
text=[f['properties'].get('ref_pmz', 'N/A') for f in geojson_zapm['features']],
hovertemplate="<b>ZAPM Ref:</b> %{text}<br><extra></extra>"
))
except Exception as e:
print(f"Erro ao renderizar GeoJSON de ZAPM: {e}")
# 4. Plotar Pontos de Mutualização (PMZs)
if "pmz" in layers:
# Filtrar PMZs pelo departamento e operadoras
if dep_value == "national":
df_pmz_filtered = df_pmz.copy()
else:
df_pmz_filtered = df_pmz[df_pmz['dep_code'] == dep_value].copy()
if not df_pmz_filtered.empty:
# Mapear operadoras
def normalize_op(op):
op_str = str(op).strip()
if "Orange" in op_str: return "Orange"
if "SFR" in op_str: return "SFR"
if "Free" in op_str: return "Free"
if "Bouygues" in op_str: return "Bouygues Telecom"
return "Outros"
df_pmz_filtered['operator_group'] = df_pmz_filtered['operator'].apply(normalize_op)
df_pmz_filtered = df_pmz_filtered[df_pmz_filtered['operator_group'].isin(selected_ops)]
# Cores por operadora
op_colors = {
"Orange": "#ff7900",
"SFR": "#e2001a",
"Free": "#bd00ff",
"Bouygues Telecom": "#009ebd",
"Outros": "#39ff14"
}
for op, group in df_pmz_filtered.groupby('operator_group'):
fig.add_trace(go.Scattermap(
lat=group['lat'],
lon=group['lon'],
mode='markers',
marker=go.scattermap.Marker(
size=4.5 if dep_value != "national" else 2.5,
color=op_colors.get(op, "#888"),
opacity=0.75
),
name=f"PMZ {op}",
text=group['commune_name'],
hoverinfo='text'
))
# 5. Adicionar Nós de Rede Óptica (NROs)
if "nro" in layers and not df_nro.empty:
df_nro_filtered = df_nro if "national" in dep_list else df_nro[df_nro['dep_code'].isin(dep_list)]
if not df_nro_filtered.empty:
fig.add_trace(go.Scattermap(
lat=df_nro_filtered['lat'],
lon=df_nro_filtered['lon'],
mode='markers',
marker=go.scattermap.Marker(
size=6 if dep_value != "national" else 7,
color=custom_styles["accent-purple"],
opacity=0.9
),
name="NRO (Nó Óptico)",
text=df_nro_filtered['commune_name'],
hoverinfo='text'
))
# 6. Adicionar Centrais Telefônicas (NRAs)
if "nra" in layers and not df_nra.empty:
df_nra_filtered = df_nra if "national" in dep_list else df_nra[df_nra['dep_code'].isin(dep_list)]
if not df_nra_filtered.empty:
fig.add_trace(go.Scattermap(
lat=df_nra_filtered['lat'],
lon=df_nra_filtered['lon'],
mode='markers',
marker=go.scattermap.Marker(
size=5 if dep_value != "national" else 6,
color=custom_styles["accent-orange"],
opacity=0.9
),
name="NRA (Central Cobre)",
text=df_nra_filtered['commune_name'],
hoverinfo='text'
))
# 7. Adicionar Redes de Cobre Legadas (COPPER)
if "copper" in layers and not df_copper.empty:
df_copper_filtered = df_copper if "national" in dep_list else df_copper[df_copper['dep_code'].isin(dep_list)]
if not df_copper_filtered.empty:
fig.add_trace(go.Scattermap(
lat=df_copper_filtered['lat'],
lon=df_copper_filtered['lon'],
mode='markers',
marker=go.scattermap.Marker(
size=4 if dep_value != "national" else 5,
color=custom_styles["accent-red"],
opacity=0.8
),
name="COPPER (Rede Cobre)",
text=df_copper_filtered['commune_name'],
hoverinfo='text'
))
# 8. Adicionar Pontos de Transporte (PT)
if "pt" in layers and not df_pt.empty:
df_pt_filtered = df_pt if "national" in dep_list else df_pt[df_pt['dep_code'].isin(dep_list)]
if not df_pt_filtered.empty:
fig.add_trace(go.Scattermap(
lat=df_pt_filtered['lat'],
lon=df_pt_filtered['lon'],
mode='markers',
marker=go.scattermap.Marker(
size=7 if dep_value != "national" else 8,
color=custom_styles["accent-blue"],
opacity=0.9
),
name="PT (Ponto Transporte)",
text=df_pt_filtered['commune_name'],
hoverinfo='text'
))
fig.update_layout(
map_style="carto-darkmatter",
map=dict(
center=dict(lat=lat_center, lon=lon_center),
zoom=zoom_level
),
uirevision="constant",
margin={"r": 0, "t": 0, "l": 0, "b": 0},
paper_bgcolor=custom_styles["bg"],
plot_bgcolor=custom_styles["bg"],
showlegend=True,
legend=dict(
yanchor="top",
y=0.97,
xanchor="left",
x=0.01,
bgcolor="rgba(18, 22, 32, 0.85)",
bordercolor="#1e2538",
borderwidth=1,
font=dict(color="#fff", size=9)
)
)
return fig
# 7. Q&A Assistente com dados reais e relatórios Markdown
@app.callback(
Output("qa-output-area-v2", "children"),
[Input("qa-btn-v2", "n_clicks")],
[State("qa-input-v2", "value")]
)
def process_qa_system(n_clicks, query):
if not query or n_clicks is None:
return html.Div(
[
html.Span("Aguardando pergunta... Exemplo de perguntas suportadas:"),
html.Ul(
[
html.Li("Qual a cobertura do departamento 13?"),
html.Li("Quem é o operador dominante em Paris (75)?"),
html.Li("Quais são os departamentos críticos?"),
html.Li("Qual o total de PMZs em Vaucluse?")
],
style={"margin-top": "8px", "font-size": "12px", "color": custom_styles["text-muted"]}
)
],
style={"color": custom_styles["text-muted"], "font-size": "13px"}
)
# Normalizar busca
q_norm = unicodedata.normalize('NFKD', query).encode('ASCII', 'ignore').decode('utf-8').lower()
q_norm = re.sub(r'[^\w\s]', '', q_norm)
# Tentar extrair código de depto de 2 ou 3 dígitos
num_match = re.findall(r'\b\d{2,3}\b', q_norm)
selected_row = None
if num_match:
dep_code = num_match[0].zfill(2)
rows = df_summary[df_summary['dep_code'] == dep_code]
if not rows.empty:
selected_row = rows.iloc[0]
else:
# Tentar buscar por correspondência no nome do departamento
for _, r in df_summary.iterrows():
dep_name_norm = unicodedata.normalize('NFKD', r['dep_name']).encode('ASCII', 'ignore').decode('utf-8').lower()
if dep_name_norm in q_norm:
selected_row = r
break
# 1. Caso: Solicitação de relatório de cobertura / dados do depto
if selected_row is not None:
dep_code = selected_row['dep_code']
estim = int(selected_row['locaux_estim'])
racc = int(selected_row['locaux_raccordables'])
insee_log = int(selected_row['logement_insee_2016'])
ans = f"### 📊 Relatório de Infraestrutura - Dept {dep_code} ({selected_row['dep_name']})\n"
ans += f"- **Região Administrativa:** `{selected_row['region_name']}`\n"
if estim > 0:
ans += f"- **Taxa de Cobertura FTTH:** `{selected_row['cobertura_pct']}%` (ARCEP)\n"
ans += f"- **Locais Conectáveis (Fibra):** `{racc:,}`\n"
ans += f"- **Total de Domicílios Estimados:** `{estim:,}` (ARCEP)\n"
else:
ans += f"- **Taxa de Cobertura FTTH:** `N/D` (Dados ARCEP indisponíveis)\n"
ans += f"- **Domicílios Estimados (INSEE 2016):** `{insee_log:,}`\n"
ans += f"- **Operador Dominante de Infraestrutura:** `{selected_row['operador_dominante']}`\n"
ans += f"- **Pontos de Mutualização (PMZ):** `{int(selected_row['pmz_total']):,}` ativos mapeados em campo.\n"
# Estatísticas adicionais
dep_nro = len(df_nro[df_nro['dep_code'] == dep_code]) if not df_nro.empty else 0
dep_nra = len(df_nra[df_nra['dep_code'] == dep_code]) if not df_nra.empty else 0
ans += f"- **Ativos Físicos:** `{dep_nro}` NROs | `{dep_nra}` NRAs mapeados no mapa."
return dcc.Markdown(ans, style={"color": "#fff"})
# 2. Caso: Operador dominante ou líder genérico
if "dominante" in q_norm or "lider" in q_norm or "quem domina" in q_norm or "operador" in q_norm:
# Se não casou acima com depto específico, mostra resumo nacional do operador
op_counts = df_pmz['operator'].value_counts() if not df_pmz.empty else {}
ans = "### 🏆 Operadores Dominantes na França Metropolitana:\n\n"
ans += "A nível nacional (França metropolitana), a distribuição de PMZs por operadoras mapeadas é:\n\n"
for op, count in list(op_counts.items())[:5]:
ans += f"- **{op}:** `{count:,}` PMZs ativos mapeados.\n"
ans += "\nExperimente perguntar por um departamento específico (ex: *'Quem é o líder no departamento 83?'*)."
return dcc.Markdown(ans)
# 3. Caso: Departamentos críticos (menor cobertura, filtrando estim > 0)
if "critico" in q_norm or "pior" in q_norm or "menor" in q_norm:
df_sorted = df_summary[df_summary['locaux_estim'] > 0].sort_values(by="cobertura_pct")
ans = "### 🚨 TOP 10 Departamentos com Menor Cobertura FTTH (Monitorados):\n\n"
ans += "| Cód | Departamento | Região | Cobertura % | PMZs | Locais Totais |\n"
ans += "|---|---|---|---|---|---|\n"
for _, r in df_sorted.head(10).iterrows():
ans += f"| **{r['dep_code']}** | {r['dep_name']} | {r['region_name']} | `{r['cobertura_pct']}%` | {int(r['pmz_total'])} | {int(r['locaux_estim']):,} |\n"
return dcc.Markdown(ans)
# 4. Caso: Comparação geral/Ranking nacional
if "compar" in q_norm or "ranking" in q_norm or "melhores" in q_norm or "maior" in q_norm:
df_sorted = df_summary[df_summary['locaux_estim'] > 0].sort_values(by="cobertura_pct", ascending=False)
ans = "### 🏆 TOP 10 Departamentos com Maior Cobertura FTTH:\n\n"
ans += "| Cód | Departamento | Região | Cobertura % | PMZs | Locais Conectáveis |\n"
ans += "|---|---|---|---|---|---|\n"
for _, r in df_sorted.head(10).iterrows():
ans += f"| **{r['dep_code']}** | {r['dep_name']} | {r['region_name']} | `{r['cobertura_pct']}%` | {int(r['pmz_total'])} | {int(r['locaux_raccordables']):,} |\n"
return dcc.Markdown(ans)
return html.Div(
[
html.Span("🤖 Desculpe, não consegui identificar o departamento ou padrão na sua pergunta.", style={"color": custom_styles["accent-red"], "font-weight": "bold"}),
html.P("Experimente perguntar informando o código do departamento francês (ex: 13, 83, 75, 59) ou termos como 'críticos', 'líder', 'comparar' ou 'cobertura do Var'.", style={"margin-top": "8px", "font-size": "12.5px"})
]
)
# ─── 6. Execução do Servidor ────────────────────────────────────────────────
if __name__ == "__main__":
import os
debug_mode = os.environ.get("DASH_DEBUG", "false").lower() == "true"
print("🚀 Iniciando INFRAESTRUTURA FTTH FRANÇA — Dashboard em http://0.0.0.0:7860 ...")
app.run(host="0.0.0.0", port=7860, debug=debug_mode)