Commit 6cda0bf9 authored by Samuel GAIST's avatar Samuel GAIST

[widgets][experimenteditor] Add modelisation of the prefix content

This will allow to have a more accurate list of assets to
choose from.
parent 12d12887
......@@ -34,9 +34,14 @@ from PyQt5.QtWidgets import QComboBox
from PyQt5.QtWidgets import QCheckBox
from PyQt5.QtWidgets import QPushButton
from PyQt5.QtSql import QSqlTableModel
from PyQt5.QtSql import QSqlQuery
from beat.core.experiment import PROCESSOR_PREFIX
from beat.core.experiment import EVALUATOR_PREFIX
from beat.backend.python.algorithm import Algorithm
from ..backend.asset import Asset
from ..backend.asset import AssetType
from ..backend.assetmodel import AssetModel
......@@ -57,6 +62,9 @@ from ..widgets.experimenteditor import FieldPresenceFilterProxyModel
from ..widgets.experimenteditor import ExperimentEditor
from ..widgets.experimenteditor import typed_user_property
from ..widgets.experimenteditor import PrefixModel
from ..widgets.experimenteditor import AlgorithmModel
from .conftest import prefix
from .conftest import sync_prefix
......@@ -151,6 +159,129 @@ def choice_parameter_name(request):
# Tests
class TestPrefixModel:
"""Test that the prefix modelisation generates the expected data"""
def test_model(self, beat_context):
model = PrefixModel(beat_context)
model = QSqlTableModel()
model.setTable("algorithms")
model.select()
total = model.rowCount()
assert total > 0
model.setFilter("is_analyzer=True")
analyzer_count = model.rowCount()
assert analyzer_count > 0
assert analyzer_count < total
class TestAlgorithmModel:
"""Test the model used to generate suitable algorithm selections"""
@pytest.fixture
def prefix_model(self, beat_context):
return PrefixModel(beat_context)
def test_default(self, prefix_model):
algorithm_model = AlgorithmModel()
query = QSqlQuery()
assert query.exec_(
"SELECT COUNT(name) AS cnt FROM algorithms WHERE is_analyzer=false"
)
query.next()
assert algorithm_model.rowCount() == query.value("cnt")
def test_analyzers(self, prefix_model):
algorithm_model = AlgorithmModel()
algorithm_model.setAnalyzerEnabled(True)
query = QSqlQuery()
assert query.exec_(
"SELECT COUNT(name) AS cnt FROM algorithms WHERE is_analyzer=true"
)
query.next()
assert algorithm_model.rowCount() == query.value("cnt")
def test_types(self, prefix_model):
algorithm_model = AlgorithmModel()
algorithm_model.setTypes(
[Algorithm.LEGACY, Algorithm.SEQUENTIAL, Algorithm.AUTONOMOUS]
)
query = QSqlQuery()
assert query.exec_(
"SELECT COUNT(name) AS cnt FROM algorithms WHERE is_analyzer=false AND type IN ('legacy', 'sequential', 'autonomous')"
)
query.next()
assert algorithm_model.rowCount() == query.value("cnt")
def test_output_count(self, prefix_model):
algorithm_model = AlgorithmModel()
algorithm_model.setOutputCount(2)
query = QSqlQuery()
assert query.exec_(
"SELECT COUNT(name) AS cnt FROM algorithms WHERE is_analyzer=false AND outputs='2'"
)
query.next()
assert algorithm_model.rowCount() == query.value("cnt")
def test_input_count(self, prefix_model):
algorithm_model = AlgorithmModel()
algorithm_model.setInputCount(2)
query = QSqlQuery()
assert query.exec_(
"SELECT COUNT(name) AS cnt FROM algorithms WHERE is_analyzer=false AND inputs='2'"
)
query.next()
assert algorithm_model.rowCount() == query.value("cnt")
def test_input_output_count(self, prefix_model):
algorithm_model = AlgorithmModel()
algorithm_model.setInputCount(1)
algorithm_model.setOutputCount(1)
query = QSqlQuery()
assert query.exec_(
"SELECT COUNT(name) AS cnt FROM algorithms WHERE is_analyzer=false AND inputs='1' AND outputs='1'"
)
query.next()
assert algorithm_model.rowCount() == query.value("cnt")
def test_input_output_count_and_types(self, prefix_model):
algorithm_model = AlgorithmModel()
algorithm_model.setInputCount(1)
algorithm_model.setOutputCount(1)
algorithm_model.setTypes([Algorithm.SEQUENTIAL])
query = QSqlQuery()
assert query.exec_(
"SELECT COUNT(name) AS cnt FROM algorithms WHERE is_analyzer=false AND inputs='1' AND outputs='1' AND type IN ('sequential')"
)
query.next()
assert algorithm_model.rowCount() == query.value("cnt")
class TestIOMapperDialog:
"""Test that the dialog used for input/output mapping works as expected"""
......
......@@ -54,6 +54,10 @@ from PyQt5.QtWidgets import QTabWidget
from PyQt5.QtWidgets import QVBoxLayout
from PyQt5.QtWidgets import QWidget
from PyQt5.QtSql import QSqlDatabase
from PyQt5.QtSql import QSqlQuery
from PyQt5.QtSql import QSqlTableModel
from beat.core.experiment import PROCESSOR_PREFIX
from beat.core.experiment import EVALUATOR_PREFIX
......@@ -261,6 +265,137 @@ class DatasetModel(QStringListModel):
self.update()
# ------------------------------------------------------------------------------
# Prefix modelization
class PrefixModel:
def __init__(self, context=None):
self.context = context
database = QSqlDatabase.addDatabase("QSQLITE")
database.setDatabaseName(":memory:")
if not database.open():
raise RuntimeError(
f"Failed to open database: {database.lastError().text()}"
)
self.refresh()
def setContext(self, context):
if self.context == context:
return
self.context = context
self.refresh()
def refresh(self):
if self.context is None:
return
ALGORITHM_TABLE_CLEANUP = "DROP TABLE IF EXISTS algorithms"
ALGORITHM_TABLE = "CREATE TABLE algorithms(name varchar, type varchar, inputs integer, outputs integer, is_analyzer boolean)"
INSERT_ALGORITHM = "INSERT INTO algorithms(name, type, inputs, outputs, is_analyzer) VALUES(?, ?, ?, ?, ?)"
query = QSqlQuery()
if not query.exec_(ALGORITHM_TABLE_CLEANUP):
raise RuntimeError(f"Failed to drop table: {query.lastError().text()}")
if not query.exec_(ALGORITHM_TABLE):
raise RuntimeError(f"Failed to create table: {query.lastError().text()}")
prefix_path = self.context.meta["config"].path
model = AssetModel()
model.asset_type = AssetType.ALGORITHM
model.prefix_path = prefix_path
model.setLatestOnlyEnabled(False)
if not query.prepare(INSERT_ALGORITHM):
raise RuntimeError(f"Failed to prepare query: {query.lastError().text()}")
for algorithm in model.stringList():
asset = Asset(prefix_path, AssetType.ALGORITHM, algorithm)
try:
declaration = asset.declaration
except json.JSONDecodeError:
continue
inputs = {}
outputs = {}
for group in declaration["groups"]:
inputs.update(group.get("inputs", {}))
outputs.update(group.get("outputs", {}))
query.addBindValue(algorithm)
query.addBindValue(declaration.get("type", "legacy"))
query.addBindValue(len(inputs))
query.addBindValue(len(outputs))
query.addBindValue("results" in declaration)
if not query.exec_():
raise RuntimeError(
f"Failed to insert algorithm: {query.lastError().text()}"
)
class AlgorithmModel(QSqlTableModel):
def __init__(self, parent=None):
super().__init__(parent=parent)
self._analyzer_enabled = False
self._input_count = None
self._output_count = None
self._types = []
self.setTable("algorithms")
self.select()
self.update()
def update(self):
filter_str = f"is_analyzer={self._analyzer_enabled}"
if self._input_count is not None:
filter_str += f" AND inputs={self._input_count}"
if self._output_count is not None:
filter_str += f" AND outputs={self._output_count}"
if self._types:
filter_str += " AND type in ({})".format(
",".join([f"'{type_}'" for type_ in self._types])
)
self.setFilter(filter_str)
def setAnalyzerEnabled(self, enabled):
if self._analyzer_enabled == enabled:
return
self._analyzer_enabled = enabled
self.update()
def setInputCount(self, count):
if self._input_count == count:
return
self._input_count = count
self.update()
def setOutputCount(self, count):
if self._output_count == count:
return
self._output_count = count
self.update()
def setTypes(self, type_list):
if self._types == type_list:
return
self._types = type_list
self.update()
# ------------------------------------------------------------------------------
# Editors
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment