wolfhece.Results2DGPU ===================== .. py:module:: wolfhece.Results2DGPU .. autoapi-nested-parse:: Author: HECE - University of Liege, Pierre Archambeau Date: 2024 Copyright (c) 2024 University of Liege. All rights reserved. This script and its content are protected by copyright law. Unauthorized copying or distribution of this file, via any medium, is strictly prohibited. Module Contents --------------- .. py:function:: _load_res(x) -> tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray] .. py:function:: _load_res_h(x) -> tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray] .. py:class:: Cache_Results2DGPU(fname: str, start_idx: int, end_idx: int = -1, only_h=False, every: int = 1, mode: Literal['TILED', 'UNTILED'] = 'TILED', memory_max_size: int = 12 * 1024 * 1024 * 1024) Gestion en mémoire de plusieurs résultats GPU Stockage CSR afin d'économiser la mémoire (Scipy CSR) ou Numpy array dense .. py:attribute:: _mode :value: 'TILED' .. py:attribute:: _results :type: Union[dict[str, tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray]], dict[str, numpy.ndarray]] .. py:attribute:: _result_store .. py:attribute:: _only_h :value: False .. py:attribute:: memory_max_size :value: 12884901888 .. py:attribute:: _maximum_n_results .. py:method:: _estimate_memory_size_one_result() Read one result to estimate memory size .. py:property:: memory_size :type: int Estimation de la taille mémoire des résultats en cache :return: taille mémoire en Mega-octets .. py:property:: only_h .. py:property:: _tile_packer Retourne le tile packer de la simulation .. py:property:: list_cached :type: list[int] Retourne la liste des indices des résultats en cache :return: liste des indices (1-based) .. py:method:: check_if_cached(idx: int) -> bool Vérifie si le résultat idx est dans le cache :param idx: index du résultat (1-based) :return: True si le résultat est dans le cache, False sinon .. py:method:: get_h(idx: int, dense: bool = True) -> Union[numpy.ndarray, scipy.sparse.csr_array] Retourne la matrice de hauteur d'eau de la position idx (0-based) - en CSR (Scipy CSR) - en dense (Numpy array) :param idx: index du résultat (1-based) :param dense: si True, retourne un Numpy array dense, sinon retourne un Scipy CSR array .. py:method:: get_qx(idx: int, dense: bool = True) -> Union[numpy.ndarray, scipy.sparse.csr_array] Retourne la matrice de débit X d'eau de la position idx (0-based) - en CSR (Scipy CSR) - en dense (Numpy array) :param idx: index du résultat (1-based) :param dense: si True, retourne un Numpy array dense, sinon retourne un Scipy CSR array .. py:method:: get_qy(idx: int, dense: bool = True) -> Union[numpy.ndarray, scipy.sparse.csr_array] Retourne la matrice de débit Y d'eau de la position idx (0-based) - en CSR (Scipy CSR) - en dense (Numpy array) :param idx: index du résultat (1-based) :param dense: si True, retourne un Numpy array dense, sinon retourne un Scipy CSR array .. py:class:: wolfres2DGPU(fname: str, eps=0.0, idx: str = '', plotted: bool = True, mapviewer=None, store=None, load_from_cache: bool = True) Bases: :py:obj:`wolfhece.wolfresults_2D.Wolfresults_2D` .. autoapi-inheritance-diagram:: wolfhece.Results2DGPU.wolfres2DGPU :parts: 1 :private-bases: Gestion des résultats du code GPU 2D Surcharge de "Wolfresults_2D" .. py:attribute:: _nap :value: None .. py:attribute:: _result_store :value: None .. py:attribute:: _cache :value: None .. py:method:: setup_store(store=None) Setup results store .. py:method:: setup_cache(start_idx: int = 0, end_idx: int = -1, only_h: bool = False) Setup cache from start_idx result to end_idx result if only_h is True, only waterdepth is loaded into memory :param start_idx: start index (0-based) :param end_idx: end index (0-based) :param only_h: only waterdepth is loaded into memory .. py:method:: clear_cache() Clear cache .. py:method:: _loader(fname: str) -> int .. py:method:: get_nbresults(force_update_timessteps=True) Récupération du nombre de résultats Lecture du fichier de tracking afin de permettre une mise à jour en cours de calcul .. py:method:: danger_map_gpu_tiled(start: int = 0, end: int = -1, every: int = 1, callback=None, hmin: float = None) -> tuple[wolfhece.wolf_array.WolfArray, wolfhece.wolf_array.WolfArray, wolfhece.wolf_array.WolfArray, wolfhece.wolf_array.WolfArray, wolfhece.wolf_array.WolfArray, wolfhece.wolf_array.WolfArray, wolfhece.wolf_array.WolfArray, wolfhece.wolf_array.WolfArray, wolfhece.wolf_array.WolfArray] Create Danger Maps without untiling GPU arrays. For better performance, use tiled results directly from the GPU simulation. The array is most compact in memory and the operations are faster. Returned arrays are untiled WolfArray for easier manipulation. Take care that Tile Packer untile arrays in transposed mode. So we need to transpose the arrays to store them in WolfArray objects. Returned arrays are: - Maximum water depth H - Maximum velocity U_norm - Maximum momentum Q_norm - Maximum water level Z - Maximum total head Head - Time of arrival Time_of_arrival - Time of maximum Time_of_maximum - Duration of inundation Duration_of_inundation - Time of ending Time_of_ending :param start: start time step - 0-based :param end: end time step - 0-based :param every: step interval :param callback: optional callback to update progress :param hmin: minimum water depth to consider for time of arrival. If None, uses epsilon value. :return : tuple of WolfArray or WolfArrayMB - H, U_norm, Q_norm, Z, Head, Time_of_arrival, Time_of_maximum, Duration_of_inundation, Time_of_ending .. py:method:: _read_oneresult_tiled(which: int = -1) Read one result in tiled mode :param which: result number to read; 0-based; -1 == last one .. py:method:: read_oneresult(which: int = -1) Lecture d'un pas de sauvegarde which: result number to read; 0-based; -1 == last one .. py:method:: _read_oneresult_only_h(which: int = -1) Lecture d'un pas de sauvegarde which: result number to read; 0-based; -1 == last one .. py:method:: read_oneresult_subarray(which: int = -1, bounds: list[list[float, float], list[float, float]] | wolfhece.wolfresults_2D.vector | tuple[tuple[float, float], tuple[float, float]] = None, nullify_all_outside: bool = True, border: float = None) -> None Read one result but only in a subarray footprint defined by bounds which: result number to read; 0-based; -1 == last one .. py:method:: read_oneresult_subarrays(which: int = -1, vectors: list[wolfres2DGPU.read_oneresult_subarrays.vector] = None, nullify_all_outside: bool = True, border: float = None) -> None Read one result but only in subarray footprints defined by vectors which: result number to read; 0-based; -1 == last one .. py:method:: get_packed_indices_from_xy(x: float, y: float) Get the packed tile indices from real world coordinates x,y. :param x: real world X coordinate :param y: real world Y coordinate .. py:method:: get_packed_indices_from_ij(i: int, j: int) Get the packed tile indices from array indices i,j. :param i: array index along X (0-based) :param j: array index along Y (0-based) .. py:method:: get_packed_indices_from_ijs(ijs: tuple[numpy.ndarray]) Get the packed tile indices from array indices i,j. :param ijs: tuple of two np.ndarray (i_array, j_array), like np.where(...) output .. py:method:: _update_result_view() Procédure interne de mise à jour du pas Etapes partagées par read_next et read_previous .. py:method:: get_times_steps(nb: int = None) Récupération des temps réels et les pas de calcul de chaque résultat sur disque :param nb : nombre de résultats à lire .. py:property:: all_dt .. py:property:: all_mostly_dry_mesh .. py:property:: all_clock_time .. py:property:: all_wet_meshes .. py:method:: get_cached_h(idx) Return cached water depth according to WOLF convention .. py:method:: get_cached_qx(idx) Return cached specific discharge along X according to WOLF convention .. py:method:: get_cached_qy(idx) Return cached specific discharge along Y according to WOLF convention .. py:method:: show_tiles() Show tiles in mapviewer .. py:method:: set_hqxqy_as_initial_conditions(idx: int = None, as_multiblocks: bool = False) Set the result as IC :param idx : 0-based index .. py:method:: get_hydrographs(vect: Union[wolfhece.wolfresults_2D.vector, list[wolfhece.wolfresults_2D.vector], wolfhece.wolfresults_2D.zone], progress_callback=None, i_start: int = 0, i_end: int = -1, i_step: int = 1, to_rasterize: bool = True) -> tuple[list[float], Union[list[float], dict[str, list[float]]]] Get hydrograph across a vector or list of vectors or a zone. If you provide rastrized vectors, set to_rasterize=False for better performance. The returned hydrographs are lists of discharge values at each time step. The first returned value is the array of time values. The second returned value is either a list of discharge values (if vect is a single vector) or a dictionary of lists of discharge values (if vect is a list of vectors or a zone). The keys of the dictionary are the vectors objects. :param vect: wolf polyline or list of wolf polylines or zone :param progress_callback: optional callback to update progress :param i_start: start index (0-based) :param i_end: end index (0-based), -1 for last :param i_step: step index :param to_rasterize: force to rasterize the vector(s) along the grid