Skip to content

Commit

Permalink
feat(IBGE): continue work of AlertaDengue#192
Browse files Browse the repository at this point in the history
  • Loading branch information
luabida committed Apr 10, 2024
1 parent abf4efe commit 3fbff99
Show file tree
Hide file tree
Showing 6 changed files with 255 additions and 241 deletions.
469 changes: 240 additions & 229 deletions poetry.lock

Large diffs are not rendered by default.

6 changes: 3 additions & 3 deletions pysus/ftp/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import pathlib
from datetime import datetime
from ftplib import FTP
from typing import Any, Dict, List, Optional, Set, Union
from typing import Any, Dict, List, Optional, Set, Union, Tuple
from typing_extensions import Self

import humanize
Expand Down Expand Up @@ -53,7 +53,7 @@ class File:
basename: str
path: str
# parent: Directory # TODO: This causes too much overhead
__info__: Set[Union[int, str, datetime]]
__info__: dict

def __init__(self, path: str, name: str, info: dict) -> None:
name, extension = os.path.splitext(name)
Expand Down Expand Up @@ -461,7 +461,7 @@ class Database:

ftp: FTP
name: str
paths: List[Directory]
paths: Tuple[Directory, ...]
metadata: dict
__content__: Dict[str, Union[Directory, File]]

Expand Down
2 changes: 1 addition & 1 deletion pysus/ftp/databases/ciha.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@

class CIHA(Database):
name = "CIHA"
paths = (Directory("/dissemin/publicos/CIHA/201101_/Dados"))
paths = (Directory("/dissemin/publicos/CIHA/201101_/Dados"),)
metadata = {
"long_name": "Comunicação de Internação Hospitalar e Ambulatorial",
"source": "http://ciha.datasus.gov.br/CIHA/index.php",
Expand Down
2 changes: 1 addition & 1 deletion pysus/ftp/databases/cnes.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@

class CNES(Database):
name = "CNES"
paths = Directory("/dissemin/publicos/CNES/200508_/Dados")
paths = (Directory("/dissemin/publicos/CNES/200508_/Dados"),)
metadata = {
"long_name": "Cadastro Nacional de Estabelecimentos de Saúde",
"source": "https://cnes.datasus.gov.br/",
Expand Down
6 changes: 4 additions & 2 deletions pysus/ftp/databases/ibge_datasus.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,8 @@ def get_files(
self,
year: Optional[Union[str, int, list]] = None,
) -> List[File]:
files = [f for f in self.files if f.extension.upper() in [".ZIP", ".DBF"] and self.describe(f)["year"] == year]
files = [f for f in self.files if f.extension.upper(
) in [".ZIP", ".DBF"] and self.describe(f)["year"] == year]
# files = list(filter(
# lambda f: f.extension.upper() in [".ZIP"], self.files
# ))
Expand All @@ -62,6 +63,7 @@ def get_files(
years = (
[zfill_year(str(y)[-4:]) for y in to_list(year)]
)
files = list(filter(lambda f: zfill_year(self.format(f)) in years, files))
files = list(filter(lambda f: zfill_year(
self.format(f)) in years, files))

return files
11 changes: 6 additions & 5 deletions pysus/online_data/IBGE.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
Helper functions to download official statistics from IBGE SIDRA
"""

import ssl # Builtin
import urllib3
import requests
import pandas as pd
Expand Down Expand Up @@ -235,8 +236,8 @@ def __init__(
self, agregado: int, periodos: str, variavel: str = 'allxp', **kwargs
):
self.url = (
APIBASE
+ f'agregados/{agregado}/periodos/{periodos}/variaveis/{variavel}?'
APIBASE
+ f'agregados/{agregado}/periodos/{periodos}/variaveis/{variavel}?'
)
self.url += '&'.join([f'{k}={v}' for k, v in kwargs.items()])
self.JSON = None
Expand Down Expand Up @@ -265,8 +266,6 @@ def to_dataframe(self):
SOLUTION: https://github.com/scrapy/scrapy/issues/5491#issuecomment-1241862323
"""

import ssl # Builtin


class CustomHttpAdapter(requests.adapters.HTTPAdapter):
# "Transport adapter" that allows us to use custom ssl_context.
Expand Down Expand Up @@ -300,5 +299,7 @@ def get_population(year, source='POPTCU'):
:return: DataFrame with population data
"""
ibgedatasus = IBGEDATASUS().load()
files = [f for f in ibgedatasus.get_files(year=year) if f.path.split('/')[-2] == source]
files = [
f for f in ibgedatasus.get_files(year=year) if f.path.split('/')[-2] == source
]
return files

0 comments on commit 3fbff99

Please sign in to comment.