######################################################################################################################
# Copyright (C) 2017-2021 Spine project consortium
# This file is part of Spine Toolbox.
# Spine Toolbox is free software: you can redistribute it and/or modify it under the terms of the GNU Lesser General
# Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option)
# any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
# without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General
# Public License for more details. You should have received a copy of the GNU Lesser General Public License along with
# this program. If not, see <http://www.gnu.org/licenses/>.
######################################################################################################################
"""
Empty models for parameter definitions and values.
:authors: M. Marin (KTH)
:date: 28.6.2019
"""
from PySide2.QtCore import Qt
from ...mvcmodels.empty_row_model import EmptyRowModel
from .parameter_mixins import (
FillInParameterNameMixin,
MakeRelationshipOnTheFlyMixin,
InferEntityClassIdMixin,
FillInAlternativeIdMixin,
FillInParameterDefinitionIdsMixin,
FillInEntityIdsMixin,
FillInEntityClassIdMixin,
FillInValueListIdMixin,
ValidateValueInListForInsertMixin,
)
from ...mvcmodels.shared import PARSED_ROLE
from ...helpers import rows_to_row_count_tuples
[docs]class EmptyParameterModel(EmptyRowModel):
"""An empty parameter model."""
def __init__(self, parent, header, db_mngr):
"""Initialize class.
Args:
parent (Object): the parent object, typically a CompoundParameterModel
header (list): list of field names for the header
db_mngr (SpineDBManager)
"""
super().__init__(parent, header)
self.db_mngr = db_mngr
self.db_map = None
self.entity_class_id = None
@property
[docs] def item_type(self):
"""The item type, either 'parameter_value' or 'parameter_definition', required by the json_fields property."""
raise NotImplementedError()
@property
[docs] def entity_class_type(self):
"""Either 'object_class' or 'relationship_class'."""
raise NotImplementedError()
@property
[docs] def entity_class_id_key(self):
return {"object_class": "object_class_id", "relationship_class": "relationship_class_id"}[
self.entity_class_type
]
@property
[docs] def entity_class_name_key(self):
return {"object_class": "object_class_name", "relationship_class": "relationship_class_name"}[
self.entity_class_type
]
@property
[docs] def can_be_filtered(self):
return False
@property
[docs] def json_fields(self):
return {"parameter_definition": ["default_value"], "parameter_value": ["value"]}[self.item_type]
[docs] def accepted_rows(self):
return list(range(self.rowCount()))
[docs] def db_item(self, _index): # pylint: disable=no-self-use
return None
[docs] def item_id(self, _row): # pylint: disable=no-self-use
return None
[docs] def flags(self, index):
flags = super().flags(index)
if self.header[index.column()] == "parameter_tag_list":
flags &= ~Qt.ItemIsEditable
return flags
[docs] def data(self, index, role=Qt.DisplayRole):
if self.header[index.column()] in self.json_fields and role in (
Qt.DisplayRole,
Qt.ToolTipRole,
Qt.TextAlignmentRole,
PARSED_ROLE,
):
data = super().data(index)
parsed_value = self.db_mngr.parse_value(data)
return self.db_mngr.format_value(parsed_value, role)
return super().data(index, role)
[docs] def _make_unique_id(self, item):
"""Returns a unique id for the given model item (name-based). Used by receive_parameter_data_added."""
return (item.get(self.entity_class_name_key), item.get("parameter_name"))
[docs] def receive_parameter_data_added(self, db_map_data):
"""Runs when parameter definitions or values are added.
Finds and removes model items that were successfully added to the db."""
added_ids = set()
for db_map, items in db_map_data.items():
for item in items:
database = db_map.codename
unique_id = (database, *self._make_unique_id(item))
added_ids.add(unique_id)
removed_rows = []
for row, data in enumerate(self._main_data):
item = dict(zip(self.header, data))
database = item.get("database")
unique_id = (database, *self._make_unique_id(item))
if unique_id in added_ids:
removed_rows.append(row)
for row, count in sorted(rows_to_row_count_tuples(removed_rows), reverse=True):
self.removeRows(row, count)
[docs] def batch_set_data(self, indexes, data):
"""Sets data for indexes in batch. If successful, add items to db."""
if not super().batch_set_data(indexes, data):
return False
rows = {ind.row() for ind in indexes}
db_map_data = self._make_db_map_data(rows)
self.add_items_to_db(db_map_data)
return True
[docs] def add_items_to_db(self, db_map_data):
"""Add items to db.
Args:
db_map_data (dict): mapping DiffDatabaseMapping instance to list of items
"""
raise NotImplementedError()
[docs] def _make_db_map_data(self, rows):
"""
Returns model data grouped by database map.
Args:
rows (set): group data from these rows
Returns:
dict: mapping DiffDatabaseMapping instance to list of items
"""
items = [dict(zip(self.header, self._main_data[row]), row=row) for row in rows]
db_map_data = dict()
for item in items:
database = item.pop("database")
db_map = next(iter(x for x in self.db_mngr.db_maps if x.codename == database), None)
if not db_map:
continue
db_map_data.setdefault(db_map, []).append(item)
return db_map_data
[docs]class EmptyParameterDefinitionModel(
FillInValueListIdMixin, FillInEntityClassIdMixin, FillInParameterNameMixin, EmptyParameterModel
):
"""An empty parameter_definition model."""
@property
[docs] def item_type(self):
return "parameter_definition"
@property
[docs] def entity_class_type(self):
"""See base class."""
raise NotImplementedError()
[docs] def add_items_to_db(self, db_map_data):
"""See base class."""
self.build_lookup_dictionary(db_map_data)
db_map_param_def = dict()
db_map_error_log = dict()
for db_map, items in db_map_data.items():
for item in items:
def_item, err = self._convert_to_db(item, db_map)
if self._check_item(def_item):
db_map_param_def.setdefault(db_map, []).append(def_item)
if err:
db_map_error_log.setdefault(db_map, []).extend(err)
if any(db_map_param_def.values()):
self.db_mngr.add_parameter_definitions(db_map_param_def)
if db_map_error_log:
self.db_mngr.error_msg.emit(db_map_error_log)
[docs] def _check_item(self, item):
"""Checks if a db item is ready to be inserted."""
return self.entity_class_id_key in item and "name" in item
[docs]class EmptyObjectParameterDefinitionModel(EmptyParameterDefinitionModel):
"""An empty object parameter_definition model."""
@property
[docs] def entity_class_type(self):
return "object_class"
[docs]class EmptyRelationshipParameterDefinitionModel(EmptyParameterDefinitionModel):
"""An empty relationship parameter_definition model."""
@property
[docs] def entity_class_type(self):
return "relationship_class"
[docs] def flags(self, index):
"""Additional hack to make the object_class_name_list column non-editable."""
flags = super().flags(index)
if self.header[index.column()] == "object_class_name_list":
flags &= ~Qt.ItemIsEditable
return flags
[docs]class EmptyParameterValueModel(
ValidateValueInListForInsertMixin,
InferEntityClassIdMixin,
FillInAlternativeIdMixin,
FillInParameterDefinitionIdsMixin,
FillInEntityIdsMixin,
FillInEntityClassIdMixin,
EmptyParameterModel,
):
"""An empty parameter_value model."""
@property
[docs] def item_type(self):
return "parameter_value"
@property
[docs] def entity_type(self):
"""Either 'object' or "relationship'."""
raise NotImplementedError()
@property
[docs] def entity_id_key(self):
return {"object": "object_id", "relationship": "relationship_id"}[self.entity_type]
@property
[docs] def entity_name_key(self):
return {"object": "object_name", "relationship": "object_name_list"}[self.entity_type]
@property
[docs] def entity_name_key_in_cache(self):
return {"object": "name", "relationship": "object_name_list"}[self.entity_type]
[docs] def _make_unique_id(self, item):
"""Returns a unique id for the given model item (name-based). Used by receive_parameter_data_added."""
return (*super()._make_unique_id(item), item.get(self.entity_name_key))
[docs] def add_items_to_db(self, db_map_data):
"""See base class."""
self.build_lookup_dictionary(db_map_data)
db_map_param_val = dict()
db_map_error_log = dict()
for db_map, items in db_map_data.items():
for item in items:
param_val, convert_errors = self._convert_to_db(item, db_map)
param_val, check_errors = self._check_item(db_map, param_val)
if param_val:
db_map_param_val.setdefault(db_map, []).append(param_val)
errors = convert_errors + check_errors
if errors:
db_map_error_log.setdefault(db_map, []).extend(errors)
if any(db_map_param_val.values()):
self.db_mngr.add_parameter_values(db_map_param_val)
if db_map_error_log:
self.db_mngr.error_msg.emit(db_map_error_log)
[docs] def _check_item(self, db_map, item):
"""Checks if a db item is ready to be inserted."""
item = item.copy()
entity_class_id = item.get(self.entity_class_id_key)
entity_id = item.get(self.entity_id_key)
parameter_id = item.get("parameter_definition_id")
alternative_id = item.get("alternative_id")
has_valid_value_from_list = item.pop("has_valid_value_from_list", True)
if not all([entity_class_id, entity_id, parameter_id, alternative_id, has_valid_value_from_list]):
return None, []
existing_items = {
(x["entity_class_id"], x["entity_id"], x["parameter_id"], x["alternative_id"]): (
x.get("object_name") or x.get("object_name_list"),
x["parameter_name"],
x["alternative_name"],
)
for x in self.db_mngr.get_items(db_map, "parameter_value")
}
dupe = existing_items.get((entity_class_id, entity_id, parameter_id, alternative_id))
if dupe is not None:
entity_name, parameter_name, alternative_name = dupe
return None, [f"The '{alternative_name}' value of '{parameter_name}' for '{entity_name}' is already set"]
return item, []
[docs]class EmptyObjectParameterValueModel(EmptyParameterValueModel):
"""An empty object parameter_value model."""
@property
[docs] def entity_class_type(self):
return "object_class"
@property
[docs] def entity_type(self):
return "object"
[docs]class EmptyRelationshipParameterValueModel(MakeRelationshipOnTheFlyMixin, EmptyParameterValueModel):
"""An empty relationship parameter_value model."""
[docs] _add_entities_on_the_fly = True
@property
[docs] def entity_class_type(self):
return "relationship_class"
@property
[docs] def entity_type(self):
return "relationship"
[docs] def add_items_to_db(self, db_map_data):
"""See base class."""
# Call the super method to add whatever is ready.
# This will fill the relationship_class_name as a side effect
super().add_items_to_db(db_map_data)
# Now we try to add relationships
self.build_lookup_dictionaries(db_map_data)
db_map_relationships = dict()
db_map_error_log = dict()
for db_map, items in db_map_data.items():
for item in items:
relationship, err = self._make_relationship_on_the_fly(item, db_map)
if relationship:
db_map_relationships.setdefault(db_map, []).append(relationship)
if err:
db_map_error_log.setdefault(db_map, []).extend(err)
if any(db_map_relationships.values()):
self.db_mngr.add_relationships(db_map_relationships)
# Something might have become ready after adding the relationship(s), so we do one more pass
super().add_items_to_db(db_map_data)
if db_map_error_log:
self.db_mngr.error_msg.emit(db_map_error_log)