• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

DaveFoss / DAVE_data / 12315802845

13 Dec 2024 12:38PM UTC coverage: 90.133% (-2.7%) from 92.844%
12315802845

Pull #13

github

web-flow
Merge f93b80a1e into cfc9ad350
Pull Request #13: 12 integrate hotmaps

52 of 55 branches covered (94.55%)

Branch coverage included in aggregate %.

146 of 176 new or added lines in 2 files covered. (82.95%)

624 of 695 relevant lines covered (89.78%)

3.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.49
/src/dave_data/datapool/hotmaps/hotmaps_request.py
1
import os
5✔
2
import shutil
5✔
3
from pathlib import Path
5✔
4

5
import geopandas as gpd
5✔
6
import rasterio
5✔
7
import requests
5✔
8
from pyproj import CRS
5✔
9
from rasterio.features import shapes
5✔
10
from rasterio.mask import mask
5✔
11
from shapely import box
5✔
12
from shapely.geometry import shape
5✔
13

14
from dave_data.datapool.hotmaps.exception.hotmaps_exceptions import InvalidBbox
5✔
15
from dave_data.datapool.hotmaps.exception.hotmaps_exceptions import InvalidEPSG
5✔
16
from dave_data.datapool.hotmaps.exception.hotmaps_exceptions import OutOfBbox
5✔
17

18

19
def get_hotmaps_layer_info(layer_name):
5✔
20
    #     This function returns a dictionary with the information about hotmaps layers
21
    layer_info = {
5✔
22
        "Final energy demand density for space heating and domestic hot water - Total": {
23
            "title": "Final energy demand density for space heating and domestic hot water - Total",
24
            "tag": "heat_tot_curr_density",
25
            "key": "heat_tot_curr_density",
26
            "category": "heat",
27
        },
28
        "Final energy demand density for space heating and domestic hot water - Residential": {
29
            "title": "Final energy demand density for space heating and domestic hot water - Residential",
30
            "tag": "heat_res_curr_density",
31
            "key": "heat_res_curr_density",
32
            "category": "heat",
33
        },
34
        "Final energy demand density for space heating and domestic hot water - Non-Residential": {
35
            "title": "Final energy demand density for space heating and domestic hot water - Non-Residential",
36
            "tag": "heat_nonres_curr_density",
37
            "key": "heat_nonres_curr_density",
38
            "category": "heat",
39
        },
40
        "Space cooling needs density": {
41
            "title": "Space cooling needs density",
42
            "tag": "cool_tot_curr_density",
43
            "key": "cool_tot_curr_density",
44
            "category": "heat",
45
        },
46
        "Building gross floor area density - Residential": {
47
            "title": "Building gross floor area density - Residential",
48
            "tag": "gfa_res_curr_density",
49
            "key": "gfa_res_curr_density",
50
            "category": "",
51
        },
52
        "Building gross floor area density - Non-Residential": {
53
            "title": "Building gross floor area density - Non-Residential",
54
            "tag": "gfa_nonres_curr_density",
55
            "key": "gfa_nonres_curr_density",
56
            "category": "",
57
        },
58
        "Building gross floor area density - Total": {
59
            "title": "Building gross floor area density - Total",
60
            "tag": "gfa_tot_curr_density",
61
            "key": "gfa_tot_curr_density",
62
            "category": "",
63
        },
64
        "Building construction period - until 1975": {
65
            "title": "Building gross floor area density - Total",
66
            "tag": "ghs_built_1975_100_share",
67
            "key": "GHS_BUILT_1975_100_share",
68
            "category": "construction_periods",
69
        },
70
        "Population total": {
71
            "title": "Population total",
72
            "tag": "pop_tot_curr_density",
73
            "key": "pop_tot_curr_density",
74
            "category": "",
75
        },
76
        "Potential solar thermal collectors - rooftop": {
77
            "title": "Potential solar thermal collectors - rooftop",
78
            "tag": "potential_solarthermal_collectors_rooftop",
79
            "key": "potential_solarthermal_collectors_rooftop",
80
            "category": "potential",
81
        },
82
        "Potential solar thermal collectors - open field": {
83
            "title": "Potential solar thermal collectors - open field",
84
            "tag": "potential_solarthermal_collectors_open_field",
85
            "key": "potential_solarthermal_collectors_open_field",
86
            "category": "potential",
87
        },
88
    }
89

90
    # Check if the layer_name is valid
91
    if layer_name not in layer_info:
5✔
92
        # Prepare a message listing available layers
93
        available_layers = "\n".join(
5✔
94
            [f"{i + 1}. {layer}" for i, layer in enumerate(layer_info.keys())]
95
        )
96
        error_message = f"Layer '{layer_name}' not found. Available layers:\n{available_layers}"
5✔
97
        raise ValueError(error_message)
5✔
98

99
    # Return the information for the given type
100
    title = layer_info[layer_name]["title"]
5✔
101
    tag = layer_info[layer_name]["tag"]
5✔
102
    key = layer_info[layer_name]["key"]
5✔
103
    category = layer_info[layer_name]["category"]
5✔
104
    return title, tag, key, category
5✔
105

106

107
def validate_bbox(bbox):
5✔
108
    # Validate bbox
109
    if not (isinstance(bbox, (list, tuple)) and len(bbox) == 4):
5✔
110
        raise InvalidBbox(
5✔
111
            "bbox must be a list or tuple of four elements (min_lon, min_lat, max_lon, max_lat)."
112
        )
113
    if len(bbox) != 4:
5✔
NEW
114
        raise InvalidBbox(
×
115
            "Bounding box must have exactly four elements: [min_x, min_y, max_x, max_y]."
116
        )
117

118
    min_x, min_y, max_x, max_y = bbox
5✔
119

120
    if not (-180 <= min_x <= 180 and -180 <= max_x <= 180):
5✔
NEW
121
        raise InvalidBbox("Y values must be between -180 and 180 degrees.")
×
122

123
    if not (-90 <= min_y <= 90 and -90 <= max_y <= 90):
5✔
NEW
124
        raise InvalidBbox("X values must be between -90 and 90 degrees.")
×
125

126
    if max_x <= min_x:
5✔
NEW
127
        raise InvalidBbox("min_x should be less than max_x.")
×
128

129
    if max_y <= min_y:
5✔
NEW
130
        raise InvalidBbox("min_y should be less than max_y.")
×
131

132

133
def validate_epsg(epsg):
5✔
134
    # First, check if the EPSG code is an integer
135
    if not isinstance(epsg, int):
5✔
NEW
136
        raise InvalidEPSG("EPSG must be an integer.")
×
137

138
    try:
5✔
139
        # Attempt to create a CRS object with the EPSG code
140
        CRS.from_epsg(epsg)
5✔
141
    except Exception as e:
5✔
142
        print(f"EPSG code {epsg} is not valid: {e}")
5✔
143
        raise InvalidEPSG(f"EPSG code {epsg} is not valid: {e}") from e
5✔
144

145

146
def download_hotmaps_raster(title, tag, key, category, directory):
5✔
147
    # Gitlab link to request Hotmaps data
148
    url = f"https://gitlab.com/hotmaps//{category}/{tag}/raw/master/data/{key}.tif"
5✔
149
    response = requests.get(url, timeout=60)
5✔
150

151
    # The path to save the raster file
152
    raster_path = os.path.join(directory, title + ".tif")
5✔
153
    # Save the raster
154
    if response.status_code == 200:
5✔
155
        with Path.open(raster_path, "wb") as f:
5✔
156
            f.write(response.content)
4✔
157
    else:
NEW
158
        print("Failed to retrieve the image")
×
159
    return raster_path
4✔
160

161

162
def clip_raster(temp_directory, title, output_directory, bbox):
5✔
163
    # The path to save the saved raster file
164
    raster_path = os.path.join(temp_directory, title + ".tif")
4✔
165

166
    # Define the path to save the clipped raster file
167
    clipped_raster_path = os.path.join(output_directory, title + ".tif")
4✔
168

169
    # Prepare the bounding box to clip the raster
170
    _bbox = box(bbox[0], bbox[1], bbox[2], bbox[3])
4✔
171

172
    # Create a GeoDataFrame with the bounding box
173
    gdf = gpd.GeoDataFrame({"geometry": [_bbox]})
4✔
174

175
    # Set the original CRS (Coordinate Reference System)
176
    gdf.set_crs(epsg="4326", inplace=True)
4✔
177

178
    # Reproject to the new CRS - the CRS of the original HotMaps raster
179
    gdf = gdf.to_crs(epsg="3035")
4✔
180

181
    # Check if the raster_path is a string and the raster file exists
182
    if not isinstance(raster_path, str) or not Path(raster_path).exists():
4✔
NEW
183
        raise FileNotFoundError(
×
184
            f"Raster file '{raster_path}' does not exist or is not a valid path."
185
        )
186

187
    # Check if gdf is provided and has valid geometries
188
    if gdf is None or not hasattr(gdf, "geometry") or gdf.empty:
4✔
NEW
189
        raise ValueError(
×
190
            "GeoDataFrame 'gdf' is invalid or does not contain valid geometry."
191
        )
192

193
    # Check if clipped_raster_path is a string
194
    if not isinstance(clipped_raster_path, str):
4✔
NEW
195
        raise ValueError(
×
196
            f"Raster file '{clipped_raster_path}' is not a valid path."
197
        )
198

199
    # Load the GeoTIFF
200
    with rasterio.open(raster_path) as src:
4✔
201
        try:
4✔
202
            out_image, out_transform = mask(src, gdf.geometry, crop=True)
4✔
203
            out_meta = src.meta.copy()
4✔
204

205
            # Check if the out_image contains valid data
206
            if out_image is None or out_image.sum() == 0:
4✔
NEW
207
                print(
×
208
                    "The area does not fit within the clipping layer or is empty."
209
                )
NEW
210
                raise OutOfBbox(
×
211
                    "The area does not fit within the clipping layer or is empty."
212
                )
213

214
            # Update the metadata
215
            out_meta.update(
4✔
216
                {
217
                    "driver": "GTiff",
218
                    "height": out_image.shape[1],
219
                    "width": out_image.shape[2],
220
                    "transform": out_transform,
221
                }
222
            )
NEW
223
        except Exception as e:
×
NEW
224
            print(e.__class__.__name__)
×
225

226
    # Save the clipped GeoTIFF
227
    with rasterio.open(clipped_raster_path, "w", **out_meta) as dest:
4✔
228
        dest.write(out_image)
4✔
229
    return clipped_raster_path
4✔
230

231

232
def raster_to_vector(output_directory, title, epsg):
5✔
233
    # Define the path to save the clipped raster file
234
    clipped_raster_path = os.path.join(output_directory, title + ".tif")
4✔
235
    # Define the path to save the vector file in gpkg format
236
    vector_path = os.path.join(output_directory, title + ".gpkg")
4✔
237

238
    # Check if clipped_raster_path is a string and the file exists
239
    if (
4✔
240
        not isinstance(clipped_raster_path, str)
241
        or not Path(clipped_raster_path).exists()
242
    ):
NEW
243
        raise FileNotFoundError(
×
244
            f"Raster file '{clipped_raster_path}' does not exist or is not a valid path."
245
        )
246

247
    # Check if vector_path is a string
248
    if not isinstance(vector_path, str):
4✔
NEW
249
        raise ValueError(f"Vector path '{vector_path}' is not a valid string.")
×
250

251
    # Check if epsg is a valid integer
252
    if not isinstance(epsg, int):
4✔
NEW
253
        raise ValueError(f"EPSG code '{epsg}' is not a valid integer.")
×
254

255
    with rasterio.open(clipped_raster_path) as src:
4✔
256
        # Read the first band of the raster image
257
        image = src.read(1)
4✔
258
        # Create a mask for non-zero values
259
        mask = image != 0
4✔
260

261
        # Extracting the CRS from the raster
262
        raster_crs = src.crs
4✔
263
        raster_transform = src.transform
4✔
264

265
        # Generate shapes (polygons) from the raster
266
        results = (
4✔
267
            {"properties": {"value": value}, "geometry": geometry}
268
            for geometry, value in shapes(
269
                image, mask=mask, transform=raster_transform
270
            )
271
        )
272

273
        # Collect the features
274
        features = list(results)
4✔
275

276
        # Convert the extracted shapes to a GeoDataFrame
277
        geometries = [shape(feature["geometry"]) for feature in features]
4✔
278
        values = [feature["properties"]["value"] for feature in features]
4✔
279
        vector = gpd.GeoDataFrame(
4✔
280
            {"geometry": geometries, "value": values}, crs=raster_crs
281
        )
282
        # Reproject the vector to the EPSG of interest
283
        vector_proj = vector.to_crs(epsg=epsg)
4✔
284

285
        # Save the Geo-dataframe as a Geopackage
286
        vector_proj.to_file(vector_path, driver="GPKG")
4✔
287
        print(
4✔
288
            f"Vector file '{vector_path}' corresponding to the georeferenced raster file is "
289
            f"generated and added to the hotmaps_output folder."
290
        )
291
        return vector_path
4✔
292

293

294
def hotmaps_request(layer_name, bbox, epsg):
5✔
295
    # Validate layer_name
296
    if not isinstance(layer_name, str):
5✔
NEW
297
        raise ValueError("layer_name must be a string.")
×
298

299
    # Validate bbox
300
    validate_bbox(bbox)
5✔
301

302
    # Validate epsg
303
    validate_epsg(epsg)
5✔
304

305
    try:
5✔
306
        title = get_hotmaps_layer_info(layer_name)[0]
5✔
307
        tag = get_hotmaps_layer_info(layer_name)[1]
5✔
308
        key = get_hotmaps_layer_info(layer_name)[2]
5✔
309
        category = get_hotmaps_layer_info(layer_name)[3]
5✔
310
    except Exception as e:
5✔
311
        print(e)
5✔
312
        return
5✔
313

314
    # Create the temp and output directory if it doesn't exist
315
    output_directory = Path("hotmaps_output")
5✔
316
    temp_directory = Path("temp_output")
5✔
317

318
    output_directory.mkdir(parents=True, exist_ok=True)
5✔
319
    temp_directory.mkdir(parents=True, exist_ok=True)
5✔
320

321
    # Download HotMaps raster temporarily in the tep directory
322
    download_hotmaps_raster(title, tag, key, category, temp_directory)
5✔
323

324
    # Clip raster based on bounding box of interest
325
    clipped_raster = clip_raster(temp_directory, title, output_directory, bbox)
4✔
326

327
    # Convert clipped raster to vector
328
    vector = raster_to_vector(output_directory, title, epsg)
4✔
329

330
    # Remove temp directory and all its contents.
331
    try:
4✔
332
        shutil.rmtree(temp_directory)
4✔
333
        print(f"Directory '{temp_directory}' and all its contents removed.")
4✔
NEW
334
    except OSError as e:
×
NEW
335
        print(f"Error: {e}")
×
336

337
    return clipped_raster, vector
4✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc