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

Fix a bug in the *.spec files and also apply the following major changes:

Browse Source 
- Add support for adding the base folder to PATH.
- Only display one pop up window at a time in order to avoid confusion.
- Add automatic updates feature:
	- Windows and Linux versions will be shipped with an updater program used to
	  update both Artemis and the updater itself.
	- MacOs versions will not have the updater. Instead the user will be asked
	  to download the new software version (if present) via browser.
This commit is contained in:
Alessandro
2019-09-21 16:11:53 +02:00
parent 08b3312b23
commit 8e79bf6adf
34 changed files with 1360 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
src/artemis.py
View File

@@ -7,45 +7,58 @@ from time import sleep, time
from pandas import read_csv
from PyQt5.QtWidgets import (QMainWindow,
QApplication,
qApp,
QDesktopWidget,
QListWidgetItem,
QMessageBox,
QSplashScreen,)
from PyQt5.QtWidgets import (
QMainWindow,
QApplication,
qApp,
QDesktopWidget,
QListWidgetItem,
QMessageBox,
QSplashScreen,
)
from PyQt5.QtGui import QPixmap
from PyQt5 import uic
from PyQt5.QtCore import (QFileInfo,
Qt,
pyqtSlot,)
from PyQt5.QtCore import (
QFileInfo,
Qt,
pyqtSlot,
)
from acfvalue import ACFValue
from audio_player import AudioPlayer
from weatherdata import ForecastData
from download_window import DownloadWindow
from spaceweathermanager import SpaceWeatherManager
from constants import (Constants,
GfdType,
Database,
ChecksumWhat,
Messages,
Signal,)
from constants import (
Constants,
GfdType,
Database,
DownloadTarget,
Messages,
Signal,
__BASE_FOLDER__,
)
from themesmanager import ThemeManager
from filters import Filters
from utilities import (checksum_ok,
pop_up,
is_undef_freq,
is_undef_band,
format_numbers,
resource_path,
safe_cast,
is_mac_os)
from utilities import (
checksum_ok,
pop_up,
is_undef_freq,
is_undef_band,
format_numbers,
safe_cast,
)
from executable_utilities import IS_BINARY, resource_path
from os_utilities import IS_MAC
from web_utilities import get_db_hash_code
from downloadtargetfactory import get_download_target
from updatescontroller import UpdatesController
# import default_imgs_rc
__LATEST_VERSION__ = "3.0.1"
if hasattr(sys, '_MEIPASS'):
__LATEST_VERSION__ = "3.1.0"
if IS_BINARY:
__VERSION__ = __LATEST_VERSION__
else:
__VERSION__ = __LATEST_VERSION__ + ".Dev"
@@ -66,7 +79,7 @@ class Artemis(QMainWindow, Ui_MainWindow):
self.set_initial_size()
self.closing = False
self.download_window = DownloadWindow()
self.download_window.complete.connect(self.show_downloaded_signals)
self.download_window.complete.connect(self.action_after_download)
self.actionExit.triggered.connect(qApp.quit)
self.action_update_database.triggered.connect(self.ask_if_download)
self.action_check_db_ver.triggered.connect(self.check_db_ver)
@@ -185,13 +198,27 @@ class Artemis(QMainWindow, Ui_MainWindow):
self.main_tab.currentChanged.connect(self.hide_show_right_widget)
# Final operations.
self.updates_controller = UpdatesController(__LATEST_VERSION__, self)
self.updates_controller.start()
self.action_check_software_version.triggered.connect(
self.updates_controller.start_verify_software_version
)
# Final operations.
self.theme_manager.start()
self.load_db()
self.display_signals()
def action_after_download(self):
"""Decide what to do after a successful download.
If a new database was downloaded, show the signals."""
if self.download_window.target is DownloadTarget.DATA_FOLDER:
self.show_downloaded_signals()
@pyqtSlot()
def hide_show_right_widget(self):
"""Hide the audio player when forecast tabs are displayed."""
if self.main_tab.currentWidget() == self.forecast_tab:
self.fixed_audio_and_image.setVisible(False)
elif not self.fixed_audio_and_image.isVisible():
@@ -320,8 +347,9 @@ class Artemis(QMainWindow, Ui_MainWindow):
Do nothing if already downloading.
"""
if not self.download_window.isVisible():
self.download_window.start_download()
self.download_window.show()
self.download_window.activate(
get_download_target(DownloadTarget.DATA_FOLDER)
)
@pyqtSlot()
def ask_if_download(self):
@@ -341,7 +369,7 @@ class Artemis(QMainWindow, Ui_MainWindow):
self.download_db()
else:
try:
is_checksum_ok = checksum_ok(db, ChecksumWhat.DB)
is_checksum_ok = checksum_ok(db, get_db_hash_code())
except Exception:
pop_up(self, title=Messages.NO_CONNECTION,
text=Messages.NO_CONNECTION_MSG).show()
@@ -349,8 +377,8 @@ class Artemis(QMainWindow, Ui_MainWindow):
if not is_checksum_ok:
self.download_db()
else:
answer = pop_up(self, title=Messages.DB_UP_TO_DATE,
text=Messages.DB_UP_TO_DATE_MSG,
answer = pop_up(self, title=Messages.UP_TO_DATE,
text=Messages.UP_TO_DATE_MSG,
informative_text=Messages.DOWNLOAD_ANYWAY_QUESTION,
is_question=True,
default_btn=QMessageBox.No).exec()
@@ -381,14 +409,14 @@ class Artemis(QMainWindow, Ui_MainWindow):
self.download_db()
else:
try:
is_checksum_ok = checksum_ok(db, ChecksumWhat.DB)
is_checksum_ok = checksum_ok(db, get_db_hash_code())
except Exception:
pop_up(self, title=Messages.NO_CONNECTION,
text=Messages.NO_CONNECTION_MSG).show()
else:
if is_checksum_ok:
pop_up(self, title=Messages.DB_UP_TO_DATE,
text=Messages.DB_UP_TO_DATE_MSG).show()
pop_up(self, title=Messages.UP_TO_DATE,
text=Messages.UP_TO_DATE_MSG).show()
else:
answer = pop_up(self, title=Messages.DB_NEW_VER,
text=Messages.DB_NEW_VER_MSG,
@@ -411,12 +439,14 @@ class Artemis(QMainWindow, Ui_MainWindow):
Handle possible missing file error.
"""
try:
self.db = read_csv(os.path.join(Constants.DATA_FOLDER, Database.NAME),
sep=Database.DELIMITER,
header=None,
index_col=0,
dtype={name: str for name in Database.STRINGS},
names=Database.NAMES)
self.db = read_csv(
os.path.join(Constants.DATA_FOLDER, Database.NAME),
sep=Database.DELIMITER,
header=None,
index_col=0,
dtype={name: str for name in Database.STRINGS},
names=Database.NAMES
)
except FileNotFoundError:
self.search_bar.setDisabled(True)
answer = pop_up(self, title=Messages.NO_DB,
@@ -449,9 +479,9 @@ class Artemis(QMainWindow, Ui_MainWindow):
)
def collect_list(self, list_property, separator=Constants.FIELD_SEPARATOR):
"""Collect all the entrys of a QListWidget.
"""Collect all the entries of a QListWidget.
Handle multiple entries in one item seprated by a separator.
Handle multiple entries in one item separated by a separator.
Keyword argument:
separator -- the separator character for multiple-entries items.
"""
@@ -581,7 +611,6 @@ class Artemis(QMainWindow, Ui_MainWindow):
if item:
spectrogram_name = item.text()
path_spectr = os.path.join(
Constants.DATA_FOLDER,
Constants.SPECTRA_FOLDER,
spectrogram_name + Constants.SPECTRA_EXT
)
@@ -658,7 +687,7 @@ class Artemis(QMainWindow, Ui_MainWindow):
if __name__ == '__main__':
# For executables running on Mac Os systems.
if hasattr(sys, "_MEIPASS") and is_mac_os():
if IS_BINARY and IS_MAC and __BASE_FOLDER__ == os.curdir:
os.chdir(sys._MEIPASS)
my_app = QApplication(sys.argv)

6
src/artemis.ui
View File

@@ -9514,6 +9514,7 @@ QSlider::handle:horizontal {
</property>
<addaction name="action_check_db_ver"/>
<addaction name="action_update_database"/>
<addaction name="action_check_software_version"/>
</widget>
<widget class="QMenu" name="menu_themes">
<property name="title">
@@ -9592,6 +9593,11 @@ QSlider::handle:horizontal {
<string>GitHub</string>
</property>
</action>
<action name="action_check_software_version">
<property name="text">
<string>Check software version</string>
</property>
</action>
</widget>
<customwidgets>
<customwidget>

3
src/audio_player.py
View File

@@ -2,8 +2,8 @@ import os
from pygame import mixer
from PyQt5.QtCore import QTimer, pyqtSlot, QObject
from constants import Constants
import qtawesome as qta
from constants import Constants
class AudioPlayer(QObject):
@@ -127,7 +127,6 @@ class AudioPlayer(QObject):
"""Set the current audio sample."""
self._reset_audio_widget()
full_name = os.path.join(
Constants.DATA_FOLDER,
Constants.AUDIO_FOLDER,
fname + '.ogg'
)

106
src/constants.py
View File

@@ -1,13 +1,21 @@
from collections import namedtuple
from enum import Enum, auto
import os.path
from executable_utilities import get_executable_path
__BASE_FOLDER__ = get_executable_path()
class SupportedOs:
"""Supported operating systems."""
WINDOWS = "windows"
LINUX = "linux"
MAC = "mac"
class Ftype:
"""Container class to differentiate between frequency and band.
Used in reset_fb_filters.
"""
"""Container class to differentiate between frequency and band."""
FREQ = "freq"
BAND = "band"
@@ -20,30 +28,13 @@ class GfdType(Enum):
LOC = auto()
class ChecksumWhat(Enum):
"""Enum class to distinguish the object you want to verify the checksum."""
class DownloadTarget(Enum):
"""Enum class to distinguish the object being downloaded."""
FOLDER = auto()
DB = auto()
class Messages:
"""Container class for messages to be displayed."""
DB_UP_TO_DATE = "Already up to date"
DB_UP_TO_DATE_MSG = "No newer version to download."
DB_NEW_VER = "New version available"
DB_NEW_VER_MSG = "A new version of the database is available for download."
NO_DB_AVAIL = "No database detected."
NO_DB = "No database"
DOWNLOAD_NOW_QUESTION = "Do you want to download it now?"
DOWNLOAD_ANYWAY_QUESTION = "Do you want to download it anyway?"
NO_CONNECTION = "No connection"
NO_CONNECTION_MSG = "Unable to establish an internet connection."
BAD_DOWNLOAD = "Something went wrong"
BAD_DOWNLOAD_MSG = "Something went wrong with the download.\nCheck your internet connection and try again."
SLOW_CONN = "Slow internet connection"
SLOW_CONN_MSG = "Your internet connection is unstable or too slow."
DATA_FOLDER = auto()
DB = auto()
SOFTWARE = auto()
UPDATER = auto()
class Signal:
@@ -109,7 +100,10 @@ _Band = namedtuple("Band", ["lower", "upper"])
class Constants:
"""Container class for several constants of the software."""
EXECUTABLE_NAME = os.path.join(__BASE_FOLDER__, "Artemis")
UPDATER_SOFTWARE = os.path.join(__BASE_FOLDER__, "_ArtemisUpdater")
CLICK_TO_UPDATE_STR = "Click to update"
VERSION_LINK = "https://aresvalley.com/Storage/Artemis/Package/latest_versions.json"
SIGIDWIKI = "https://www.sigidwiki.com/wiki/Signal_Identification_Guide"
ADD_SIGNAL_LINK = "https://www.sigidwiki.com/index.php/Special:FormEdit/Signal/?preload=Signal_Identification_Wiki:Signal_form_preload_text"
FORUM_LINK = "https://aresvalley.com/community/"
@@ -136,10 +130,7 @@ class Constants:
"https://amunters.home.xs4all.nl/aurorastatus.gif"]
SEARCH_LABEL_IMG = "search_icon.png"
VOLUME_LABEL_IMG = "volume.png"
DATA_FOLDER = "Data"
SPECTRA_FOLDER = "Spectra"
SPECTRA_EXT = ".png"
AUDIO_FOLDER = "Audio"
ACTIVE = "active"
INACTIVE = "inactive"
LABEL_ON_COLOR = "on"
@@ -184,6 +175,61 @@ class Constants:
NOT_SELECTED = "nosignalselected.png"
FIELD_SEPARATOR = ";"
ACF_SEPARATOR = " - "
DATA_FOLDER = os.path.join(__BASE_FOLDER__, "Data")
SPECTRA_FOLDER = os.path.join(DATA_FOLDER, "Spectra")
AUDIO_FOLDER = os.path.join(DATA_FOLDER, "Audio")
DEFAULT_IMGS_FOLDER = os.path.join(":", "pics", "default_pics")
DEFAULT_NOT_SELECTED = os.path.join(DEFAULT_IMGS_FOLDER, NOT_SELECTED)
DEFAULT_NOT_AVAILABLE = os.path.join(DEFAULT_IMGS_FOLDER, NOT_AVAILABLE)
class Messages:
"""Container class for messages to be displayed."""
FEATURE_NOT_AVAILABLE = "Feature not available"
SCRIPT_NOT_UPDATE = "When running from source, software updates\ncannot be checked."
UPDATES_AVAILABALE = "Updates available"
UPDATES_MSG = "Do you want to install the updates now?"
UP_TO_DATE = "Already up to date"
UP_TO_DATE_MSG = "No newer version to download."
DB_NEW_VER = "New version available"
DB_NEW_VER_MSG = "A new version of the database is available for download."
NO_DB_AVAIL = "No database detected."
NO_DB = "No database"
DOWNLOAD_NOW_QUESTION = "Do you want to download it now?"
DOWNLOAD_ANYWAY_QUESTION = "Do you want to download it anyway?"
NO_CONNECTION = "No connection"
NO_CONNECTION_MSG = "Unable to establish an internet connection."
BAD_DOWNLOAD = "Something went wrong"
BAD_DOWNLOAD_MSG = "Something went wrong with the download.\nCheck your internet connection and try again."
SLOW_CONN = "Slow internet connection"
SLOW_CONN_MSG = "Your internet connection is unstable or too slow."
NEW_VERSION_AVAILABLE = "New software version"
NEW_VERSION_MSG = lambda v: f"The software version {v} is available." # noqa: E731
DOWNLOAD_SUGG_MSG = "Download new version now?"
class ThemeConstants:
"""Container class for all the theme-related constants."""
EXTENSION = ".qss"
ICONS_FOLDER = "icons"
DEFAULT = "dark"
CURRENT = "__current_theme"
COLORS = "colors.txt"
COLOR_SEPARATOR = "="
DEFAULT_ACTIVE_COLOR = "#000000"
DEFAULT_INACTIVE_COLOR = "#9f9f9f"
DEFAULT_OFF_COLORS = "#000000", "#434343"
DEFAULT_ON_COLORS = "#4b79a1", "#283e51"
DEFAULT_TEXT_COLOR = "#ffffff"
THEME_NOT_FOUND = "Theme not found"
MISSING_THEME = "Missing theme folder."
MISSING_THEME_FOLDER = "Themes folder not found.\nOnly the basic theme is available."
THEME_FOLDER_NOT_FOUND = "Themes folder not found"
FOLDER = os.path.join(__BASE_FOLDER__, "themes")
DEFAULT_ICONS_PATH = os.path.join(FOLDER, DEFAULT, ICONS_FOLDER)
DEFAULT_SEARCH_LABEL_PATH = os.path.join(DEFAULT_ICONS_PATH, Constants.SEARCH_LABEL_IMG)
DEFAULT_VOLUME_LABEL_PATH = os.path.join(DEFAULT_ICONS_PATH, Constants.VOLUME_LABEL_IMG)
CURRENT_THEME_FILE = os.path.join(FOLDER, CURRENT)
DEFAULT_THEME_PATH = os.path.join(FOLDER, DEFAULT)

8
src/download_db_window.ui
View File

@@ -7,11 +7,11 @@
<x>0</x>
<y>0</y>
<width>400</width>
<height>160</height>
<height>185</height>
</rect>
</property>
<property name="windowTitle">
<string>Download database</string>
<string>Downloading</string>
</property>
<property name="windowIcon">
<iconset resource="default_imgs.qrc">
@@ -29,7 +29,7 @@
</font>
</property>
<property name="text">
<string>Downloading database
<string>Downloading updates
Please wait...
</string>
</property>
@@ -69,7 +69,7 @@ Please wait...
</widget>
</item>
<item>
<widget class="QProgressBar" name="progressBar">
<widget class="QProgressBar" name="_progress_bar">
<property name="minimum">
<number>0</number>
</property>

25
src/download_window.py
View File

@@ -2,7 +2,8 @@ from PyQt5 import uic
from PyQt5.QtCore import Qt, pyqtSlot, pyqtSignal
from PyQt5.QtWidgets import QWidget
from threads import DownloadThread, ThreadStatus
from utilities import pop_up, resource_path
from utilities import pop_up
from executable_utilities import resource_path
from constants import Constants, Messages
@@ -12,10 +13,12 @@ Ui_Download_window, _ = uic.loadUiType(
class DownloadWindow(QWidget, Ui_Download_window):
"""Subclass QWidget and Ui_Download_window. It is the window displayed during the database download."""
"""Subclass QWidget and Ui_Download_window. It is the window displayed during
downloads and software updates."""
complete = pyqtSignal()
closed = pyqtSignal()
_PROGRESS_CONEVERSION_FACTOR = 1024
def __init__(self):
"""Initialize the window."""
@@ -47,10 +50,22 @@ class DownloadWindow(QWidget, Ui_Download_window):
self._download_thread.speed_progress.connect(self._display_speed)
self.closed.connect(self._download_thread.set_exit)
self.cancel_btn.clicked.connect(self._terminate_process)
self._size = 0
self.target = None
def start_download(self):
def _prepare_progress_bar(self, size):
"""Prepare the progress bar for the upcoming download."""
self._progress_bar.setMinimum(0)
self._progress_bar.setMaximum(size)
self._progress_bar.setValue(0)
def activate(self, target):
"""Start the download thread."""
self._download_thread.start()
self._size = target.size
self.target = target.target
self._prepare_progress_bar(target.size)
self._download_thread.start(target)
self.show()
def _download_format_str(self, n):
"""Return a well-formatted string with the downloaded MB."""
@@ -77,6 +92,8 @@ class DownloadWindow(QWidget, Ui_Download_window):
self.status_lbl.setText(self._download_format_str(progress))
elif progress == Constants.EXTRACTING_CODE:
self.status_lbl.setText(Constants.EXTRACTING_MSG)
if self._size > 0:
self._progress_bar.setValue(progress * self._PROGRESS_CONEVERSION_FACTOR)
def show(self):
"""Extends QWidget.show. Set downloaded MB and speed to zero."""

164
src/downloadtargetfactory.py Normal file
View File

@@ -0,0 +1,164 @@
from contextlib import contextmanager
from shutil import rmtree
from os import remove
import os.path
import stat
from constants import (
Constants,
Database,
__BASE_FOLDER__,
ThemeConstants,
DownloadTarget,
SupportedOs,
)
from os_utilities import get_os
from web_utilities import get_folder_hash_code
from zipfile import ZipFile
from tarfile import TarFile
class _ZipExtractor:
"""Extractor class for zip files.
Exposes a static method which can be used as a context manager."""
@staticmethod
@contextmanager
def open(fileobj):
zipped = ZipFile(fileobj)
try:
yield zipped
finally:
zipped.close()
class _TarExtractor:
"""Extractor class for tar files.
Exposes a static method which can be used as a context manager."""
@staticmethod
@contextmanager
def open(fileobj):
tarfile = TarFile.open(fileobj=fileobj)
try:
yield tarfile
finally:
tarfile.close()
EXTRACTORS = {
SupportedOs.WINDOWS: _ZipExtractor,
SupportedOs.LINUX: _TarExtractor,
# No extractor for MacOs, just download the file through the browser.
}
def _on_rmtree_error(func, path, excinfo):
"""Function called whenever rmtree fails."""
os.chmod(path, stat.S_IWRITE)
func(path)
def _delete_data_folder():
"""Delete the Data folder."""
if os.path.exists(Constants.DATA_FOLDER):
rmtree(Constants.DATA_FOLDER, onerror=_on_rmtree_error)
def _delete_updater():
"""Delete the updater program."""
if os.path.exists(Constants.UPDATER_SOFTWARE):
remove(Constants.UPDATER_SOFTWARE)
def _delete_software():
"""Delete the main program and the themes folder."""
if os.path.exists(Constants.EXECUTABLE_NAME):
remove(Constants.EXECUTABLE_NAME) # Remove Artemis executable.
if os.path.exists(ThemeConstants.FOLDER): # One could not have the theme folder for some reason.
rmtree(ThemeConstants.FOLDER, onerror=_on_rmtree_error)
class _DataFolderInfo:
"""Simple class to implement the interface of a 'target' object for the data folder:
- url;
- hash_code;
- size."""
def __init__(self):
self.url = Database.LINK_LOC
self.hash_code = get_folder_hash_code()
self.size = 0
class _BaseDownloadTarget:
"""Base class for the '_Download*Target' objects.
Contains all the attributes needed by DownloadWindow and DownloadThread
to do the job."""
def __init__(self, target, dest_path, target_enum, Extractor, delete_files):
self.url = target.url
self.hash_code = target.hash_code
self.size = target.size
self.dest_path = dest_path
self.target = target_enum
self.Extractor = Extractor
self.delete_files = delete_files
class _DownloadDataFolderTarget(_BaseDownloadTarget):
"""Extend _BaseDownloadTarget. Represent the data folder."""
def __init__(self, data_folder_info, dest_path=__BASE_FOLDER__):
super().__init__(
target=data_folder_info,
dest_path=dest_path,
target_enum=DownloadTarget.DATA_FOLDER,
Extractor=_ZipExtractor,
delete_files=_delete_data_folder
)
class _DownloadSoftwareTarget(_BaseDownloadTarget):
"""Extends _BaseDownloadTarget. Represents the main software."""
def __init__(self, software, dest_path=__BASE_FOLDER__):
super().__init__(
target=software,
dest_path=dest_path,
target_enum=DownloadTarget.SOFTWARE,
Extractor=EXTRACTORS[get_os()],
delete_files=_delete_software
)
class _DownloadUpdaterTarget(_BaseDownloadTarget):
"""Extends _BaseDownloadTarget. Represents the updater software."""
def __init__(self, updater, dest_path=__BASE_FOLDER__):
super().__init__(
target=updater,
dest_path=dest_path,
target_enum=DownloadTarget.UPDATER,
Extractor=EXTRACTORS[get_os()],
delete_files=_delete_updater
)
def get_download_target(target_type, target=None):
"""Return a Download*Obj based on the target download.
These objects expose a common interface:
Attributes:
- url;
- hash_code;
- dest_path;
- target: an element of the enum DownloadTarget;
- Extractor: an object which exposes an 'open(fileobj)' method
to extract compressed files;
- delete_files: a function to remove the old files."""
if target_type is DownloadTarget.DATA_FOLDER:
return _DownloadDataFolderTarget(_DataFolderInfo())
elif target_type is DownloadTarget.UPDATER and target is not None:
return _DownloadUpdaterTarget(target)
elif target_type is DownloadTarget.SOFTWARE and target is not None:
return _DownloadSoftwareTarget(target)
else:
raise Exception("ERROR: Invalid download target!")

33
src/executable_utilities.py Normal file
View File

@@ -0,0 +1,33 @@
import sys
from shutil import which
import os
import os.path
def _is_executable_version():
"""Return whether the binary version is running."""
return hasattr(sys, "_MEIPASS")
IS_BINARY = _is_executable_version()
def get_executable_path():
"""Check whether the executable is in the PATH folder.
Return the full path or just an ampty string if it is not found
in the PATH folder."""
path = which("Artemis")
if path is not None:
return os.path.dirname(path)
else: # Assume that the executable is in the cwd.
return os.curdir
def resource_path(relative_path):
"""Get absolute path to resource, works for dev and for PyInstaller."""
try:
base_path = sys._MEIPASS
except Exception:
base_path = os.path.abspath(".")
return os.path.join(base_path, relative_path)

28
src/filters.py
View File

@@ -7,21 +7,23 @@ The only class exposed is Filters which provides the following methods:
from functools import partial
import webbrowser
from PyQt5.QtWidgets import QListWidgetItem, QTreeWidgetItem
from PyQt5.QtCore import pyqtSlot, QObject
from constants import (Constants,
Ftype,
Signal,)
from utilities import (uncheck_and_emit,
connect_events_to_func,
filters_limit,
is_undef_freq,
is_undef_band,
safe_cast,
show_matching_strings,
get_field_entries,)
from constants import (
Constants,
Ftype,
Signal,
)
from utilities import (
uncheck_and_emit,
connect_events_to_func,
filters_limit,
is_undef_freq,
is_undef_band,
safe_cast,
show_matching_strings,
get_field_entries,
)
class _BaseFilter(QObject):

34
src/os_utilities.py Normal file
View File

@@ -0,0 +1,34 @@
import sys
from constants import SupportedOs
def _is_mac_os():
"""Return True if running OS is Mac."""
return sys.platform == 'darwin'
def _is_win_os():
"""Return True if running OS is Windows."""
return sys.platform == 'win32'
def _is_linux_os():
"""Return True if running OS is Linux."""
return sys.platform == 'linux'
IS_MAC = _is_mac_os()
IS_LINUX = _is_linux_os()
IS_WINDOWS = _is_win_os()
def get_os():
"""Get the name of the current running operating system."""
if IS_WINDOWS:
return SupportedOs.WINDOWS
elif IS_LINUX:
return SupportedOs.LINUX
elif IS_MAC:
return SupportedOs.MAC
else:
raise Exception("ERROR: OS not recognized.")

2
src/themes/elegant_dark/elegant_dark.qss
View File

@@ -404,4 +404,4 @@ QScrollBar::add-line:horizontal, QScrollBar::sub-line:horizontal,
QScrollBar::add-line:vertical, QScrollBar::sub-line:vertical {
border: none;
background: none;
}
}

66
src/themesmanager.py
View File

@@ -4,36 +4,10 @@ import re
from PyQt5.QtWidgets import QAction, QActionGroup
from PyQt5.QtCore import pyqtSlot
from PyQt5.QtGui import QPixmap
from constants import Constants
from constants import Constants, ThemeConstants
from utilities import pop_up
class ThemeConstants:
"""Container class for all the theme-related constants."""
FOLDER = "themes"
EXTENSION = ".qss"
ICONS_FOLDER = "icons"
DEFAULT = "dark"
CURRENT = ".current_theme"
COLORS = "colors.txt"
COLOR_SEPARATOR = "="
DEFAULT_ACTIVE_COLOR = "#000000"
DEFAULT_INACTIVE_COLOR = "#9f9f9f"
DEFAULT_OFF_COLORS = "#000000", "#434343"
DEFAULT_ON_COLORS = "#4b79a1", "#283e51"
DEFAULT_TEXT_COLOR = "#ffffff"
THEME_NOT_FOUND = "Theme not found"
MISSING_THEME = "Missing theme in '" + FOLDER + "' folder."
MISSING_THEME_FOLDER = "'" + FOLDER + "'" + " folder not found.\nOnly the basic theme is available."
THEME_FOLDER_NOT_FOUND = "'" + FOLDER + "'" + " folder not found"
DEFAULT_ICONS_PATH = os.path.join(FOLDER, DEFAULT, ICONS_FOLDER)
DEFAULT_SEARCH_LABEL_PATH = os.path.join(DEFAULT_ICONS_PATH, Constants.SEARCH_LABEL_IMG)
DEFAULT_VOLUME_LABEL_PATH = os.path.join(DEFAULT_ICONS_PATH, Constants.VOLUME_LABEL_IMG)
CURRENT_THEME_FILE = os.path.join(FOLDER, CURRENT)
DEFAULT_THEME_PATH = os.path.join(FOLDER, DEFAULT)
class _ColorsHandler:
"""Manage the theme's secondary colors.
@@ -125,7 +99,6 @@ class ThemeManager:
self._owner.spaceweather_screen.refreshable_labels.set(
"switch_off_colors", ThemeConstants.DEFAULT_OFF_COLORS
)
self._theme_names = {}
@pyqtSlot()
@@ -179,7 +152,8 @@ class ThemeManager:
new_theme = ag.addAction(
QAction(
theme_name,
self._owner, checkable=True
self._owner,
checkable=True
)
)
self._owner.menu_themes.addAction(new_theme)
@@ -295,31 +269,37 @@ class ThemeManager:
try:
with open(ThemeConstants.CURRENT_THEME_FILE, "w") as current_theme:
current_theme.write(self._theme_path)
current_theme.write(os.path.basename(self._theme_path))
except Exception:
pass
def apply_default_theme(self):
"""Apply the default theme if no theme is set or the theme name is invalid."""
try:
self._theme_names[
self._pretty_name(ThemeConstants.DEFAULT)
].setChecked(True)
except Exception:
pop_up(
self._owner,
title=ThemeConstants.THEME_NOT_FOUND,
text=ThemeConstants.MISSING_THEME
).show()
else:
self._apply(ThemeConstants.DEFAULT_THEME_PATH)
def start(self):
"""Start the theme manager."""
self._detect_themes()
if os.path.exists(ThemeConstants.CURRENT_THEME_FILE):
with open(ThemeConstants.CURRENT_THEME_FILE, "r") as current_theme_path:
theme_path = current_theme_path.read()
with open(ThemeConstants.CURRENT_THEME_FILE, "r") as current_theme_name:
theme_path = os.path.join(ThemeConstants.FOLDER, current_theme_name.read())
theme_name = self._pretty_name(os.path.basename(theme_path))
try:
self._theme_names[theme_name].setChecked(True)
except Exception:
pop_up(self._owner, title=ThemeConstants.THEME_NOT_FOUND,
text=ThemeConstants.MISSING_THEME).show()
self.apply_default_theme()
else:
self._apply(theme_path)
else:
try:
self._theme_names[
self._pretty_name(ThemeConstants.DEFAULT)
].setChecked(True)
except Exception:
pop_up(self._owner, title=ThemeConstants.THEME_NOT_FOUND,
text=ThemeConstants.MISSING_THEME).show()
else:
self._apply(ThemeConstants.DEFAULT_THEME_PATH)
self.apply_default_theme()

190
src/threads.py
View File

@@ -2,18 +2,18 @@ import asyncio
from enum import Enum, auto
from io import BytesIO
from math import ceil
import os.path
from shutil import rmtree
from time import perf_counter
from zipfile import ZipFile
import aiohttp
from PyQt5.QtCore import QThread, pyqtSignal
from constants import Constants, Database, ChecksumWhat
from utilities import checksum_ok, get_pool_manager, get_cacert_file
import ssl
from time import perf_counter
import aiohttp
from PyQt5.QtCore import QThread, pyqtSignal, pyqtSlot
from constants import Constants
from utilities import checksum_ok
from web_utilities import (
get_cacert_file,
get_pool_manager,
)
# Needed for pyinstaller compilation.
import encodings.idna
import encodings.idna # noqa: 401
class ThreadStatus(Enum):
@@ -39,30 +39,31 @@ class BaseDownloadThread(QThread):
super().__init__(parent)
self.status = ThreadStatus.UNDEFINED
def __del__(self):
"""Force the termination of the thread."""
self.terminate()
self.wait()
# def __del__(self):
# """Force the termination of the thread."""
# self.terminate()
# self.wait()
class DownloadThread(BaseDownloadThread):
"""Subclass BaseDownloadThread. Download the database, images and audio samples."""
"""Subclass BaseDownloadThread. Download the database folder. Used also for software updates."""
progress = pyqtSignal(int)
speed_progress = pyqtSignal(float)
_CHUNK = 128 * 1024
_MEGA = 1024**2
_DELTAT = 2
def __init__(self):
def __init__(self, min_bytes=1024**2):
"""Just call super().__init__."""
self._db = None
self._exit_call = False
super().__init__()
self._min_bytes = min_bytes
self._data = None
self._exit_call = False
self._target = None
def _pretty_len(self, byte_obj):
"""Return a well-formatted number of downloaded MB."""
mega = len(byte_obj) / self._MEGA
mega = len(byte_obj) / self._min_bytes
if mega.is_integer():
return int(mega)
else:
@@ -71,87 +72,144 @@ class DownloadThread(BaseDownloadThread):
def _get_download_speed(self, data, delta):
"""Return the download speed in MB/s."""
return round(
(len(data) / self._MEGA) / delta, 2
(len(data) / self._min_bytes) / delta, 2
)
@pyqtSlot()
def set_exit(self):
"""Time to shutdown the thread.
Executed in the main thread."""
self._exit_call = True
def start(self, target):
"""Start the thread. Set the correct download options first."""
self._target = target
super().start()
def _download_loop(self):
"""Read a chunck of the downloaded data at every iteration."""
raw_data = bytes(0)
sub_data = bytes(0)
start = perf_counter()
prev_downloaded = 0
while True:
try:
data = self._data.read(self._CHUNK)
except Exception:
raise _SlowConnError
else:
delta = perf_counter() - start
if not data:
break
raw_data += data
sub_data += data
# Emit a progress signal only if at least self._min_bytes has been downloaded.
if len(raw_data) - prev_downloaded >= self._min_bytes:
prev_downloaded = len(raw_data)
self.progress.emit(self._pretty_len(raw_data))
if delta >= self._DELTAT:
self.speed_progress.emit(
self._get_download_speed(sub_data, delta)
)
sub_data = bytes(0)
start = perf_counter()
if self._exit_call:
self._data.release_conn()
break
return raw_data
def run(self):
"""Override QThread.run. Download the database, images and audio samples.
Handle all possible exceptions. Also extract the files
in the local folder."""
in the destination folder."""
self.status = ThreadStatus.UNDEFINED
self._db = None
raw_data = bytes(0)
sub_data = bytes(0)
self._data = None
try:
self._db = get_pool_manager().request(
self._data = get_pool_manager().request(
'GET',
Database.LINK_LOC,
self._target.url,
preload_content=False,
timeout=4.0
)
start = perf_counter()
prev_downloaded = 0
while True:
try:
data = self._db.read(self._CHUNK)
except Exception:
raise _SlowConnError
else:
delta = perf_counter() - start
if not data:
break
raw_data += data
sub_data += data
# Emit a progress signal only if at least 1 MB has been downloaded.
if len(raw_data) - prev_downloaded >= self._MEGA:
prev_downloaded = len(raw_data)
self.progress.emit(self._pretty_len(raw_data))
if delta >= self._DELTAT:
self.speed_progress.emit(
self._get_download_speed(sub_data, delta)
)
sub_data = bytes(0)
start = perf_counter()
if self._exit_call:
self._exit_call = False
self._db.release_conn()
return
raw_data = self._download_loop()
if self._exit_call:
self._exit_call = False
return
except Exception as e: # No (or bad) internet connection.
self._db.release_conn()
self._data.release_conn()
if isinstance(e, _SlowConnError):
self.status = ThreadStatus.SLOW_CONN_ERR
else:
self.status = ThreadStatus.NO_CONNECTION_ERR
return
if self._db.status != 200:
if self._data.status != 200:
self.status = ThreadStatus.BAD_DOWNLOAD_ERR
return
try:
is_checksum_ok = checksum_ok(raw_data, ChecksumWhat.FOLDER)
except Exception: # checksum_ok unable to connect to the reference.
self.status = ThreadStatus.NO_CONNECTION_ERR
if self._wrong_checksum(raw_data):
return
self._target.delete_files()
self._extract(raw_data)
def _wrong_checksum(self, raw_data):
"""Verify the checksum of the downloaded data and set the status accordingly."""
try:
is_checksum_ok = checksum_ok(raw_data, self._target.hash_code)
except Exception: # Invalid hash code.
self.status = ThreadStatus.NO_CONNECTION_ERR
return True
else:
if not is_checksum_ok:
self.status = ThreadStatus.BAD_DOWNLOAD_ERR
return
if os.path.exists(Constants.DATA_FOLDER):
rmtree(Constants.DATA_FOLDER)
return True
return False
def _extract(self, raw_data):
"""Unzip and save the downloaded data into the destination folder."""
try:
self.progress.emit(Constants.EXTRACTING_CODE)
self.speed_progress.emit(Constants.ZERO_FINAL_SPEED)
with ZipFile(BytesIO(raw_data)) as zipped:
zipped.extractall()
with self._target.Extractor.open(fileobj=BytesIO(raw_data)) as zipped:
zipped.extractall(path=self._target.dest_path)
except Exception:
self.status = ThreadStatus.UNKNOWN_ERR
else:
self.status = ThreadStatus.OK
class UpdatesControllerThread(BaseDownloadThread):
on_success = pyqtSignal(bool)
def __init__(self, version_controller):
super().__init__()
self.version_controller = version_controller
def run(self):
if self.version_controller.update():
self.on_success.emit(True)
else:
self.on_success.emit(False)
# class GenercWorkerThread(BaseDownloadThread):
# def __init__(self, func, *args, **kwargs):
# super().__init__()
# self._args = args
# self._kwargs = kwargs
# self._func
# def run(self):
# self.status = ThreadStatus.UNDEFINED
# try:
# self._func(self._args, self._kwargs)
# except Exception:
# self.status = ThreadStatus.UNKNOWN_ERR
# else:
# self.status = ThreadStatus.OK
class _AsyncDownloader:
"""Mixin class for asynchronous threads."""

200
src/updater.py Normal file
View File

@@ -0,0 +1,200 @@
import argparse
import os
import os.path
import sys
from PyQt5.QtCore import QObject, QProcess
from PyQt5.QtGui import QPixmap
from PyQt5.QtWidgets import QApplication, qApp
from download_window import DownloadWindow
from constants import Constants, DownloadTarget
from downloadtargetfactory import get_download_target
__VERSION__ = "0.0.1"
# Global stylesheet.
stylesheet = """
/*************************************
Main Window and Splitters
**************************************/
QWidget:window {
background-color: #29353B;
}
/*************************************
Main menu (Bar)
**************************************/
QMenuBar {
background-color: transparent;
color: #AFBDC4;
}
QMenuBar::item {
background-color: transparent;
}
QMenuBar::item:disabled {
color: gray;
}
QMenuBar::item:selected {
color: #FFFFFF;
border-bottom: 2px solid #88cc00;
}
QMenuBar::item:pressed {
color: #FFFFFF;
border-bottom: 2px solid #88cc00;
}
QToolBar {
background-color: transparent;
border: 1px solid transparent;
}
QToolBar:handle {
background-color: transparent;
border-left: 2px dotted #80CBC4;
color: transparent;
}
QToolBar::separator {
border: 0;
}
QMenu {
background-color: #263238;
color: #AFBDC4;
}
QMenu::item:selected {
color: #FFFFFF;
}
QMenu::item:pressed {
color: #FFFFFF;
}
QMenu::separator {
background-color: transparent;
height: 1px;
margin-left: 10px;
margin-right: 10px;
margin-top: 5px;
margin-bottom: 5px;
}
/*************************************
Progressbar
**************************************/
QProgressBar
{
border: 2px solid grey;
border-radius: 5px;
text-align: center;
}
QProgressBar::chunk
{
background-color: #88cc00;
width: 2.15px;
margin: 0.5px;
}
/*************************************
Labels and Rich Text boxes
**************************************/
QLabel {
background-color: transparent;
color: #CFD8DC;
}
QDialog {
background-color: transparent;
color: #949a9c;
}
QTextBrowser {
background-color: transparent;
color: #949a9c;
}
/*************************************
Buttons
**************************************/
QPushButton {
background-color: transparent;
color: #AFBDC4;
border: 1px solid transparent;
padding: 4px 22px;
}
QPushButton:hover {
border-left: 2px solid #88cc00;
border-right: 2px solid #88cc00;
color: #FFFFFF;
}
QPushButton:pressed {
color: #FFFFFF;
}
QPushButton:disabled {
color:#546E7A;
}
QPushButton:checked {
color: #88cc00;
}
"""
class _ArtemisUpdater(QObject):
"""Updater of the main software."""
def __init__(self, target):
super().__init__()
self.target = get_download_target(DownloadTarget.SOFTWARE, target)
self.download_window = DownloadWindow()
self.download_window.setStyleSheet(stylesheet)
self.download_window.cancel_btn.clicked.connect(qApp.quit)
self.download_window.complete.connect(self.start_main_program)
def start(self):
"""Close the main program and start the download."""
self.download_window.activate(self.target)
def init_ok(self):
return self.target.url and self.target.hash_code and self.target.size > 0
def start_main_program(self):
"""Restart the (updated) main program and close the updater."""
self.download_window.setVisible(False)
artemis = QProcess()
try:
artemis.startDetached(Constants.EXECUTABLE_NAME)
except BaseException:
pass
qApp.quit()
if __name__ == '__main__':
parser = argparse.ArgumentParser(prog='Artemis Updater')
parser.add_argument("url", nargs="?", default="", type=str, help="Download url")
parser.add_argument("hash_code", nargs="?", default="", type=str, help="sha256 of the file")
parser.add_argument("size", nargs="?", default=0, type=int, help="Size (KB) of the file")
parser.add_argument('--version', action='version', version=__VERSION__)
args = parser.parse_args()
my_app = QApplication(sys.argv)
ARTEMIS_ICON = os.path.join(":", "icon", "default_pics", "Artemis3.500px.png")
img = QPixmap(ARTEMIS_ICON)
updater = _ArtemisUpdater(args)
if not updater.init_ok():
updater.start_main_program()
else:
updater.start()
sys.exit(my_app.exec_())

141
src/updatescontroller.py Normal file
View File

@@ -0,0 +1,141 @@
import subprocess as sp
import webbrowser
from PyQt5.QtCore import QObject, pyqtSlot, QProcess
from PyQt5.QtWidgets import QMessageBox, qApp
from constants import Constants, Messages, DownloadTarget
from downloadtargetfactory import get_download_target
from utilities import pop_up
from os_utilities import IS_MAC
from executable_utilities import IS_BINARY
from threads import UpdatesControllerThread
from versioncontroller import VersionController
class UpdatesController(QObject):
def __init__(self, current_version, owner):
super().__init__()
self._owner = owner
self._download_window = self._owner.download_window
self._current_version = current_version
self.version_controller = VersionController()
self._updates_thread = UpdatesControllerThread(self.version_controller)
self._updates_thread.on_success.connect(self._startup_updates_check)
def start(self):
"""Start the thread."""
if IS_BINARY:
self._updates_thread.start()
@pyqtSlot()
def start_verify_software_version(self):
if not IS_BINARY:
pop_up(
self._owner,
title=Messages.FEATURE_NOT_AVAILABLE,
text=Messages.SCRIPT_NOT_UPDATE
).show()
return
if not self._download_window.isVisible():
self._updates_thread.start()
@pyqtSlot(bool)
def _verify_software_version(self, success):
"""Verify if there is a new software version.
Otherwise notify the user that the software is up to date."""
if not self._download_window.isVisible():
if success:
new_version_found = self._check_new_version()
if not new_version_found:
pop_up(
self._owner,
title=Messages.UP_TO_DATE,
text=Messages.UP_TO_DATE_MSG
).show()
else:
pop_up(
self._owner,
title=Messages.NO_CONNECTION,
text=Messages.NO_CONNECTION_MSG
).show()
@pyqtSlot(bool)
def _startup_updates_check(self, success):
self._updates_thread.on_success.disconnect()
self._updates_thread.on_success.connect(self._verify_software_version)
if success:
if not self._check_new_version():
# Check for a new version of the updater only if Artemis is up to date.
self._check_updater_version()
def _check_new_version(self):
"""Check whether there is a new software version available.
Does something only if the running program is a compiled version."""
if not IS_BINARY:
return None
latest_version = self.version_controller.software.version
if latest_version is None:
return False
if latest_version == self._current_version:
return False
answer = pop_up(
self._owner,
title=Messages.NEW_VERSION_AVAILABLE,
text=Messages.NEW_VERSION_MSG(latest_version),
informative_text=Messages.DOWNLOAD_SUGG_MSG,
is_question=True,
).exec()
if answer == QMessageBox.Yes:
if IS_MAC:
webbrowser.open(self.version_controller.software.url)
else:
updater = QProcess()
command = Constants.UPDATER_SOFTWARE + " " + \
self.version_controller.software.url + \
" " + self.version_controller.software.hash_code + \
" " + str(self.version_controller.software.size)
try:
updater.startDetached(command)
except BaseException:
pass
else:
qApp.quit()
return True
def _check_updater_version(self):
"""Check is a new version of the updater is available.
If so, ask to download the new version.
If the software is not a compiled version, the function is a NOP."""
if not IS_BINARY or IS_MAC:
return
latest_updater_version = self.version_controller.updater.version
try:
with sp.Popen(
[Constants.UPDATER_SOFTWARE, "--version"],
encoding="UTF-8",
stdout=sp.PIPE,
stderr=sp.STDOUT,
stdin=sp.DEVNULL # Needed to avoid OsError: [WinError 6] The handle is invalid.
) as proc:
updater_version = proc.stdout.read().rstrip("\r\n") # Strip any possible newline, to be sure.
except Exception:
updater_version = latest_updater_version
if latest_updater_version is None:
return
if updater_version != latest_updater_version:
answer = pop_up(
self._owner,
title=Messages.UPDATES_AVAILABALE,
text=Messages.UPDATES_MSG,
is_question=True,
).exec()
if answer == QMessageBox.Yes:
self._download_window.activate(
get_download_target(
DownloadTarget.UPDATER,
self.version_controller.updater
)
)

79
src/utilities.py
View File

@@ -1,19 +1,29 @@
from functools import partial
import hashlib
import sys
import os
from PyQt5.QtWidgets import QMessageBox
import urllib3
from constants import Constants, Signal, Database, ChecksumWhat
from constants import Constants, Signal
def resource_path(relative_path):
"""Get absolute path to resource, works for dev and for PyInstaller."""
try:
base_path = sys._MEIPASS
except Exception:
base_path = os.path.abspath(".")
return os.path.join(base_path, relative_path)
class UniqueMessageBox(QMessageBox):
"""Subclass of QMessageBox. Overrides only the exec method.
Only one instance of this class can execute super().exec() exec at a given time.
If another instance is the the exec loop, calling exec simply return None."""
_open_message = False
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def exec(self):
"""Overrides QMessageBox.exec. Call the parent method if there are no
other instances executing exec. Otherwise return None,"""
if UniqueMessageBox._open_message:
return None
UniqueMessageBox._open_message = True
answer = super().exec()
UniqueMessageBox._open_message = False
return answer
def uncheck_and_emit(button):
@@ -46,7 +56,7 @@ def get_field_entries(db_entry, separator=Constants.FIELD_SEPARATOR):
]
def pop_up(cls, title, text,
def pop_up(instance, title, text,
informative_text=None,
connection=None,
is_question=False,
@@ -58,7 +68,7 @@ def pop_up(cls, title, text,
connection -- a callable to connect the message when emitting the finished signal.
is_question -- whether the message contains a question.
default_btn -- the default button for the possible answer to the question."""
msg = QMessageBox(cls)
msg = UniqueMessageBox(instance)
msg.setWindowTitle(title)
msg.setText(text)
if informative_text:
@@ -72,46 +82,15 @@ def pop_up(cls, title, text,
return msg
def is_mac_os():
"""Return True if running OS is Mac."""
return sys.platform == 'darwin'
def checksum_ok(data, reference_hash_code):
"""Check whether the checksum of the 'data' argument is correct.
def get_cacert_file():
"""Return the path to the cacert.pem file."""
if hasattr(sys, "_MEIPASS"):
ca_certs = os.path.join(sys._MEIPASS, 'cacert.pem')
else:
ca_certs = 'cacert.pem'
return ca_certs
def get_pool_manager():
"""Return a urllib3.PoolManager object."""
return urllib3.PoolManager(ca_certs=get_cacert_file())
def checksum_ok(data, what):
"""Check whether the checksum of the 'data' argument is correct."""
Expects a sha256 code as argument."""
if reference_hash_code is None:
raise Exception("ERROR: Invalid hash code.")
code = hashlib.sha256()
code.update(data)
if what is ChecksumWhat.FOLDER:
n = 0
elif what is ChecksumWhat.DB:
n = 1
else:
raise ValueError("Wrong entry name.")
try:
# The downloaded file is a csv file with columns (last version == last line):
# data.zip_SHA256 | db.csv_SHA256 | Version | Creation_date
reference = get_pool_manager().request(
'GET',
Database.LINK_REF,
timeout=4.0
).data.decode("utf-8").splitlines()[-1].split(Database.DELIMITER)[n]
except Exception:
raise
return code.hexdigest() == reference
return code.hexdigest() == reference_hash_code
def connect_events_to_func(events_to_connect, fun_to_connect, fun_args):

111
src/versioncontroller.py Normal file
View File

@@ -0,0 +1,111 @@
from io import BytesIO
import json
from constants import Constants
from os_utilities import get_os
from web_utilities import download_file
"""This module exposes just one class: VersionController.
All the relevant information can be accessed with the dot notation on an instance of such class, e.g.:
version_controller.software.hash_code
is the hash_code of the latest release of the software running on the current OS."""
def _download_versions_file():
"""Download the json file containing all the information
about the latest version of the software. Return a dictionary
containing only the information for the running OS.
Return a dictionary from a json with the following structure:
{
"windows": {
"software": {
"version": "...",
"url": "...",
"hash_code": "...",
"size": ...
},
"updater": {
"version": "...",
"url": "...",
"hash_code": "...",
"size": ...
}
},
"linux": {
"software": {
"version": "...",
"url": "...",
"hash_code": "...",
"size": ...
},
"updater": {
"version": "...",
"url": "...",
"hash_code": "...",
"size": ...
}
},
"mac": {
"software": {
"version": "...",
"url": "...",
"hash_code": "...",
"size": ...
},
"updater": {
"version": "...",
"url": "...",
"hash_code": "...",
"size": ...
}
}
}
"""
try:
version_dict = json.load(
BytesIO(download_file(Constants.VERSION_LINK))
)[get_os()]
except Exception:
return None
else:
return version_dict
class VersionController:
"""Dynamically create attributes corresponding to elements of a dictionary.
Used to get updates information."""
def __init__(self, dct=None):
"""Initialize the dictionary"""
super().__init__()
self._dct = dct
def __getattr__(self, attr):
"""Override super().__getattr__. Dynamically create new attributes
corresponding to elements of the diciotnary."""
if self._dct is None:
if not self.update():
return None
try:
dct_element = self._dct[attr]
except Exception("ERROR: Invalid attribute!"):
return None
else:
if isinstance(dct_element, dict):
setattr(self, attr, type(self)(dct_element))
else:
setattr(self, attr, dct_element)
return getattr(self, attr)
def update(self):
"""Reset the dictionary to the correspondig json file containing
the latest version information. Call this function inside a Qthread."""
dct = _download_versions_file()
if dct is not None:
self._dct = dct
return True
else:
return False

10
src/weatherdata.py
View File

@@ -1,10 +1,12 @@
import re
from PyQt5.QtGui import QPixmap
from PyQt5.QtCore import pyqtSlot, pyqtSignal, QObject
from threads import (BaseDownloadThread,
UpdateSpaceWeatherThread,
ThreadStatus,
UpdateForecastThread)
from threads import (
BaseDownloadThread,
UpdateSpaceWeatherThread,
ThreadStatus,
UpdateForecastThread
)
from constants import Constants
from switchable_label import MultiColorSwitchableLabel
from utilities import safe_cast

51
src/web_utilities.py Normal file
View File

@@ -0,0 +1,51 @@
import os
import sys
import urllib3
from constants import Database
from executable_utilities import IS_BINARY
def get_cacert_file():
"""Return the path to the cacert.pem file."""
if IS_BINARY:
ca_certs = os.path.join(sys._MEIPASS, 'cacert.pem')
else:
ca_certs = 'cacert.pem'
return ca_certs
def get_pool_manager():
"""Return a urllib3.PoolManager object."""
return urllib3.PoolManager(ca_certs=get_cacert_file())
def download_file(url, encoding=""):
resp = get_pool_manager().request(
'GET',
url,
preload_content=True,
timeout=4.0
).data
if encoding:
return resp.decode(encoding)
return resp
def _download_multiline_file_as_list(url=Database.LINK_REF):
"""Download a text file and return the last line as a list.
The downloaded file is a csv file with columns (last version == last line):
data.zip_SHA256 | db.csv_SHA256 | Version | Creation_date"""
try:
f = download_file(url, encoding="UTF-8").splitlines()[-1].split(Database.DELIMITER)
except Exception:
return None
return f
def get_folder_hash_code():
return _download_multiline_file_as_list()[0]
def get_db_hash_code():
return _download_multiline_file_as_list()[1]