from ...conesFunctions import printCones, mergeClips, is_cell_set
from ..conesClip import conesClip
from ...conesErrors import conesPathNotFound
import numpy as np
import os
import sys
[docs]
class conesSettingsAbstract:
"""
A class that groups settings for cones
"""
def __init__(self, case_path, model, nens, n_model_procs):
""" Initialize the class
:param case_path: the path of the source OpenFOAM case
:type case_path: str
:param model: the coupled model, 'OF' (OpenFOAM) or 'MNH' (Meso-NH)
:type model: str
:param nens: the number of ensemble members
:type nens: int
:param n_model_procs: the number of processors used by each model simulation
:type n_model_procs: int
"""
self.model = str(model)
self.verbose = True
self.case_path = os.getcwd() + "/" + case_path
self.checkPath(self.case_path)
self.nens = nens
if int(n_model_procs) <= 0:
raise ValueError(
f"--nmodelProcs must be a positive integer, got {n_model_procs}. "
"Check that this option is passed when you launch cones"
)
self.nRanks = n_model_procs
self.obs_file = str()
self.global_cells = list()
self.local_cells = list()
self.globalStateCells = list()
self.localStateCells = list()
self.stateVar = str()
self.obs_var = str()
self.nParams = int()
self.stateEstSwitch = bool()
self.paramEstSwitch = bool()
self.inflationType = str()
self.stateInflation = float()
self.parametersInflation = float()
self.stateCovarianceLocalisation = bool()
self.localisationLength = float()
self.Clips = list()
self.mergedClips = list()
self.regions = list()
self.obsWindow = int()
self.endTimeSim = float()
self.deltaTSim = float()
self.total_it = int()
self.mda_iterations = int()
[docs]
def checkPath(self, path):
"""
Check if path exists
:param path: The file directory
:type path: str
:returns: Boolean
:rtype: bool
"""
printCones("Checking ", path)
if not os.path.exists(path):
try:
raise conesPathNotFound(path)
except conesPathNotFound as e:
print(e, file=sys.stderr)
sys.exit(1)
printCones("Found!")
return True
[docs]
def mergeClips(self):
"""
Given a list of conesClip, finds clippings that shares cells and merge them
:returns: the list of conesClip after merge
:rtype: list
"""
results = mergeClips(self.Clips, self.global_cells)
clipList = []
# Mapping of global cell id to rank
cell_to_rank_local = {}
for rank, cellList in enumerate(self.global_cells):
for localID, cell in enumerate(cellList):
cell_to_rank_local[cell] = (rank, localID)
for i, newClip in enumerate(results):
clipGlobalList = [[] for _ in range(self.nRanks)]
clipLocalList = [[] for _ in range(self.nRanks)]
for cell in newClip:
rankCell, localID = cell_to_rank_local[cell]
clipGlobalList[rankCell].append(cell)
clipLocalList[rankCell].append(localID)
c = conesClip(f"newClip{i}")
c.globalCellList = clipGlobalList
c.localCellList = clipLocalList
clipList.append(c)
return clipList
[docs]
def get_clippings(self):
"""
Function that reads the clippings in polyMesh/sets/
Returns a list of clippings
:returns: list of clippings
:rtype: list
"""
path = self.case_path + "constant/polyMesh/sets/"
flist = sorted(os.listdir(path))
clipList = []
for ifile, file in enumerate(flist):
if not is_cell_set(os.path.join(path,file)):
if self.verbose:
printCones(file, "is not a cellSet, skipping")
continue
clipList.append(conesClip(file))
clipList[ifile].path = self.case_path
clipList[ifile].nRanks = self.nRanks
clipList[ifile].get_clip_data(self.global_cells)
if (self.verbose):
printCones("found the following clippings:")
if (self.verbose):
for ii, clip in enumerate(clipList):
printCones(clip.name)
printCones(ii, "Global", clip.globalCellList)
printCones(ii, "Local", clip.localCellList)
return clipList
[docs]
def setGlobalStateCells(self):
"""
Sets a list of global cell ids from which state variables are extracted
:returns: None
"""
stateCellsList = []
for irank in range(0, self.nRanks):
rankCellList = []
for region in self.regions:
rankCellList.append(region.globalCells[irank])
rankCellList = [cell for cellRank in rankCellList for cell in cellRank]
stateCellsList.append(sorted(rankCellList))
self.globalStateCells = stateCellsList
return
[docs]
def setLocalStateCells(self):
"""
Sets a list of local (rank) cell ids belonging from which state variables are extracted
:returns: None
"""
self.localStateCells = []
for irank in range(0, self.nRanks):
idx = np.where(np.isin(self.global_cells[irank], self.globalStateCells[irank]))[0]
self.localStateCells.append(idx.tolist())
return
[docs]
def setGlobalCells(self) -> list[list[int]]:
"""
Generate a list of global cell ids splitted by ranks
:returns: list of global cells ids
:rtype: list
"""
globalCellList = []
for irank in range(0, self.nRanks):
cellList = []
# Be sure that files are not binary written!
fpath = self.case_path + 'processor'+str(irank)+'/constant/polyMesh/cellProcAddressing'
with open(fpath) as file:
lines = file.readlines()
nCells = int(lines[17].strip())
for icell in range(0, nCells):
cellList.append(int(lines[19+icell].strip()))
globalCellList.append(cellList)
if (self.verbose):
printCones("Global Cell List :", globalCellList)
return globalCellList
[docs]
def setLocalCells(self):
"""
Generate a list of local cells ids splitted by ranks
:returns: list of local cells ids
:rtype: list
"""
return [list(range(len(cellList))) for cellList in self.global_cells]
[docs]
def report_regions(self):
"""
Prints to stdout reports of conesRegions
"""
for region in self.regions:
region.report()