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

Artemis 4 RC1

Browse Source
This commit is contained in:
Marco Dalla Tiezza
2024-05-28 22:40:45 +02:00
parent acc44c93b3
commit 528c816508
254 changed files with 14757 additions and 30137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5848
artemis/resources.py Normal file
View File

File diff suppressed because it is too large Load Diff

354
artemis/ui/artemis.py Normal file
View File

@@ -0,0 +1,354 @@
import uuid
from PySide6.QtQml import QQmlApplicationEngine
from PySide6.QtCore import QObject, Slot, Signal
from artemis.utils.constants import Constants, Messages
from artemis.utils.sys_utils import open_directory, pack_db, unpack_db
from artemis.utils.sql_utils import ArtemisDatabase, ArtemisSignal
from artemis.utils.path_utils import check_data_dir
from artemis.utils.network_utils import NetworkManager
from artemis.utils.generic_utils import generate_filter_query
from artemis.utils.path_utils import normalize_dialog_path
from artemis.ui.preferences import UIPreferences
from artemis.ui.dbmanager import UIdbmanager
from artemis.ui.signaleditor import UIsignaleditor
from artemis.ui.downloader import UIDownloader
from artemis.ui.spaceweather import UIspaceweather
from artemis.ui.documentsmanager import UIdocumentsmanager
from artemis.ui.categoryeditor import UIcategoryeditor
import artemis.resources
class UIArtemis(QObject):
# Python > QML Signals
populate_sig_list = Signal(list)
populate_sig_details = Signal(list)
populate_filter_modulation = Signal(list)
clear_list = Signal()
clear_signal_page = Signal()
clear_filter_page = Signal()
lock_audio_player = Signal()
lock_menu = Signal(bool)
show_dialog_popup = Signal(str, str, str)
show_dialog_download_db = Signal(str, str, str)
show_dialog_download_art = Signal(str, str, str)
update_info_bar = Signal(str, str)
def __init__(self):
super().__init__()
# Main UI initialization
self._engine = QQmlApplicationEngine()
self._engine.rootContext().setContextProperty('APPLICATION_VERSION', Constants.APPLICATION_VERSION)
self._engine.rootContext().setContextProperty('PYTHON_VERSION', Constants.PYTHON_VERSION)
self._engine.rootContext().setContextProperty('QT_VERSION', Constants.QT_VERSION)
self._engine.load('qrc:/ui/Artemis.qml')
self._window = self._engine.rootObjects()[0]
self._window_filter = self._window.findChild(QObject, "filterPageObj")
self._window_signal = self._window.findChild(QObject, "signalPageObj")
self.loaded_db = None
self._connect()
# Creating istances for other windows
self.preferences = UIPreferences(self)
self.dbmanager = UIdbmanager(self)
self.downloader = UIDownloader(self)
self.spaceweather = UIspaceweather(self)
self.docmanager = UIdocumentsmanager(self)
self.sigeditor = UIsignaleditor(self)
self.cateditor = UIcategoryeditor(self)
self.network_manager = NetworkManager(self)
check_data_dir()
def _connect(self):
# QML > Python connections
self._window.showDBmanager.connect(self.show_dbmanager_ui)
self._window.loadSignal.connect(self.load_sig)
self._window.showPref.connect(self.show_pref_ui)
self._window.openSigEditor.connect(self.open_sig_editor)
self._window.startDownloader.connect(self.start_download_db)
self._window.checkDbUpdates.connect(self.check_update_db)
self._window.showSpaceWeather.connect(self.show_space_weather_ui)
self._window.openDbDirectory.connect(self.open_db_directory)
self._window.showCatManager.connect(self.open_cat_manager)
self._window.newDb.connect(self.new_db)
self._window.exportDb.connect(self.export_db)
self._window.importDb.connect(self.import_db)
self._window_filter.applyFilter.connect(self.apply_filter)
self._window_filter.sendBottomAlert.connect(self.bottom_info_bar)
self._window_signal.openDocManager.connect(self.show_documentsmanager_ui)
self._window_signal.openSigEditor.connect(self.open_sig_editor)
self._window_signal.deleteCatTag.connect(self.delete_cat_tag)
self._window_signal.addCatTag.connect(self.add_cat_tag)
# Python > QML connections
self.populate_sig_list.connect(self._window.populateList)
self.clear_list.connect(self._window.clearList)
self.update_info_bar.connect(self._window.bottomInfoBar)
self.show_dialog_popup.connect(self._window.openGeneralDialog)
self.show_dialog_download_db.connect(self._window.openDialogDownloadDb)
self.show_dialog_download_art.connect(self._window.openDialogDownloadArtemis)
self.lock_menu.connect(self._window.lockMenu)
self.populate_sig_details.connect(self._window_signal.populateSignalParam)
self.lock_audio_player.connect(self._window_signal.lockPlayer)
self.clear_signal_page.connect(self._window_signal.resetAll)
self.clear_filter_page.connect(self._window_filter.resetAll)
self.populate_filter_modulation.connect(self._window_filter.loadLists)
def load_db(self, db_dir_name):
""" Load the DB and populate the signals list
Args:
db_dir_name (str): folder name in the data folder
"""
# Loading DB
self.loaded_db = ArtemisDatabase(db_dir_name)
self.loaded_db.load()
# Clearing UI
self.lock_menu.emit(False)
self.clear_signal_page.emit()
self.clear_filter_page.emit()
# Populating UI
self.load_filter_lists()
self.populate_sig_list.emit(self.loaded_db.all_signals)
# Updating status bar
total_signals = len(self.loaded_db.all_signals)
self.bottom_info_bar("Database loaded with {} signals".format(total_signals), "info")
@Slot(int)
def load_sig(self, sig_id):
""" Load the selected signal and populate the SignalPage
Args:
sig_id (int): SIG_ID of the signal to be loaded
"""
self.loaded_sig = ArtemisSignal(self.loaded_db)
self.loaded_sig.load(sig_id)
sig_dic = self.loaded_sig.generate_dic()
self.populate_sig_details.emit([sig_dic])
def load_filter_lists(self):
""" Populates the 3 listviews in the FilterPage
"""
self.populate_filter_modulation.emit([{
'modulation': self.loaded_db.all_modulation,
'location': self.loaded_db.all_location,
'category': self.loaded_db.all_category_labels
}])
@Slot(dict)
def apply_filter(self, filter_status):
""" Update the signal list according to the selected filters in the FilterPage.
Args:
filter_status (dic): dictionary containing the active filters with all
the details to generate a search query
"""
filter_status = filter_status.toVariant()
if filter_status != {}:
filter_query = generate_filter_query(filter_status)
self.loaded_db.select_by_filter(filter_query)
self.clear_signal_page.emit()
self.populate_sig_list.emit(self.loaded_db.all_signals)
total_signals = len(self.loaded_db.all_signals)
self.bottom_info_bar("FILTERS ACTIVE: {} signals found".format(total_signals), "warning")
else:
self.load_db(self.loaded_db.db_dir_name)
def show_pref_ui(self):
""" Load the preference windows
"""
self.preferences.load_preferences_ui()
def show_dbmanager_ui(self):
""" Load the DB manager windows
"""
self.dbmanager.load_dbmanager_ui()
@Slot(str, list, bool)
def open_sig_editor(self, type, sig_param, is_new):
""" Open the signal editor windows
Called when the user want to add, edit or delete the signal or its parametes.
"""
self.sigeditor.load_signaleditor_ui(type, sig_param, is_new)
def show_space_weather_ui(self):
""" Open the space weather windows
"""
self.spaceweather.load_spaceweather_ui()
def show_documentsmanager_ui(self):
""" Open the documents manager windows
"""
self.docmanager.load_documentsmanager_ui()
def check_update_db(self):
""" User manual check for updates db updates
"""
self.network_manager.show_popup = True
self.network_manager.check_updates()
def start_download_db(self):
""" Show the downloader and start the download of the sigid db
"""
self.downloader.show_ui.emit()
self.downloader.on_start()
def dialog_download_db(self, message_type, title, message):
""" Dialog popup for DB download confirmation
"""
self.show_dialog_download_db.emit(message_type, title, message)
def dialog_download_artemis(self, message_type, title, message):
""" Dialog popup for artemis download confirmation
"""
self.show_dialog_download_art.emit(message_type, title, message)
def open_db_directory(self):
""" Open the local folder of the loaded DB
"""
open_directory(self.loaded_db.db_dir)
@Slot(str)
def new_db(self, name):
""" Create a new local DB
Args:
name (str): name of the new DB, hardcoded in sql info table
"""
try:
new_db = ArtemisDatabase(str(uuid.uuid4()))
new_db.create(name)
self.load_db(new_db.db_dir_name)
self.dialog_popup(
Messages.DIALOG_TYPE_INFO,
Messages.GENERIC_SUCCESS,
Messages.DB_CREATION_SUCCESS_MSG
)
except Exception as e:
self.dialog_popup(
Messages.DIALOG_TYPE_ERROR,
Messages.GENERIC_ERROR,
Messages.GENERIC_ERROR_MSG.format(e)
)
@Slot(str)
def export_db(self, save_path):
""" Export the load DB in a tar file. Does not use compression
Args:
save_path (str): destination path of the generated .tar file
"""
try:
dest_path = normalize_dialog_path(save_path)
pack_db(dest_path, self.loaded_db.db_dir)
self.dialog_popup(
Messages.DIALOG_TYPE_INFO,
Messages.GENERIC_SUCCESS,
Messages.EXPORTING_SUCCESS_MSG
)
except Exception as e:
self.dialog_popup(
Messages.DIALOG_TYPE_ERROR,
Messages.GENERIC_ERROR,
Messages.GENERIC_ERROR_MSG.format(e)
)
@Slot(str)
def import_db(self, tar_path):
""" Import a new DB in the Artemis data folder
Args:
tar_path (str): Path of the archive to be imported
"""
try:
origin_path = normalize_dialog_path(tar_path)
unpack_db(origin_path, str(uuid.uuid4()))
self.dialog_popup(
Messages.DIALOG_TYPE_INFO,
Messages.GENERIC_SUCCESS,
Messages.IMPORTING_SUCCESS_MSG
)
except Exception as e:
self.dialog_popup(
Messages.DIALOG_TYPE_ERROR,
Messages.GENERIC_ERROR,
Messages.GENERIC_ERROR_MSG.format(e)
)
@Slot(int)
def add_cat_tag(self, clb_id):
self.loaded_sig.insert_category(clb_id)
self.load_db(self.loaded_db.db_dir_name)
@Slot(int)
def delete_cat_tag(self, cat_id):
self.loaded_sig.delete_category(cat_id)
self.load_db(self.loaded_db.db_dir_name)
def open_cat_manager(self):
""" Open the category manager windows
"""
self.cateditor.load_cateditor_ui()
def dialog_popup(self, message_type, title, message):
""" Opens a general dialog popup
Args:
message_type (str): 'info', 'question', 'warn', 'error'
title (str): header of the dialoog
message (sstr): description inside the dialog
"""
self.show_dialog_popup.emit(message_type, title, message)
@Slot(str, str)
def bottom_info_bar(self, message, message_type):
""" Manage the footer info bar
Args:
message (str): text to be shown in the info bar
message_type (str): 'info', 'warning'
"""
self.update_info_bar.emit(message, message_type)

67
artemis/ui/categoryeditor.py Normal file
View File

@@ -0,0 +1,67 @@
from PySide6.QtQml import QQmlApplicationEngine
from PySide6.QtCore import QObject, Signal, Slot
from artemis.utils.path_utils import *
from artemis.utils.generic_utils import *
class UIcategoryeditor(QObject):
# Python > QML Signals
show_ui = Signal()
load = Signal(list)
def __init__(self, parent):
super().__init__()
self._parent = parent
self._engine = QQmlApplicationEngine()
self._engine.load('qrc:/ui/CategoryEditor.qml')
self._window = self._engine.rootObjects()[0]
self._connect()
def _connect(self):
# QML > Python connections
self._window.saveParam.connect(self.save)
self._window.deleteParam.connect(self.delete)
# Python > QML connections
self.show_ui.connect(self._window.show)
self.load.connect(self._window.loadList)
def load_cateditor_ui(self):
""" Load the list with existing category tags and show the UI
"""
all_cat = self._parent.loaded_db.all_category_labels
self.load.emit(all_cat)
self.show_ui.emit()
@Slot(list, bool)
def save(self, data, is_new):
""" Save new category tag or update the existing ones.
"""
data = data.toVariant()
if is_new:
self._parent.loaded_db.insert_category_label(data[0])
else:
self._parent.loaded_db.update_category_label(data[1], data[0])
self._parent.load_db(self._parent.loaded_db.db_dir_name)
self.load_cateditor_ui()
@Slot(int)
def delete(self, clb_id):
""" Delete a database category tag.
All the entries in the documents table are automatically beign deleted due to
foreign-key cascade propagation
"""
self._parent.loaded_db.delete_category_label(clb_id)
self._parent.load_db(self._parent.loaded_db.db_dir_name)
self.load_cateditor_ui()

105
artemis/ui/dbmanager.py Normal file
View File

@@ -0,0 +1,105 @@
from PySide6.QtQml import QQmlApplicationEngine
from PySide6.QtCore import QObject, Signal, Slot
from artemis.utils.path_utils import *
from artemis.utils.generic_utils import *
from artemis.utils.sql_utils import ArtemisDatabase
from artemis.utils.constants import Constants
class UIdbmanager(QObject):
# Python > QML Signals
show_ui = Signal()
close_ui = Signal()
populate_db_list = Signal(list)
def __init__(self, parent):
super().__init__()
self._parent = parent
self._engine = QQmlApplicationEngine()
self._engine.load('qrc:/ui/DbManager.qml')
self._window = self._engine.rootObjects()[0]
self._connect()
def _connect(self):
# QML > Python connections
self._window.loadDB.connect(self.load_db)
self._window.deleteDB.connect(self.delete_db)
self._window.renameDB.connect(self.rename_db)
# Python > QML connections
self.show_ui.connect(self._window.show)
self.close_ui.connect(self._window.close)
self.populate_db_list.connect(self._window.loadList)
def load_dbmanager_ui(self):
self.load_local_db_list()
self.show_ui.emit()
def load_local_db_list(self):
""" Scan for all the valid DBs in the data folder and show them on the list
"""
valid_db_list = self.scan_db_dir()
self.populate_db_list.emit(valid_db_list)
def load_db(self, db_dir_name):
""" Load the selected DB (from the DB Manager list) in the main artemis window
"""
self._parent.load_db(db_dir_name)
self.close_ui.emit()
@Slot(str)
def delete_db(self, db_dir_name):
""" Delete the DB folder.
Clear the main UI if the database to be deleted is the selected one
"""
if self._parent.loaded_db is not None:
if self._parent.loaded_db.db_dir_name == db_dir_name:
self._parent.lock_menu.emit(True)
self._parent.clear_list.emit()
self._parent.clear_signal_page.emit()
delete_db_dir(db_dir_name)
self.load_local_db_list()
@Slot(str, str)
def rename_db(self, db_dir_name, new_name):
""" Rename db in the data folder
"""
database = ArtemisDatabase(db_dir_name)
database.rename(new_name)
self.load_local_db_list()
def scan_db_dir(self):
""" Scans the data directory for valid databases and
return a dictionary containing only the valid ones with a summary
"""
valid_db_list = []
db_dirs = next(os.walk(Constants.DB_DIR))[1]
for db_dir_name in db_dirs:
if valid_db(db_dir_name):
database = ArtemisDatabase(db_dir_name)
database.load()
valid_db_list.append(
{
'name': database.name,
'db_dir_name': database.db_dir_name,
'documents_n': database.stats['documents'],
'signals_n': database.stats['signals'],
'images_n': database.stats['images'],
'audio_n': database.stats['audio']
}
)
return valid_db_list

128
artemis/ui/documentsmanager.py Normal file
View File

@@ -0,0 +1,128 @@
from PySide6.QtQml import QQmlApplicationEngine
from PySide6.QtCore import QObject, Signal, Slot
from artemis.utils.path_utils import *
from artemis.utils.generic_utils import *
from artemis.utils.sys_utils import *
class UIdocumentsmanager(QObject):
# Python > QML Signals
show_ui = Signal()
close_ui = Signal()
populate_documents_list = Signal(list)
def __init__(self, parent):
super().__init__()
self._parent = parent
self._engine = QQmlApplicationEngine()
self._engine.load('qrc:/ui/DocumentsManager.qml')
self._window = self._engine.rootObjects()[0]
self._connect()
def _connect(self):
# QML > Python connections
self._window.saveNewDoc.connect(self.save_new_doc)
self._window.deleteDoc.connect(self.delete_doc)
self._window.updateDoc.connect(self.update_doc)
self._window.openDoc.connect(self.open_doc)
# Python > QML connections
self.show_ui.connect(self._window.show)
self.close_ui.connect(self._window.close)
self.populate_documents_list.connect(self._window.loadList)
def load_documentsmanager_ui(self):
self.load_documents_list()
self.show_ui.emit()
def load_documents_list(self):
""" Load the documents of the selected signal and populate the documents list
"""
self._parent.loaded_sig.select_documents()
all_documents = self._parent.loaded_sig.documents
keys = (
'doc_id',
'extension',
'name',
'description',
'type',
'preview'
)
doc_lst = [dict(zip(keys, values)) for values in all_documents]
self.populate_documents_list.emit(doc_lst)
@Slot(list)
def save_new_doc(self, doc_lst):
""" Save the new document (identified by the DOC_ID = -1) and reload the document list.
doc_param contains all the details of the new documents.
"""
doc_param = doc_lst.toVariant()
file_extension = os.path.splitext(doc_param[0])[1][1:]
doc_id = self._parent.loaded_sig.insert_document([
-1,
file_extension,
doc_param[1],
doc_param[2],
doc_param[3],
0
])
local_file_name = '{}.{}'.format(str(doc_id), file_extension)
origin_path = normalize_dialog_path(doc_param[0])
copy_file(origin_path, self._parent.loaded_db.media_dir / local_file_name)
self.load_documents_list()
@Slot(list)
def update_doc(self, doc_lst):
""" Update the details of the existent document
"""
doc_list = doc_lst.toVariant()
for doc in doc_list:
self._parent.loaded_sig.update_documents(doc[0], doc[1], doc[2], doc[3], doc[4])
self.load_documents_list()
@Slot(str, str)
def open_doc(self, doc_id, extension):
""" Open the selected document with the proper system application (if any)
"""
try:
open_file(self._parent.loaded_db.media_dir / '{}.{}'.format(doc_id, extension))
except Exception as e:
self.close_ui.emit()
self._parent.dialog_popup(
Messages.DIALOG_TYPE_ERROR,
Messages.GENERIC_ERROR,
str(e)
)
@Slot(str, str, str, bool)
def delete_doc(self, doc_id, doc_extension, doc_type, doc_preview):
""" Delete the selected document
"""
doc_file_name = '{}.{}'.format(doc_id, doc_extension)
doc_file_path = self._parent.loaded_db.media_dir / doc_file_name
self._parent.loaded_sig.delete_document(doc_id)
if doc_preview:
if doc_type == 'Audio':
self._parent.lock_audio_player.emit()
delete_file(doc_file_path)
self.load_documents_list()

133
artemis/ui/downloader.py Normal file
View File

@@ -0,0 +1,133 @@
from PySide6.QtQml import QQmlApplicationEngine
from PySide6.QtCore import QObject, Slot, Signal, QUrl, QSaveFile, QDir, QIODevice
from PySide6.QtNetwork import QNetworkReply, QNetworkRequest, QNetworkAccessManager
from artemis.utils.config_utils import *
from artemis.utils.sys_utils import delete_file, match_hash, unpack_db
from artemis.utils.constants import Messages
from artemis.utils.sys_utils import delete_db_dir
class UIDownloader(QObject):
# Python > QML Signals
show_ui = Signal()
close_ui = Signal()
def __init__(self, parent):
super().__init__()
self._parent = parent
self._engine = QQmlApplicationEngine()
self._engine.load('qrc:/ui/Downloader.qml')
self._window = self._engine.rootObjects()[0]
self._progress_bar = self._window.findChild(QObject, "progressBar")
self._label_progress = self._window.findChild(QObject, "labelProgress")
self._connect()
def _connect(self):
# QML > Python connections
self._window.onAbort.connect(self.on_abort)
# Python > QML connections
self.show_ui.connect(self._window.show)
self.close_ui.connect(self._window.close)
@Slot()
def on_start(self):
""" Start the download of the DB taking the needed url and size from
the attributes of the UpdatesController class
"""
url_file = QUrl(self._parent.network_manager.remote_db_url)
dest_path = QDir(Constants.DB_DIR)
self.dest_file = dest_path.filePath(url_file.fileName())
self.file = QSaveFile(self.dest_file)
if self.file.open(QIODevice.WriteOnly):
# Start a GET HTTP request
self.manager = QNetworkAccessManager(self)
self.reply = self.manager.get(QNetworkRequest(url_file))
self.reply.downloadProgress.connect(self.on_progress)
self.reply.finished.connect(self.on_finished)
self.reply.readyRead.connect(self.on_ready_read)
self.reply.errorOccurred.connect(self.on_error)
else:
self.close_ui.emit()
self.show_popup_error(
self.file.errorString()
)
@Slot()
def on_abort(self):
""" Stop the download when user press abort button """
if self.reply:
self.reply.abort()
self._progress_bar.setProperty("value", 0)
if self.file:
self.file.cancelWriting()
self.close_ui.emit()
@Slot()
def on_ready_read(self):
""" Get available bytes and store them into the file """
if self.reply:
if self.reply.error() == QNetworkReply.NoError:
self.file.write(self.reply.readAll())
@Slot()
def on_finished(self):
""" Delete reply, close the file, check the hash for integrity,
extract the database and delete the downloaded zip
"""
if self.reply:
self.reply.deleteLater()
if self.file:
self.file.commit()
self._label_progress.setProperty("text", "Checking DB integrity (SHA-256)")
if match_hash(self.dest_file, self._parent.network_manager.remote_db_hash):
self._label_progress.setProperty("text", "Unpacking archive...")
delete_db_dir('SigID')
unpack_db(self.dest_file, 'SigID')
delete_file(self.dest_file)
self._parent.load_db('SigID')
self.close_ui.emit()
@Slot(int, int)
def on_progress(self, bytesReceived: int):
""" Update progress bar and label
"""
total_bytes = self._parent.network_manager.remote_db_size
self._label_progress.setProperty("text", "{:.1f} Mb / {:.1f} Mb".format(bytesReceived/10**6, total_bytes/10**6))
self._progress_bar.setProperty("to", total_bytes)
self._progress_bar.setProperty("value", bytesReceived)
@Slot(QNetworkReply.NetworkError)
def on_error(self, code: QNetworkReply.NetworkError):
""" Show a message if an error happen during download
"""
if self.reply:
self.close_ui.emit()
self.show_popup_error(
self.reply.errorString()
)
def show_popup_error(self, error_msg):
self._parent.dialog_popup(
Messages.DIALOG_TYPE_ERROR,
Messages.GENERIC_ERROR,
Messages.GENERIC_ERROR_MSG.format(error_msg)
)

56
artemis/ui/preferences.py Normal file
View File

@@ -0,0 +1,56 @@
from PySide6.QtQml import QQmlApplicationEngine
from PySide6.QtCore import QObject, Slot, Signal
from artemis.utils.config_utils import *
class UIPreferences(QObject):
# Python > QML Signals
show_ui = Signal()
load_material_accent = Signal(str)
load_material_theme = Signal(str)
def __init__(self, parent):
super().__init__()
self._parent = parent
self._engine = QQmlApplicationEngine()
self._engine.load('qrc:/ui/Preferences.qml')
self._window = self._engine.rootObjects()[0]
self._connect()
def _connect(self):
# QML > Python connections
self._window.saveMaterialAccent.connect(self.save_material_accent)
self._window.saveMaterialTheme.connect(self.save_material_theme)
# Python > QML connections
self.show_ui.connect(self._window.show)
self.load_material_accent.connect(self._window.loadMaterialAccent)
self.load_material_theme.connect(self._window.loadMaterialTheme)
def load_preferences_ui(self):
""" Loading all the initial preferences from the conf file to the UI
"""
self.load_material_accent.emit(CONFIGURE_QT.get_or_default("Material", "Accent", "Green"))
self.load_material_theme.emit(CONFIGURE_QT.get_or_default("Material", "Theme", "System"))
self.show_ui.emit()
@Slot(str)
def save_material_accent(self, material_accent):
""" Saving material accent setting
"""
CONFIGURE_QT.set("Material", "Accent", material_accent)
@Slot(str)
def save_material_theme(self, material_theme):
""" Saving material theme setting
"""
CONFIGURE_QT.set("Material", "Theme", material_theme)

124
artemis/ui/signaleditor.py Normal file
View File

@@ -0,0 +1,124 @@
from PySide6.QtQml import QQmlApplicationEngine
from PySide6.QtCore import QObject, Signal, Slot
from artemis.utils.path_utils import *
from artemis.utils.generic_utils import *
from artemis.utils.sql_utils import ArtemisSignal
class UIsignaleditor(QObject):
# Python > QML Signals
show_ui = Signal()
load = Signal(str, list, bool)
def __init__(self, parent):
super().__init__()
self._parent = parent
self._engine = QQmlApplicationEngine()
self._engine.load('qrc:/ui/SignalEditor.qml')
self._window = self._engine.rootObjects()[0]
self._connect()
def _connect(self):
# QML > Python connections
self._window.saveParam.connect(self.save)
self._window.deleteParam.connect(self.delete)
# Python > QML connections
self.show_ui.connect(self._window.show)
self.load.connect(self._window.load)
def load_signaleditor_ui(self, param_type, sig_param, is_new):
""" Load all the details of the selected signal
Args:
param_type (str): Signal, Frequency, Bandwidth, Modulation, Mode,
ACF, Location
sig_param (list): a list formed as [id, value, description]
is_new (bool): If true, the windows open in an empty state ready to
be compiled by the user. If false, the windows will open all the
current parameter for the loaded signal, for editing or deleting purposes.
"""
if param_type == 'Signal' and not is_new:
sig_param = [
self._parent.loaded_sig.sig_id,
self._parent.loaded_sig.name,
self._parent.loaded_sig.description
]
self.load.emit(param_type, sig_param, is_new)
self.show_ui.emit()
@Slot(str, list, bool)
def save(self, param_type, data, is_new):
""" Save new signal parameters or update the existing ones.
"""
data = data.toVariant()
if is_new:
if param_type == 'Signal':
self._parent.loaded_sig = ArtemisSignal(self._parent.loaded_db)
self._parent.loaded_sig.insert_signal(data[1], data[2])
elif param_type == 'Frequency':
self._parent.loaded_sig.insert_frequency(int(data[1]), data[2])
elif param_type == 'Bandwidth':
self._parent.loaded_sig.insert_bandwidth(int(data[1]), data[2])
elif param_type == 'Modulation':
self._parent.loaded_sig.insert_modulation(data[1], data[2])
elif param_type == 'Mode':
self._parent.loaded_sig.insert_mode(data[1], data[2])
elif param_type == 'ACF':
self._parent.loaded_sig.insert_acf(data[1], data[2])
elif param_type == 'Location':
self._parent.loaded_sig.insert_location(data[1], data[2])
else:
if param_type == 'Signal':
self._parent.loaded_sig.update_signal(data[0], data[1], data[2])
elif param_type == 'Frequency':
self._parent.loaded_sig.update_frequency(data[0], int(data[1]), data[2])
elif param_type == 'Bandwidth':
self._parent.loaded_sig.update_bandwidth(data[0], int(data[1]), data[2])
elif param_type == 'Modulation':
self._parent.loaded_sig.update_modulation(data[0], data[1], data[2])
elif param_type == 'Mode':
self._parent.loaded_sig.update_mode(data[0], data[1], data[2])
elif param_type == 'ACF':
self._parent.loaded_sig.update_acf(data[0], data[1], data[2])
elif param_type == 'Location':
self._parent.loaded_sig.update_location(data[0], data[1], data[2])
self._parent.load_db(self._parent.loaded_db.db_dir_name)
@Slot(str, int)
def delete(self, param_type, id):
""" Delete a signal parameter or the signal itself (with all the parameters and documents).
All the entries in the documents table are automatically beign deleted due to
foreign-key cascade propagation
"""
if param_type == 'Signal':
self._parent.loaded_sig.delete_signal()
for doc in self._parent.loaded_sig.documents:
doc_file_name = '{}.{}'.format(str(doc[0]), doc[1])
doc_file_path = self._parent.loaded_db.media_dir / doc_file_name
delete_file(doc_file_path)
elif param_type == 'Frequency':
self._parent.loaded_sig.delete_frequency(id)
elif param_type == 'Bandwidth':
self._parent.loaded_sig.delete_bandwidth(id)
elif param_type == 'Modulation':
self._parent.loaded_sig.delete_modulation(id)
elif param_type == 'Mode':
self._parent.loaded_sig.delete_mode(id)
elif param_type == 'ACF':
self._parent.loaded_sig.delete_acf(id)
elif param_type == 'Location':
self._parent.loaded_sig.delete_location(id)
self._parent.load_db(self._parent.loaded_db.db_dir_name)

65
artemis/ui/spaceweather.py Normal file
View File

@@ -0,0 +1,65 @@
from PySide6.QtQml import QQmlApplicationEngine
from PySide6.QtCore import QObject, Signal
from artemis.utils.path_utils import *
from artemis.utils.generic_utils import *
from artemis.utils.constants import Constants
class UIspaceweather(QObject):
# Python > QML Signals
show_ui = Signal()
load_poseidon_report = Signal(dict)
load_poseidon_forecast_report = Signal(dict)
update_bottom_bar = Signal(str)
def __init__(self, parent):
super().__init__()
self._parent = parent
self._engine = QQmlApplicationEngine()
self._engine.load('qrc:/ui/SpaceWeather.qml')
self._window = self._engine.rootObjects()[0]
self._window_current = self._window.findChild(QObject, "spaceWeatherCurrentObj")
self._window_forecast = self._window.findChild(QObject, "spaceWeatherForecastObj")
self._connect()
def _connect(self):
# QML > Python connections
# Python > QML connections
self.show_ui.connect(self._window.show)
self.update_bottom_bar.connect(self._window.updateBottomBar)
self.load_poseidon_report.connect(self._window_current.loadReport)
self.load_poseidon_forecast_report.connect(self._window_forecast.loadForecastReport)
def load_spaceweather_ui(self):
""" Before opening the windows, poseidon report (data.json) is read online
"""
self.download_poseidon_report()
def download_poseidon_report(self):
network_manager = self._parent.network_manager
network_manager.show_popup = True
poseidon_data = network_manager.fetch_remote_json(
Constants.POSEIDON_REPORT
)
if poseidon_data:
self.load_poseidon_report.emit(poseidon_data)
self.load_poseidon_forecast_report.emit(poseidon_data)
self.update_bottom_bar.emit(
'Loaded Poseidon report issued on {} at {} UTC'.format(
poseidon_data['JSON_INFO']['utc_date'],
poseidon_data['JSON_INFO']['utc_time']
)
)
self.show_ui.emit()

32
artemis/utils/config_utils.py Normal file
View File

@@ -0,0 +1,32 @@
from configparser import ConfigParser
from artemis.utils.constants import Constants
class Config(ConfigParser):
""" Custom configuration class derived from ConfigParser.
Used to get, set, save and remove any configuration from the conf file
"""
def __init__(self, config_file_path, space_around_delimiters=False):
super().__init__()
self._config_file_path = config_file_path
self.read(self._config_file_path)
self._space_around_delimiters = space_around_delimiters
def get_or_default(self, section, option, default_value):
value = super().get(section, option)
return value if value else default_value
def set(self, section, option, value=None):
super().set(section, option, value)
self.save()
def remove(self, section, option):
super().remove_option(section, option)
self.save()
def save(self):
with open(self._config_file_path, 'w') as f:
self.write(f, space_around_delimiters=self._space_around_delimiters)
CONFIGURE_QT = Config((Constants.PREFERENCES_DIR / 'qtquickcontrols2.conf').resolve().as_posix())

499
artemis/utils/constants.py Normal file
View File

@@ -0,0 +1,499 @@
import os
import locale
import sys
from PySide6.QtCore import qVersion
from pathlib import Path
class Constants():
""" Container class for several constants of the software """
APPLICATION_NAME = 'Artemis'
ORGANIZATION_NAME = 'AresValley'
ORGANIZATION_DOMAIN = 'aresvalley.com'
APPLICATION_VERSION = '4.0.0'
BASE_DIR = Path(os.path.dirname(__file__)) / '../..'
PREFERENCES_DIR = BASE_DIR / 'config'
DB_DIR = BASE_DIR / 'data'
UI_DIR = BASE_DIR / 'ui'
IMAGES_DIR = BASE_DIR / 'images'
LOGS_DIR = BASE_DIR / 'logs'
SQL_NAME = 'data.sqlite'
DB_LATEST_VERSION = 'https://www.aresvalley.com/artemis/v4/latest.json'
POSEIDON_REPORT = 'https://www.aresvalley.com/poseidon_engine/data.json'
DEFAULT_ENCODING = 'utf-8'
SYSTEM_LANGUAGE = 'en_US' # locale.getdefaultlocale()[0]
PYTHON_VERSION = '.'.join(str(v) for v in sys.version_info[:3])
QT_VERSION = qVersion()
class Messages:
""" Container class for messages to be displayed """
# Type
DIALOG_TYPE_INFO = 'info'
DIALOG_TYPE_QUEST = 'question'
DIALOG_TYPE_WARN = 'warn'
DIALOG_TYPE_ERROR = 'error'
# Titles
GENERIC_SUCCESS = "Success!"
GENERIC_ERROR = "Something went wrong!"
NO_DB_DETECTED = "No SigID database detected..."
NO_CONNECTION = "Connection Error!"
UP_TO_DATE = "You're up to date!"
DB_NEW_VER = "New SigID DB version available!"
ART_NEW_VER = "New Artemis version available!"
# Messages
DB_CREATION_SUCCESS_MSG = "The new database has been created succesfully."
GENERIC_ERROR_MSG = "An error occurred during the process. Details: {}"
IMPORTING_SUCCESS_MSG = "Database importing has been succesfully completed!"
EXPORTING_SUCCESS_MSG = "Database exporting has been succesfully completed!"
FILE_NOT_FOUND_ERR_MSG = "The file you are trying to access cannot be located. This may be because the file has been moved or deleted."
NO_DB_DETECTED_MSG = "Do you want to download it now?"
NO_CONNECTION_MSG = "Unable to check for updates. It appears that there is a problem with your internet connection. Please check your network settings and try again later. {}"
UP_TO_DATE_MSG = "The latest version of Artemis and SigID wiki is installed on your computer."
DB_NEW_VER_MSG = "A new version of the database ({}) is available for download. Download now?"
ART_NEW_VER_MSG = "A new version of Artemis ({}) is available for download. Check GitHub page now?"
DOWNLOAD_CORRUPTED_MSG = "Downloaded data corrupted or invalid. Please retry."
class Query():
""" Container class for all the sqlite queries """
############################## SELECT
SELECT_ALL_SIGNALS = "SELECT SIG_ID, NAME FROM signals ORDER BY NAME ASC"
SELECT_ALL_MODULATION = "SELECT DISTINCT VALUE FROM modulation ORDER BY VALUE ASC"
SELECT_ALL_LOCATION = "SELECT DISTINCT VALUE FROM location ORDER BY VALUE ASC"
SELECT_SIG_ID = "SELECT SIG_ID, NAME FROM signals WHERE SIG_ID IN ({}) ORDER BY NAME ASC"
SELECT_ALL_CAT_LABELS = "SELECT CLB_ID, VALUE FROM category_label ORDER BY VALUE ASC"
SELECT_INFO = """
SELECT
NAME,
DATE,
VERSION,
EDITABLE
FROM info
"""
SELECT_SIGNAL = """
SELECT
NAME,
DESCRIPTION,
URL
FROM signals WHERE SIG_ID = ?
"""
SELECT_CATEGORY = """
SELECT
category.CAT_ID,
category_label.VALUE
FROM category
INNER JOIN category_label ON category.CLB_ID = category_label.CLB_ID
WHERE SIG_ID = ?
"""
SELECT_DOCUMENTS = """
SELECT
DOC_ID,
EXTENSION,
NAME,
DESCRIPTION,
TYPE,
PREVIEW
FROM documents WHERE SIG_ID = ?
ORDER BY TYPE ASC
"""
SELECT_FREQUENCY = """
SELECT
FREQ_ID,
VALUE,
DESCRIPTION
FROM frequency WHERE SIG_ID = ?
"""
SELECT_BANDWIDTH = """
SELECT
BAND_ID,
VALUE,
DESCRIPTION
FROM bandwidth WHERE SIG_ID = ?
"""
SELECT_MODULATION = """
SELECT
MDL_ID,
VALUE,
DESCRIPTION
FROM modulation WHERE SIG_ID = ?
"""
SELECT_MODE= """
SELECT
MOD_ID,
VALUE,
DESCRIPTION
FROM mode WHERE SIG_ID = ?
"""
SELECT_LOCATION = """
SELECT
LOC_ID,
VALUE,
DESCRIPTION
FROM location WHERE SIG_ID = ?
"""
SELECT_ACF = """
SELECT
ACF_ID,
VALUE,
DESCRIPTION
FROM acf WHERE SIG_ID = ?
"""
SELECT_STAT_DOCS = """
SELECT COUNT(*)
FROM documents
"""
SELECT_STAT_IMAGES = """
SELECT COUNT(*)
FROM documents
WHERE type IS 'Image'
"""
SELECT_STAT_AUDIO = """
SELECT COUNT(*)
FROM documents
WHERE type IS 'Audio'
"""
############################## CREATE
CREATE_SIGNALS = """
CREATE TABLE signals (
SIG_ID INTEGER PRIMARY KEY AUTOINCREMENT,
NAME TEXT,
DESCRIPTION TEXT,
URL TEXT
)
"""
CREATE_CATEGORY = """
CREATE TABLE category (
CAT_ID INTEGER PRIMARY KEY AUTOINCREMENT,
SIG_ID INTEGER,
CLB_ID INTEGER,
FOREIGN KEY (SIG_ID) REFERENCES signals (SIG_ID) ON DELETE CASCADE ON UPDATE CASCADE,
FOREIGN KEY (CLB_ID) REFERENCES category_label (CLB_ID) ON DELETE CASCADE ON UPDATE CASCADE
)
"""
CREATE_CATEGORY_LABELS = """
CREATE TABLE category_label (
CLB_ID INTEGER PRIMARY KEY AUTOINCREMENT,
VALUE TEXT
)
"""
CREATE_DOCUMENTS = """
CREATE TABLE documents (
DOC_ID INTEGER PRIMARY KEY AUTOINCREMENT,
SIG_ID INTEGER REFERENCES signals (SIG_ID) ON DELETE CASCADE,
EXTENSION TEXT,
NAME TEXT,
DESCRIPTION TEXT,
TYPE TEXT,
PREVIEW INTEGER
)
"""
CREATE_FREQUENCY = """
CREATE TABLE frequency (
FREQ_ID INTEGER PRIMARY KEY AUTOINCREMENT,
SIG_ID INTEGER REFERENCES signals (SIG_ID) ON DELETE CASCADE,
VALUE INTEGER,
DESCRIPTION TEXT
)
"""
CREATE_BANDWIDTH = """
CREATE TABLE bandwidth (
BAND_ID INTEGER PRIMARY KEY AUTOINCREMENT,
SIG_ID INTEGER REFERENCES signals (SIG_ID) ON DELETE CASCADE,
VALUE INTEGER,
DESCRIPTION TEXT
)
"""
CREATE_MODULATION = """
CREATE TABLE modulation (
MDL_ID INTEGER PRIMARY KEY AUTOINCREMENT,
SIG_ID INTEGER REFERENCES signals (SIG_ID) ON DELETE CASCADE,
VALUE TEXT,
DESCRIPTION TEXT
)
"""
CREATE_MODE = """
CREATE TABLE mode (
MOD_ID INTEGER PRIMARY KEY AUTOINCREMENT,
SIG_ID INTEGER,
VALUE TEXT,
DESCRIPTION TEXT,
FOREIGN KEY (SIG_ID) REFERENCES signals (SIG_ID) ON DELETE CASCADE ON UPDATE CASCADE
)
"""
CREATE_LOCATION = """
CREATE TABLE location (
LOC_ID INTEGER PRIMARY KEY AUTOINCREMENT,
SIG_ID INTEGER,
VALUE TEXT,
DESCRIPTION TEXT,
FOREIGN KEY (SIG_ID) REFERENCES signals (SIG_ID) ON DELETE CASCADE ON UPDATE CASCADE
)
"""
CREATE_ACF = """
CREATE TABLE acf (
ACF_ID INTEGER PRIMARY KEY AUTOINCREMENT,
SIG_ID INTEGER REFERENCES signals (SIG_ID) ON DELETE CASCADE,
VALUE FLOAT,
DESCRIPTION TEXT
)
"""
CREATE_INFO = """
CREATE TABLE info (
NAME TEXT,
DATE TEXT,
VERSION INTEGER,
EDITABLE INTEGER
)
"""
CREATE_VIEW_FREQ = """
CREATE VIEW FREQ_RANGE AS
SELECT SIG_ID,
MIN(VALUE) AS MIN_VALUE,
MAX(VALUE) AS MAX_VALUE
FROM frequency
GROUP BY SIG_ID
"""
CREATE_VIEW_BAND = """
CREATE VIEW BAND_RANGE AS
SELECT SIG_ID,
MIN(VALUE) AS MIN_VALUE,
MAX(VALUE) AS MAX_VALUE
FROM bandwidth
GROUP BY SIG_ID
"""
############################## INSERT
INSERT_SIGNAL = """
INSERT INTO signals (
NAME,
DESCRIPTION
) VALUES (?,?)
"""
INSERT_CATEGORY = """
INSERT INTO category (
SIG_ID,
CLB_ID
) VALUES (?,?)
"""
INSERT_CATEGORY_LABEL = """
INSERT INTO category_label (
VALUE
) VALUES (?)
"""
INSERT_INFO = """
INSERT INTO info (
NAME,
DATE,
VERSION,
EDITABLE
) VALUES (?,?,?,?)
"""
INSERT_DOCUMENTS = """
INSERT INTO documents (
SIG_ID,
EXTENSION,
NAME,
DESCRIPTION,
TYPE,
PREVIEW
) VALUES (?,?,?,?,?,?)
"""
INSERT_FREQUENCY = """
INSERT INTO frequency (
SIG_ID,
VALUE,
DESCRIPTION
) VALUES (?,?,?)
"""
INSERT_BANDWIDTH = """
INSERT INTO bandwidth (
SIG_ID,
VALUE,
DESCRIPTION
) VALUES (?,?,?)
"""
INSERT_MODULATION = """
INSERT INTO modulation (
SIG_ID,
VALUE,
DESCRIPTION
) VALUES (?,?,?)
"""
INSERT_MODE = """
INSERT INTO mode (
SIG_ID,
VALUE,
DESCRIPTION
) VALUES (?,?,?)
"""
INSERT_LOCATION = """
INSERT INTO location (
SIG_ID,
VALUE,
DESCRIPTION
) VALUES (?,?,?)
"""
INSERT_ACF = """
INSERT INTO acf (
SIG_ID,
VALUE,
DESCRIPTION
) VALUES (?,?,?)
"""
############################## UPDATE
RENAME_DB = "UPDATE info SET NAME = ?"
UPDATE_SIGNAL = """
UPDATE signals SET
NAME = ?,
DESCRIPTION = ?
WHERE SIG_ID = ?
"""
UPDATE_CATEGORY_LABEL = """
UPDATE category_label SET
VALUE = ?
WHERE CLB_ID = ?
"""
UPDATE_FREQUENCY = """
UPDATE frequency SET
VALUE = ?,
DESCRIPTION = ?
WHERE FREQ_ID = ?
"""
UPDATE_BANDWIDTH = """
UPDATE bandwidth SET
VALUE = ?,
DESCRIPTION = ?
WHERE BAND_ID = ?
"""
UPDATE_ACF = """
UPDATE acf SET
VALUE = ?,
DESCRIPTION = ?
WHERE ACF_ID = ?
"""
UPDATE_MODE = """
UPDATE mode SET
VALUE = ?,
DESCRIPTION = ?
WHERE MOD_ID = ?
"""
UPDATE_LOCATION = """
UPDATE location SET
VALUE = ?,
DESCRIPTION = ?
WHERE LOC_ID = ?
"""
UPDATE_MODULATION = """
UPDATE modulation SET
VALUE = ?,
DESCRIPTION = ?
WHERE MDL_ID = ?
"""
UPDATE_DOCUMENTS = """
UPDATE documents SET
NAME = ?,
DESCRIPTION = ?,
TYPE = ?,
PREVIEW = ?
WHERE DOC_ID = ?
"""
############################## DELETE
DELETE_SIGNAL = "DELETE FROM signals WHERE SIG_ID = ?"
DELETE_DOCUMENT = "DELETE FROM documents WHERE DOC_ID = ?"
DELETE_FREQUENCY = "DELETE FROM frequency WHERE FREQ_ID = ?"
DELETE_BANDWIDTH = "DELETE FROM bandwidth WHERE BAND_ID = ?"
DELETE_MODULATION = "DELETE FROM modulation WHERE MDL_ID = ?"
DELETE_MODE = "DELETE FROM mode WHERE MOD_ID = ?"
DELETE_LOCATION = "DELETE FROM location WHERE LOC_ID = ?"
DELETE_ACF = "DELETE FROM acf WHERE ACF_ID = ?"
DELETE_CATEGORY = "DELETE FROM category WHERE CAT_ID = ?"
DELETE_CATEGORY_LABEL = "DELETE FROM category_label WHERE CLB_ID = ?"
############################## FILTER QUERY
FILTER_FREQ = "SELECT SIG_ID FROM FREQ_RANGE WHERE ({} >= MIN_VALUE) AND ({} <= MAX_VALUE)"
FILTER_BAND = "SELECT SIG_ID FROM BAND_RANGE WHERE ({} >= MIN_VALUE) AND ({} <= MAX_VALUE)"
FILTER_ACF = "SELECT SIG_ID FROM acf WHERE ({} >= VALUE) AND ({} <= VALUE)"
FILTER_MODULATION = "SELECT SIG_ID FROM modulation WHERE VALUE IN ({})"
FILTER_LOCATION = "SELECT SIG_ID FROM location WHERE VALUE IN ({})"
FILTER_CATEGORY = "SELECT SIG_ID FROM category WHERE CLB_ID IN ({})"

77
artemis/utils/generic_utils.py Normal file
View File

@@ -0,0 +1,77 @@
from artemis.utils.constants import Query
def format_frequency(freq_hz):
""" Return frequency in a human-readable format
Args:
freq_hz (int): frequency in Hz
"""
scale = _change_unit_freq(freq_hz)
formatted_freq = f'{freq_hz / scale[0]} {scale[1]}'
return formatted_freq
def _change_unit_freq(freq_hz):
""" Return a scale factor and unit based on the number of digits in the frequency
Args:
freq_hz (int): frequency in Hz
"""
digits = len(str(freq_hz))
if digits < 4:
return 1, 'Hz'
elif digits < 7:
return 10**3, 'kHz'
elif digits < 10:
return 10**6, 'MHz'
else:
return 10**9, 'GHz'
def generate_filter_query(filer_status):
""" Returns the sql query according to the selected filter parameters
Args:
filer_status (dic): dictionary containing a summary of the active
filtering options with the related parametes.
"""
query = []
for key, val in filer_status.items():
if key == 'frequency':
query.append(Query.FILTER_FREQ.format(
val['upper_band'],
val['lower_band']
))
elif key == 'bandwidth':
query.append(Query.FILTER_BAND.format(
val['upper_band'],
val['lower_band']
))
elif key == 'acf':
query.append(Query.FILTER_ACF.format(
val['upper_band'],
val['lower_band']
))
elif key == 'modulation':
query.append(Query.FILTER_MODULATION.format(
', '.join(f"'{mod}'" for mod in val)
))
elif key == 'location':
query.append(Query.FILTER_LOCATION.format(
', '.join(f"'{loc}'" for loc in val)
))
elif key == 'category':
query.append(Query.FILTER_CATEGORY.format(
', '.join(f"{cat}" for cat in val)
))
return ' INTERSECT '.join(query)

134
artemis/utils/network_utils.py Normal file
View File

@@ -0,0 +1,134 @@
import os
import requests
from packaging.version import Version
from artemis.utils.constants import Constants, Messages
from artemis.utils.sql_utils import ArtemisDatabase
from artemis.utils.sys_utils import is_windows, is_linux, is_macos
class NetworkManager:
""" Class that checks for DB or software updates
"""
def __init__(self, parent):
self._parent = parent
self.sigid_db_path = Constants.DB_DIR / 'sigID' / Constants.SQL_NAME
self.show_popup = False
self.db_update = None
self.art_update = None
self.remote_db_url = None
self.remote_db_hash = None
self.remote_db_version = None
self.remote_db_size = None
self.remote_art_version = None
self.check_updates()
def check_updates(self):
""" Checks if a new DB update is available.
Args:
popup (bool, optional): Suppress the "already up-to-date" message on startup.
Defaults to False.
"""
latest_json = self.fetch_remote_json(Constants.DB_LATEST_VERSION)
if latest_json:
local_db = self.load_local_db()
remote_db = latest_json['sigID_DB']
self.remote_db_version = remote_db['version']
self.remote_db_url = remote_db['url']
self.remote_db_hash = remote_db['sha256_hash']
self.remote_db_size = remote_db['total_bytes']
if is_windows():
self.remote_art_version = latest_json['windows']['version']
elif is_linux():
self.remote_art_version = latest_json['linux']['version']
elif is_macos():
self.remote_art_version = latest_json['mac']['version']
if Version(self.remote_art_version) > Version(Constants.APPLICATION_VERSION):
self.art_update = True
else:
self.art_update = False
if self.art_update:
self.show_popup_art_update()
else:
if local_db:
if self.remote_db_version > local_db.version:
self.show_popup_db_update()
elif self.show_popup:
self.show_popup_up_to_date()
else:
self.show_popup_initial_db_download()
def fetch_remote_json(self, url):
""" Fetches the remote json from a url
"""
try:
response = requests.get(url)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
if self.show_popup:
self._parent.dialog_popup(
Messages.DIALOG_TYPE_ERROR,
Messages.NO_CONNECTION,
Messages.NO_CONNECTION_MSG.format(e)
)
return None
def load_local_db(self):
""" Loads the local database if exists
"""
if os.path.exists(self.sigid_db_path):
local_db = ArtemisDatabase('sigID')
local_db.load()
return local_db
return None
def show_popup_db_update(self):
"""Prompts the user to download the updated version of the database."""
self._parent.dialog_download_db(
Messages.DIALOG_TYPE_WARN,
Messages.DB_NEW_VER,
Messages.DB_NEW_VER_MSG.format(self.remote_db_version)
)
def show_popup_art_update(self):
"""Prompts the user to download the updated version of the database."""
self._parent.dialog_download_artemis(
Messages.DIALOG_TYPE_WARN,
Messages.ART_NEW_VER,
Messages.ART_NEW_VER_MSG.format(self.remote_art_version)
)
def show_popup_up_to_date(self):
"""Notifies the user that the database is up to date."""
self._parent.dialog_popup(
Messages.DIALOG_TYPE_INFO,
Messages.UP_TO_DATE,
Messages.UP_TO_DATE_MSG
)
def show_popup_initial_db_download(self):
"""Prompts the user to download the database for the first time."""
self._parent.dialog_download_db(
Messages.DIALOG_TYPE_QUEST,
Messages.NO_DB_DETECTED,
Messages.NO_DB_DETECTED_MSG
)

56
artemis/utils/path_utils.py Normal file
View File

@@ -0,0 +1,56 @@
import os
from pathlib import Path
from artemis.utils.sql_utils import ArtemisDatabase
from artemis.utils.constants import Constants
from artemis.utils.sys_utils import *
def check_data_dir():
if not os.path.exists(Constants.DB_DIR):
os.makedirs(Constants.DB_DIR)
def normalize_dialog_path(path):
if is_windows():
norm_path = path.replace('file:///', '')
elif is_linux() or is_macos():
norm_path = path.replace('file:///', '/')
return norm_path
def logs_dir():
if is_macos():
logs_dir_path = Path.home() / 'Library/Logs/' / Constants.ORGANIZATION_NAME / Constants.APPLICATION_NAME
elif is_windows():
logs_dir_path = Path.home() / 'AppData/Local/' / Constants.ORGANIZATION_NAME / Constants.APPLICATION_NAME / 'logs'
elif is_linux():
logs_dir_path = Path.home() / '/var/log/' / Constants.ORGANIZATION_NAME / Constants.APPLICATION_NAME
else:
logs_dir_path = Constants.LOGS_DIR
if not logs_dir_path.exists():
logs_dir_path.mkdir(parents=True)
return logs_dir_path
def valid_db(db_dir_name):
""" Checks if db_dir_name is a valid db dir containing a `data.sqlite` file.
Db must be valid as well and should be properly initialized and loaded with
no errors.
Args:
db_dir_name (str): name of the db folder
"""
if os.path.exists(Constants.DB_DIR / db_dir_name / Constants.SQL_NAME):
try:
database = ArtemisDatabase(db_dir_name)
database.load()
return True
except Exception as e:
# Invalid or corrupted DB
return False
else:
# The dir is not containing a data.sqlite file
return False

416
artemis/utils/sql_utils.py Normal file
View File

@@ -0,0 +1,416 @@
import sqlite3
import os
from PySide6.QtCore import QUrl
from operator import itemgetter
from datetime import datetime
from artemis.utils.constants import Query, Constants
from artemis.utils.generic_utils import *
from contextlib import closing
class Database():
""" General superclass for SQLite DB manipulation.
Foreign keys are activated (otherwise disabled by default for compatibility purposes)
"""
def __init__(self, sql_path):
self.sql_path = sql_path
def execute(self, query, parameters=None, last_rowid=False):
""" Open a connection, execute the given query with optional parameters and close the connection.
In the case of a SELECT query, returns the results as a fetchall().
If last_rowid == True, this function returns a tuple with the result of the fetchall() and
the latest modified row id of the current connection.
"""
with closing(sqlite3.connect(self.sql_path, check_same_thread=False)) as conn:
conn.execute('PRAGMA foreign_keys = ON;')
curs = conn.cursor()
if parameters:
curs.execute(query, parameters)
else:
curs.execute(query)
conn.commit()
if last_rowid:
result = (curs.fetchall(), curs.lastrowid)
else:
result = curs.fetchall()
return result
################################## MARK: >>> DATABASE <<<
class ArtemisDatabase(Database):
""" General CRUD class for SQLite DB manipulation.
Foreign keys are activated (otherwise disabled by default for compatibility purposes)
"""
def __init__(self, db_dir_name):
self.db_dir_name = db_dir_name
self.db_dir = Constants.DB_DIR / db_dir_name
self.sql_path = self.db_dir / Constants.SQL_NAME
self.media_dir = self.db_dir / 'media'
super().__init__(self.sql_path)
self.name = None
self.date = None
self.version = None
self.editable = None
self.all_signals = None
self.all_modulation = None
self.all_location = None
self.all_category_labels = None
self.filtered_signals = None
self.stats = {}
def load(self):
self._select_info()
self._select_all()
self._select_all_modulation()
self._select_all_location()
self._select_all_category_labels()
self._select_stats()
def _select_info(self):
""" Load the DB meta INFO from the table 'info'
"""
result = self.execute(Query.SELECT_INFO)[0]
self.name = result[0]
self.date = result[1]
self.version = result[2]
self.editable = result[3]
def _select_all(self):
""" Load a list of tuple for all signals. Each tuple (representing a signal)
contains the SIG_ID and the NAME of the signal
"""
self.all_signals = self.execute(Query.SELECT_ALL_SIGNALS)
keys = ('SIG_ID', 'name')
result = [dict(zip(keys, values)) for values in self.all_signals]
self.all_signals = result
def _select_all_modulation(self):
self.all_modulation = self.execute(Query.SELECT_ALL_MODULATION)
self.all_modulation = [{'value': item[0]} for item in self.all_modulation]
def _select_all_location(self):
self.all_location = self.execute(Query.SELECT_ALL_LOCATION)
self.all_location = [{'value': item[0]} for item in self.all_location]
def _select_all_category_labels(self):
self.all_category_labels = self.execute(Query.SELECT_ALL_CAT_LABELS)
self.all_category_labels = [{'clb_id': item[0], 'value': item[1]} for item in self.all_category_labels]
def _select_stats(self):
tot_docs = self.execute(Query.SELECT_STAT_DOCS)[0][0]
tot_images = self.execute(Query.SELECT_STAT_IMAGES)[0][0]
tot_audio = self.execute(Query.SELECT_STAT_AUDIO)[0][0]
self.stats['documents'] = tot_docs
self.stats['images'] = tot_images
self.stats['audio'] = tot_audio
self.stats['signals'] = len(self.all_signals)
def select_by_filter(self, filter_query):
matching_sig_ids = self.execute(filter_query)
sig_ids = ",".join(str(num[0]) for num in matching_sig_ids)
self.all_signals = self.execute(Query.SELECT_SIG_ID.format(sig_ids))
keys = ('SIG_ID', 'name')
result = [dict(zip(keys, values)) for values in self.all_signals]
self.all_signals = result
def create(self, name):
""" Create new db in the data folder.
The name of folder containing the new db has a unique id as name (db_dir_name).
"""
meta = [name, datetime.now(), 0, 0]
os.makedirs(self.db_dir)
os.makedirs(self.media_dir)
self.execute(Query.CREATE_INFO)
self.execute(Query.INSERT_INFO, meta)
self.execute(Query.CREATE_SIGNALS)
self.execute(Query.CREATE_CATEGORY)
self.execute(Query.CREATE_CATEGORY_LABELS)
self.execute(Query.CREATE_FREQUENCY)
self.execute(Query.CREATE_BANDWIDTH)
self.execute(Query.CREATE_MODULATION)
self.execute(Query.CREATE_MODE)
self.execute(Query.CREATE_LOCATION)
self.execute(Query.CREATE_ACF)
self.execute(Query.CREATE_DOCUMENTS)
self.execute(Query.CREATE_VIEW_FREQ)
self.execute(Query.CREATE_VIEW_BAND)
def rename(self, name):
self.execute(Query.RENAME_DB, [name])
def insert_category_label(self, value):
self.execute(Query.INSERT_CATEGORY_LABEL, [value])
def update_category_label(self, clb_id, value):
self.execute(Query.UPDATE_CATEGORY_LABEL, [value, clb_id])
def delete_category_label(self, clb_id):
self.execute(Query.DELETE_CATEGORY_LABEL, [clb_id])
################################## MARK: >>> SIGNAL <<<
class ArtemisSignal():
""" Main class of the object signal
"""
def __init__(self, loaded_db):
self.db = loaded_db
self.sig_id = None
self.name = None
self.description = None
self.url = None
self.category = None
self.frequency = None
self.bandwidth = None
self.modulation = None
self.mode = None
self.location = None
self.acf = None
self.documents = None
self.spectrum_path = None
self.audio_path = None
def load(self, sig_id):
self.sig_id = sig_id
self._select_signals()
self._select_category()
self._select_frequency()
self._select_bandwidth()
self._select_modulation()
self._select_mode()
self._select_location()
self._select_acf()
self.select_documents()
def generate_dic(self):
dic = {
'name': self.name,
'description': self.description,
'url': self.url,
'category': self.category,
'frequency': self.frequency,
'bandwidth': self.bandwidth,
'modulation': self.modulation,
'mode': self.mode,
'location': self.location,
'acf': self.acf,
'spectrum_path': self.spectrum_path,
'audio_path': self.audio_path,
'all_category': self.db.all_category_labels
}
return dic
################################## MARK: SELECT Methods
def _select_signals(self):
signal = self.db.execute(Query.SELECT_SIGNAL, [self.sig_id])[0]
self.name = signal[0]
self.description = signal[1]
self.url = signal[2]
def _select_category(self):
self.category = self.db.execute(Query.SELECT_CATEGORY, [self.sig_id])
self.category = [list(x) for x in self.category]
def _select_frequency(self):
result = self.db.execute(Query.SELECT_FREQUENCY, [self.sig_id])
sorted_list = sorted(result, key=itemgetter(1))
self.frequency = [list(x) + [format_frequency(x[1])] for x in sorted_list]
def _select_bandwidth(self):
result = self.db.execute(Query.SELECT_BANDWIDTH, [self.sig_id])
sorted_list = sorted(result, key=itemgetter(1))
self.bandwidth = [list(x) + [format_frequency(x[1])] for x in sorted_list]
def _select_acf(self):
self.acf = self.db.execute(Query.SELECT_ACF, [self.sig_id])
self.acf = [list(x) for x in self.acf]
def _select_modulation(self):
self.modulation = self.db.execute(Query.SELECT_MODULATION, [self.sig_id])
self.modulation = [list(x) for x in self.modulation]
def _select_mode(self):
self.mode = self.db.execute(Query.SELECT_MODE, [self.sig_id])
self.mode = [list(x) for x in self.mode]
def _select_location(self):
self.location = self.db.execute(Query.SELECT_LOCATION, [self.sig_id])
self.location = [list(x) for x in self.location]
def select_documents(self):
self.documents = self.db.execute(Query.SELECT_DOCUMENTS, [self.sig_id])
default_spectrum = [doc for doc in self.documents if doc[4] == 'Image' and doc[5] == 1]
default_audio = [doc for doc in self.documents if doc[4] == 'Audio' and doc[5] == 1]
if default_spectrum != []:
default_spectrum_filename = '{}.{}'.format(str(default_spectrum[0][0]), default_spectrum[0][1])
self.spectrum_path = self.db.media_dir / default_spectrum_filename
self.spectrum_path = QUrl.fromLocalFile(self.spectrum_path.resolve())
else:
self.spectrum_path = 'qrc:///images/spectrum_not_available.svg'
if default_audio != []:
default_audio_filename = '{}.{}'.format(str(default_audio[0][0]), default_audio[0][1])
self.audio_path = self.db.media_dir / default_audio_filename
self.audio_path = QUrl.fromLocalFile(self.audio_path.resolve())
else:
self.audio_path = ''
################################## MARK: UPDATE Methods
def update_signal(self, sig_id, value, description):
self.db.execute(Query.UPDATE_SIGNAL, [value, description, sig_id])
def update_frequency(self, freq_id, value, description):
self.db.execute(Query.UPDATE_FREQUENCY, [value, description, freq_id])
def update_bandwidth(self, band_id, value, description):
self.db.execute(Query.UPDATE_BANDWIDTH, [value, description, band_id])
def update_modulation(self, modu_id, value, description):
self.db.execute(Query.UPDATE_MODULATION, [value, description, modu_id])
def update_mode(self, mode_id, value, description):
self.db.execute(Query.UPDATE_MODE, [value, description, mode_id])
def update_acf(self, acf_id, value, description):
self.db.execute(Query.UPDATE_ACF, [value, description, acf_id])
def update_location(self, loc_id, value, description):
self.db.execute(Query.UPDATE_LOCATION, [value, description, loc_id])
def update_documents(self, doc_id, name, description, type, is_preview):
self.db.execute(Query.UPDATE_DOCUMENTS, [name, description, type, is_preview, doc_id])
################################## MARK: INSERT Methods
def insert_signal(self, value, description):
self.db.execute(Query.INSERT_SIGNAL, [value, description])
def insert_frequency(self, value, description):
self.db.execute(Query.INSERT_FREQUENCY, [self.sig_id, value, description])
def insert_bandwidth(self, value, description):
self.db.execute(Query.INSERT_BANDWIDTH, [self.sig_id,value, description])
def insert_modulation(self, value, description):
self.db.execute(Query.INSERT_MODULATION, [self.sig_id,value, description])
def insert_mode(self, value, description):
self.db.execute(Query.INSERT_MODE, [self.sig_id,value, description])
def insert_acf(self, value, description):
self.db.execute(Query.INSERT_ACF, [self.sig_id,value, description])
def insert_location(self, value, description):
self.db.execute(Query.INSERT_LOCATION, [self.sig_id,value, description])
def insert_category(self, clb_id):
self.db.execute(Query.INSERT_CATEGORY, [self.sig_id, clb_id])
def insert_document(self, doc_lst):
row_id = self.db.execute(Query.INSERT_DOCUMENTS, [self.sig_id] + doc_lst[1:], True)[1]
return row_id
################################## MARK: DELETE Methods
def delete_signal(self):
self.db.execute(Query.DELETE_SIGNAL, [self.sig_id])
def delete_frequency(self, freq_id):
self.db.execute(Query.DELETE_FREQUENCY, [freq_id])
def delete_bandwidth(self, band_id):
self.db.execute(Query.DELETE_BANDWIDTH, [band_id])
def delete_modulation(self, modu_id):
self.db.execute(Query.DELETE_MODULATION, [modu_id])
def delete_mode(self, mode_id):
self.db.execute(Query.DELETE_MODE, [mode_id])
def delete_acf(self, acf_id):
self.db.execute(Query.DELETE_ACF, [acf_id])
def delete_location(self, loc_id):
self.db.execute(Query.DELETE_LOCATION, [loc_id])
def delete_document(self, doc_id):
self.db.execute(Query.DELETE_DOCUMENT, [doc_id])
def delete_category(self, cat_id):
self.db.execute(Query.DELETE_CATEGORY, [cat_id])

91
artemis/utils/sys_utils.py Normal file
View File

@@ -0,0 +1,91 @@
import os
import platform
import subprocess
import hashlib
from shutil import rmtree, copyfile, make_archive, unpack_archive
from pathlib import Path
from artemis.utils.constants import Constants, Messages
def is_windows():
return platform.system() == 'Windows'
def is_macos():
return platform.system() == 'Darwin'
def is_linux():
return platform.system() == 'Linux'
def open_file(file_path, timeout=3):
try:
if is_windows():
os.startfile(file_path)
elif is_macos():
subprocess.call(['open', file_path], timeout=timeout)
elif is_linux():
subprocess.call(['xdg-open', file_path], timeout=timeout)
else:
return
except FileNotFoundError:
raise Exception(Messages.FILE_NOT_FOUND_ERR_MSG)
except Exception as e:
raise Exception(Messages.GENERIC_ERROR_MSG.format(e))
def open_directory(directory, timeout=3):
if is_windows():
subprocess.call(['explorer', str(Path(directory))], timeout=timeout)
elif is_macos():
subprocess.call(['open', directory], timeout=timeout)
elif is_linux():
subprocess.call(['xdg-open', directory], timeout=timeout)
else:
return
def delete_db_dir(db_dir_name):
"""Delete the db folder"""
db_dir = Constants.DB_DIR / db_dir_name
if os.path.exists(db_dir):
rmtree(db_dir)
def copy_file(src_file_path, dst_file_path):
copyfile(src_file_path, dst_file_path)
def delete_file(file_path):
if os.path.exists(file_path):
os.remove(file_path)
def pack_db(save_path, db_dir):
make_archive(save_path, 'tar', db_dir.resolve().as_posix())
def unpack_db(tar_path, db_dir_name):
db_dir = Constants.DB_DIR / db_dir_name
unpack_archive(tar_path, db_dir, 'tar')
def match_hash(data, reference_hash):
""" Check whether the checksum of 'data' match the reference one.
Args:
data (str): Path of the file to be checked
reference_hash (str): Reference SHA-256 hash
"""
if reference_hash is None:
raise ValueError("ERROR: Invalid hash code.")
code = hashlib.sha256()
b = bytearray(128*1024)
mv = memoryview(b)
with open(data, 'rb', buffering=0) as f:
while n := f.readinto(mv):
code.update(mv[:n])
return code.hexdigest() == reference_hash

21
artemis/utils/ui_utils.py Normal file
View File

@@ -0,0 +1,21 @@
import os
from artemis.utils.sys_utils import is_windows, is_linux, is_macos
from artemis.utils.config_utils import CONFIGURE_QT
def set_ui():
os.environ['QT_QUICK_CONTROLS_STYLE'] = CONFIGURE_QT.get_or_default('Controls', 'style', 'Material')
os.environ['QT_QUICK_CONTROLS_MATERIAL_VARIANT'] = CONFIGURE_QT.get_or_default('Material', 'variant', 'Dense')
os.environ['QT_QUICK_CONTROLS_MATERIAL_THEME'] = CONFIGURE_QT.get_or_default('Material', 'theme', 'System')
os.environ['QT_QUICK_CONTROLS_MATERIAL_ACCENT'] = CONFIGURE_QT.get_or_default('Material', 'accent', 'Green')
if is_windows():
os.environ['QSG_RHI_BACKEND'] = 'opengl'
if is_linux():
os.environ['GDK_BACKEND'] = 'x11'
os.environ['QT_QPA_PLATFORM'] = 'xcb'
os.environ['QT_ENABLE_GLYPH_CACHE_WORKAROUND'] = '1'
os.environ['QML_USE_GLYPHCACHE_WORKAROUND'] = '1'