# BoundaryConditionsAdisTS.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, old_pamhyr_date_to_timestamp, date_iso_to_timestamp, date_dmy_to_timestamp, ) from Model.Tools.PamhyrDB import SQLSubModel from Model.Except import NotImplementedMethodeError logger = logging.getLogger() class BoundaryConditionAdisTS(SQLSubModel): _sub_classes = [] _id_cnt = 0 def __init__(self, id: int = -1, pollutant: int = -1, status=None): super(BoundaryConditionAdisTS, self).__init__() self._status = status if id == -1: self.id = BoundaryConditionAdisTS._id_cnt else: self.id = id self._type = "" self._node = None self._pollutant = pollutant self._data = [] self._header = [] self._types = [self.time_convert, float] BoundaryConditionAdisTS._id_cnt = max( BoundaryConditionAdisTS._id_cnt + 1, self.id) @classmethod def _db_create(cls, execute): execute(""" CREATE TABLE boundary_condition_adists( id INTEGER NOT NULL PRIMARY KEY, pollutant INTEGER NOT NULL, type TEXT NOT NULL, node INTEGER, FOREIGN KEY(pollutant) REFERENCES Pollutants(id), FOREIGN KEY(node) REFERENCES river_node(id) ) """) execute(""" CREATE TABLE boundary_condition_data_adists( id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, data0 TEXT NOT NULL, data1 TEXT NOT NULL, bc INTEGER, FOREIGN KEY(bc) REFERENCES boundary_condition_adists(id) ) """) return cls._create_submodel(execute) @classmethod def _db_update(cls, execute, version): return True @classmethod def _db_load(cls, execute, data=None): new = [] table = execute( "SELECT id, pollutant, type, node " + "FROM boundary_condition_adists" ) if table is not None: for row in table: bc = cls( id=row[0], pollutant=row[1], status=data['status'] ) bc.type = row[2] bc.node = None if row[3] != -1: tmp = next(filter( lambda n: n.id == row[3], data["nodes"]), None) if tmp is not None: bc.node = tmp.id else: bc.node = -1 values = execute( "SELECT data0, data1 FROM " + "boundary_condition_data_adists " + f"WHERE bc = '{bc.id}'" ) # Write data for v in values: data0 = bc._types[0](v[0]) data1 = bc._types[1](v[1]) # Replace data at pos ind bc._data.append((data0, data1)) new.append(bc) return new def _db_save(self, execute, data=None): execute(f"DELETE FROM boundary_condition_adists WHERE id = {self.id}") execute(f"DELETE FROM boundary_condition_data_adists" + f" WHERE bc = {self.id}") node = -1 if self._node is not None: node = self._node sql = ( "INSERT INTO " + "boundary_condition_adists(id, pollutant, type, node) " + "VALUES (" + f"{self.id}, {self._pollutant}, " + f"'{self._db_format(self._type)}', {node}" + ")" ) execute(sql) for d in self._data: data0 = self._db_format(str(d[0])) data1 = self._db_format(str(d[1])) sql = ( "INSERT INTO " + "boundary_condition_data_adists(data0, data1, bc) " + f"VALUES ('{data0}', {data1}, {self.id})" ) execute(sql) return True def __len__(self): return len(self._data) @classmethod def time_convert(cls, data): if type(data) is str: if data.count("-") == 2: return date_iso_to_timestamp(data) if data.count("/") == 2: return date_dmy_to_timestamp(data) if data.count(":") == 3: return old_pamhyr_date_to_timestamp(data) if data.count(":") == 2: return old_pamhyr_date_to_timestamp("00:" + data) if data.count(".") == 1: return round(float(data)) return int(data) @property def node(self): return self._node @node.setter def node(self, node): self._node = node self._status.modified() @property def header(self): return self._header.copy() @header.setter def header(self, header): self._header = header self._status.modified() @property def pollutant(self): return self._pollutant @pollutant.setter def pollutant(self, pollutant): self._pollutant = pollutant self._status.modified() @property def type(self): return self._type @type.setter def type(self, type): self._type = type self._status.modified() @property def data(self): return self._data.copy() @property def _default_0(self): return self._types[0](0) @property def _default_1(self): return self._types[1](0.0) def new_from_data(self, header, data): new_0 = self._default_0 new_1 = self._default_1 if len(header) != 0: for i in [0, 1]: for j in range(len(header)): if self._header[i] == header[j]: if i == 0: new_0 = self._types[i](data[j].replace(",", ".")) else: new_1 = self._types[i](data[j].replace(",", ".")) else: new_0 = self._types[0](data[0].replace(",", ".")) new_1 = self._types[1](data[1].replace(",", ".")) return (new_0, new_1) def add(self, index: int): value = (self._default_0, self._default_1) self._data.insert(index, value) self._status.modified() return value def insert(self, index: int, value): 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 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() def index(self, bc): self._data.index(bc) def get_i(self, index): return self.data[index] def get_range(self, _range): lst = [] for r in _range: lst.append(r) return lst def _set_i_c_v(self, index, column, value): v = list(self._data[index]) v[column] = self._types[column](value) self._data[index] = tuple(v) self._status.modified() def set_i_0(self, index: int, value): self._set_i_c_v(index, 0, value) def set_i_1(self, index: int, value): self._set_i_c_v(index, 1, value)