[b4a150]: / ReadersWriters / _TextFile.py

Download this file

224 lines (206 with data), 8.8 kB

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
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