# 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 functools import reduce from tools import trace, timer, old_pamhyr_date_to_timestamp from Model.Tools.PamhyrDB import SQLSubModel from Model.Scenario import Scenario from Model.Except import NotImplementedMethodeError from Model.HydraulicStructures.Basic.HydraulicStructures import BasicHS from Model.HydraulicStructures.Basic.Types import ( NotDefined, ) logger = logging.getLogger() class HydraulicStructure(SQLSubModel): _sub_classes = [ BasicHS, ] _id_cnt = 0 def __init__(self, id: int = -1, name: str = "", status=None): super(HydraulicStructure, self).__init__(id) self._status = status self._name = name self._input_rk = None self._output_rk = None self._input_reach = None self._output_reach = None self._enabled = True self._data = [] @classmethod def _db_create(cls, execute, ext=""): execute(f""" CREATE TABLE hydraulic_structures{ext} ( {cls.create_db_add_pamhyr_id()}, name TEXT NOT NULL, enabled BOOLEAN NOT NULL, input_rk REAL NOT NULL, output_rk REAL NOT NULL, input_reach INTEGER, output_reach INTEGER, {Scenario.create_db_add_scenario()}, {Scenario.create_db_add_scenario_fk()}, FOREIGN KEY(input_reach) REFERENCES river_reach(pamhyr_id), FOREIGN KEY(output_reach) REFERENCES river_reach(pamhyr_id), PRIMARY KEY(pamhyr_id, scenario) ) """) return cls._create_submodel(execute) @classmethod def _db_update(cls, execute, version, data=None): major, minor, release = version.strip().split(".") if major == minor == "0": rl = int(release) if rl < 6: cls._db_create(execute) if rl < 11: execute( """ ALTER TABLE hydraulic_structures RENAME COLUMN input_kp TO input_rk """ ) execute( """ ALTER TABLE hydraulic_structures RENAME COLUMN output_kp TO output_rk """ ) cls._db_update_to_0_1_0(execute, data) return cls._update_submodel(execute, version, data) @classmethod def _db_update_to_0_1_0(cls, execute, data): table = "hydraulic_structures" cls.update_db_add_pamhyr_id(execute, table, data) Scenario.update_db_add_scenario(execute, table) cls._db_create(execute, ext="_tmp") execute( f"INSERT INTO {table}_tmp " + "(pamhyr_id, name, enabled, input_rk, output_rk, " + "input_reach, output_reach, scenario) " + "SELECT pamhyr_id, name, enabled, input_rk, output_rk, " + "input_reach, output_reach, scenario " + f"FROM {table}" ) execute(f"DROP TABLE {table}") execute(f"ALTER TABLE {table}_tmp RENAME TO {table}") @classmethod def _db_load(cls, execute, data=None): new = [] table = execute( "SELECT pamhyr_id, name, enabled, " + "input_rk, output_rk, " + "input_reach, output_reach " + "FROM hydraulic_structures" ) for row in table: it = iter(row) hs_id = next(it) name = next(it) enabled = (next(it) == 1) input_rk = next(it) output_rk = next(it) input_reach_id = next(it) output_reach_id = next(it) hs = cls( id=hs_id, name=name, status=data['status'] ) hs.enabled = enabled hs.input_rk = input_rk if input_rk != -1 else None hs.output_rk = output_rk if output_rk != -1 else None hs.input_reach, hs.output_reach = reduce( lambda acc, n: ( n if n.id == input_reach_id else acc[0], n if n.id == output_reach_id else acc[1] ), data["edges"], [None, None] ) data['hs_id'] = hs_id hs._data = BasicHS._db_load(execute, data) new.append(hs) return new def _db_save(self, execute, data=None): execute( "DELETE FROM hydraulic_structures " + f"WHERE pamhyr_id = {self.pamhyr_id}" ) input_reach_id = -1 if self._input_reach is not None: input_reach_id = self._input_reach.pamhyr_id output_reach_id = -1 if self._output_reach is not None: output_reach_id = self._output_reach.pamhyr_id input_rk = -1 if self.input_rk is not None: input_rk = self.input_rk output_rk = -1 if self.output_rk is not None: output_rk = self.output_rk execute( "INSERT INTO " + "hydraulic_structures(" + " pamhyr_id, name, enabled, input_rk, output_rk, " + " input_reach, output_reach" + ") " + "VALUES (" + f"{self.pamhyr_id}, '{self._db_format(self._name)}', " + f"{self._db_format(self.enabled)}, " + f"{input_rk}, {output_rk}, " + f"{input_reach_id}, {output_reach_id}" + ")" ) data['hs_id'] = self.pamhyr_id execute( "DELETE FROM hydraulic_structures_basic " + f"WHERE hs = {self.pamhyr_id}" ) for basic in self._data: basic._db_save(execute, data) return True def __len__(self): return len(self._data) @property def name(self): if self._name == "": return f"HS #{self.pamhyr_id}" return self._name @name.setter def name(self, name): self._name = name self._status.modified() @property def input_rk(self): return self._input_rk @input_rk.setter def input_rk(self, input_rk): if input_rk is not None: input_rk = float(input_rk) self._input_rk = input_rk self._status.modified() @property def output_rk(self): return self._output_rk @output_rk.setter def output_rk(self, output_rk): self._output_rk = output_rk self._status.modified() @property def enabled(self): return self._enabled @enabled.setter def enabled(self, enabled): self._enabled = enabled self._status.modified() @property def input_reach(self): return self._input_reach @input_reach.setter def input_reach(self, input_reach): self._input_reach = input_reach self._status.modified() @property def output_reach(self): return self._output_reach @output_reach.setter def output_reach(self, output_reach): self._output_reach = output_reach self._status.modified() @property def basic_structures(self): return self._data.copy() def basic_structure(self, index: int): return self._data[index] def add(self, index: int): value = NotDefined(status=self._status) self._data.insert(index, value) self._status.modified() return value def insert(self, index: int, value: BasicHS): self._data.insert(index, value) self._status.modified() def delete_i(self, indexes): self._data = list( map( lambda e: e[1], filter( lambda e: e[0] not in indexes, enumerate(self._data) ) ) ) self._status.modified() def delete(self, els): self._data = list( filter( lambda e: e not in els, self._data ) ) self._status.modified() def sort(self, _reverse=False, key=None): if key is None: self._data.sort(reverse=_reverse) else: self._data.sort(reverse=_reverse, key=key) self._status.modified()