diff --git a/hera/datalayer/collection.py b/hera/datalayer/collection.py index f62bdccd7..0182d579f 100644 --- a/hera/datalayer/collection.py +++ b/hera/datalayer/collection.py @@ -168,17 +168,6 @@ def addDocument(self,projectName,resource="",dataFormat="string",type="",desc={} "or one of the fields type is not proper. %s " % str(e)) return obj - def addDocumentFromJSON(self, json_data): - """ - Adds a document from a JSON string representation. - - Parameters - ---------- - json_data : str - A JSON string representing the document. - """ - self._metadataCol.from_json(json_data).save() - def deleteDocuments(self, projectName, **query): """ Deletes documents that satisfy the given query. diff --git a/hera/measurements/GIS/raster/landcover.py b/hera/measurements/GIS/raster/landcover.py index 3215bd9f6..28451eab8 100644 --- a/hera/measurements/GIS/raster/landcover.py +++ b/hera/measurements/GIS/raster/landcover.py @@ -485,43 +485,6 @@ def getRoughness(self, minx, miny, maxx, maxy, dxdy=30, inputCRS=WSG84, dataSour ) return landcover - def _handleType1(self, landcover): - """ - Converting land type of Type-1 to roughness. - Based on the paper: - * https://wes.copernicus.org/articles/6/1379/2021/ table a2 - * https://doi.org/10.5194/wes-6-1379-2021 Satellite-based estimation of roughness lengths and displacement heights for wind resource modelling, Rogier Floors, Merete Badger, Ib Troen, Kenneth Grogan, and Finn-Hendrik Permien - - Parameters - ---------- - landcover : int - Landcover type value. - - Returns - ------- - float - """ - roughnessDict = { - 0: 0.0001, # Water - 1: 1, # Evergreen needleleaf forest - 2: 1, # Evergreen broadleaf forest - 3: 1, # Deciduous needleleaf forest - 4: 1, # Deciduous broadleaf forest - 5: 1, # Mixed forests - 6: 0.05, # Closed shrubland - 7: 0.06, # Open shrublands - 8: 0.05, # Woody savannas - 9: 0.15, # Savannas - 10: 0.12, # Grasslands - 11: 0.3, # Permanent wetlands - 12: 0.15, # Croplands - 13: 0.8, # Urban and built-up - 14: 0.14, # Cropland/natural vegetation mosaic - 15: 0.001, # Snow and ice - 16: 0.01 # Barren or sparsely vegetated - } - return roughnessDict.get(landcover, 0.05) - def getCodingMap(self, datasourceName): """ Returns dictionary that maps landcover int value to string of landcover. @@ -557,35 +520,6 @@ def getCodingMap(self, datasourceName): } return {} - @staticmethod - def roughnesslength2sandgrainroughness(rl): - """ - Converts roughness length to equivalent sand grain roughness. - - Based on: - Desmond, C. J., Watson, S. J., & Hancock, P. E. (2017). - Modelling the wind energy resource in complex terrain and atmospheres. - Numerical simulation and wind tunnel investigation of non-neutral forest canopy flow. - Journal of wind engineering and industrial aerodynamics, 166, 48-60. - https://www.sciencedirect.com/science/article/pii/S0167610516300083#bib12 - - Equation 5: Equivalent sand grain roughness (m) is z0 * 30 - - We can use it for "nutkRoughWallFunction" boundary condition for Ks (sand grain roughness) parameter. - Cs value can be set as 0.5. - - Parameters - ---------- - rl : float - Roughness length. - - Returns - ------- - float - Equivalent sand grain roughness (Ks). - """ - return rl * 30.0 # return Ks value - def _getUrbanRoughnessFromLandCover(self, landcover, windMeteorologicalDirection, resolution, dataSourceName, GIS_BUILDINGS_dataSourceName): """ Add Roughness for Urban areas to landcover Xarray. diff --git a/hera/measurements/GIS/raster/tiles.py b/hera/measurements/GIS/raster/tiles.py index 93e0396c6..58ca4a8cf 100644 --- a/hera/measurements/GIS/raster/tiles.py +++ b/hera/measurements/GIS/raster/tiles.py @@ -50,27 +50,6 @@ def __init__(self, projectName, filesDirectory=None,connectionName=None): self._presentation = presentation(dataLayer=self) - def tileScaleAtLatLonZoom(self,latitude,longitude,zoomlevel): - """ - Returns the scale of a tile im meters at the location and zoom level - - Parameters - ---------- - latitude : float - The latitude in WGS84 - - longitude : float - The longitude in WGS - - zoomlevel : int - The zoom to retrieve. usually up to ~19. (highest). - - Returns - ------- - float - """ - return self.Z0RES / (2 ** zoomlevel) * numpy.cos(numpy.deg2rad(latitude)) - def getImageFromCorners(self, minx, miny, maxx, maxy, zoomlevel, tileServer=None, inputCRS=WSG84, outputCRS=WSG84): """ Gets the image from the lower left corner and upper right cornet - [left,right,bottom,top] in the coordinate system of the outputCRS. @@ -258,20 +237,6 @@ def deg2tile(self, lat_deg, lon_deg, zoom): ytile = int((1.0 - math.asinh(math.tan(lat_rad)) / math.pi) / 2.0 * n) return (xtile, ytile) - def listImages(self,**filters): - """Return stored tile image documents matching the given filters.""" - return self.getMeasurementsDocuments(type=self.doctype, **filters) - - def setDefaultTileServer(self,server): - """Set the default tile server URL in the project config. - - Parameters - ---------- - server : str - URL template for the tile server. - """ - self.setConfig(**{"defaultTileServer": server}) - class presentation: """ diff --git a/hera/measurements/GIS/vector/buildings/toolkit.py b/hera/measurements/GIS/vector/buildings/toolkit.py index e123241e7..cfa7c869d 100644 --- a/hera/measurements/GIS/vector/buildings/toolkit.py +++ b/hera/measurements/GIS/vector/buildings/toolkit.py @@ -197,97 +197,4 @@ def getBuildingsFromRectangle(self, minx, miny, maxx, maxy, dataSourceName=None, if withElevation: buildings = self.getBuildingHeightFromRasterTopographyToolkit(buildings) - return buildings - @staticmethod - def get_buildings_height(gdf): - """ - Extract building names, geometries(coordination), and height information from a GeoDataFrame. - if there is no height available - it will calculate the number of levels in the building * 3/ - else: none - - Parameters: - gdf (GeoDataFrame): A GeoPandas DataFrame containing building geometries - and associated properties. - - Returns: - geopandas.DataFrame - - """ - building_info = [] - - for index, row in gdf.iterrows(): - name = row.get('name', 'Unnamed') # Extract the building name - - # Retain the original geometry - geometry = row.geometry - - # Try to get the height or number of levels - height = row.get('height') - levels = row.get('building:levels') - - # Determine the height value - if height is None and levels is not None: - height = levels * 3 # Estimate height based on number of levels - - # Append the building information - building_info.append({ - 'name': name, - 'geometry': geometry, # Keep the original geometry - 'height': height - }) - - return gpd.GeoDataFrame(building_info) - @staticmethod - def filter_buildings_in_area(buildings_data, min_longitude, min_latitude, max_longitude, max_latitude): - """ - Filter building features by a specified geographic area using a bounding box. - - Parameters: - ---------- - buildings_data : dict - A GeoJSON-like dictionary containing building features to filter. - - min_longitude : float - Minimum longitude defining the bounding box. - - min_latitude : float - Minimum latitude defining the bounding box. - - max_longitude : float - Maximum longitude defining the bounding box. - - max_latitude : float - Maximum latitude defining the bounding box. - - Returns: - ------- - gpd.GeoDataFrame - A GeoDataFrame containing buildings that are located within the specified area. - """ - # Create a polygon from the bounding box - bbox_polygon = Polygon([(min_longitude, min_latitude), - (max_longitude, min_latitude), - (max_longitude, max_latitude), - (min_longitude, max_latitude), - (min_longitude, min_latitude)]) # Close the polygon - - # List to hold filtered building data - filtered_buildings = [] - - for feature in buildings_data['features']: - geometry = feature['geometry'] - properties = feature['properties'] - - # Create a geometry shape from the geometry data - geom = shape(geometry) # Handles Point, Polygon, MultiPolygon, etc. - - # Check if the geometry intersects with the bounding box polygon - if geom.intersects(bbox_polygon): - # Add all properties and the geometry - properties['geometry'] = geom - filtered_buildings.append(properties) - - # Create a GeoDataFrame - gdf = gpd.GeoDataFrame(filtered_buildings) - - return gdf \ No newline at end of file + return buildings \ No newline at end of file diff --git a/hera/measurements/GIS/vector/toolkit.py b/hera/measurements/GIS/vector/toolkit.py index 1246027c5..09b1f4b40 100644 --- a/hera/measurements/GIS/vector/toolkit.py +++ b/hera/measurements/GIS/vector/toolkit.py @@ -1,4 +1,3 @@ -import io from hera import toolkit import geopandas from shapely.geometry import Polygon, box @@ -6,9 +5,6 @@ from ....utils.logging import get_classMethod_logger -TOOLKIT_VECTOR_REGIONNAME = "regionName" - - class VectorToolkit(toolkit.abstractToolkit): """Base toolkit for vector GIS data operations.""" @@ -34,28 +30,6 @@ def __init__(self, projectName, toolkitName = 'VectorToolkit', filesDirectory=No """ super().__init__(projectName=projectName,toolkitName=toolkitName,filesDirectory=filesDirectory,connectionName=connectionName) - @staticmethod - def geopandasToGeoJson(geoData): - """Convert a GeoDataFrame to a GeoJSON string. - - Parameters - ---------- - geoData : geopandas.GeoDataFrame - The GeoDataFrame to convert. - - Returns - ------- - str - GeoJSON string representation. - """ - if isinstance(geoData,geopandas.GeoDataFrame): - dataHandler = io.BytesIO() - geoData.to_file(dataHandler,driver='GeoJSON') - return dataHandler.getvalue().decode('ascii') - - else: - raise ValueError("Function receives only GeoDataFrame") - def _RegionToGeopandas(self, regionData, crs = None): """ Converts a shape to geopandas. diff --git a/hera/measurements/GIS/vector/topography.py b/hera/measurements/GIS/vector/topography.py index 14041d337..ff515b00f 100644 --- a/hera/measurements/GIS/vector/topography.py +++ b/hera/measurements/GIS/vector/topography.py @@ -97,22 +97,6 @@ def cutRegionFromSource(self, shapeDataOrName, datasourceName, isBounds = False, - def geoPandasToSTL(self,gpandas, dxdy=50, solidName="Topography"): - """ - Transforsm the gpandas to STL. - - Parameters - ---------- - gpandas - dxdy - solidName - - Returns - ------- - - """ - return self.stlFactory.vectorToSTL(gpandas, dxdy=dxdy, solidName="Topography") - def regionToSTL(self, shapeDataOrName, dxdy, datasourceName, crs=None): """ Converts a region in a vector height map (contours) to STL at requested resolution diff --git a/hera/measurements/experiment/analysis.py b/hera/measurements/experiment/analysis.py index 323eefbdc..0fee7a02a 100644 --- a/hera/measurements/experiment/analysis.py +++ b/hera/measurements/experiment/analysis.py @@ -26,64 +26,6 @@ def __init__(self,datalayer): """ self._datalayer = datalayer - def getDeviceLocations(self,entityTypeName,trialName,trialSetName=None): - """ - Returns a pandas with the device locations. - - Parameters - ---------- - entityTypeName: str - The entity type. - - trialName: str - The trial name. - - trialSetName: str - The trial set name. - - - Returns - ------- - pandas.DataFrame - """ - trialSetName = self.datalayer.defaultTrialSet if trialSetName is None else trialSetName - if trialSetName is None: - raise ValueError("trialSetName is None. Either set default with the getExperiment, or set explicitly here") - return self.datalayer.trialSet[trialSetName][trialName].entitiesTable().query("entityType==@entityTypeName") - - - def getTurbulenceStatistics(self,sonicData,samplingWindow,height=1): - """ - Returns the turbulence analysis of the sonic/kaijo data for the start->end times. - - Parameters - ---------- - sonicData: pandas/Dask - the data of the sonic to analyze. - - samplingWindow: str - The window width for analysis - - height: int - Height - - Returns - ------- - singlePointTurbulenceStatistics class - """ - - highfreqtk = self.datalayer.toolkitExtension.sonicHighFreqToolkit - analysis = highfreqtk.analysis.singlePointTurbulenceStatistics(sonicData=kaijoData, - start=None, - end=None, - height=kaijoHeight, - samplingWindow=samplingWindow, - buildingHeight= 0, - averagedHeight=0, - isMissingData=True) - - return analysis - def _splitName(self,x): """Extract the device identifier from a space-separated name string. @@ -216,9 +158,11 @@ def getDeviceTypeTransmissionFrequencyOfTrial(self, pvt = docList[0].getData() if normalize: - total_expected_messages = self.getDeviceTypePlannedMessageCount(deviceType=deviceType, - samplingWindow=samplingWindow) - pvt = pvt / total_expected_messages + raise NotImplementedError( + "Normalizing to the optimal sampling rate is not implemented " + "(depends on getDeviceTypePlannedMessageCount/getOptimalFrequencyHz, " + "which were removed as dead/broken code)." + ) if not wideFormat: @@ -229,33 +173,6 @@ def getDeviceTypeTransmissionFrequencyOfTrial(self, return ret - def getDeviceTypePlannedMessageCount(self,deviceType,samplingWindow="1min"): - """ - Returns the number of messages that should be obtained in a sampling window,according to the device type. - - Parameters - ---------- - deviceType : string - The type of the device to present. - - samplingWindow : string - A time string (of pandas). i.e. '1min' and ect. - - """ - - - - samplingWindow_seconds = pandas.to_timedelta(samplingWindow).total_seconds() - total_expected_messages = self.getOptimalFrequencyHz(deviceType)*samplingWindow_seconds - - if total_expected_messages<1: - raise ValueError(f"Sampling window is too short. Use sampling window of at least {1/self.getOptimalFrequencyHz(deviceType)}s") - - return total_expected_messages - - - - def addMetadata(self,dataset,trialName,trialSetName=None): """ diff --git a/hera/measurements/experiment/dataEngine.py b/hera/measurements/experiment/dataEngine.py index bf29cf7e9..342b82671 100644 --- a/hera/measurements/experiment/dataEngine.py +++ b/hera/measurements/experiment/dataEngine.py @@ -1,5 +1,3 @@ -import json -import pymongo import pandas from ... import datalayer from ...utils.logging import get_classMethod_logger @@ -35,370 +33,13 @@ def getDataEngine(self,projectName, datasourceConfiguration,experimentObj, dataT pandasDataEngineDB or parquetDataEngineHera or daskDataEngineDB """ - if dataType == PANDASDB: - return pandasDataEngineDB(projectName,datasourceConfiguration) - elif dataType == PARQUETHERA: + if dataType == PARQUETHERA: return parquetDataEngineHera(projectName,datasourceConfiguration,experimentObj) - elif dataType == DASKDB: - return daskDataEngineDB(projectName,datasourceConfiguration) + elif dataType in (PANDASDB, DASKDB): + raise NotImplementedError(f"Hera datalayer {dataType} is not implemeneted yet. ") else: raise NotImplementedError(f"Hera datalayer {dataType} is not implemeneted yet. ") -class pandasDataEngineDB: - """Data engine that retrieves experiment data from MongoDB using pandas.""" - - _mongo_client = None - _dbConfiguration = None - - @property - def DBconfiguration(self): - """Return the database configuration dictionary.""" - self._dbConfiguration - - def dbConnect(self): - """Establish a connection to the MongoDB server.""" - # print("mongodb://" + self._DB['login']['username'] + ":" + self._DB['login']['password'] + "@" + - # self._DB['login']['ip'] + ":" + self._DB['login']['port'] + "/") - try: - self._mongo_client = pymongo.MongoClient(f"mongodb://{self._dbConfiguration['login']['username']}:{self._dbConfiguration['login']['password']}@{self._dbConfiguration['login']['ip']}:{self._dbConfiguration['login']['port']}/") - try: - self._mongo_client.server_info() - except OperationFailure as e: - return e - except Exception as e: - return e - - def __init__(self, projectName,datasourceConfiguration): - """Initialize the pandas DB engine and connect to MongoDB. - - Parameters - ---------- - projectName : str - Name of the project. - datasourceConfiguration : dict - Configuration dictionary containing 'DB' connection details. - """ - if 'DB' not in datasourceConfiguration: - raise ValueError(f"The configuration file does not have 'DB' definitions\n Got: {json.dumps(datasourceConfiguration,indent=4)}") - - self._dbConfiguration = datasourceConfiguration['DB'] - self.dbConnect() - - def getDataFromTrial(self, deviceType, trialName, trialSet=None, deviceName=None, withMetadata=True, startTime=None, endTime=None): - """ - Return the device data from the set. Use the default trial set if it is None. - - deviceType: str - The device type KAIJO,PT1000,GC,NDIR - - trialName: str - The name of the trial - - trialSet: str [optional] - The trial set. Use default if None - - deviceName: str [optional] - Filter according to one specific Device. - - startTime: str [optional] - Use as start time if exists, otherwise use the trial start Time. - - endTime: str [optional] - Use as end time if exists, otherwise use the trial end Time. - - withMetadata: bool - If true adds the following Fields: - * TimeFromReleaseStart: - The time elasped from the release start - * TimeFromReleaseEnd: - The time elasped from the release end - * longitude - The position X of the device. - Use the trial state to determine the location - * latitute - The position Y of the device - Use the trial state to determine the location - - :return: pandas - Pandas with the data - - """ - trialSet = self.experimentObj.trialSet if trialSet is None else trialSet - - trial = self._experimentObj.experimentSetup.trialSet[trialSet][trialName] - startTime = trial.properties['TrialStart'] if startTime is None else startTime - endTime = trial.properties['TrialEnd'] if endTime is None else endTime - - data = self.getData(deviceType=deviceType, - deviceName=deviceName, - startTime=startTime, - endTime=endTime) - - if len(data) == 0: - raise ValueError(f"There is no data for {deviceType} between the dates {startTime} and {endTime}") - - if withMetadata: - devicemetadata = self.experimentObj.experimentSetup.trialSet[trialSet][trialName].entitiesTable() - if len(devicemetadata) > 0: - data = data.reset_index().merge(devicemetadata, left_on="deviceName", right_on="entityName").set_index( - "timestamp") - - return data - - def getData(self, deviceType, deviceName=None, startTime=None, endTime=None): - """Query MongoDB and return device data as a pandas DataFrame. - - Parameters - ---------- - deviceType : str - The device type (collection name). - deviceName : str, optional - Filter by specific device name. - startTime : datetime-like or float, optional - Start of the time range. - endTime : datetime-like or float, optional - End of the time range. - - Returns - ------- - pandas.DataFrame - """ - collectionList = [x['name'] for x in self._mongo_client[self._dbConfiguration['db_name']].list_collections()] - if deviceType not in collectionList: - raise ValueError(f"device type {deviceType} not found. Should be one of {','.join(collectionList)}") - - collection = self._mongo_client[self._dbConfiguration['db_name']][deviceType] - - full_qry = {"deviceName": deviceName} if deviceName is not None else {} - - timeFilter = {} - if startTime is not None: - if not isinstance(startTime, float): - try: - startTime = pandas.to_datetime(startTime).timestamp() * 1000 - except Exception: - raise ValueError(f"{startTime} is not a valid date") - - timeFilter['$gte'] = startTime - - if endTime is not None: - if not isinstance(endTime, float): - try: - endTime = pandas.to_datetime(endTime).timestamp() * 1000 - except Exception: - raise ValueError(f"{endTime} is not a valid date") - - timeFilter['$lte'] = endTime - - if len(timeFilter) > 0: - full_qry["timestamp"] = timeFilter - - ret = pandas.DataFrame(list(collection.find(full_qry))) - if not ret.empty: - # When the ts is saved in UTC time. use this line: - # ret.timestamp.apply(lambda x: pandas.to_datetime(x, unit="ms",utc=True).tz_convert("israel")) - ret.timestamp = ret.timestamp.apply(lambda x: pandas.to_datetime(x, unit="ms",utc=True).tz_convert("israel")) - ret = ret.set_index("timestamp") - - return ret - - def getDeviceList(self, device): - """Return the list of entities for the given device type. - - Parameters - ---------- - device : str - The device type name. - """ - devicesList = self._experimentObj.experimentSetup.entityType - return devicesList[device] - - def getDeviceTable(self, device): - """Return the entity type table for the given device type. - - Parameters - ---------- - device : str - The device type name. - """ - Table = self._experimentObj.experimentSetup.entityTypeTable - return Table[device] - -class daskDataEngineDB: - """Data engine that retrieves experiment data from MongoDB using dask.""" - - _mongo_client = None - _dbConfiguration = None - - @property - def DBconfiguration(self): - """Return the database configuration dictionary.""" - self._dbConfiguration - - @property - def connectionString(self): - """Return the MongoDB connection URI string.""" - return f"mongodb://{self._dbConfiguration['login']['username']}:{self._dbConfiguration['login']['password']}@{self._dbConfiguration['login']['ip']}:{self._dbConfiguration['login']['port']}/" - - def __init__(self, projectName, datasourceConfiguration): - """Initialize the dask DB engine with connection configuration. - - Parameters - ---------- - projectName : str - Name of the project. - datasourceConfiguration : dict - Configuration dictionary containing 'DB' connection details. - """ - - if 'DB' not in datasourceConfiguration: - raise ValueError(f"The configuration file does not have 'DB' definitions\n Got: {json.dumps(datasourceConfiguration,indent=4)}") - - self._dbConfiguration = datasourceConfiguration['DB'] - - def getDataFromTrial(self, deviceType, trialName, trialSet=None, deviceName=None, withMetadata=True, startTime=None, endTime=None): - """ - Return the device data from the set. Use the default trial set if it is None. - - deviceType: str - The device type KAIJO,PT1000,GC,NDIR - - trialName: str - The name of the trial - - trialSet: str [optional] - The trial set. Use default if None - - deviceName: str [optional] - Filter according to one specific Device. - - startTime: str [optional] - Use as start time if exists, otherwise use the trial start Time. - - endTime: str [optional] - Use as end time if exists, otherwise use the trial end Time. - - withMetadata: bool - If true adds the following Fields: - * TimeFromReleaseStart: - The time elasped from the release start - * TimeFromReleaseEnd: - The time elasped from the release end - * longitude - The position X of the device. - Use the trial state to determine the location - * latitute - The position Y of the device - Use the trial state to determine the location - - - - :return: pandas - Pandas with the data - - """ - trialSet = self.experimentObj.trialSet if trialSet is None else trialSet - - trial = self._experimentObj.experimentSetup.trialSet[trialSet][trialName] - startTime = trial.properties['TrialStart'] if startTime is None else startTime - endTime = trial.properties['TrialEnd'] if endTime is None else endTime - - data = self.getData(deviceType=deviceType, - deviceName=deviceName, - startTime=startTime, - endTime=endTime) - - if len(data) == 0: - raise ValueError(f"There is no data for {deviceType} between the dates {startTime} and {endTime}") - - if withMetadata: - devicemetadata = self.experimentObj.experimentSetup.trialSet[trialSet][trialName].entitiesTable() - if len(devicemetadata) > 0: - data = data.reset_index().merge(devicemetadata, left_on="deviceName", right_on="entityName").set_index( - "timestamp") - - return data - - def getData(self, deviceType, deviceName=None, startTime=None, endTime=None): - """Query MongoDB and return device data as a dask DataFrame. - - Parameters - ---------- - deviceType : str - The device type (collection name). - deviceName : str, optional - Filter by specific device name. - startTime : datetime-like or float, optional - Start of the time range. - endTime : datetime-like or float, optional - End of the time range. - - Returns - ------- - dask.DataFrame or None - """ - # collectionList = [x['name'] for x in self._mongo_client[self._dbConfiguration['db_name']].list_collections()] - # if deviceType not in collectionList: - # raise ValueError(f"device type {deviceType} not found. Should be one of {','.join(collectionList)}") - - full_qry = {"deviceName": deviceName} if deviceName is not None else {} - - timeFilter = {} - if startTime is not None: - if not isinstance(startTime, float): - try: - startTime = pandas.to_datetime(startTime).timestamp() * 1000 - except Exception: - raise ValueError(f"{startTime} is not a valid date") - - timeFilter['$gte'] = startTime - - if endTime is not None: - if not isinstance(endTime, float): - try: - endTime = pandas.to_datetime(endTime).timestamp() * 1000 - except Exception: - raise ValueError(f"{endTime} is not a valid date") - - timeFilter['$lte'] = endTime - - if len(timeFilter) > 0: - full_qry["timestamp"] = timeFilter - - try: - ret = dask_mongo.read_mongo(database="expradmin", - collection=deviceType, - connection_kwargs=dict(host=self.connectionString), - chunksize=10, - match=full_qry - ) - except StopIteration: - ret = None - - return ret - - def getDeviceList(self, device): - """Return the list of entities for the given device type. - - Parameters - ---------- - device : str - The device type name. - """ - devicesList = self._experimentObj.experimentSetup.entityType - return devicesList[device] - - def getDeviceTable(self, device): - """Return the entity type table for the given device type. - - Parameters - ---------- - device : str - The device type name. - """ - Table = self._experimentObj.experimentSetup.entityTypeTable - return Table[device] - class parquetDataEngineHera(datalayer.Project): """Data engine that retrieves experiment data from Hera parquet storage.""" diff --git a/hera/measurements/experiment/experiment.py b/hera/measurements/experiment/experiment.py index 260a89474..222ba6a8e 100644 --- a/hera/measurements/experiment/experiment.py +++ b/hera/measurements/experiment/experiment.py @@ -218,13 +218,6 @@ def __getitem__(self, item): """ return self.getExperiment(item) - def experimentDataType(self): - """ - Backward-compatibility hook for experiment data type. - """ - return getattr(self, "_experimentDataType", None) - - class experimentSetupWithData(argosDataObjects.ExperimentZipFile, toolkit.abstractToolkit): """ A class that unifies the argos.experiment setup with the data. @@ -379,25 +372,6 @@ def defaultTrialSet(self): """Return the name of the default trial set.""" return self._defaultTrialSetName - @property - def trialsOfDefaultTrialSet(self): - """Return the trials belonging to the default trial set.""" - return self.trialSet[self.defaultTrialSet] - - def _initAnalysisAndPresentation(self, analysisCLS, presentationCLS): - """ - Initialize the analysis and presentation classes and set the data layer. - - Parameters - ---------- - analysisCLS : class - The analysis class, recommended to inherit from .analysis.experimentAnalysis. - presentationCLS : class - The presentation class, recommended to inherit from .presentation.experimentPresentation. - """ - self._analysis = analysisCLS(self) - self._presentation = presentationCLS(self, self._analysis) - def getDataFromDateRange( self, deviceType, diff --git a/hera/measurements/experiment/parsers.py b/hera/measurements/experiment/parsers.py index 44cdb5827..81549866e 100644 --- a/hera/measurements/experiment/parsers.py +++ b/hera/measurements/experiment/parsers.py @@ -349,26 +349,6 @@ def getData(self, file, fromTime, toTime): return ts, cbi.columnsNames, retVal -class Parser_TOA5(object): - """Parser for Campbell Scientific TOA5 ASCII data files.""" - - def __init__(self): - """Initialize the TOA5 parser.""" - pass - - def parse(self, file): - """ - Parse a TOA5 file. - - Parameters - ---------- - file : str - Path to the TOA5 file. - """ - pass - - - ############################## Private ############################### class CampbellBinaryInterface(object): """Low-level interface for reading Campbell Scientific TOB1 binary files.""" diff --git a/hera/measurements/experiment/presentation.py b/hera/measurements/experiment/presentation.py index 523864bfe..b38a69ef5 100644 --- a/hera/measurements/experiment/presentation.py +++ b/hera/measurements/experiment/presentation.py @@ -3,11 +3,10 @@ from hera import toolkitHome # lazy singleton from hera.utils.lazy import _LazyModule -# numpy, pandas, jinja2 are deferred to the methods that use them. +# numpy, pandas are deferred to the methods that use them. numpy = _LazyModule("numpy") np = numpy pd = _LazyModule("pandas") -jinja2 = _LazyModule("jinja2") class experimentPresentation: """ @@ -216,37 +215,6 @@ def plotImage(self, imageName, ax=None, xlabel=True, ylabel=True, withGrid = Tru return ax - def plotMap(self,trialSetName, trialName): - """ - Plots the tile map (sattelite) of the devices in that region. - - NOT COMPLETE!!. - Parameters - ---------- - trialSetName - trialName - - Returns - ------- - - """ - import matplotlib.pyplot as plt - devices_df = self.trialSet[trialSetName][trialName].entitiesTable - - if ax is None: - plot_kwargs = plot_kwargs or {} - fig, ax = plt.subplots(1, 1, **plot_kwargs) - else: - fig = ax.figure - - tiles_tk = toolkitHome.getToolkit(toolkitHome.GIS_TILES, projectName=self.datalayer.projectName) - devices_df[['ITM_Latitude', 'ITM_Longitude']] = devices_df.apply(self.datalayer._process_row, axis=1) - minx, miny, maxx, maxy = self.datalayer.get_devices_image_coordinates(trialSetName, trialName, deviceType) - region = dict(minx=minx, maxx=maxx, maxy=maxy, miny=miny, zoomlevel=17, inputCRS=ITM, tileServer=toolkitDataSource) - img = tiles_tk.getImageFromCorners(**region) - - plot = tiles_tk.presentation.plot(img, ax=ax, display=True) - def _plotEntityLocationScatter(self, entityTypeName, trialSet, trialName, status, floorName, ax=None, plotNameMode=None, scatter_kw=dict()): """ @@ -446,56 +414,6 @@ def plotDevicesOnImage(self, trialSetName, trialName, deviceType,mapName, ax=Non return fig, ax - def plotDevices(self, trialSetName, trialName, deviceType,mapName, ax=None, plotkwargs=None): - """ - Plot map of devices type places in a specific trial set and trial on the requested map. - - When the map is the world map, the tiletoolkit is initialized and the self.tileDataSourceName is used instead. - The plotting of the map itself is perfomed in the plot map function. - Parameters - ---------- - trialSetName : str - Trial Set Name. - trialName: str - Trial Name. - deviceType: str - Device type name. - mapName : str - The name of the map to plot on. - plotkwargs: dict - Parameters for matplotlib.pyplot subplot. - - Returns - ------- - fig - ax - """ - import matplotlib.pyplot as plt - if ax is None: - plot_kwargs = plot_kwargs or {} - fig, ax = plt.subplots(1, 1, **plot_kwargs) - else: - fig = ax.figure - - devices_df = self.trialSet[trialSetName][trialName].entitiesTable.query("deviceTypeName==@deviceType") - - devices_df[['ITM_Latitude', 'ITM_Longitude']] = devices_df.apply(self.datalayer._process_row, axis=1) - minx,miny,maxx,maxy = self.datalayer.get_devices_image_coordinates(trialSetName,trialName,deviceType) - d = {} - for row in devices_df.itertuples(): - x = row.ITM_Latitude - y = row.ITM_Longitude - stationCount = d.get(row.stationName,0)+1 - d[row.stationName] = stationCount - num_of_devices_in_station = d[row.stationName] - delta = num_of_devices_in_station * 0.02 - - ax.scatter(x, y, color='red', marker='o', s=50) # 's' controls size - ax.text(x, y + (maxy - miny) * delta, f"{row.deviceItemName}", color='red', fontsize=20, ha='center', - bbox=dict(facecolor='white', edgecolor='none', alpha=0.8)) - - return fig, ax - ######################## ### ### Technical plots @@ -645,61 +563,6 @@ def plotDeviceTypeFunctionality(self, - def generateLatexTable(self, latex_template, folder_path): - """ - Save folder for overleaf website upload to transform to PDF. - - Parameters - ---------- - latex_template: str - Latex Template. - folder_path: str - Path to save folder - - Returns - ------- - """ - data = {} - data['trialSets'] = [] - os.makedirs(folder_path, exist_ok=True) - for trialSet in self.datalayer.setup['trialSets']: - trialSet_dict = {} - trialSet_dict['trialSet_name'] = trialSet['name'] - trialSet_dict['trials'] = [] - for trial in trialSet['trials']: - trial_dict = {} - trial_dict['trial_name'] = trial['name'] - devices_df = self.datalayer.trialSet['Measurements']['Measurements'].entitiesTable - trial_dict['devices'] = [] - for device_name in devices_df['deviceTypeName'].unique(): - device_dict = {} - fig , _ = self.plotDevices(trialSetName=trialSet['name'], trialName=trial['name'], device=device_name, display=False) - image_path = os.path.join(folder_path,f"{device_name}.png") - fig.savefig(image_path) - device_dict['device_name'] = device_name - device_dict['map_image_path'] = f"{device_name}.png" - device_dict['locations_table'] = [] - device_df = devices_df[devices_df['deviceTypeName'] == device_name] - for row in device_df.itertuples(): - location = {} - location["latitude"] = row.Latitude - location['longitude'] = row.Longitude - location['device_name'] = str(row.deviceItemName).replace("_", " ") - location['station'] = str(row.stationName).replace("_", " ") - device_dict['locations_table'].append(location) - trial_dict['devices'].append(device_dict) - trialSet_dict['trials'].append(trial_dict) - data['trialSets'].append(trialSet_dict) - - - template = jinja2.Template(latex_template) - latex_content = template.render(trialSets=data["trialSets"]) - tex_path = os.path.join(folder_path,f"latex_document.tex") - with open(tex_path, "w", encoding="utf-8") as file: - file.write(latex_content) - print(f"LaTeX document generated at: {tex_path}") - - # def plotNDIRFrequencyDistribution(self, # trialName, # trialSetName, diff --git a/hera/measurements/meteorology/highfreqdata/analysis/abstractcalculator.py b/hera/measurements/meteorology/highfreqdata/analysis/abstractcalculator.py index dd6c231c8..d0426e80e 100644 --- a/hera/measurements/meteorology/highfreqdata/analysis/abstractcalculator.py +++ b/hera/measurements/meteorology/highfreqdata/analysis/abstractcalculator.py @@ -58,11 +58,6 @@ def __init__(self, rawData, metadata): self._AllCalculatedParams = [] self._joinmethod = "left" - @property - def JoinMethod(self): - """str : Join method used when merging computed columns (default ``'left'``).""" - return self._joinmethod - @property def RawData(self): """pandas.DataFrame or dask.dataframe.DataFrame : The original raw data.""" diff --git a/hera/measurements/meteorology/highfreqdata/analysis/meandatacalculator.py b/hera/measurements/meteorology/highfreqdata/analysis/meandatacalculator.py index 3ec1ca6c8..cde5e3cfe 100644 --- a/hera/measurements/meteorology/highfreqdata/analysis/meandatacalculator.py +++ b/hera/measurements/meteorology/highfreqdata/analysis/meandatacalculator.py @@ -194,9 +194,6 @@ def horizontalSpeed(self): return self - def _UV_to_SpdDir(self,U, V): - return (U ** 2 + V ** 2) ** 0.5, (-numpy.degrees(numpy.arctan2(V, U)) + 90) % 360 - def alignedStress(self): """Rotate the Reynolds stress tensor to align with the mean wind direction. diff --git a/hera/measurements/meteorology/highfreqdata/analysis/turbulencestatistics.py b/hera/measurements/meteorology/highfreqdata/analysis/turbulencestatistics.py index 8f7e2ab85..409d5acce 100644 --- a/hera/measurements/meteorology/highfreqdata/analysis/turbulencestatistics.py +++ b/hera/measurements/meteorology/highfreqdata/analysis/turbulencestatistics.py @@ -1,5 +1,3 @@ -import os -import json import numpy import pandas import dask.dataframe @@ -1350,82 +1348,6 @@ def ThirdStrucFun(self, tau_range = None, ubar_data = None, u_bar = "u_bar", v_b return self -class SinglePointStatisticsSpark(singlePointTurbulenceStatistics): - """Spark-compatible single-point turbulence statistics calculator. - - Extends `singlePointTurbulenceStatistics` with a ``fluctuations`` - implementation that handles Spark-style repartitioning and an alternative - wind-direction convention. - """ - - def fluctuations(self, inMemory=None): - """Calculate mean values and fluctuations for u, v, w, T, and wind direction. - - This override adjusts the wind-direction computation to use a - [0, 360) degree convention and corrects the first raw-data index to - align with the averaged data. - - Parameters - ---------- - inMemory : InMemoryAvgData or None, optional - In-memory reference for averaged data storage. Default is ``None``. - - Returns - ------- - SinglePointStatisticsSpark - The instance itself, allowing method chaining. - """ - if self._InMemoryAvgRef is None: - self._InMemoryAvgRef = inMemory - - if 'up' not in self._RawData.columns: - avg = self._RawData - if self.SamplingWindow is None: - avg = avg.mean() - if self._DataType == 'pandas': - avg = pandas.DataFrame(avg).T - avg.index = [self._RawData.index[0]] - else: - avg = pandas.DataFrame(avg.compute()).T - avg.index = self._RawData.head(1).index - npartitions = self._RawData.npartitions - avg = dask.dataframe.from_pandas(avg, npartitions=npartitions) - else: - avg = avg.resample(self.SamplingWindow).mean() - - avg = avg.rename(columns={'u': 'u_bar', 'v': 'v_bar', 'w': 'w_bar', 'T': 'T_bar'}) - - avg['wind_dir_bar'] = numpy.arctan2(avg['v_bar'], avg['u_bar']) - avg['wind_dir_bar'] = (2 * numpy.pi + avg['wind_dir_bar']) % (2 * numpy.pi) - avg['wind_dir_bar'] = numpy.rad2deg(avg['wind_dir_bar']) - - avg['wind_dir_bar'] = avg['wind_dir_bar'].apply(toMeteorologicalAngle) - - self._TemporaryData = avg - self._CalculatedParams += [['u_bar',{}], ['v_bar',{}], ['w_bar',{}], ['T_bar',{}]] - - # correcting the first index to be the same as the avg. - self._RawData = self._RawData.reset_index() - self._RawData.at[0,'Time'] = avg.index[0] - self._RawData = self._RawData.set_index("Time") - - self._RawData = self._RawData.merge(avg, how='left', left_index=True, right_index=True) - self._RawData = self._RawData.ffill() - - self._RawData['wind_dir'] = numpy.arctan2(self._RawData['v'], self._RawData['u']) - self._RawData['wind_dir'] = (2 * numpy.pi + self._RawData['wind_dir']) % (2 * numpy.pi) - self._RawData['wind_dir'] = numpy.rad2deg(self._RawData['wind_dir']) - self._RawData['wind_dir'] = self._RawData['wind_dir'].apply(toMeteorologicalAngle) - - self._RawData['up'] = self._RawData['u'] - self._RawData['u_bar'] - self._RawData['vp'] = self._RawData['v'] - self._RawData['v_bar'] - self._RawData['wp'] = self._RawData['w'] - self._RawData['w_bar'] - self._RawData['Tp'] = self._RawData['T'] - self._RawData['T_bar'] - self._RawData['wind_dir_p'] = (180 - (180 - (self._RawData['wind_dir'] - self._RawData['wind_dir_bar']).abs()).abs()).abs() - - return self - - class InMemoryRawData(pandas.DataFrame): """In-memory container for raw high-frequency data. @@ -1455,100 +1377,6 @@ def __init__(self, data=None, index=None, columns=None, dtype=None, copy=False): super(InMemoryRawData, self).__init__(data=data, index=index, columns=columns, dtype=dtype, copy=copy) self._Attrs = {} - def append(self, other, ignore_index=False, verify_integrity=False): - """Append rows of *other* and merge attribute dictionaries. - - Parameters - ---------- - other : InMemoryRawData - The data to append. - ignore_index : bool, optional - If ``True``, the resulting index will be relabelled 0, 1, ... - Default is ``False``. - verify_integrity : bool, optional - If ``True``, raise ``ValueError`` on duplicate index. Default is - ``False``. - - Returns - ------- - InMemoryRawData - A new ``InMemoryRawData`` instance containing the combined rows and - merged attributes. - """ - ret = super(InMemoryRawData, self).append(other, ignore_index=ignore_index, verify_integrity=verify_integrity) - ret = InMemoryRawData(ret) - ret._Attrs = other._Attrs - ret._Attrs.update(self._Attrs) - - return ret - - @classmethod - def read_hdf(cls, path_or_buf, key=None, **kwargs): - """Read an HDF5 file into an ``InMemoryRawData`` instance. - - If a JSON sidecar file (same base name, ``.json`` extension) exists - alongside the HDF5 file, its contents are loaded as the instance - attributes. - - Parameters - ---------- - path_or_buf : str - Path to the HDF5 file. - key : str or None, optional - The group identifier in the HDF store. - **kwargs - Additional keyword arguments forwarded to - ``pandas.read_hdf``. - - Returns - ------- - InMemoryRawData - A new instance populated with data from the file and, when - available, its sidecar attributes. - """ - ret = InMemoryRawData(pandas.read_hdf(path_or_buf, key, **kwargs)) - path_or_buf = '%s%s' % (path_or_buf.rpartition('.')[0], '.json') - - if os.path.isfile(path_or_buf): - with open(path_or_buf, 'r') as jsonFile: - ret._Attrs = json.load(jsonFile) - - return ret - - def to_hdf(self, path_or_buf, key, **kwargs): - """Write the data to an HDF5 file with an optional JSON sidecar. - - The data is saved to an ``.hdf`` file. If the instance carries - non-empty attributes, they are written (or merged) into a JSON - sidecar file with the same base name. - - Parameters - ---------- - path_or_buf : str - Path for the output file (extension is replaced with ``.hdf``). - key : str - The group identifier in the HDF store. - **kwargs - Additional keyword arguments forwarded to - ``pandas.DataFrame.to_hdf``. - """ - pandasCopy = self.copy() - path_or_buf = '%s%s' % (path_or_buf.rpartition('.')[0], '.hdf') - pandasCopy.to_hdf(path_or_buf, key, **kwargs) - path_or_buf = '%s%s' % (path_or_buf.rpartition('.')[0], '.json') - attrsToSave = self._Attrs - - if len(self._Attrs) > 0: - if os.path.isfile(path_or_buf): - with open(path_or_buf, 'r') as jsonFile: - attrsFile = json.load(jsonFile) - attrsFile.update(attrsToSave) - attrsToSave = attrsFile - - with open(path_or_buf, 'w') as jsonFile: - json.dump(attrsToSave, jsonFile, indent=4, sort_keys=True) - - class InMemoryAvgData(InMemoryRawData): """In-memory container for averaged turbulence data. diff --git a/hera/measurements/meteorology/highfreqdata/parsers/CampbellBinary.py b/hera/measurements/meteorology/highfreqdata/parsers/CampbellBinary.py index 083126c75..df4460686 100644 --- a/hera/measurements/meteorology/highfreqdata/parsers/CampbellBinary.py +++ b/hera/measurements/meteorology/highfreqdata/parsers/CampbellBinary.py @@ -218,16 +218,6 @@ def headers(self): self._headers = self._getHeaders() return self._headers - @property - def station(self): - """str : Station name extracted from the first header line.""" - return self.headers[0].split(',')[1] - - @property - def instrument(self): - """str : Instrument name extracted from the first header line.""" - return self.headers[0].split(',')[-1] - @property def heights(self): """list of int : Measurement heights in metres derived from column layout.""" @@ -265,20 +255,6 @@ def format(self): self._format = self._getFormat() return self._format - @property - def firstTime(self): - """pandas.Timestamp : Timestamp of the first record in the file.""" - if self._firstTime is None: - self._firstTime = self._getFirstTime() - return self._firstTime - - @property - def lastTime(self): - """pandas.Timestamp : Timestamp of the last record in the file.""" - if self._lastTime is None: - self._lastTime = self._getLastTime() - return self._lastTime - @property def columnsNames(self): """list of list of str : Column names grouped by measurement height.""" @@ -448,24 +424,6 @@ def getRecordByIndex(self, i): time = pandas.Timestamp(1990, 1, 1) + pandas.Timedelta(days=lastSec / 86400.0, milliseconds=lastmili) return time, line - def getRecordByTime(self, time): - """Retrieve a record by its timestamp. - - Parameters - ---------- - time : pandas.Timestamp - Exact timestamp of the desired record. - - Returns - ------- - time : pandas.Timestamp - Timestamp of the record. - line : list - Unpacked data values for the record. - """ - i = self.getRecordIndexByTime(time) - return self.getRecordByIndex(i) - def _getDataFromStream(self, partStream): retval = list(struct.unpack(self.format, partStream)) for i in range(3, len(retval)): diff --git a/hera/measurements/meteorology/highfreqdata/toolkit.py b/hera/measurements/meteorology/highfreqdata/toolkit.py index b135b212b..237ba70be 100644 --- a/hera/measurements/meteorology/highfreqdata/toolkit.py +++ b/hera/measurements/meteorology/highfreqdata/toolkit.py @@ -23,9 +23,6 @@ class HighFreqToolKit(toolkit.abstractToolkit): - **Campbell Scientific TOA5 ASCII** — via :class:`ASCIIParser` """ - DOCTYPE_STATIONS = 'StationsData' - DOCTYPE_MEASUREMENTS = 'MeasurementsData' - def __init__(self, projectName, filesDirectory=None, connectionName=None): """Initialise the high-frequency meteorology toolkit. diff --git a/hera/riskassessment/agents/Agents.py b/hera/riskassessment/agents/Agents.py index d06f9e830..581c9c2ee 100644 --- a/hera/riskassessment/agents/Agents.py +++ b/hera/riskassessment/agents/Agents.py @@ -57,28 +57,6 @@ def physicalproperties(self): """ return self._physicalproperties - @property - def fullDescription(self): - """ - The full JSON descriptor used to initialize this agent. - - Returns - ------- - dict - """ - return self._agentconfig - - @property - def effectproperties(self): - """ - The effect parameters dictionary (e.g. tenbergeCoefficient). - - Returns - ------- - dict - """ - return self._effectParameters - @property def tenbergeCoefficient(self): """ diff --git a/hera/riskassessment/agents/effects/Injury.py b/hera/riskassessment/agents/effects/Injury.py index eecd8e25f..49e637b0c 100755 --- a/hera/riskassessment/agents/effects/Injury.py +++ b/hera/riskassessment/agents/effects/Injury.py @@ -187,13 +187,6 @@ def _postCalculate(self,retList,time): """ raise NotImplementedError("Abstract class") - def _postCalculatePointWise(self, retList): - """ - apply some post calculations on the results - - """ - pass - def calculate(self, concentrationField, field, time="datetime", x="x", y="y", breathingRate=10 * ureg.L / ureg.min, sel={},isel={}): """Deprecated. Use ``calculateRegionOfInjured`` instead.""" @@ -282,81 +275,6 @@ def calculateToxicLoads(self,concentrationField,time="datetime",breathingRate=10 return self.calculator.calculate(concentrationField, field, breathingRate=breathingRate, time=time) - def calculatePointWiseFractionInjured(self,timeConcentration,time="datetime",breathingRate=10*ureg.L / ureg.min,field=None): - """ - Calculates the fraction of injury over time in each point. - - Can be used with pandas.DataFrame or xarray.Dataset [not implemented yet]. - - Parameters - ---------- - - timeConcentration : pandas.DataFrame, or xarray.DataFrame. - Holds the concentration in time. - - If xarray, also has a 'time' coordinate that will be calculated. [not implemented yet] - - If DataFrame: - Each point is represeneted by a column, and the time is an index (with the name 'datetime'). - - So the structure of the input is : - P1 P2 P3 - time - 00:00 0 0 0 - 00:01 0.1 0.1 0.01 - 00:02 0.4 0.1 0.01 - 00:05 0.5 0.1 0.01 - - time : str - The name of the time column (or the name of the index). - - breathingRate : unum, L/min - The breathing rate of the population. - - parameters: kwargs. - Additional parameter to the calculator (for example ten-berge coefficient). - - selection parameters (xarray only): - - sel - select according to the coordinates (see sel funcion of the xarray). - isel - select according to the coordinate index (see isel function of the xarray). - - """ - - - - - # 2. For each injury level: - # Create a dataframe with the fields: - # - # P1 injury - # datetime - # 0 [fraction] level 1 name - # 1 [fraction] level 1 name - # ... - - if not isinstance(timeConcentration,pandas.DataFrame): - raise ValueError("Still not implemented....") - - # 1. Calculate the toxic load for each point. - toxicLoads = self.calculateToxicLoads(concentrationField=timeConcentration, - time=time, - breathingRate=breathingRate, - field=field) - retList = [] - for lvl in self.levels: - data = pandas.DataFrame() - - for device in toxicLoads: - prct = toxicLoads[device].apply(lambda x: self.getPercent(lvl.name,x)) - data = prct.to_frame("injuryPercent").assign(deviceName=device,level=lvl.name) - if data is not None: - retList.append(data) - - ret = pandas.concat(retList) - return ret - - def calculateThresholdPolygon(self,data,time): """ Calculates the diff of the polygon based on the toxic load. diff --git a/hera/riskassessment/protectionpolicy/ProtectionPolicy.py b/hera/riskassessment/protectionpolicy/ProtectionPolicy.py index 7df1faa5a..d80fd0b04 100644 --- a/hera/riskassessment/protectionpolicy/ProtectionPolicy.py +++ b/hera/riskassessment/protectionpolicy/ProtectionPolicy.py @@ -205,18 +205,6 @@ def compute(self,data,C="C", lazy=False): self._data.compute() return self.data - @property - def hdfkey(self): - """Combined HDF key for all actions in this policy. - - Returns - ------- - str - """ - return "/".join([action.hdfkey for action in self._actionList]) - - - class abstractAction(object): """Abstract base class for a single protection action in a policy pipeline. diff --git a/hera/riskassessment/riskToolkit.py b/hera/riskassessment/riskToolkit.py index 180f1e189..304d09abf 100644 --- a/hera/riskassessment/riskToolkit.py +++ b/hera/riskassessment/riskToolkit.py @@ -124,33 +124,6 @@ def getAgent(self, nameOrDesc, version=None): return Agent(descriptor) - def listAgentsNames(self): - """ - Lists the agents that are currently loaded in the DB (both local and public). - - :return: list - A list of agent names. - - """ - return [x.desc["datasourceName"] for x in self.getDataSourceDocumentsList()] - - def loadAgent(self, name, agentDescription, version,saveMode=TOOLKIT_SAVEMODE_FILEANDDB): - """ - Adds the agent to the DB. Either to the public or to the local DB. - Equivalent to loadData - - :param name: str - Agent name - :param agentDescription: dict - The agent description - - :return: - None - """ - agentDescription['name'] = name - agentDescription['version'] = version - return self.loadData(agentDescription,saveMode=saveMode) - def loadData(self, fileNameOrData, saveMode=TOOLKIT_SAVEMODE_FILEANDDB,**kwargs): """ Abstract loading a data from file. Manages the parsing of the diff --git a/hera/simulations/CLI.py b/hera/simulations/CLI.py index 13c7fade3..1c462ce63 100644 --- a/hera/simulations/CLI.py +++ b/hera/simulations/CLI.py @@ -477,96 +477,6 @@ def workflow_list(arguments): print(f"\t\t\t - {pname}") -def workflowNodes_list(arguments): - """ - Lists the nodes in the requested workflow. The workflow can be a file on the disk or a name - of a simulation in the database. - - Parameters - ---------- - arguments - projectName: str - The name of the project. - workflowName: str - A file on the disk or a simulation in the DB. - Returns - ------- - prints a list of all the nodes of the workflow. - """ - from hera import toolkitHome - logger = logging.getLogger("hera.bin.hera_workflows.listNodes") - logger.info(f" -- Starting: Listing workflow nodes --") - - if os.path.exists(arguments.workflowName) and not os.path.isdir(arguments.workflowName): - from hermes import workflow - from ..utils import loadJSON - json = loadJSON(arguments.workflowName) - hermesObject = workflow(json, Resource_path=arguments.workflowName) - else: - if arguments.projectName is None: - raise ValueError("Must supply a project name for a non-file workflow") - - wftk = toolkitHome.getToolkit(toolkitName=toolkitHome.SIMULATIONS_WORKFLOWS, projectName=arguments.projectName) - - hermesObject = wftk.getHermesWorkflowFromDB(arguments.workflowName) - - tlte = f"The nodes of the {arguments.workflowName}" - print(tlte) - print("-"*len(tlte)) - - if arguments.parameters: - for hnodeName,hnodeData in hermesObject.items(): - print(f"\t * {hnodeName}") - for prop in hnodeData.parameters.keys(): - print(f"\t\t + {prop}") - else: - print("\t * "+"\n\t * ".join(hermesObject.nodeList)) - -def workflowNodes_listParameters(arguments): - """ - List the parameters of the node. - - Parameters - ---------- - arguments - projectName: str - The name of the project. - nodename : str - The name of the node to list. - - workflowName: str - A file on the disk or a simulation in the DB. - - Returns - ------- - - """ - from hera import toolkitHome - logger = logging.getLogger("hera.bin.hera_workflows.listNodeParameters") - logger.info(f" -- Starting: Listing node parameters --") - - if os.path.isfile(arguments.workflowName): - from hermes import workflow - from ..utils import loadJSON - json = loadJSON(arguments.workflowName) - hermesObject = workflow(json, Resource_path=arguments.workflowName) - else: - wftk = toolkitHome.getToolkit(toolkitName=toolkitHome.SIMULATIONS_WORKFLOWS, projectName=arguments.projectName) - - hermesObject = wftk.getHermesWorkflowFromDB(arguments.workflowName) - - if arguments.nodeName not in hermesObject.nodeList: - raise ValueError(f" Node {arguments.nodeName} not found in workflow {arguments.workflowName}. Existing nodes are: {','.join(hermesObject.nodeList)}") - - tlte = f"The parameters of the node {arguments.nodeName} in the workflow {arguments.workflowName}" - print(tlte) - print("-"*len(tlte)) - import json - - for nd,pm in hermesObject[arguments.nodeName].parameters.items(): - vls = json.dumps(pm,indent=4) - print(f"-\t {nd}: {vls}") - def workflow_compare(arguments): """ Compares the parameters of the list of simulations that were supplied. diff --git a/hera/simulations/analysis/errorCalculation.py b/hera/simulations/analysis/errorCalculation.py deleted file mode 100644 index 1831d88ad..000000000 --- a/hera/simulations/analysis/errorCalculation.py +++ /dev/null @@ -1,99 +0,0 @@ -class errorCalculation(): - - def calculateFB(self,data, modelColumn='model', measureColumn='measure'): - """ - Calculates the fractional mean bias. - - :param data: pandas of the model and raw data together. - :param modelColumn: The name of the model's data column. - :param measureColumn: The name of the experimental measured data. - :return: float FB. - """ - - FB = 2*(data[modelColumn]-data[measureColumn]).mean()/(data[modelColumn].mean()+data[measureColumn].mean()) - return FB - - - def calculateNMSE(self,data, modelColumn='model', measureColumn='measure'): - """ - Calculates the normalized mean-square error. - - :param data: pandas of the model and raw data together. - :param modelColumn: The name of the model's data column. - :param measureColumn: The name of the experimental measured data. - :return: float NMSE. - """ - - NMSE = ((data[modelColumn]-data[measureColumn])**2).mean()/(data[modelColumn].mean()*data[measureColumn].mean()) - return NMSE - - - def calculateFAC(self,data, relation=2, modelColumn='model', measureColumn='measure'): - """ - Calculates the FAC criteria. - - :param data: pandas of the model and raw data together. - :param modelColumn: The name of the model's data column. - :param measureColumn: The name of the experimental measured data. - :return: float FAC2. - """ - - model_over_measure = data[modelColumn]/data[measureColumn] - FAC2 = model_over_measure.apply(lambda x: 1/relation5) and (kplus < 70): - ret = 8-1/kappa*numpy.log(3.4+kplus) - else: - ret = 8 - - return ret - - def ReynoldsUm(self,ustar,Um): - """ - Return the reynolds based on the Um and hydraulic diameter. - :return: - """ - return ustar*self.hydraulicHeight/self.kinematicViscosity - - - def __init__(self,Ra,nu): - """ - - Parameters - ---------- - - Ra: float/unit - Arithmetic mean deviations of surface asperities. default unit [m] - - nu: float/ unit - The viscosity of the fluid. default unit [m^2/s] - """ - self._nu = tounit(nu, ureg.m**2/ureg.s) - self._Ra = tounit(Ra, ureg.m) - - - -class channelFlow(nearWallFlow): - """ - Defines a list of function fr the anaytical management of rough cannel. - See chapter 16 and 17. - - Units are mks. - - This is the model 'flow' for the atmosphere. - - """ - def __init__(self,Ra,nu,channelHeight): - """ - - Parameters - ---------- - - Ra: float/unit - Arithmetic mean deviations. default unit [m] - - nu: float/ unit - The viscosity of the fluid. default unit [m^2/s] - - channelHeight: float/unit - The height of the channel. default unit [m] - - Note that in schlichting H is channelHeight/2. - - """ - super().__init__(Ra,nu) - self._channelHeight = tounit(channelHeight, ureg.m) - self._functionG = functionG() - self.C_bar_plus_C_barbar = -1.7 # Cbar + Cbar bar (equation 17.91). - self.C_bar = 0.94 # Cbar + Cbar bar (equation 17.91). - - - - def skin_friction(self,ustar,Um): - """ - Skin friction (c_f) is given by eqn 17.92. - - - ustar: float/unum. - The guess for the current Ustar. units [m**2/s] - - Um: float/unum - The flow velocity at the center of the channel. (u_m). - - :return: float - The skin friction. - """ - kappa = 0.41 # von karman constant. - Re_dh = self.ReynoldsUm(ustar, Um) - Lambda = 2*numpy.log() - Cplus = self.Cplus(ustar) - D = 0.82*Cplus - 4.56 - G = self._functionG.solve(Lambda,D) - - ret = 2*(kappa/numpy.log(Re_dh)*G)**2 - return ret - - - def Re_tau(self,ustar): - """ - Returns the Reynolds number with ustar and half channel height as measurements.old. - - :param ustar: - :return: - """ - ustar = tounit(ustar, ureg.m/ureg.s) - return ((ustar*self._channelHeight/2)/self.kinematicViscosity).magnitude - - - def get_Umean_from_Ustar(self,ustar): - """ - Calculate the U mean that is needed in order to obtain the requested Ustar. - - Parameters - ----------- - - ustar: float, unum - The ustar, default units [m/s] - - Returns - ------- - the velocity in the center of the channel (Um) [m/s] - """ - kappa = 0.41 # von karman constant. - - ustar = tounit(ustar, ureg.m/ureg.s) - - Re_tau = self.Re_tau(ustar) - - Um_plus = 1/kappa*numpy.log(Re_tau) + self.Cplus(ustar) + self.C_bar_plus_C_barbar - - Um = Um_plus*ustar ## see Eqn. 17.86 - - return Um - - def get_Ucenter_from_Ustar(self,ustar): - """ - Calculate the U at the center of the channel that is needed in order to obtain the requested Ustar. - - Eqn 17.53. - - Parameters - ----------- - - ustar: float, unum - The ustar, default units [m/s] - - Returns - ------- - the velocity in the center of the channel (Um) [m/s] - """ - kappa = 0.41 # von karman constant. - - ustar = tounit(ustar, ureg.m/ureg.s) - - Re_tau = self.Re_tau(ustar) - - Um_plus = 1/kappa*numpy.log(Re_tau) + self.Cplus(ustar) + self.C_bar - - Um = Um_plus*ustar ## see Eqn. 17.86 - - return Um - - -class couetteFlow(nearWallFlow): - """ - Defines a list of function fr the anaytical management of rough cannel. - See chapter 16 and 17. - - Units are mks. - - This is the model 'flow' for indoors. - - """ - def __init__(self, Ra, nu, channelHeight): - """ - - Parameters - ---------- - - Ra: float/unit - Arithmetic mean deviations. default unit [m] - - nu: float/ unit - The viscosity of the fluid. default unit [m^2/s] - - channelHeight: float/unit - The height of the channel. default unit [m] - - Note that in schlichting H is channelHeight/2. - - """ - super().__init__(Ra,nu) - self._channelHeight = tounit(channelHeight, ureg.m) - self._functionG = functionG() - - - def skin_friction(self, ustar, Um): - """ - Skin friction (c_f) is given by eqn 17.92. - - - ustar: float/unum. - The guess for the current Ustar. units [m**2/s] - - Um: float/unum - The flow velocity at the center of the channel. (u_m). - - :return: float - The skin friction. - """ - kappa = 0.41 # von karman constant. - Re_dh = self.ReynoldsUm(ustar, Um) - Lambda = 2 * numpy.log() - Cplus = self.Cplus(ustar) - D = 0.82 * Cplus - 4.56 - G = self._functionG.solve(Lambda, D) - - ret = 2 * (kappa / numpy.log(Re_dh) * G) ** 2 - return ret - - def Re_tau(self, ustar): - """ - Returns the Reynolds number with ustar and half channel height as measurements.old. - - :param ustar: - :return: - """ - ustar = tounit(ustar, ureg.m/ureg.s) - return ((ustar * self._channelHeight / 2) / self.kinematicViscosity).magnitude - - def get_Um_from_Ustar(self, ustar): - """ - Calculate the Um that is needed in order to obtain the requested Ustar. - - 17.21 - - Parameters - ----------- - - ustar: float, unum - The ustar, default units [m/s] - - Returns - ------- - the velocity in the center of the channel (Um) [m/s] - """ - kappa = 0.41 # von karman constant. - - ustar = tounit(ustar, ureg.m/ureg.s) - - Re_tau = self.Re_tau(ustar) - - Um_plus = 1 / kappa * numpy.log(Re_tau) + self.Cplus(ustar) - - Um = Um_plus * ustar ## see Eqn. 17.86 - - return Um diff --git a/hera/simulations/openFoam/lagrangian/LSM/toolkit.py b/hera/simulations/openFoam/lagrangian/LSM/toolkit.py index 4a4a4a844..6bd1034ef 100644 --- a/hera/simulations/openFoam/lagrangian/LSM/toolkit.py +++ b/hera/simulations/openFoam/lagrangian/LSM/toolkit.py @@ -698,43 +698,6 @@ def _writeScalarField(headerLines, cellStart, nCells, boundaryLines, f.write("\n)\n;\n\n") f.write(boundarySection) - def createRootCaseMeshLink(self, rootCase): - """ - Creates the directories for run (currently only parallel). - - For each processorXX in the rootCase: - - 1. Copy the timestep - - If parallel, create all the processor** and link it. - - :param rootCase: - :param parallel: - :return: - """ - for fl in glob.glob(os.path.join(rootCase, "processor*")): - print(fl) - fullpath = os.path.join(os.path.abspath(fl), lastTS) - - proc = os.path.split(fl)[-1] - destination = os.path.join(os.path.abspath(proc), "3600") - os.makedirs(os.path.dirname(destination), exist_ok=True) - if os.path.exists(destination): - shutil.rmtree(destination) - shutil.copytree(fullpath, destination) - - fullpath = os.path.abspath(os.path.join(fl, "constant", "polyMesh")) - destination = os.path.join(os.path.abspath(proc), "constant", "polyMesh") - os.makedirs(os.path.dirname(destination), exist_ok=True) - if not os.path.exists(destination): - os.symlink(fullpath, destination) - - # link the root dir . - curdir = os.path.abspath(os.path.join("rootCase", os.path.basename(fl))) - targetdir = os.path.abspath(os.path.join(fl, "rootCase")) - if not os.path.exists(targetdir): - os.symlink(curdir, targetdir) - def to_paraview_CSV(self, data, outputdirectory, filename, timeFactor=1): """ Writes the globalPositions (globalX,globalY,globalZ) as CSV for visualization in paraview. diff --git a/hera/simulations/openFoam/lagrangian/abstractLagrangianSolver.py b/hera/simulations/openFoam/lagrangian/abstractLagrangianSolver.py index 470be5a35..575d2ac35 100644 --- a/hera/simulations/openFoam/lagrangian/abstractLagrangianSolver.py +++ b/hera/simulations/openFoam/lagrangian/abstractLagrangianSolver.py @@ -847,33 +847,6 @@ def addToLists(time, name, action, m, parcel): "mass": mass}) - def getOriginalFlowFieldMesh(self,nameOrWorkflowFileOrJSONOrResource,readParallel=True, time=0): - """ - Returns the mesh of the original flow field. name from the workflow - - Parameters - ---------- - nameOrWorkflowFileOrJSONOrResource : string or dict - The name/dict that defines the item - - readParallel: bool - If parallel case exists, read it . - - time : float - The time to read the mesh from. (relevant for mesh moving cases). - - Returns - ------- - - """ - logger = get_classMethod_logger(self,"getMeshFromLagrangianName") - logger.info(f"Getting the mesh for {nameOrWorkflowFileOrJSONOrResource}") - logger.debug(f"Getting the original flow field") - originalFlowField = self.getOriginalFlowDocument(nameOrWorkflowFileOrJSONOrResource) - logger.debug(f"Getting the mesh from the original flow field") - return self.toolkit.getMesh(originalFlowField.getData()) - - def getCaseResults(self, caseDescriptor, timeList=None, withVelocity=True, withReleaseTimes=False, withMass=True, cloudName="kinematicCloud", forceSingleProcessor=False, cache=True, overwrite=False): """ @@ -1153,51 +1126,6 @@ def _saveToCacheNetCDF(self, data, cacheDoc, caseDescriptor, ret.to_netcdf(fullname) return ret - def getDispersionDocument(self, nameOrDispersionWorkflow): - """Return the DB document for a dispersion simulation. - - Parameters - ---------- - nameOrDispersionWorkflow : str or workflow_StochasticLagrangianSolver - The name of the workflow or a workflow instance. - - Returns - ------- - document or None - The DB document, or None if not found. - """ - # TODO: This method was incomplete in the original code (bare reference - # to getWorkflowDocumentFromDB without calling it). Implemented as a - # simple delegation to the toolkit's workflow document lookup. - return self.toolkit.getWorkflowDocumentFromDB(nameOrDispersionWorkflow) - - def getDispersionFlowDocument(self,nameOrDispersionWorkflow): - """ - Returns the DB document of the dispersion workflow. - We assume that it is a name or a nameOrDispersionWorkflow. - - Parameters - ---------- - nameOrWorkflow : str, workflow_StochasticLagrangianSolver - The name of the workflow or an instance of the hermes workflow of the StochasticLagrangianSolver). - - Returns - ------- - DB document. - """ - logger = get_classMethod_logger(self,"getDispersionFlowDocument") - if isinstance(nameOrDispersionWorkflow,str): - wf = self.toolkit.getHermesWorkflowFromDB(nameOrDispersionWorkflow) - dffname = wf.dispersionFlowFieldName - elif isinstance(nameOrDispersionWorkflow,workflow_StochasticLagrangianSolver): - dffname = nameOrDispersionWorkflow.dispersionFlowFieldName - else: - err = f"{nameOrDispersionWorkflow} must be of type str or workflow_StochasticLagrangianSolver, got {type(nameOrDispersionWorkflow)}" - - logger.info(f"Trying to retireve the document for {dffname}") - ret = self.toolkit.getWorkflowDocumentFromDB(dffname, doctype=self.toolkit.DOCTYPE_OF_FLOWDISPERSION) - return ret[0] if len(ret) > 0 else None - def getOriginalFlowDocument(self,nameOrDispersionWorkflow): """ Returns the flow document of the original workflow. diff --git a/hera/simulations/openFoam/postProcess/VTKPipeline.py b/hera/simulations/openFoam/postProcess/VTKPipeline.py index a1b9e8237..de41c5d2e 100644 --- a/hera/simulations/openFoam/postProcess/VTKPipeline.py +++ b/hera/simulations/openFoam/postProcess/VTKPipeline.py @@ -8,11 +8,8 @@ import pydoc from hera import get_classMethod_logger import hera.simulations.openFoam.postProcess.VTKPipeline as pipelineModule -from hera.simulations.openFoam import CASETYPE_DECOMPOSED, CASETYPE_RECONSTRUCTED, TYPE_VTK_FILTER -from hera.utils import dictToMongoQuery from hera.simulations.openFoam.postProcess.pvOpenFOAMBase import paraviewOpenFOAM import paraview.simple as pvsimple -from deprecated import deprecated import os import shutil @@ -86,19 +83,6 @@ def addFilterFromObj(self,newFilter): """ self[newFilter.name] = newFilter - @deprecated("Use addFilterFromObj") - def addExistingFilter(self, newFilter): - """ - Adds a filter to the pipeline using an already existent instance. - - Parameters - - filter(VTKFilter) - an instance of a filter - - """ - self[newFilter.name] = newFilter - def __setitem__(self, key, value): """Set a filter in the pipeline by key.""" self.filters[key] = value @@ -126,15 +110,6 @@ def __getitem__(self, item): raise KeyError(f"The filter {item} is not found in the current pipeline") return val - def registerPipeline(self, nameOrWorkflowFileOrJSONOrResource, serverName=None, caseType=CASETYPE_DECOMPOSED): - """Bind this pipeline to a simulation case and return a registered pipeline.""" - - return registeredVTKPipeLine(datalayer=self.datalayer, - vtkpipeline=self, - nameOrWorkflowFileOrJSONOrResource=nameOrWorkflowFileOrJSONOrResource, - serverName=serverName, - caseType=caseType) - def toJSON(self): """ Converts the pipeline to a VTK JSON of the executions. @@ -198,426 +173,6 @@ def recurseAllNames(fatherPath, filtersList): return recurseAllNames(None, self.filters) -class registeredVTKPipeLine: - """ - Represents binding of a vtk pipline to a case. - """ - vtkpipeline = None - datalayer = None - casePath = None - pvOFBase = None - - def __init__(self, datalayer, vtkpipeline, nameOrWorkflowFileOrJSONOrResource, serverName=None, - caseType=CASETYPE_DECOMPOSED): - """Initialize by binding a VTK pipeline to a simulation case from DB or directory.""" - logger = get_classMethod_logger(self, "__init__") - self.datalayer = datalayer - self.vtkpipeline = vtkpipeline - self.tsBlockNum = 50 - - simulationDocumentList = [] - simulationDocumentList += self.datalayer.getWorkflowDocumentFromDB(nameOrWorkflowFileOrJSONOrResource) - if len(simulationDocumentList) != 0: - logger.info(f"found {nameOrWorkflowFileOrJSONOrResource} in database") - simulationDocument = simulationDocumentList[0] - self.casePath = simulationDocument.resource - self.simulationDocument = simulationDocument - simulationProperties = self.simulationDocument['desc'].copy() - self.simulationParams = { - "workflowName": simulationProperties['workflowName'], - "groupName": simulationProperties['groupName'], - "workflowParameters": simulationProperties['parameters'] - } - elif os.path.isdir(nameOrWorkflowFileOrJSONOrResource): - logger.info(f"{nameOrWorkflowFileOrJSONOrResource} is a directory, creating a psuedo document") - self.casePath = nameOrWorkflowFileOrJSONOrResource - self.simulationDocument = None - simName = os.path.basename(self.casePath) - groupName = simName.split("_")[0] if '_' in simName else "" - self.simulationParams = { - "workflowName": os.path.basename(self.casePath), - "groupName": groupName, - "workflowParameters": {} - } - - else: - raise ValueError(f"Simulation {nameOrWorkflowFileOrJSONOrResource} is not in the DB, and does not represent a valid case directory") - - self.pvOFBase = paraviewOpenFOAM(casePath=self.casePath, - caseType=caseType, - servername=serverName) - - def clearCache(self, regularMesh=None,filterName=None): - """Remove cached filter output documents and files from disk.""" - # 1. Get the potential filters to process - logger = get_classMethod_logger(self, "clearCache") - if filterName is None: - requestedFiltersToProcess = self.vtkpipeline.allFilterNames() - else: - requestedFiltersToProcess = list(numpy.atleast_1d(filterName)) - logger.info(f"Removing the cache for filters {requestedFiltersToProcess}") - - paramDict = dict() - if regularMesh is not None: - paramDict['regularMesh'] = regularMesh - - for filterName in requestedFiltersToProcess: - logger.debug(f"Removing {filterName}") - - qry = self._buildFilterQuery(filterName=filterName,**paramDict) - docList = self.datalayer.deleteSimulationsDocuments(type=TYPE_VTK_FILTER, **dictToMongoQuery(qry)) - logger.info(f"Found {len(docList)} documents to delete. ") - for doc in docList: - logger.debug(f"Deleting resource {doc['desc']['workflowName']} : {doc['desc']['pipeline']['filters']} ") - outputFile = doc['resource'] - - if os.path.exists(outputFile): - if os.path.isfile(outputFile): - os.remove(outputFile) - else: - shutil.rmtree(outputFile) - - def getData(self, regularMesh, filterName=None, timeList=None, latestTime=False, fieldNames=None, overwrite=False): - """ - Return pipeline filter results as a dict keyed by filter name. - - Orchestrates: time resolution, cache lookup, ParaView execution, and - DB persistence. Each logical step is delegated to a private helper. - - Parameters - ---------- - regularMesh : bool - If True, output as zarr (xarray); otherwise parquet (pandas). - filterName : str, list of str, or None - Filter(s) to retrieve. None means all write-enabled filters. - timeList : None, str, or list - Timesteps to process. None = all; str = "start:end" range; list = explicit. - latestTime : bool - If True, restrict to only the last available timestep. - fieldNames : list or None - Optional field-name whitelist to limit reader I/O. - overwrite : bool - If True, recompute even when cached results exist. - """ - logger = get_classMethod_logger(self, "getData") - filext = self.getFilterOutputFileExt(regularMesh) - - # Step 1: Determine which filters to process. - requestedFilters = self._resolveRequestedFilters(filterName) - logger.info(f"The requested filters are : {requestedFilters}") - - # Step 2: Resolve the list of timesteps to compute. - caseTimeList = self.datalayer.getTimeList(self.casePath) - timeList = self._parseTimeList(timeList, caseTimeList) - if latestTime: - timeList = [timeList[-1]] - logger.debug(f"Getting timeList {timeList}") - - # Step 3: Check cache — remove already-computed timesteps per filter. - timeList, filtersToProcess, filtersOutputFilename, DBDocumentsDict = \ - self._filterCachedTimesteps(requestedFilters, timeList, regularMesh, filext, overwrite) - logger.info(f"Computing filters {filtersToProcess}") - - # Step 4: Build ParaView pipeline and execute for uncached timesteps. - if len(filtersToProcess) > 0: - filtersToComputeDict = self._buildAndExecuteParaViewPipeline( - filtersToProcess, filtersOutputFilename, timeList, - fieldNames, overwrite, regularMesh) - - # Step 5: Persist newly computed timesteps into the DB cache. - self._updateCacheDB( - filtersToProcess, timeList, DBDocumentsDict, - filtersOutputFilename if len(filtersToProcess) > 0 else {}, - regularMesh) - - # Step 6: Load and return cached data for every requested filter. - ret = {} - for fName in requestedFilters: - ret[fName] = DBDocumentsDict[fName].getData() - return ret - - # ------------------------------------------------------------------ - # Private helpers — each encapsulates one logical step of getData - # ------------------------------------------------------------------ - - def _resolveRequestedFilters(self, filterName): - """Return the list of filter names to process. - - If *filterName* is None, return every filter in the pipeline that has - ``write=True``. Otherwise, coerce *filterName* (str or list) into a - list. - """ - if filterName is None: - return self.vtkpipeline.allFilterNames(writeOnly=True) - return list(numpy.atleast_1d(filterName)) - - def _parseTimeList(self, timeList, caseTimeList): - """Normalise the caller-supplied *timeList* into a concrete list. - - Handles three forms: - - ``None`` — use every timestep available in the case. - - ``str`` — a colon-separated ``"start:end"`` range where either - bound may be omitted (defaults to first/last case time). - - ``list`` — returned as-is. - """ - if timeList is None: - # No restriction — use the full case time range. - return caseTimeList - - if isinstance(timeList, str): - # Parse "start:end" range; missing sides default to case bounds. - bounds = [caseTimeList[0], caseTimeList[-1]] - for i, val in enumerate(timeList.split(":")): - bounds[i] = bounds[i] if len(val) == 0 else float(val) - tl = pandas.Series(caseTimeList) - return tl[tl.between(*bounds)].values - - # Explicit list — pass through unchanged. - return timeList - - def _filterCachedTimesteps(self, requestedFilters, timeList, regularMesh, filext, overwrite): - """Check the DB cache and strip already-computed timesteps. - - For each requested filter, look up an existing cache document. If one - exists, remove its timesteps from *timeList* so only the delta needs - to be computed (incremental strategy). - - Returns - ------- - timeList : list - Timesteps still requiring computation after cache subtraction. - filtersToProcess : list[str] - Subset of *requestedFilters* that actually need (re-)computation. - filtersOutputFilename : dict[str, str] - Mapping from filter name to its output file path on disk. - DBDocumentsDict : dict - Mapping from filter name to its existing cache document (if any). - """ - logger = get_classMethod_logger(self, "_filterCachedTimesteps") - filtersToProcess = [] - filtersOutputFilename = dict() - DBDocumentsDict = dict() - - for fName in requestedFilters: - qry = self._buildFilterQuery(filterName=fName, regularMesh=regularMesh) - docList = self.datalayer.getCacheDocuments(type=TYPE_VTK_FILTER, **dictToMongoQuery(qry)) - - if len(docList) > 0: - # Cache hit — reuse output path and subtract known timesteps. - logger.info("Found existing filter output in cache") - cached_filter = docList[0] - filtersOutputFilename[fName] = cached_filter.resource - dbTimeList = cached_filter['desc']['simulation']['timeList'] - timeList = [ts for ts in timeList if ts not in dbTimeList] - DBDocumentsDict[fName] = cached_filter - else: - # Cache miss — generate a fresh output file path. - outputFilePath = self.getFilterOutputFilePath(fName, filext, generate_new=True) - filtersOutputFilename[fName] = outputFilePath - - # A filter needs computation if: forced overwrite, no cache, or - # there are timesteps not yet in the cache. - logger.debug( - "Compute the filter if you need to overwrite the results, it is not in the DB, or there are times not in the DB") - if overwrite or len(docList) == 0 or len(timeList) > 0: - logger.debug(f"{fName} added to process because overwrite=True or filter not in DB") - filtersToProcess.append(fName) - - return timeList, filtersToProcess, filtersOutputFilename, DBDocumentsDict - - def _buildAndExecuteParaViewPipeline(self, filtersToProcess, filtersOutputFilename, - timeList, fieldNames, overwrite, regularMesh): - """Instantiate the ParaView filter tree and run it over *timeList*. - - Steps: - 1. Create the OpenFOAM reader. - 2. Optionally restrict the reader to *fieldNames*. - 3. Build the full filter tree from the pipeline JSON. - 4. Execute and write results to disk (parquet or zarr). - 5. Clean up ParaView proxies to free server-side memory. - - Returns the dict mapping filter names to their output file paths. - """ - logger = get_classMethod_logger(self, "_buildAndExecuteParaViewPipeline") - filtersToComputeDict = dict() - - logger.info(f"Building the vtk objects from the JSON") - # The reader is the root of the ParaView pipeline graph. - reader = self.pvOFBase.initializeReader(readerName="reader") - - # Restrict to specific field arrays to reduce memory and I/O. - if fieldNames is not None: - reader.CellArrays = fieldNames - - # Recursively create ParaView filter proxies from the pipeline JSON. - filtersToCompute = self._buildFilterLayer( - fatherName=None, father=reader, - structureJson=self.vtkpipeline.toJSON()['filters']) - logger.info(f"Added all filters to the layer. Computing filters {filtersToCompute}") - - # Map each filter that needs computation to its output file path. - for fName in filtersToProcess: - logger.debug(f"\t{fName} will be saved in {filtersOutputFilename[fName]}") - filtersToComputeDict[fName] = filtersOutputFilename[fName] - - # Execute the pipeline over the remaining timesteps and write output. - self.pvOFBase.writeCase(filtersDict=filtersToComputeDict, - timeList=timeList, - fieldnames=fieldNames, - tsBlockNum=self.tsBlockNum, - overwrite=overwrite, regularMesh=regularMesh) - - # Clean up all ParaView sources/filters to free server-side memory. - for name, proxy in list(pvsimple.GetSources().items()): - logger.debug(f"Deleting source {name}") - pvsimple.Delete(proxy) - - return filtersToComputeDict - - def _updateCacheDB(self, filtersToProcess, timeList, DBDocumentsDict, - filtersToComputeDict, regularMesh): - """Merge newly computed timesteps into the DB cache. - - For filters that already have a cache document, append the new - timesteps and save. For first-time filters, create a brand-new - cache document pointing to the output file on disk. - """ - logger = get_classMethod_logger(self, "_updateCacheDB") - - for fName in filtersToProcess: - logger.debug(f"Updating times {timeList} to filter {fName}") - - if fName in DBDocumentsDict: - # Incremental update: merge new timesteps with existing ones. - doc = DBDocumentsDict[fName] - fullTime = sorted(timeList + doc['desc']['simulation']['timeList']) - doc.desc['simulation']['timeList'] = fullTime - doc.save() - else: - # First computation — create a new cache record in the DB. - logger.debug("...Adding a new record to the DB") - recordData = self._buildFilterQuery(filterName=fName, regularMesh=regularMesh) - recordData['simulation']['timeList'] = timeList - dataFormat = self.datalayer.datatypes.ZARR_XARRAY if regularMesh else self.datalayer.datatypes.PARQUET - doc = self.datalayer.addCacheDocument( - dataFormat=dataFormat, - resource=os.path.abspath(filtersToComputeDict[fName]), - type=TYPE_VTK_FILTER, - desc=recordData) - DBDocumentsDict[fName] = doc - - logger.debug(f"Reading filter {fName} data") - - def getFilterOutputFileExt(self, regularMesh): - """Return the file extension based on mesh regularity.""" - return "zarr" if regularMesh else "parquet" - - def getFilterOutputFilePath(self, filterName, filext, generate_new=True): - """Generate or retrieve the output file path for a filter.""" - workflow_name = self.simulationParams['workflowName'] - counter_name = f"{workflow_name}_{filterName}_counter" # more consistent to have a counter per filter name - curr_number = self.datalayer.getCounter(counter_name) - if curr_number is None and not generate_new: # means there are none - return None - counter = self.datalayer.getCounterAndAdd(counter_name) if generate_new else curr_number - outputFileName = f"{filterName.replace('.','_')}_{counter}.{filext}" - outputFilePath = os.path.join(os.path.abspath(self.casePath), "vtkpipelinedata", outputFileName) - return outputFilePath - - def getRegularData(self, filterName=None, timeList=None, fieldNames=None, overwrite=False): - """Retrieve pipeline data as regular mesh (xarray/zarr) format.""" - return self.getData(regularMesh=True, filterName=filterName, timeList=timeList, - fieldNames=fieldNames, - overwrite=overwrite) - - def getNonRegularData(self, filterName=None, timeList=None, fieldNames=None, overwrite=False): - """Retrieve pipeline data as non-regular mesh (pandas/parquet) format.""" - return self.getData(regularMesh=False, filterName=filterName, timeList=timeList, - fieldNames=fieldNames, - overwrite=overwrite) - - def _buildFilterQuery(self, filterName, regularMesh=None): - """Build a database query dict for a specific filter.""" - qry = dict(simulation=self.simulationParams, - pipeline=self.vtkpipeline.toJSON()) - - if regularMesh is not None: - qry['simulation']['regularMesh'] = regularMesh - - qry['filterName'] = filterName - return qry - - def _buildFilterLayer(self, fatherName, father, structureJson): - """ - Recursively builds the structure of the leaf. - Populates the self._filterWrite map - - Since the order of setting the params might be of importance (for example, setting the - plane type determine the rest of the parameters), we set it as a list. - - :param father: - The current filter father of the layer. - - :param structureJson: - The portion of Json to build. - - :param[output] filterWrite - an dictionary with the names of the filters that are about - to be printed according to format. - - """ - logger = get_classMethod_logger(self, "_buildFilterLayer") - logger.debug(f"Initialized logger {logger}") - logger.info(f"building Filter layer {json.dumps(structureJson, indent=4)}") - # Accumulates the fully-qualified names (e.g. "ExtractBlock.CellCenters") of - # every filter created during this recursive traversal, so the caller knows - # which ParaView sources exist and can be looked up via pvsimple.FindSource(). - ret = [] - - - if structureJson is not None: - # Iterate over each sibling filter at this level of the tree. - for filterGuiName in structureJson: - # params is deliberately a *list* of (key, value) pairs rather than a - # dict, because the order in which VTK filter properties are set can - # change behaviour (e.g. setting SliceType before SliceType.Origin). - paramPairList = structureJson[filterGuiName]['params'] # must be a list to enforce order in setting. - filtertype = structureJson[filterGuiName]['filterType'] - # Build the dot-separated full name: root filters use their own name, - # child filters prepend their parent's name (e.g. "parent.child"). - newFilterName = filterGuiName if fatherName is None else f"{fatherName}.{filterGuiName}" - # Instantiate the actual ParaView filter object. `father` is the - # upstream pipeline source (reader or another filter) that feeds data - # into this filter. - filter = getattr(pvsimple, filtertype)(Input=father, guiName=newFilterName) - logger.debug( - f"Adding filter {filterGuiName} of type {filtertype} to {'Reader' if fatherName is None else fatherName}") - - # Apply each parameter in order. Parameters may use dot-notation - # (e.g. "SliceType.Origin") to reach nested sub-proxy attributes. - for param, pvalue in paramPairList: - logger.debug(f"...Adding parameters {param} with value {pvalue}") - # pvalue = str(pvalue) if isinstance(pvalue, unicode) else pvalue # python2, will be removed in python3. - # Split the param name on "." to traverse nested proxy attributes. - # For "SliceType.Origin", we first resolve filter.SliceType, then - # set the "Origin" attribute on that sub-proxy. - paramnamelist = param.split(".") - paramobj = filter - for pname in paramnamelist[:-1]: - paramobj = getattr(paramobj, pname) - setattr(paramobj, paramnamelist[-1], pvalue) - # Force the filter to execute so downstream filters see updated data. - filter.UpdatePipeline() - logger.debug(f"Filter {newFilterName} added to the pipeline. Now adding its downstream filters.") - ret.append(newFilterName) - # Recurse into the downstream children of this filter, building the - # tree depth-first. The returned names are appended so the final list - # is in creation (topological) order. - ret += self._buildFilterLayer(newFilterName, filter, - structureJson[filterGuiName].get("downstream", None)) - - return ret - class VTKFilter: """Base class representing a single VTK filter node in a pipeline tree.""" @@ -657,23 +212,6 @@ def __init__(self, name, filterType, write, params): self.params = params self.downstream = dict() - @property - def fullName(self): - """ - - Returns - ------- - full path of the filter from the father. - """ - - def traverse(filter): - """Traverse up the filter tree to build the full path.""" - if filter.father is not None: - fatherName = filter.father.fullName() - return [fatherName, self.name] - - return ".".join(traverse(self)) - def toJSON(self): """converts the node diff --git a/hera/simulations/openFoam/preprocessOFObjects/OFObject.py b/hera/simulations/openFoam/preprocessOFObjects/OFObject.py index bdcfb5b62..70d82ab60 100644 --- a/hera/simulations/openFoam/preprocessOFObjects/OFObject.py +++ b/hera/simulations/openFoam/preprocessOFObjects/OFObject.py @@ -19,9 +19,6 @@ class OFObject: fileName = None # The name of the file ont the disk dimensions = None - REGION_INTERNSALFIELD = 'internalField' - REGION_BOUNDARYFIELD = 'boundaryField' - @staticmethod def getDimensions(kg=0, m=0, s=0, K=0, mol=0, A=0, cd=0): """ @@ -56,30 +53,6 @@ def componentNames(self): return ret - def internalField(self,processorName='singleProcessor'): - """ - Return the interinal field data - Returns - ------- - - """ - return self.data[processorName]['internalField'] - - @property - def processors(self): - """Return the processor names in this field.""" - return self.data.keys() - - @property - def processorItems(self): - """Return the processor name-data pairs.""" - return self.data.items() - - @property - def dimensionsStr(self): - """Return the dimensions as an OpenFOAM-formatted string.""" - return self.getDimensions(**self.dimensions) - @property def dimensionsList(self): """Return the dimensions as an ordered list of exponents.""" diff --git a/hera/simulations/openFoam/toolkit.py b/hera/simulations/openFoam/toolkit.py index 781b81d1f..6ed603848 100644 --- a/hera/simulations/openFoam/toolkit.py +++ b/hera/simulations/openFoam/toolkit.py @@ -13,7 +13,6 @@ from collections.abc import Iterable from dask.delayed import delayed from hera.utils import dictToMongoQuery, slurm -from hera.simulations.openFoam.OFWorkflow import workflow_Eulerian from hera.simulations.openFoam.preprocessOFObjects import OFObjectHome from hera.simulations.hermesWorkflowToolkit import hermesWorkflowToolkit from hera.simulations.openFoam.postProcess.VTKPipeline import VTKPipeLine @@ -204,45 +203,6 @@ def processorList(self, caseDirectory): """ return [os.path.basename(proc) for proc in glob.glob(os.path.join(caseDirectory, "processor*"))] - def getHermesWorkflow_Flow(self, workflowfile): - """ - Returns the workflow of the requested JSON file. - Parameters - ---------- - workflowfile - - Returns - ------- - - """ - return workflow_Eulerian(workflowfile) - - def getMeshFromName(self,nameOrWorkflowFileOrJSONOrResource,readParallel=True, time=0): - """ - Returns the name from the workflow - Parameters - ---------- - nameOrWorkflowFileOrJSONOrResource : string or dict - The name/dict that defines the item - - readParallel: bool - If parallel case exists, read it . - - time : float - The time to read the mesh from. (relevant for mesh moving cases). - - Returns - ------- - - """ - docList = self.getWorkflowDocumentFromDB(nameOrWorkflowFileOrJSONOrResource) - if len(docList)==0: - return None - else: - doc = docList[0] - - return self.getMesh(doc.getData()) - def getMesh(self, caseDirectory, readParallel=True, time=0): """ Reads the mesh from the mesh directory. @@ -320,32 +280,6 @@ def getMesh(self, caseDirectory, readParallel=True, time=0): readParallel=readParallel) return cellCenters - def getMeshExtentFromName(self,nameOrWorkflowFileOrJSONOrResource,readParallel=True, time=0): - """ - Returns the name from the workflow - Parameters - ---------- - nameOrWorkflowFileOrJSONOrResource : string or dict - The name/dict that defines the item - - readParallel: bool - If parallel case exists, read it . - - time : float - The time to read the mesh from. (relevant for mesh moving cases). - - Returns - ------- - - """ - docList = self.getWorkflowDocumentFromDB(nameOrWorkflowFileOrJSONOrResource) - if len(docList)==0: - return None - else: - doc = docList[0] - - return self.getMeshExtent(doc.getData()) - def read_points_file(self,path): """Parse an OpenFOAM points file and return coordinates as a numpy array.""" pts = [] @@ -499,31 +433,6 @@ def writeEmptyField(self,fieldName,flowType,caseDirectory,timeOrLocation=0,readB ############################################################# - def template_add(self, name, objFile, workflowObj=None): - """ - Adds a templates to the toolkit. - - Templates can be - - Flow : Holds Hermes flow templates. - - Node : Holds a hermes node objects - - Field : Holds a field templates. - This can be - * xarray - * pandas/dask - * constant - - Parameters - ---------- - name - objFile - workflowObj - - Returns - ------- - - """ - pass - def xarrayToSetFieldsDictDomain(self, xarrayData, xColumnName="x", yColumnName="y", zColumnName="z", time=None, timeColumn="time", **kwargs): """ @@ -676,60 +585,6 @@ def getVTKPipelineCacheDocuments(self, regularMesh=None, filterName=None, workfl return self.getCacheDocuments(type=TYPE_VTK_FILTER, **dictToMongoQuery(qry)) - def getVTKPipelineCacheTable(self,regularMesh=None, filterName=None, workflowName=None, groupName=None): - """ - Return the table. - Parameters - ---------- - regularMesh - filterName - workflowName - groupName - - Returns - ------- - - """ - docList = self.getVTKPipelineCacheDocuments(regularMesh=regularMesh, filterName=filterName, - workflowName=workflowName, groupName=groupName) - cacheDict = [dict(filterName=doc.desc['filterName'],workflowName=doc.desc['simulation']['workflowName'],groupName=doc.desc['simulation']['groupName']) for doc in docList] - return pandas.DataFrame(cacheDict) - - def clearVTKPipelineCache(self, regularMesh=None, filterName=None, workflowName=None, groupName=None): - """ - deletes the cache documents and the data from the disk. - Use with care!. - Parameters - ---------- - regularMesh - filterName - workflowName - groupName - - Returns - ------- - - """ - - # 1. Get the potential filters to process - logger = get_classMethod_logger(self, "clearCache") - - docList = self.getVTKPipelineCacheDocuments(regularMesh = regularMesh, filterName = filterName, workflowName= workflowName, groupName= groupName) - - logger.info(f"Found {len(docList)} documents to delete. ") - for doc in docList: - logger.debug(f"Deleting resource {doc['desc']['filterName']} : {doc['desc']['pipeline']['filters']} ") - outputFile = doc['resource'] - - if os.path.exists(outputFile): - if os.path.isfile(outputFile): - os.remove(outputFile) - else: - shutil.rmtree(outputFile) - - for doc in docList: - doc.delete() - def getTimeList(self,nameOrWorkflowFileOrJSONOrResourceorDirectory,singleProcessor=False,returnFirst=True): """ Extract the computed Time steps from the case. diff --git a/hera/simulations/windProfile/toolkit.py b/hera/simulations/windProfile/toolkit.py index 5cc7559f0..0bfda1692 100644 --- a/hera/simulations/windProfile/toolkit.py +++ b/hera/simulations/windProfile/toolkit.py @@ -112,21 +112,6 @@ def getSpatialWind(self,minlat,minlon,maxlat,maxlon,IMS_TOKEN,dxdy=30,inputCRS=W stations_with_data = self._getWindSpeedDirection(stations,IMS_TOKEN) return xarray,stations_with_data - def _getStationsInRegion(self,minlon,minlat,maxlon,maxlat,inputCRS): - min_pp = convertCRS(points=[[minlon, minlat]], inputCRS=inputCRS, outputCRS=ITM)[0] - max_pp = convertCRS(points=[[maxlon, maxlat]], inputCRS=inputCRS, outputCRS=ITM)[0] - with open('wind_stations.json', 'r') as json_file: - wind_stations = json.load(json_file) - stations_in_region = [] - for station in wind_stations: - lat = station['attributes'][2]['value']['latitude'] - lon = station['attributes'][2]['value']['longitude'] - point_ITM = convertCRS(points=[[lon, lat]], inputCRS=WSG84, outputCRS=ITM)[0] - if min_pp.x <= point_ITM.x and max_pp.x >= point_ITM.x and min_pp.y <= point_ITM.y and max_pp.y >= point_ITM.y: - stations_in_region.append(station) - - return stations_in_region - def _getWindSpeedDirection(self,stations,IMS_TOKEN): stations_with_data = [] headers = {'Authorization': IMS_TOKEN['Authorization']} diff --git a/hera/toolkit.py b/hera/toolkit.py index dc642edea..21c8a49ca 100644 --- a/hera/toolkit.py +++ b/hera/toolkit.py @@ -128,17 +128,6 @@ def __init__(self, toolkitName: str, projectName: Optional[str] = None, logger = get_classMethod_logger(self, "init") self._toolkitname = toolkitName - @property - def classLoggerName(self): - """ - The logger name for the current class and method context. - - Returns - ------- - str - """ - return str(get_classMethod_logger(self, "{the_function_name}")).split(" ")[1] - # ------------------------------------------------------------------ # Document overrides — automatically tag with toolkit name # ------------------------------------------------------------------ diff --git a/hera/utils/__init__.py b/hera/utils/__init__.py index aff994684..1a11a2721 100644 --- a/hera/utils/__init__.py +++ b/hera/utils/__init__.py @@ -54,13 +54,13 @@ "processJSONToPandas", "convertJSONtoPandas", "setJSONPath", "JSONVariations", "JSONvariationItem", # query - "andClause", "dictToMongoQuery", + "dictToMongoQuery", # matplotlibCountour "standardize_polygon", "toGeopandas", # angle "toMeteorologicalAngle", "toMathematicalAngle", "toAzimuthAngle", # zipUtils - "add_directory_to_zip", "zip_items", "list_json_files_in_zip", + "zip_items", "list_json_files_in_zip", ] _NOT_FOUND = object() diff --git a/hera/utils/query.py b/hera/utils/query.py index f9962ebeb..ce4ebc133 100644 --- a/hera/utils/query.py +++ b/hera/utils/query.py @@ -1,38 +1,3 @@ -def andClause(excludeFields=None, **kwargs): - """ - Builds a pandas query str - Parameters - ---------- - excludeFields - kwargs - - Returns - ------- - - """ - if excludeFields is None: - excludeFields = [] - - L = [] - for key, value in kwargs.items(): - if key in excludeFields: - continue - - if isinstance(value, list): - conditionStr = "%s in %s" - elif isinstance(value, str): - conditionStr = "%s == '%s'" - elif isinstance(value, dict): - conditionStr = "%s " + value['operator'] + " %s" - value = value['value'] - else: - conditionStr = "%s == %s" - - L.append(conditionStr % (key, value)) - - return " and ".join(L) - - def dictToMongoQuery(dictObj,prefix="",prefixExclude="desc"): """ Converts a dict object to a mongodb query. diff --git a/hera/utils/unitHandler.py b/hera/utils/unitHandler.py index 8e27ff11d..13e0e5879 100644 --- a/hera/utils/unitHandler.py +++ b/hera/utils/unitHandler.py @@ -298,16 +298,6 @@ def strToUnum(value): # Callers should migrate to ureg.Quantity(value) return value - def extractUnumUnitsFromPint(pint_quantity): - """Extract unum unit equivalent from a pint Quantity.""" - units = pint_quantity._units - unum_unit = 1 * _m / _m # unitless - for unit_name, power in units.items(): - if unit_name not in PINT_TO_UNUM_MAP: - raise ValueError(f"Unit '{unit_name}' not mapped to Unum.") - unum_unit *= PINT_TO_UNUM_MAP[unit_name] ** power - return unum_unit - def pintToUnum(pint_quantity): """ Convert a pint Quantity to a Unum object. @@ -351,8 +341,3 @@ def unumToPint(unum_obj, value=1.0): pint_str = convert_unum_units_to_eval_str(unit_str) return value * ureg.parse_expression(pint_str) - def unumToBaseUnits(unum_obj): - """Convert a Unum object to MKS base units.""" - pint_obj = unumToPint(unum_obj) - standardize = pint_obj.to_base_units() - return pintToUnum(standardize) diff --git a/hera/utils/zipUtils.py b/hera/utils/zipUtils.py index 39358c278..1829ada47 100644 --- a/hera/utils/zipUtils.py +++ b/hera/utils/zipUtils.py @@ -6,19 +6,6 @@ from hera.utils.jsonutils import loadJSON -def add_directory_to_zip(zipf, folder_path, zip_path=""): - """ - Recursively adds a folder to the zip file. - zip_path is the path inside the zip. - """ - for root, dirs, files in os.walk(folder_path): - for file in files: - full_path = os.path.join(root, file) - # Compute the relative path to keep folder structure - relative_path = os.path.relpath(full_path, folder_path) - zipf.write(full_path, os.path.join(zip_path, relative_path)) - - def zip_items(zip_filename, items): """ Create a zip file and add files, directories, dictionaries.