--- a +++ b/ReadersWriters/_TextFile.py @@ -0,0 +1,223 @@ +#!/usr/bin/env python +# -*- coding: UTF-8 -*- +# +# Copyright 2017 University of Westminster. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ============================================================================== +""" It is an interface for reading and writing text files. +""" + +from typing import Dict, TypeVar, Callable +import os +import sys +from typing import List +from collections import OrderedDict +import pandas as pd +import numpy as np +import pprint as pp +import logging +from Configs.CONSTANTS import CONSTANTS + +PandasDataFrame = TypeVar('DataFrame') +CollectionsOrderedDict = TypeVar('OrderedDict') + +__author__ = "Mohsen Mesgarpour" +__copyright__ = "Copyright 2016, https://github.com/mesgarpour" +__credits__ = ["Mohsen Mesgarpour"] +__license__ = "GPL" +__version__ = "1.1" +__maintainer__ = "Mohsen Mesgarpour" +__email__ = "mohsen.mesgarpour@gmail.com" +__status__ = "Release" + + +class TextFile: + def __init__(self, + max_width: int=100000000): + """Initialise the objects and constants. + """ + self.__logger = logging.getLogger(CONSTANTS.app_name) + self.__logger.debug(__name__) + self.__path = None + self.__max_width = max_width + + def set(self, + path: str, + title: str, + ext: str): + """Set the text file for reading or writing. + :param path: the directory path of the text file. + :param title: the file name of the text file. + :param ext: the extension of the text file. + """ + self.__logger.debug("Set the text file.") + self.__path = os.path.join(path, title + "." + ext) + + def reset(self): + """Reset the text file reader/writer. + """ + self.__logger.debug("Reset the text file.") + try: + open(self.__path, 'w').close() + except (OSError, IOError) as e: + self.__logger.error(__name__ + " - Can not open the file: \n" + self.__path + "\n" + str(e)) + sys.exit() + except(): + self.__logger.error(__name__ + " - Could not create the file: \n" + self.__path) + sys.exit() + + def exists(self) -> bool: + """Check if the text file exists. + :return: indicates if the file exists. + """ + self.__logger.debug("Check if the text file exists.") + return os.path.isfile(self.__path) + + def read(self, + skip: int) -> List: + """Read the text file into list. + :param skip: lines to skip before reading. + :return: the read file contents. + """ + self.__logger.debug("Read the text file.") + rows = self.__read_array(skip) + return rows + + def __read_array(self, + skip: int) -> List: + """Read the text file into array. + :param skip: lines to skip before reading. + :return: the read file contents. + """ + self.__logger.debug("Read the text file into array.") + rows = [] + i = 0 + try: + with open(self.__path, "r") as f: + for line in f: + i += 1 + if i > skip: + rows.append(line) + except (OSError, IOError) as e: + self.__logger.error(__name__ + " - Can not open the file: \n" + self.__path + "\n" + str(e)) + sys.exit() + except(): + self.__logger.error(__name__ + " - Can not read the Text file: \n" + self.__path) + sys.exit() + return rows + + def append(self, + data: Callable[[List, Dict, PandasDataFrame], None]): + """Append to text file using dataframe, dictionary or list. + :param data: the data to write. + """ + self.__logger.debug("Append to the text file.") + # Set numpy writing options + np.set_printoptions(threshold=sys.maxsize) + # Set Pandas writing options + pd.set_option("display.width", None) + pd.set_option("display.max_rows", None) + pd.set_option("display.max_columns", None) + + if isinstance(data, pd.DataFrame): + self.__append_dataframe(data) + elif isinstance(data, list) or isinstance(data, str): + self.__append_pretty(data) + elif isinstance(data, dict) or isinstance(data, OrderedDict): + self.__append_pretty_dict(data) + else: + self.__logger.error(__name__ + " - Invalid object to write into Text file: \n" + str(type(data))) + sys.exit() + + # Reset numpy writing options + np.set_printoptions(threshold=None) + # Reset Pandas writing options + pd.reset_option("display.width") + pd.reset_option("display.max_rows") + pd.reset_option("display.max_columns") + + def __append_dataframe(self, + data: PandasDataFrame, + label: str = ''): + """Append to text file using dataframe. + :param data: the dataframe to write. + """ + try: + with open(self.__path, 'a') as f: + pp.pprint(label + ":", stream=f) + data.to_string(f, columns=data.columns.values, header=True, index=True) + except (OSError, IOError) as e: + self.__logger.error(__name__ + " - Can not open the file: \n" + self.__path + "\n" + str(e)) + sys.exit() + except(): + self.__logger.error(__name__ + " - Can not append dataframe to file: \n" + self.__path) + sys.exit() + + def __append_pretty(self, + data: Callable[[List, str], None]): + """Append to text file using list. + :param data: the data to write. + """ + self.__logger.debug("Append a Dataframe to the text file.") + try: + with open(self.__path, 'a') as f: + pp.pprint(data, stream=f, width=self.__max_width, depth=self.__max_width, compact=True) + except (OSError, IOError) as e: + self.__logger.error(__name__ + " - Can not open the file: \n" + self.__path + "\n" + str(e)) + sys.exit() + except(): + self.__logger.error(__name__ + " - Can not append dictionary to file: \n" + self.__path) + sys.exit() + + def __append_pretty_dict(self, + data: Callable[[Dict, CollectionsOrderedDict], None], + label: str = ''): + """Append to text file using dictionary. + :param data: the data to write. + """ + self.__logger.debug("Append a Dictionary to the text file.") + try: + for label, value in data.items(): + if isinstance(value, dict) or isinstance(value, OrderedDict): + self.__append_pretty_dict(value, label) + elif isinstance(value, pd.DataFrame): + self.__append_dataframe(value, "") + else: + with open(self.__path, 'a') as f: + pp.pprint(label + ":", stream=f) + pp.pprint(value, stream=f, width=self.__max_width, depth=self.__max_width, compact=True) + except (OSError, IOError) as e: + self.__logger.error(__name__ + " - Can not open the file: \n" + self.__path + "\n" + str(e)) + sys.exit() + except(): + self.__logger.error(__name__ + " - Can not append dictionary to file: \n" + self.__path) + sys.exit() + + def size(self) -> int: + """Check number of lines in the text file. + :return: number of lines in the file. + """ + self.__logger.debug("Check number of lines in the CSV file.") + cnt_lines = 0 + try: + with open(self.__path, "r") as f: + for _ in f: + cnt_lines += 1 + except (OSError, IOError) as e: + self.__logger.error(__name__ + " - Can not open the file: \n" + self.__path + "\n" + str(e)) + sys.exit() + except(): + self.__logger.error(__name__ + " - Can not read the Text file: \n" + self.__path) + sys.exit() + return cnt_lines