wouter/Artemis
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add docstrings. Also add safe_cast function. Finally

Browse Source 
fix some minor issues.
This commit is contained in:
alessandro90
2019-05-25 15:18:06 +02:00
parent a7e36505e9
commit 43a9ce954e
14 changed files with 531 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
weatherdata.py
View File

@@ -7,27 +7,39 @@ from threads import (BaseDownloadThread,
UpdateForecastThread)
from constants import Constants
from switchable_label import MultiColorSwitchableLabel
from utilities import safe_cast
class _BaseWeatherData(QObject):
"""Base class for the weather data. Extends QObject."""
update_complete = pyqtSignal(bool)
def __init__(self):
"""Create a BaseDownloadThread object."""
super().__init__()
self._update_thread = BaseDownloadThread()
@property
def is_updating(self):
"""Return whether the thread is running."""
return self._update_thread.isRunning()
def update(self):
"""Start the thread."""
self._update_thread.start()
def _parse_data(self):
"""Dummy function. Must be overrided by subclasses."""
pass
@pyqtSlot()
def _parse_and_emit_signal(self):
"""Parse the data and emit an 'update_complete' signal.
If the download was not successful, do not parse the data.
The 'update_complete' signal propagates the thread status up to the
calling slot."""
status_ok = False
if self._update_thread.status is ThreadStatus.OK:
status_ok = True
@@ -35,15 +47,22 @@ class _BaseWeatherData(QObject):
self.update_complete.emit(status_ok)
def _double_split(self, string):
"""Given a string, return a list of lists.
First split on each line. Then split each line on whitespaces."""
return [i.split() for i in string.splitlines()]
def shutdown_thread(self):
"""Terminate the download thread."""
self._update_thread.terminate()
self._update_thread.wait()
class SpaceWeatherData(_BaseWeatherData):
"""Space weather class. Extends _BaseWeatherData."""
def __init__(self):
"""Set all attributes and connect the thread to _parse_and_emit_signal."""
super().__init__()
self.xray = ''
self.prot_el = ''
@@ -65,6 +84,9 @@ class SpaceWeatherData(_BaseWeatherData):
self._update_thread.finished.connect(self._parse_and_emit_signal)
def _parse_data(self):
"""Override _BaseWeatherData._parse_data.
Set all the data."""
self.xray = self._double_split(self.xray)
self.prot_el = self._double_split(self.prot_el)
self.ak_index = self._double_split(self.ak_index)
@@ -72,6 +94,7 @@ class SpaceWeatherData(_BaseWeatherData):
self.geo_storm = self._double_split(self.geo_storm)
def remove_data(self):
"""Remove the reference to all the data."""
self.xray = ''
self.prot_el = ''
self.ak_index = ''
@@ -91,6 +114,7 @@ class SpaceWeatherData(_BaseWeatherData):
class ForecastData(_BaseWeatherData):
"""3-day forecast class. Extends _BaseWeatherData."""
ROW_KEYWORDS = {
"solar_row": "S1 or greater",
@@ -101,6 +125,7 @@ class ForecastData(_BaseWeatherData):
}
def __init__(self, parent):
"""Initialize all attributes and connect the thread to _parse_and_emit_signal."""
super().__init__()
self.forecast = ''
self.probabilities = ''
@@ -112,16 +137,17 @@ class ForecastData(_BaseWeatherData):
self.__kp_index_row = None
self._update_thread = UpdateForecastThread(self)
self._update_thread.finished.connect(self._parse_and_emit_signal)
self.today_lbl = parent.today_lbl
self.today_p1_lbl = parent.today_p1_lbl
self.today_p2_lbl = parent.today_p2_lbl
# Cannot use '__' here because of the for loop below.
self._today_lbl = parent.today_lbl
self._today_p1_lbl = parent.today_p1_lbl
self._today_p2_lbl = parent.today_p2_lbl
self.__today_lbls = []
self.__today_p1_lbls = []
self.__today_p2_lbls = []
self.__all_lbls = []
flags = ['', 'p1_', 'p2_']
for flag in flags:
title_lbl = getattr(self, "today_" + flag + "lbl")
title_lbl = getattr(self, "_today_" + flag + "lbl")
title_lbl.setText("-")
for index in range(20):
label = getattr(
@@ -143,6 +169,9 @@ class ForecastData(_BaseWeatherData):
]
def _parse_data(self):
"""Override _BaseWeatherData._parse_data.
Set all the relevant data."""
# Remove possible '(G\d)' from the kp_index table
self.forecast = re.sub(
'\(G\d\)', lambda obj: '', self.forecast
@@ -154,16 +183,21 @@ class ForecastData(_BaseWeatherData):
self.probabilities = self.probabilities.splitlines()
def __split_lists(self):
"""Split the elements of forecast and probabilities."""
self.forecast = [i.split() for i in self.forecast]
self.probabilities = [i.split() for i in self.probabilities]
def __find_row_with(self, data, text):
"""Given a list of strings, return the index of the first string containing the target text."""
for i, row in enumerate(data):
if text in row:
return i
return None
def __get_rows(self):
"""Set all the rows needed for updating the screen.
Raise an exception if something goes wrong."""
self.__solar_row = self.__find_row_with(
self.forecast,
self.ROW_KEYWORDS["solar_row"]
@@ -185,26 +219,27 @@ class ForecastData(_BaseWeatherData):
self.ROW_KEYWORDS["kp_index_row"]
)
is_none = lambda x: x is None
if any([
is_none(self.__solar_row),
is_none(self.__event_row),
is_none(self.__rb_now_row),
is_none(self.__ga_now_row),
is_none(self.__kp_index_row)
self.__solar_row is None,
self.__event_row is None,
self.__rb_now_row is None,
self.__ga_now_row is None,
self.__kp_index_row is None
]):
raise Exception('Missing Rows')
def __set_dates(self):
"""Set the date labels."""
month = self.forecast[self.__solar_row - 1][0]
today = self.forecast[self.__solar_row - 1][1]
today_p1 = self.forecast[self.__solar_row - 1][3]
today_p2 = self.forecast[self.__solar_row - 1][5]
self.today_lbl.setText(month + ' ' + today)
self.today_p1_lbl.setText(month + ' ' + today_p1)
self.today_p2_lbl.setText(month + ' ' + today_p2)
self._today_lbl.setText(month + ' ' + today)
self._today_p1_lbl.setText(month + ' ' + today_p1)
self._today_p2_lbl.setText(month + ' ' + today_p2)
def __make_labels_table(self):
"""Organize all the arguments to feed __get_lbl_value."""
get_first_split = lambda x: x.split("/")[0]
get_second_split = lambda x: x.split("/")[1]
get_third_split = lambda x: x.split("/")[2]
@@ -277,34 +312,32 @@ class ForecastData(_BaseWeatherData):
]
]
def __get_lbl_value(self, data, row, col, f = None):
def __get_lbl_value(self, data, row, col, f=None):
"""Return the well-formatted string-value of the label."""
val = data[row][col]
if f is not None:
val = f(val)
val = val.lstrip('0').rstrip('%')
val = val.rstrip('%')
if len(val) > 1:
val = val.lstrip('0')
return val
def __is_integer(self, s):
try:
int(s)
except Exception:
return False
else:
return True
def __set_labels_values(self):
"""Set all the labels values."""
for lbl_list, table in zip(self.__all_lbls, self.__labels_table):
for lbl, row in zip(lbl_list, table):
lbl.switch_off()
value = self.__get_lbl_value(*row)
if self.__is_integer(value):
lbl.level = int(value)
lbl.level = safe_cast(value, int)
if not isinstance(lbl, MultiColorSwitchableLabel):
value += '%'
lbl.setText(value)
lbl.switch_on()
def update_all_labels(self):
"""Update all the labels values.
If an exception is raised in the process, do nothing."""
try:
self.__get_rows()
self.__split_lists()
@@ -315,5 +348,6 @@ class ForecastData(_BaseWeatherData):
pass
def remove_data(self):
"""Remove the reference to the downloaded data."""
self.forecast = ''
self.probabilities = ''