# HydraulicStructures.py -- Pamhyr # Copyright (C) 2023-2024 INRAE # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU 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 General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . # -*- coding: utf-8 -*- import logging from tools import trace, timer from Model.Tools.PamhyrDB import SQLSubModel from Model.Scenario import Scenario from Model.Except import NotImplementedMethodeError from Model.HydraulicStructures.Basic.Value import ( BHSValue ) logger = logging.getLogger() class BasicHS(SQLSubModel): _sub_classes = [ BHSValue, ] def __init__(self, id: int = -1, name: str = "", status=None, owner_scenario=-1): super(BasicHS, self).__init__( id=id, status=status, owner_scenario=owner_scenario ) self._name = name self._type = "" self._enabled = True self._data = [] @classmethod def _db_create(cls, execute, ext=""): execute(f""" CREATE TABLE hydraulic_structures_basic{ext} ( {cls.create_db_add_pamhyr_id()}, deleted BOOLEAN NOT NULL DEFAULT FALSE, name TEXT NOT NULL, type TEXT NOT NULL, enabled BOOLEAN NOT NULL, hs INTEGER, {Scenario.create_db_add_scenario()}, {Scenario.create_db_add_scenario_fk()}, FOREIGN KEY(hs) REFERENCES hydraulic_structures(pamhyr_id), PRIMARY KEY(pamhyr_id, scenario) ) """) if ext == "_tmp": return True return cls._create_submodel(execute) @classmethod def _db_update(cls, execute, version, data=None): major, minor, release = version.strip().split(".") if major == minor == "0": if int(release) < 6: cls._db_create(execute) return True else: cls._db_update_to_0_1_0(execute, data) if major == "0" and minor == "1": if release < 2: execute( "ALTER TABLE hydraulic_structures_basic " + "ADD COLUMN deleted BOOLEAN NOT NULL DEFAULT FALSE" ) return cls._update_submodel(execute, version, data) @classmethod def _db_update_to_0_1_0(cls, execute, data): table = "hydraulic_structures_basic" hs = data['id2pid']['hydraulic_structures'] cls.update_db_add_pamhyr_id(execute, table, data) cls._db_update_to_0_1_0_set_hs_pid(execute, table, hs) Scenario.update_db_add_scenario(execute, table) cls._db_create(execute, ext="_tmp") execute( f"INSERT INTO {table}_tmp " + "(pamhyr_id, name, type, enabled, hs, scenario) " + "SELECT pamhyr_id, name, type, enabled, hs, scenario " + f"FROM {table}" ) execute(f"DROP TABLE {table}") execute(f"ALTER TABLE {table}_tmp RENAME TO {table}") @classmethod def _db_update_to_0_1_0_set_hs_pid(cls, execute, table, hs): els = execute( f"SELECT pamhyr_id, hs FROM {table}" ) for row in els: it = iter(row) pid = next(it) hs_id = next(it) execute( f"UPDATE {table} " + f"SET hs = {hs[hs_id]} " + f"WHERE pamhyr_id = {pid}" ) @classmethod def _get_ctor_from_type(cls, t): from Model.HydraulicStructures.Basic.Types import ( BHS_types, NotDefined, ) res = NotDefined if t in BHS_types.keys(): res = BHS_types[t] return res @classmethod def _db_load(cls, execute, data=None): new = [] status = data["status"] scenario = data["scenario"] loaded = data['loaded_pid'] if scenario is None: return [] table = execute( "SELECT pamhyr_id, deleted, name, type, enabled, hs, scenario " + "FROM hydraulic_structures_basic " + f"WHERE hs = {data['hs_id']} " + f"AND scenario = {scenario.id} " + f"AND pamhyr_id NOT IN ({', '.join(map(str, loaded))})" ) for row in table: it = iter(row) bhs_pid = next(it) deleted = (next(it) == 1) name = next(it) type = next(it) enabled = (next(it) == 1) hs_id = next(it) owner_scenario = next(it) ctor = cls._get_ctor_from_type(type) bhs = ctor( id=bhs_pid, name=name, status=status, owner_scenario=owner_scenario ) if deleted: bhs.set_as_deleted() bhs.enabled = enabled data['bhs_id'] = bhs_pid bhs._data = BHSValue._db_load( execute, data ) print(f"{bhs_pid} : {deleted}") if deleted: bhs.set_as_deleted() new.append(bhs) loaded.add(bhs_pid) data["scenario"] = scenario.parent new += cls._db_load(execute, data) data["scenario"] = scenario return new def _db_save(self, execute, data=None): if not self.must_be_saved(): return True hs_id = data['hs_id'] execute( "INSERT INTO " + "hydraulic_structures_basic(pamhyr_id, deleted, " + "name, type, enabled, hs, scenario) " + "VALUES (" + f"{self.pamhyr_id}, " + f"{self._db_format(self.is_deleted())}, " + f"'{self._db_format(self._name)}', " + f"'{self._db_format(self._type)}', " + f"{self._db_format(self.enabled)}, " + f"{hs_id}, " + f"{self._status.scenario_id}" ")" ) data['bhs_id'] = self.pamhyr_id execute( "DELETE FROM hydraulic_structures_basic_value " + f"WHERE bhs = {self.pamhyr_id} " + f"AND scenario = {self._status.scenario_id}" ) for values in self._data: values._db_save(execute, data) return True def __len__(self): return len(self._data) @property def name(self): if self._name == "": return f"{self._type} #{self.pamhyr_id}" return self._name @name.setter def name(self, name): self._name = name self.modified() @property def type(self): return self._type @type.setter def type(self, type): self._type = type self.modified() @property def enabled(self): return self._enabled @enabled.setter def enabled(self, enabled): self._enabled = enabled self.modified() @property def parameters(self): return self._data def convert(self, new_type): return new_type(id=self.id, name=self.name, status=self._status)