Skip to content

Commit

Permalink
Some optimizations performed
Browse files Browse the repository at this point in the history
  • Loading branch information
jzsmoreno committed Dec 5, 2024
1 parent 188b81d commit bc00097
Show file tree
Hide file tree
Showing 3 changed files with 197 additions and 271 deletions.
5 changes: 4 additions & 1 deletion pydbsmgr/lightest.py
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,10 @@ def clean_frame(
"""
table = self.df.copy()
cols = table.columns
table_sample = table.sample(frac=sample_frac, replace=False)
if sample_frac != 1.0:
table_sample = table.sample(frac=sample_frac, replace=False)
else:
table_sample = table.copy()
errors = kwargs.get("errors", "ignore")

for column_index, datatype in enumerate(table.dtypes):
Expand Down
162 changes: 85 additions & 77 deletions pydbsmgr/logs.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,113 +2,121 @@
from datetime import datetime
from typing import List

import numpy as np
import pandas as pd
from pandas.core.frame import DataFrame


class EventLogger:
"""Allows you to store screen prints in a plain text file."""

def __init__(self, FILE_NAME: str, FILE_PATH: str) -> None:
self.FILE_NAME = FILE_NAME + ".txt"
self.FILE_PATH = FILE_PATH
if os.path.isfile(self.FILE_PATH + self.FILE_NAME):
def __init__(self, file_name: str, file_path: str) -> None:
self.file_name = f"{file_name}.txt"
self.file_path = file_path
self._check_file_existence()

def _check_file_existence(self):
if os.path.isfile(self.full_path):
warning_type = "UserWarning"
msg = "The log file {%s} already exists, the changes will be added." % self.FILE_NAME
msg = f"The log file {self.file_name} already exists, the changes will be added."
print(f"{warning_type}: {msg}")
self.FILE = open(self.FILE_PATH + self.FILE_NAME, "a")

@property
def full_path(self) -> str:
return os.path.join(self.file_path, self.file_name)

def audit(self):
if os.path.isfile(self.FILE_PATH + self.FILE_NAME):
msg = (
"The log file {%s} already exists, do you want to delete it? (True/False): "
% self.FILE_NAME
)
delete = input(msg)
if delete == "True":
os.remove(self.FILE_PATH + self.FILE_NAME)
warning_type = "UserWarning"
msg = "The log file {%s} has been deleted successfully" % self.FILE_NAME
print(f"{warning_type}: {msg}")
else:
msg = "Reading the log file: {%s}" % self.FILE_NAME
print(f"{msg}")
with open(self.FILE_PATH + self.FILE_NAME, "r") as file:
log = file.read()
return log
else:
if not os.path.isfile(self.full_path):
warning_type = "UserWarning"
msg = "The log file {%s} does not exist" % self.FILE_NAME
msg = f"The log file {self.file_name} does not exist."
print(f"{warning_type}: {msg}")
return

delete = (
input(f"Do you want to delete the log file {self.file_name}? (y/N): ").strip().lower()
)
if delete == "y":
os.remove(self.full_path)
warning_type = "UserWarning"
msg = f"The log file {self.file_name} has been deleted successfully."
print(f"{warning_type}: {msg}")
else:
msg = f"Reading the log file: {self.file_name}"
print(msg)
with open(self.full_path, "r") as file:
return file.read()

def writer(self, *chars: str) -> None:
self.FILE = open(self.FILE_PATH + self.FILE_NAME, "a")
date = str(datetime.now().strftime("%d/%m/%Y %H:%M:%S "))
print(date, end="")
for item in chars:
print(item, end="")
print("\n")
print(date, end="", file=self.FILE)
for item in chars:
print(item, end="", file=self.FILE)
print("\n", file=self.FILE)
self.FILE.close()
date = datetime.now().strftime("%d/%m/%Y %H:%M:%S ")
output_line = date + " ".join(chars) + "\n"

print(output_line, end="")
with open(self.full_path, "a") as file:
file.write(output_line)


class EventLogBook:
"""Allows you to create and write new lines in a logbook."""

def __init__(self, FILE_NAME: str, FILE_PATH: str) -> None:
self.FILE_NAME = FILE_NAME + ".csv"
self.FILE_PATH = FILE_PATH
if os.path.isfile(self.FILE_PATH + self.FILE_NAME):
def __init__(self, file_name: str, file_path: str) -> None:
self.file_name = f"{file_name}.csv"
self.file_path = file_path
self._check_file_existence()

@property
def full_path(self) -> str:
return os.path.join(self.file_path, self.file_name)

def _check_file_existence(self):
if os.path.isfile(self.full_path):
warning_type = "UserWarning"
msg = (
"The logbook file {%s} already exists, the changes will be added." % self.FILE_NAME
)
msg = f"The logbook file {self.file_name} already exists, the changes will be added."
print(f"{warning_type}: {msg}")

def create(self, df: DataFrame, encoding: str = "latin1") -> None:
def create(self, df: pd.DataFrame, encoding: str = "latin1") -> None:
self._encoding = encoding
df.to_csv(self.FILE_PATH + self.FILE_NAME, index=False, encoding=self._encoding)
df.to_csv(self.full_path, index=False, encoding=self._encoding)

def audit(self, encoding: str = "latin1"):
self._encoding = encoding
if os.path.isfile(self.FILE_PATH + self.FILE_NAME):
msg = (
"The log file {%s} already exists, do you want to delete it? (True/False): "
% self.FILE_NAME
)
delete = input(msg)
if delete == "True":
os.remove(self.FILE_PATH + self.FILE_NAME)
warning_type = "UserWarning"
msg = "The log file {%s} has been deleted successfully" % self.FILE_NAME
print(f"{warning_type}: {msg}")
else:
msg = "Reading the log file: {%s}" % self.FILE_NAME
print(f"{msg}")
return pd.read_csv(self.FILE_PATH + self.FILE_NAME, encoding=self._encoding)
else:
if not os.path.isfile(self.full_path):
warning_type = "UserWarning"
msg = "The log file {%s} does not exist" % self.FILE_NAME
msg = f"The logbook file {self.file_name} does not exist."
print(f"{warning_type}: {msg}")
return

def update(self, rows: List[str]) -> None:
delete = (
input(f"Do you want to delete the logbook file {self.file_name}? (y/N): ")
.strip()
.lower()
)
if delete == "y":
os.remove(self.full_path)
warning_type = "UserWarning"
msg = f"The logbook file {self.file_name} has been deleted successfully."
print(f"{warning_type}: {msg}")
else:
return pd.read_csv(self.full_path, encoding=self._encoding)

def update(self, rows: List[List[str]]) -> None:
try:
df = pd.read_csv(self.FILE_PATH + self.FILE_NAME, encoding=self._encoding)
to_add = rows
df_to_add = pd.DataFrame(np.array(to_add).reshape((1, -1)), columns=df.columns)
df = pd.concat([df, df_to_add])
df.to_csv(self.FILE_PATH + self.FILE_NAME, index=False, encoding=self._encoding)
except:
df = pd.read_csv(self.full_path, encoding=self._encoding)
new_df = pd.DataFrame(rows, columns=df.columns)
df = pd.concat([df, new_df], ignore_index=True)
df.to_csv(self.full_path, index=False, encoding=self._encoding)

except FileNotFoundError:
warning_type = "UserWarning"
msg = (
"The logbook {%s} has not been created, so it is going to be created"
% self.FILE_NAME
f"The logbook {self.file_name} has not been created, so it is going to be created."
)
print(f"{warning_type}: {msg}")
to_add = rows
df = pd.DataFrame(np.array(to_add).reshape((1, -1)), columns=df.columns)
df.to_csv(to_add=(rows), index=False, encoding=self._encoding)
new_df = pd.DataFrame(rows)
new_df.to_csv(self.full_path, index=False, encoding=self._encoding)


# Usage example
if __name__ == "__main__":
logger = EventLogger("example_log", "./")
logger.writer("This is a test log entry.")

logbook = EventLogBook("example_logbook", "./")
logbook.create(pd.DataFrame({"Column1": [1, 2], "Column2": ["A", "B"]}))
logbook.update([[3, "C"]])
Loading

0 comments on commit bc00097

Please sign in to comment.