add typing
This commit is contained in:
parent
a3df014ec5
commit
fb104f9790
@ -32,7 +32,7 @@ INFLATION_RATES = {
|
|||||||
}
|
}
|
||||||
|
|
||||||
class Company:
|
class Company:
|
||||||
def __init__(self, data, report, writer):
|
def __init__(self, data, report, writer) -> None:
|
||||||
self.data = data
|
self.data = data
|
||||||
self.writer = writer
|
self.writer = writer
|
||||||
self.report = report
|
self.report = report
|
||||||
@ -47,7 +47,7 @@ class Company:
|
|||||||
|
|
||||||
|
|
||||||
|
|
||||||
def clean_simple(self, year: int, type: str):
|
def clean_simple(self, year: int, type: str) -> None:
|
||||||
"""Clean simple data. This means tax and capital."""
|
"""Clean simple data. This means tax and capital."""
|
||||||
try:
|
try:
|
||||||
self.cleaned_data[f"{self.get_simple_suffix(type)}{year}"] = int(self.data[f"{type} EUR {year}"])
|
self.cleaned_data[f"{self.get_simple_suffix(type)}{year}"] = int(self.data[f"{type} EUR {year}"])
|
||||||
@ -59,7 +59,7 @@ class Company:
|
|||||||
return "ek" if type == "Eigenkapital" else "st"
|
return "ek" if type == "Eigenkapital" else "st"
|
||||||
|
|
||||||
|
|
||||||
def clean_complex(self, year: int, state: str):
|
def clean_complex(self, year: int, state: str) -> None:
|
||||||
"""Clean the complex data. This means earnings before/after tax."""
|
"""Clean the complex data. This means earnings before/after tax."""
|
||||||
try:
|
try:
|
||||||
if f"Gewinn/(Verlust) {state} Steuern EUR {year}" in self.data.keys() and self.data[f"Gewinn/(Verlust) {state} Steuern EUR {year}"] != '' and not self.cleaned_data.get(f"gn{year}"):
|
if f"Gewinn/(Verlust) {state} Steuern EUR {year}" in self.data.keys() and self.data[f"Gewinn/(Verlust) {state} Steuern EUR {year}"] != '' and not self.cleaned_data.get(f"gn{year}"):
|
||||||
@ -76,7 +76,7 @@ class Company:
|
|||||||
return "n" if state == "nach" else "v"
|
return "n" if state == "nach" else "v"
|
||||||
|
|
||||||
|
|
||||||
def calculate_all_tax(self):
|
def calculate_all_tax(self) -> None:
|
||||||
"""Calculate tax for all relevant years."""
|
"""Calculate tax for all relevant years."""
|
||||||
self.calculate_tax(2020)
|
self.calculate_tax(2020)
|
||||||
self.calculate_tax(2021)
|
self.calculate_tax(2021)
|
||||||
@ -84,14 +84,14 @@ class Company:
|
|||||||
self.calculate_tax(2023)
|
self.calculate_tax(2023)
|
||||||
self.calculate_tax(2024)
|
self.calculate_tax(2024)
|
||||||
|
|
||||||
def calculate_tax(self, year: int):
|
def calculate_tax(self, year: int) -> None:
|
||||||
"""Calculate simple tax from provided values."""
|
"""Calculate simple tax from provided values."""
|
||||||
if not self.cleaned_data.get(f"st{year}") and self.cleaned_data.get(f"gv{year}") != None and self.cleaned_data.get(f"gn{year}") != None:
|
if not self.cleaned_data.get(f"st{year}") and self.cleaned_data.get(f"gv{year}") != None and self.cleaned_data.get(f"gn{year}") != None:
|
||||||
self.cleaned_data[f"st{year}"] = self.cleaned_data.get(f"gv{year}") - self.cleaned_data.get(f"gn{year}")
|
self.cleaned_data[f"st{year}"] = self.cleaned_data.get(f"gv{year}") - self.cleaned_data.get(f"gn{year}")
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def reporter(self):
|
def reporter(self) -> None:
|
||||||
"""Simple class to report valid and invalid data to the main import class."""
|
"""Simple class to report valid and invalid data to the main import class."""
|
||||||
for year in YEARS:
|
for year in YEARS:
|
||||||
if self.cleaned_data.get(f"st{year}") and self.cleaned_data.get(f"ek{year}"):
|
if self.cleaned_data.get(f"st{year}") and self.cleaned_data.get(f"ek{year}"):
|
||||||
@ -100,7 +100,7 @@ class Company:
|
|||||||
self.report.invalid_data +=1
|
self.report.invalid_data +=1
|
||||||
|
|
||||||
|
|
||||||
def calculate_data(self):
|
def calculate_data(self) -> None:
|
||||||
"""Calculate data relevant to the project."""
|
"""Calculate data relevant to the project."""
|
||||||
for year in YEARS:
|
for year in YEARS:
|
||||||
if self.cleaned_data.get(f"st{year}") and self.cleaned_data.get(f"gv{year}") and self.cleaned_data.get(f"gn{year}") and self.cleaned_data.get(f"ek{year}"):
|
if self.cleaned_data.get(f"st{year}") and self.cleaned_data.get(f"gv{year}") and self.cleaned_data.get(f"gn{year}") and self.cleaned_data.get(f"ek{year}"):
|
||||||
@ -109,7 +109,7 @@ class Company:
|
|||||||
self.cleaned_data[f"realefftax{year}"] = (self.cleaned_data.get(f"st{year}") + (INFLATION_RATES[year] * self.cleaned_data.get(f"gv{year}")) + (INFLATION_RATES[year] * self.cleaned_data.get(f"ek{year}"))) / self.cleaned_data.get(f"gv{year}")
|
self.cleaned_data[f"realefftax{year}"] = (self.cleaned_data.get(f"st{year}") + (INFLATION_RATES[year] * self.cleaned_data.get(f"gv{year}")) + (INFLATION_RATES[year] * self.cleaned_data.get(f"ek{year}"))) / self.cleaned_data.get(f"gv{year}")
|
||||||
|
|
||||||
|
|
||||||
def write(self):
|
def write(self) -> None:
|
||||||
"""Write the current dataset to CSV"""
|
"""Write the current dataset to CSV"""
|
||||||
with open(self.report.output) as out_csv:
|
with open(self.report.output) as out_csv:
|
||||||
try:
|
try:
|
||||||
@ -122,7 +122,7 @@ class Company:
|
|||||||
|
|
||||||
|
|
||||||
class dataimport:
|
class dataimport:
|
||||||
def __init__(self, filename, logfile, output, seek=0):
|
def __init__(self, filename, logfile, output, seek=0) -> None:
|
||||||
self.seek = seek
|
self.seek = seek
|
||||||
self.progress = progress.Progress(
|
self.progress = progress.Progress(
|
||||||
*progress.Progress.get_default_columns(),
|
*progress.Progress.get_default_columns(),
|
||||||
@ -159,7 +159,7 @@ class dataimport:
|
|||||||
self.invalid_data = 0
|
self.invalid_data = 0
|
||||||
self.importer()
|
self.importer()
|
||||||
|
|
||||||
def importer(self):
|
def importer(self) -> None:
|
||||||
"""Start the actual import process. Seperates process and setup."""
|
"""Start the actual import process. Seperates process and setup."""
|
||||||
with self.progress:
|
with self.progress:
|
||||||
with open(self.filename, mode='r', encoding='utf-8-sig', newline='') as csv_file:
|
with open(self.filename, mode='r', encoding='utf-8-sig', newline='') as csv_file:
|
||||||
@ -194,11 +194,11 @@ class dataimport:
|
|||||||
else:
|
else:
|
||||||
self.log.critical("ERROR CALCULATION EXCEPTION")
|
self.log.critical("ERROR CALCULATION EXCEPTION")
|
||||||
|
|
||||||
def get_total(self, file):
|
def get_total(self, file) -> int:
|
||||||
"""Get total data rows in the input file. Corrected for header row."""
|
"""Get total data rows in the input file. Corrected for header row."""
|
||||||
return sum(1 for _ in open(file, mode='r')) - 1
|
return sum(1 for _ in open(file, mode='r')) - 1
|
||||||
|
|
||||||
def generate_fieldnames(self):
|
def generate_fieldnames(self) -> list:
|
||||||
"""Generate fieldnames for the export CSV."""
|
"""Generate fieldnames for the export CSV."""
|
||||||
fieldnames = ['bvd_id', 'name']
|
fieldnames = ['bvd_id', 'name']
|
||||||
for year in YEARS:
|
for year in YEARS:
|
||||||
@ -211,7 +211,7 @@ class dataimport:
|
|||||||
fieldnames.append(f"realefftax{year}")
|
fieldnames.append(f"realefftax{year}")
|
||||||
return fieldnames
|
return fieldnames
|
||||||
|
|
||||||
def comp_import(self, data, writer):
|
def comp_import(self, data: dict, writer) -> None:
|
||||||
"""Import the active dataset as a company. Give write directive."""
|
"""Import the active dataset as a company. Give write directive."""
|
||||||
current = Company(data, report=self, writer=writer)
|
current = Company(data, report=self, writer=writer)
|
||||||
current.reporter()
|
current.reporter()
|
||||||
|
Loading…
x
Reference in New Issue
Block a user