-
Notifications
You must be signed in to change notification settings - Fork 1
/
monitor_utils.py
314 lines (254 loc) · 9.08 KB
/
monitor_utils.py
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
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
# == monitor_utils.py Author: Zuinige Rijder =========
"""
monitor utils
"""
import configparser
import errno
import sys
import os
from datetime import datetime, timezone
from pathlib import Path
import time
from typing import Generator
def log(msg: str) -> None:
"""log a message prefixed with a date/time format yyyymmdd hh:mm:ss"""
print(datetime.now().strftime("%Y%m%d %H:%M:%S") + ": " + msg)
def arg_has(string: str) -> bool:
"""arguments has string"""
for i in range(1, len(sys.argv)):
if sys.argv[i].lower() == string:
return True
return False
def get_vin_arg() -> str:
"""get vin argument"""
for i in range(1, len(sys.argv)):
if "vin=" in sys.argv[i].lower():
vin = sys.argv[i]
vin = vin.replace("vin=", "")
vin = vin.replace("VIN=", "")
return vin
return ""
def sleep(retries: int) -> int:
"""sleep when retries > 0"""
if retries > 0:
retries -= 1
if retries > 0:
log("Sleeping a minute")
time.sleep(60)
return retries
def get_filepath(filename: str) -> str:
"""get_filepath"""
if os.path.isfile(filename): # current directory
filepath = filename
else: # script directory
script_dirname = os.path.abspath(os.path.dirname(__file__))
filepath = f"{script_dirname}/{filename}"
if not os.path.isfile(filepath):
raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), filepath)
return filepath
def km_to_mile(kilometer: float) -> float:
"""Convert km to mile"""
mile = kilometer * 0.6213711922
return mile
def get(dictionary: dict, key: str, default: str = "") -> str:
"""get key from dictionary"""
if key in dictionary:
return dictionary[key].strip()
return default
def safe_divide(numerator: float, denumerator: float) -> float:
"""safe_divide"""
if denumerator == 0.0:
return 0.0
return numerator / denumerator
def to_int(string: str) -> int:
"""convert to int"""
if "None" in string:
return -1
return round(to_float(string))
def to_float(string: str) -> float:
"""convert to float"""
if "None" in string:
return 0.0
return float(string.strip())
def get_safe_float(number: float) -> float:
"""get safe float"""
if number is None:
return 0.0
return number
def float_to_string_no_trailing_zero(input_value: float) -> str:
"""float to string without trailing zero"""
return (f"{input_value:.1f}").rstrip("0").rstrip(".")
def is_true(string: str) -> bool:
"""return if string is true (True or not 0 digit)"""
if "None" in string:
return False
tmp = string.strip().lower()
if tmp == "true":
return True
elif tmp == "false":
return False
else:
return tmp.isdigit() and tmp != "0"
def same_year(d_1: datetime, d_2: datetime) -> bool:
"""return if same year"""
return d_1.year == d_2.year
def same_month(d_1: datetime, d_2: datetime) -> bool:
"""return if same month"""
if d_1.month != d_2.month:
return False
return d_1.year == d_2.year
def same_week(d_1: datetime, d_2: datetime) -> bool:
"""return if same week"""
if d_1.isocalendar().week != d_2.isocalendar().week:
return False
return d_1.year == d_2.year
def same_day(d_1: datetime, d_2: datetime) -> bool:
"""return if same day"""
if d_1.day != d_2.day:
return False
if d_1.month != d_2.month:
return False
return d_1.year == d_2.year
def split_on_comma(text: str) -> list[str]:
"""split string on comma and strip spaces around strings"""
result = [x.strip() for x in text.split(",")]
return result
def split_output_to_sheet_list(text: str) -> list[list[str]]:
"""split output to sheet list"""
return [split_on_comma(text)]
def split_output_to_sheet_float_list(text: str) -> list[list[float]]:
"""split output to sheet float list"""
result = [float(x.strip()) for x in text.split(",")]
return [result]
def get_last_line(filename: Path) -> str:
"""get last line of filename"""
last_line = ""
if filename.is_file():
with open(filename.name, "rb") as file:
try:
file.seek(-2, os.SEEK_END)
while file.read(1) != b"\n":
file.seek(-2, os.SEEK_CUR)
except OSError:
file.seek(0)
last_line = file.readline().decode().strip()
return last_line
def get_safe_datetime(date: datetime, tzinfo: timezone) -> datetime:
"""get safe datetime"""
if date is None:
return datetime(2000, 1, 1, tzinfo=tzinfo)
return date
def get_last_date(filename: str) -> str:
"""get last date of filename"""
last_date = "20000101" # millenium
last_line = get_last_line(Path(filename))
if last_line.startswith("20"): # year starts with 20
last_date = last_line.split(",")[0].strip()
return last_date
def read_reverse_order(file_name: str) -> Generator[str, None, None]:
"""read in reverse order"""
# Open file for reading in binary mode
with open(file_name, "rb") as read_obj:
# Move the cursor to the end of the file
read_obj.seek(0, os.SEEK_END)
# Get the current position of pointer i.e eof
pointer_location = read_obj.tell()
# Create a buffer to keep the last read line
buffer = bytearray()
# Loop till pointer reaches the top of the file
while pointer_location >= 0:
# Move the file pointer to the location pointed by pointer_location
read_obj.seek(pointer_location)
# Shift pointer location by -1
pointer_location = pointer_location - 1
# read that byte / character
new_byte = read_obj.read(1)
# If the read byte is newline character then one line is read
if new_byte == b"\n":
# Fetch the line from buffer and yield it
yield buffer.decode()[::-1]
# Reinitialize the byte array to save next line
buffer = bytearray()
else:
# If last read character is not eol then add it in buffer
buffer.extend(new_byte)
# If there is still data in buffer, then it is first line.
if len(buffer) > 0:
# Yield the first line too
yield buffer.decode()[::-1]
def read_reverse_order_init(
path: Path,
) -> tuple[bool, str, Generator[str, None, None]]:
""" "read_reverse_order_init"""
eof = False
last_read_line = ""
if path.is_file():
reverse_order_iterator = read_reverse_order(path.name)
return eof, last_read_line, reverse_order_iterator
else:
eof = True
# avoid mypy type error
empty_list: list[str] = []
return eof, last_read_line, (item for item in empty_list)
def reverse_read_next_line(
reverse_order_generator: Generator[str, None, None],
eof: bool,
last_read_line: str,
) -> tuple[bool, str]:
"""reverse_read_next_line"""
stop_value = None
while not eof:
line = next(reverse_order_generator, stop_value)
if line is stop_value:
eof = True
last_read_line = ""
else:
line = line.strip()
if (
line != "" and "Date" not in line and "date" not in line
): # skip header/empty lines
last_read_line = line
return eof, last_read_line
return eof, last_read_line
def read_translations() -> dict:
"""read translations"""
translations: dict = {}
parser = configparser.ConfigParser()
parser.read(get_filepath("monitor.cfg"))
monitor_settings = dict(parser.items("monitor"))
language = monitor_settings["language"].lower().strip()
translations_csv_file = Path(get_filepath("monitor.translations.csv"))
with translations_csv_file.open("r", encoding="utf-8") as inputfile:
linecount = 0
column = 1
for line in inputfile:
linecount += 1
split = split_on_comma(line)
if len(split) < 15:
log(f"Error: unexpected translation csvline {linecount}: {line}")
continue
key = split[0]
translation = split[1]
if linecount == 1:
continue # skip first line
elif linecount == 2:
# determine column index
for index, value in enumerate(split):
if value.lower() == language:
column = index
break
else:
current = split[column]
if current != "":
translation = current
translations[key] = translation
return translations
def get_translation(translations: dict[str, str], text: str) -> str:
"""get translation"""
if text in translations:
translation = translations[text]
# print(f"found translation: {translation}")
else:
print(f"fallback translation: {text}")
translation = text
return translation