Merge pull request #190 from ResidenciaTICBrisa/script_anuario

Refatoração e estruturação dos scripts de extração do PNE 2050 e do BEN

.github/workflows/atualiza_tabelas.yml

@@ -1,7 +1,7 @@
 name: run main.py
 
 on: 
-  schedule: [{cron: "0 0 * * 4"}] #agendamento
+  schedule: [{cron: "0 15 * * 5"}] #agendamento
 
 jobs:
   build:
@@ -15,11 +15,11 @@ jobs:
     - name: Install dependencies
       run: |
         pip install --upgrade pip
-        pip install -r requirements.txt
+        pip install requests==2.31.0
+        pip install pandas==2.0.2
+        pip install bs4==0.0.1
     - name: execute py script # run main.py
-        env:
-          SOME_SECRET: ${{ secrets.SOME_SECRET }}
-        run: python 04_PipelineTCU/main.py
+      run: python 04_PipelineTCU/main.py
     - name: commit files
       run: |
         git config --local user.email "[email protected]"
@@ -30,4 +30,4 @@ jobs:
       uses: ad-m/[email protected]
       with:
         github_token: ${{ secrets.GITHUB_TOKEN }}
-        branch: main    
+        branch: main    

04_PipelineTCU/anuarioHandler.py

@@ -1,6 +1,5 @@
 import math
 import pandas as pd
-from pathlib import Path
 
 class AnuarioHandler:
     # Uma classe que lê o arquivo correspondente ao Anuário Estatístico de Energia Elétrica, seleciona e transforma as tabelas desejadas e salva as tabelas resultantes em arquivos csv.
@@ -20,14 +19,14 @@ class AnuarioHandler:
     #     def generateTables(self): 
     #         Formata e salva todas as tabelas do Anuário Estatístico de Energia Elétrica em arquivos csv
 
-    def __init__(self, file):
-        # Construtor da classe anuarioHandler.
+    def __init__(self, file, path_final):
+        # Construtor da classe AnuarioHandler.
         # Args:
         #     file (str): nome do arquivo correspondente ao  arquivo do Anuário Estatístico de Eletricidade.
+        #     path_final (str): caminho até a pasta ondes os arquivos serão salvos.
         self.file = file
         self.n_tabelas = 0
-        self.path = str(Path(__file__).parent.resolve()) 
-        self.path += "/constants/" 
+        self.path = path_final
 
     def formataSalvaTabela(self, df, nome_tab, cols):
         # Formata uma tabela para adicionar o ano do dado como uma coluna e salva essa tabela como arquivo csv
@@ -199,5 +198,5 @@ def generateTables(self):
                     df = self.tabelasConsRegioes(df, regioes, False)
                     self.formataSalvaTabela(df,'ConsMedAnPerCapReg(kWh_hab).csv', cols3)
                     self.n_tabelas += 1
-            except Exception as e:
+            except Exception:
                 continue

04_PipelineTCU/benHandler.py

@@ -0,0 +1,87 @@
+import math
+import pandas as pd
+
+class BenHandler:
+    # Uma classe que lê o arquivo correspondente às matrizes energéticas consolidadas do Balanço Energético Nacional (BEN), seleciona e transforma as tabelas desejadas, além de salvar o resultado em uma única tabela em arquivo csv com os dados da matrizes consolidades do Balanço Energético Nacional de 1970 até o ano mais atual.
+
+    # Atributos:
+    #   file (str): nome do arquivo que será lido
+    #   path (str): caminho até a pasta onde os arquivos serão salvos
+
+    # Métodos:
+    #   def drop_row_by_text(self, df, text):
+    #       Exclui uma linha de um Data Frame com base no texto da sua primeira célula
+    #   def generateTable(self):   
+    #       Gera a tabela com os dados das matrizes consolidadas do BEN de 1970 até o ano mais atual
+
+
+    def __init__(self, file, path_final):
+        # Construtor da classe BenHandler.
+        # Args:
+        #     file (str): nome do arquivo correspondente às matrizes energéticas consolidadas do Balanço Energético Nacional (BEN),
+        #     path_final (str): caminho até a pasta ondes os arquivos serão salvos.
+        self.file = file
+        self.path = path_final
+
+    def drop_row_by_text(self, df, text):
+        # Exclui uma linha de um Data Frame com base no texto da sua primeira célula
+        # Args:
+        #     df (DataFrame): Data Frame que armazena a linha que se deseja apagar
+        #     text (str): texto que a linha a ser excluída contém
+        index_to_drop = df[df.iloc[:, 0].str.contains(text)].index
+        df.drop(index=index_to_drop, inplace=True)
+        return df
+
+    def generateTable(self):   
+        # Gera a tabela com os dados das matrizes consolidadas do BEN de 1970 até o ano mais atual
+        cont = False
+        df_total = pd.DataFrame()
+        titulos = ['CONTA',
+                   'PRODUÇÃO',
+                   'IMPORTAÇÃO',
+                   'VARIAÇÃO DE ESTOQUES',
+                   'EXPORTAÇÃO',
+                   'NÃO-APROVEITADA',
+                   'REINJEÇÃO',
+                   'CONSUMO FINAL NÃO-ENERGÉTICO',
+                   'SETOR ENERGÉTICO',
+                   'RESIDENCIAL',
+                   'COMERCIAL',
+                   'PÚBLICO',
+                   'AGROPECUÁRIO',
+                   'RODOVIÁRIO',
+                   'FERROVIÁRIO',
+                   'AÉREO',
+                   'HIDROVIÁRIO',
+                   'INDUSTRIAL - TOTAL']
+        for sheet_name, df in pd.read_excel(self.path + self.file, sheet_name=None).items():
+            df.drop('Unnamed: 0', axis=1, inplace=True)
+            df.dropna(inplace=True)
+            i = 0
+            while len(df.index) != 18:
+                titulo = df.iloc[i,0]
+                while titulo[0] == ' ':
+                    titulo = titulo[1:]
+                while titulo[-1] == ' ':
+                    titulo = titulo[0:-1]             
+                if titulo in titulos:
+                    df.iloc[i,0] = titulo
+                    i += 1
+                else:
+                    df.drop(index=df.iloc[i].name, inplace=True)
+            df = df.transpose()
+            num_rows = len(df)
+            new_column_data = [sheet_name] * num_rows
+            df['Ano'] = new_column_data
+            df.iloc[0, -1] = 'Ano'
+            if cont:
+                df.drop(index=df.iloc[0].name, inplace=True)
+            cont = True
+            df = self.drop_row_by_text(df, 'ENERGIA SECUNDÁRIA TOTAL ')
+            df = self.drop_row_by_text(df,'ENERGIA PRIMÁRIA TOTAL')
+            df = self.drop_row_by_text(df,'TOTAL')
+            df_total = pd.concat([df_total,df])
+        signal = lambda y : 1 if (y >= 0) else -1
+        df_total.iloc[1:,1:-1] = df_total.iloc[1:,1:-1].applymap(lambda x : math.floor(abs(x))*signal(x) if (abs(x)*10 - int(abs(x))*10) < 5 else math.ceil(abs(x))*signal(x))
+        df_total.iloc[1:,1:-1]=df_total.iloc[1:,1:-1].astype('int64') 
+        df_total.to_csv(self.path + "BEN_total.csv", index=False, header=False)