Coverage for src/rok4_tools/tmsizer_utils/processors/reduce.py: 62%
86 statements
« prev ^ index » next coverage.py v7.6.1, created at 2024-11-06 17:15 +0000
1"""Provide processor combining data. Input data is transformed and accumulated and the result is yielded juste once, at the end.
3The module contains the following classes:
5- `CountProcessor` - Count the number of item read from the input processor
6- `HeatmapProcessor` - Generate an heat map with all point coordinate read from the input processor
7"""
8import sys
9import numpy as np
10from typing import Dict, List, Tuple, Union, Iterator
11from math import floor
13import rasterio
14from rasterio.transform import from_origin
15from rasterio.io import MemoryFile
16from rasterio import logging
17logging.getLogger().setLevel(logging.ERROR)
19from rok4_tools.tmsizer_utils.processors.processor import Processor
21class CountProcessor(Processor):
22 """Processor counting the number of item read from the input processor
24 All input formats are allowed and output format is "COUNT"
26 Attributes:
27 __input (Processor): Processor from which data is read
28 """
30 def __init__(self, input: Processor):
31 """Constructor method
33 Args:
34 input (Processor): Processor from which data is read
35 """
37 super().__init__("COUNT")
39 self.__input = input
41 def process(self) -> Iterator[int]:
42 """Count number of input item
44 Yield count only once at the end
46 Examples:
48 Get input items count
50 from rok4_tools.tmsizer_utils.processors.reduce import CountProcessor
52 try:
53 # Creation of Processor source_processor
54 processor = CountProcessor(source_processor, level="15", format="GeoJSON" )
55 count = processor.process().__next__()
56 print(f"{count} items in source_processor")
58 except Exception as e:
59 print("{e}")
61 Yields:
62 Iterator[int]: the count of input items
63 """
65 for item in self.__input.process():
66 self._processed += 1
68 yield self._processed
70 def __str__(self) -> str:
71 return f"CountProcessor : {self._processed} {self.__input.format} items counted"
74class HeatmapProcessor(Processor):
75 """Processor counting the number of item read from the input processor
77 Accepted input format is "POINT" and output format is "FILELIKE". Output file-like object is an in-memory GeoTIFF
79 Attributes:
80 __input (Processor): Processor from which data is read
81 __bbox (Tuple[float, float, float, float]): Bounding box of the heat map (xmin,ymin,xmax,ymax)
82 __dimensions (Tuple[int, int]): Pixel dimensions of the heat map (width, height)
83 __resolutions (Tuple[float, float]): Pixel resolution (x resolution, y resolution)
84 """
86 input_formats_allowed = ["POINT"]
88 areas = {
89 "EPSG:3857": {
90 "FXX": [-649498, 5048729, 1173394, 6661417]
91 }
92 }
94 def __init__(self, input: Processor, **options):
95 """Constructor method
97 Args:
98 input (Processor): Processor from which data is read
99 **bbox (str): Bounding box of the heat map. Format "<xmin>,<ymin>,<xmax>,<ymax>". Coordinates system have to be the pivot TMS' one
100 **dimensions (str): Pixel dimensions of the heat map.Format "<width>x<height>"
102 Raises:
103 ValueError: Input format is not allowed
104 KeyError: A mandatory option is missing
105 ValueError: A mandatory option is not valid
106 ValueError: Provided level is not in the pivot TMS
107 """
109 if input.format not in self.input_formats_allowed:
110 raise Exception(f"Input format {input.format} is not handled for HeatmapProcessor : allowed formats are {self.input_formats_allowed}")
112 super().__init__("FILELIKE")
114 self.__input = input
116 if "bbox" in options:
117 try:
118 self.__bbox = [float(c) for c in options["bbox"].split(",")]
119 self.__bbox = tuple(self.__bbox)
120 except ValueError as e:
121 raise ValueError(f"Option 'bbox' contains non float values : {e}")
123 if len(self.__bbox) != 4 or self.__bbox[0] >= self.__bbox[2] or self.__bbox[1] >= self.__bbox[3]:
124 raise ValueError(f"Option 'bbox' have to be provided with format <xmin>,<ymin>,<xmax>,<ymax> (floats, min < max)")
126 elif "area" in options:
127 try:
128 self.__bbox = self.areas[self.tms.srs][options["area"]]
129 except KeyError as e:
130 if self.tms.srs in self.areas:
131 raise ValueError(f"Area '{options['area']}' is not available for the TMS coordinates system ({self.tms.srs}): available areas are {', '.join(self.areas[self.tms.srs].keys())}")
132 else :
133 raise ValueError(f"No defined areas for the TMS coordinates system ({self.tms.srs})")
134 else:
135 raise KeyError(f"Option 'bbox' or 'area' is required for a heatmap processing")
137 if "dimensions" in options:
138 try:
139 self.__dimensions = [int(d) for d in options["dimensions"].split("x")]
140 self.__dimensions = tuple(self.__dimensions)
141 except ValueError as e:
142 raise ValueError(f"Option 'dimensions' contains non integer values : {e}")
144 if len(self.__dimensions) != 2 or self.__dimensions[0] <= 0 or self.__dimensions[1] <= 0:
145 raise ValueError(f"Option 'dimensions' have to be provided with format <width>x<height> (positive integers)")
147 self.__resolutions = (
148 (self.__bbox[2] - self.__bbox[0]) / self.__dimensions[0],
149 (self.__bbox[3] - self.__bbox[1]) / self.__dimensions[1]
150 )
151 elif "level" in options:
152 level = self.tms.get_level(options["level"])
153 if level is None:
154 raise ValueError(f"The provided level '{options['dimensions']}' (to have one pixel per tile) is not in the TMS")
156 # On va caler la bbox pour qu'elle coïncide avec les limites de tuiles du niveau demandé
157 (col_min, row_min, col_max, row_max) = level.bbox_to_tiles(self.__bbox)
159 # Calage du coin en bas à gauche
160 (xmin, ymin, xmax, ymax) = level.tile_to_bbox(col_min, row_max)
161 self.__bbox[0] = xmin
162 self.__bbox[1] = ymin
164 # Calage du coin en haut à droite
165 (xmin, ymin, xmax, ymax) = level.tile_to_bbox(col_max, row_min)
166 self.__bbox[2] = xmax
167 self.__bbox[3] = ymax
169 self.__resolutions = (
170 xmax - xmin,
171 ymax - ymin
172 )
174 self.__dimensions = (
175 int((self.__bbox[2] - self.__bbox[0]) / self.__resolutions[0]),
176 int((self.__bbox[3] - self.__bbox[1]) / self.__resolutions[1])
177 )
179 else:
180 raise KeyError(f"Option 'dimensions' or 'level' is required for a heatmap processing")
182 if self.__dimensions[0] > 10000 or self.__dimensions[1] > 10000:
183 raise ValueError(f"Heatmap dimensions have to be less than 10 000 x 10 000: here it's {self.__dimensions}")
185 def process(self) -> Iterator[MemoryFile]:
186 """Read point coordinates from the input processor and accumule them as a heat map
188 Points outsides the provided bounding box are ignored.
190 Examples:
192 Get intersecting tiles' indices
194 from rok4_tools.tmsizer_utils.processors.reduce import HeatmapProcessor
196 try:
197 # Creation of Processor source_processor with format POINT
199 processor = HeatmapProcessor(source_processor, bbox="65000,6100000,665000,6500000", dimensions="600x400" )
200 f = processor.process().__next__()
202 with open("hello.txt", "w") as my_file:
203 my_file.write(f.read())
205 except Exception as e:
206 print("{e}")
208 Yields:
209 Iterator[rasterio.io.MemoryFile]: In-memory GeoTIFF
210 """
212 data = np.zeros((self.__dimensions[1], self.__dimensions[0]), dtype=np.uint32)
214 if self.__input.format == "POINT":
216 for item in self.__input.process():
217 self._processed += 1
219 (x_center, y_center) = item
221 if x_center > self.__bbox[2] or y_center > self.__bbox[3] or x_center < self.__bbox[0] or y_center < self.__bbox[1]:
222 continue
224 pcol = floor((x_center - self.__bbox[0]) / self.__resolutions[0])
225 prow = floor((self.__bbox[3] - y_center) / self.__resolutions[1])
227 data[prow][pcol] += 1
229 memfile = MemoryFile()
230 with memfile.open(
231 driver='GTiff',
232 height=data.shape[0],
233 width=data.shape[1],
234 count=1,
235 dtype=data.dtype,
236 crs=rasterio.CRS.from_string(self.tms.srs),
237 nodata=0,
238 transform=from_origin(self.__bbox[0], self.__bbox[3], self.__resolutions[0], self.__resolutions[1]),
239 ) as dataset:
240 dataset.write(data, indexes=1)
242 yield memfile
244 def __str__(self) -> str:
245 return f"HeatmapProcessor : {self._processed} hits on image with dimensions {self.__dimensions} and bbox {self.__bbox} (resolutions {self.__resolutions})"