-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #63 from ImpulsoGov/painel_produtividade
ETL de dados de produção para o Painel de Produtividade do AGP
- Loading branch information
Showing
6 changed files
with
293 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
35 changes: 35 additions & 0 deletions
35
src/impulsoetl/sisab/relatorio_producao_profissional_conduta_tipo_atendimento/extracao.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,35 @@ | ||
import warnings | ||
warnings.filterwarnings("ignore") | ||
|
||
import pandas as pd | ||
from datetime import date | ||
|
||
from impulsoetl.loggers import logger, habilitar_suporte_loguru | ||
from impulsoetl.sisab.utilitarios_sisab_relatorio_producao import extrair_producao_por_municipio | ||
from impulsoetl.sisab.utilitarios_sisab_relatorio_producao import transformar_producao_por_municipio | ||
|
||
|
||
def extrair_relatorio( | ||
periodo_competencia: date)-> pd.DataFrame(): | ||
|
||
df_consolidado = pd.DataFrame() | ||
|
||
try: | ||
df_parcial = extrair_producao_por_municipio( | ||
tipo_producao="Atendimento individual", | ||
competencias=[periodo_competencia], | ||
selecoes_adicionais={ | ||
"Conduta":"Selecionar Todos", | ||
"Categoria do Profissional":"Selecionar Todos", | ||
"Tipo de Atendimento": "Selecionar Todos", | ||
}, | ||
|
||
).pipe(transformar_producao_por_municipio) | ||
|
||
df_consolidado = df_consolidado.append(df_parcial) | ||
|
||
except ValueError as e: | ||
logger.error(e) | ||
pass | ||
|
||
return df_consolidado |
54 changes: 54 additions & 0 deletions
54
src/impulsoetl/sisab/relatorio_producao_profissional_conduta_tipo_atendimento/principal.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,54 @@ | ||
import warnings | ||
warnings.filterwarnings("ignore") | ||
|
||
import pandas as pd | ||
from datetime import date | ||
|
||
from sqlalchemy.orm import Session | ||
from prefect import flow | ||
|
||
from impulsoetl import __VERSION__ | ||
from impulsoetl.bd import Sessao | ||
|
||
from impulsoetl.loggers import logger, habilitar_suporte_loguru | ||
from impulsoetl.sisab.relatorio_producao_profissional_conduta_tipo_atendimento.extracao import extrair_relatorio | ||
from impulsoetl.sisab.relatorio_producao_profissional_conduta_tipo_atendimento.tratamento import tratamento_dados | ||
from impulsoetl.utilitarios.bd import carregar_dataframe | ||
|
||
|
||
@flow( | ||
name="Obter Relatório de Produção por Profissional, Contuta e Tipo de Atendimento (Painel AGP)", | ||
description=( | ||
"Extrai, transforma e carrega os dados de produção da Atenção Primária à Saúde " | ||
+"por problema/condição avaliada, a partir do Sistema de Informação em Saúde da Atenção " | ||
+ "Básica do SUS." | ||
), | ||
retries=0, | ||
retry_delay_seconds=None, | ||
timeout_seconds=14400, | ||
version=__VERSION__, | ||
validate_parameters=False, | ||
) | ||
def relatorio_profissional_conduta_atendimento( | ||
sessao: Session, | ||
tabela_destino: str, | ||
periodo_competencia: date, | ||
periodo_id: str, | ||
unidade_geografica_id: str | ||
)-> None: | ||
|
||
logger.info("Extraindo relatório da competencia {}, ...".format(periodo_competencia)) | ||
|
||
df_extraido = extrair_relatorio( | ||
periodo_competencia = periodo_competencia | ||
) | ||
|
||
df_tratado = tratamento_dados( | ||
df_extraido=df_extraido, | ||
periodo_id=periodo_id, | ||
unidade_geografica_id=unidade_geografica_id, | ||
) | ||
|
||
carregar_dataframe( | ||
sessao=sessao, df=df_tratado, tabela_destino=tabela_destino | ||
) |
55 changes: 55 additions & 0 deletions
55
src/impulsoetl/sisab/relatorio_producao_profissional_conduta_tipo_atendimento/tratamento.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,55 @@ | ||
import warnings | ||
|
||
warnings.filterwarnings("ignore") | ||
from datetime import date | ||
|
||
import pandas as pd | ||
import numpy as np | ||
|
||
from typing import Final | ||
from frozendict import frozendict | ||
|
||
from impulsoetl.loggers import logger, habilitar_suporte_loguru | ||
|
||
COLUNAS_RENOMEAR: Final[dict[str, str]] = { | ||
"uf_sigla":"municipio_uf", | ||
"periodo_data_inicio":"periodo_data", | ||
"Conduta":"conduta", | ||
"Categoria do Profissional":"categoria_profissional", | ||
"Tipo de Atendimento":"tipo_atendimento", | ||
"quantidade_aprovada":"quantidade" | ||
} | ||
|
||
COLUNAS_TIPOS: Final[frozendict] = frozendict( | ||
{ | ||
"municipio_uf":"str", | ||
"periodo_data":"str", | ||
"municipio_nome":"str", | ||
"periodo_data":"str", | ||
"conduta":"str", | ||
"categoria_profissional":"str", | ||
"tipo_atendimento":"str", | ||
"quantidade":"int" | ||
} | ||
) | ||
|
||
def tratar_tipos(df_extraido: pd.DataFrame) -> pd.DataFrame: | ||
df_extraido = df_extraido.astype(COLUNAS_TIPOS, errors="ignore").where( | ||
df_extraido.notna(), None | ||
) | ||
return df_extraido | ||
|
||
def renomear_colunas(df_extraido: pd.DataFrame) -> pd.DataFrame: | ||
df_extraido.rename(columns=COLUNAS_RENOMEAR, inplace=True) | ||
return df_extraido | ||
|
||
def tratamento_dados( | ||
df_extraido: pd.DataFrame, periodo_id: str, unidade_geografica_id: str | ||
)-> pd.DataFrame: | ||
df_extraido = renomear_colunas(df_extraido) | ||
tratar_tipos(df_extraido) | ||
df_extraido["periodo_id"] = periodo_id | ||
df_extraido["unidade_geografica_id"] = unidade_geografica_id | ||
df_extraido = df_extraido.reset_index(drop=True) | ||
|
||
return df_extraido |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,94 @@ | ||
import warnings | ||
|
||
warnings.filterwarnings("ignore") | ||
from datetime import date | ||
|
||
import pandas as pd | ||
import numpy as np | ||
|
||
from prefect import task | ||
|
||
from impulsoetl.loggers import logger, habilitar_suporte_loguru | ||
from impulsoetl.sisab.utilitarios_sisab_relatorio_producao import extrair_producao_por_municipio | ||
from impulsoetl.sisab.utilitarios_sisab_relatorio_producao import transformar_producao_por_municipio | ||
from impulsoetl.sisab.relatorio_saude_producao.extracao import extrair_relatorio | ||
|
||
periodo_competencia = date(2023,1,1) | ||
|
||
|
||
CATEGORIA_PROFISSIONAL_REDUZIDA = [ | ||
'Psicólogo', | ||
'Profissional de educação física', | ||
'Nutricionista', | ||
'Médico', | ||
'Fonoaudiólogo', | ||
'Fisioterapeuta', | ||
'Farmacêutico', | ||
'Enfermeiro', | ||
'Assistente Social' | ||
] | ||
|
||
|
||
def obter_relatorio_reduzido_bloco_consolidado( | ||
periodo_competencia: date)-> pd.DataFrame(): | ||
|
||
df_consolidado = pd.DataFrame() | ||
|
||
try: | ||
df_parcial = extrair_producao_por_municipio( | ||
tipo_producao="Atendimento individual", | ||
competencias=[periodo_competencia], | ||
selecoes_adicionais={ | ||
"Problema/Condição Avaliada": "Selecionar Todos", | ||
#"Problema/Condição Avaliada": ["Diabetes"], | ||
"Conduta":"Selecionar Todos", | ||
#"Conduta":["Alta do episódio"], | ||
"Categoria do Profissional":CATEGORIA_PROFISSIONAL_REDUZIDA, | ||
#"Categoria do Profissional":["Médico"], | ||
"Tipo de Atendimento": "Selecionar Todos", | ||
#"Tipo de Atendimento":["Consulta agendada"] | ||
}, | ||
|
||
).pipe(transformar_producao_por_municipio) | ||
|
||
df_consolidado = df_consolidado.append(df_parcial) | ||
|
||
except Exception as e: | ||
logger.error(e) | ||
pass | ||
|
||
return df_consolidado | ||
|
||
|
||
|
||
def obter_relatorio_reduzido_bloco_6_tela1( | ||
periodo_competencia: date)-> pd.DataFrame(): | ||
|
||
df_consolidado = pd.DataFrame() | ||
|
||
try: | ||
df_parcial = extrair_producao_por_municipio( | ||
tipo_producao="Atendimento individual", | ||
competencias=[periodo_competencia], | ||
selecoes_adicionais={ | ||
#"Conduta":"Selecionar Todos", | ||
"Conduta":["Alta do episódio"], | ||
#"Categoria do Profissional":CATEGORIA_PROFISSIONAL_REDUZIDA, | ||
"Categoria do Profissional": ['Médico'], | ||
#"Tipo de Atendimento": "Selecionar Todos", | ||
"Tipo de Atendimento":["Consulta agendada"] | ||
}, | ||
|
||
).pipe(transformar_producao_por_municipio) | ||
|
||
df_consolidado = df_consolidado.append(df_parcial) | ||
|
||
except Exception as e: | ||
logger.error(e) | ||
pass | ||
|
||
return df_consolidado | ||
|
||
df = obter_relatorio_reduzido_bloco_6_tela1(periodo_competencia) | ||
print(df) | ||
print(df.columns) |