Skip to content

Grid

Sampling grids.

Grid dataclass

Bases: BaseMixin

A single survey grid.

Attributes:

Name Type Description
ident str

unique identifier
size int

grid size in cells
spacing float

size of individual cell (m)
lat0 float

reference latitude of cell (0, 0)
lon0 float

reference longitude of cell (0, 0)
cells list[float]

pollution measurements for cells
Source code in src/snailz/grid.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
@dataclass
class Grid(BaseMixin):
    """
    A single survey grid.

    Attributes:
        ident: unique identifier
        size: grid size in cells
        spacing: size of individual cell (m)
        lat0: reference latitude of cell (0, 0)
        lon0: reference longitude of cell (0, 0)
        cells: pollution measurements for cells
    """

    primary_key: ClassVar[str] = "ident"
    pivot_keys: ClassVar[set[str]] = {"cells"}
    _next_id: ClassVar[IdGeneratorType] = id_generator("G", 4)

    ident: str = ""
    size: int = 0
    spacing: float = 0.0
    lat0: float = 0.0
    lon0: float = 0.0
    cells: list[float] = field(default_factory=list)
    params: InitVar[Parameters | None] = None

    def __post_init__(self, params: Parameters | None):
        """
        Validate fields, generate unique identifier, and fill in cells.

        Args:
            params: Parameters object.

        Raises:
            ValueError: If validation fails.
        """

        validate(params is not None, "params required to initialize grid")
        validate(self.ident == "", "grid ID cannot be set externally")
        validate(self.size > 0, f"grid size must be positive not {self.size}")
        validate(
            self.spacing > 0.0, f"grid spacing must be positive not {self.spacing}"
        )
        validate_lat_lon("grid", self.lat0, self.lon0)
        validate(params is not None, "params required for initializing grid")

        self.ident = next(self._next_id)
        self.cells = [0.0 for _ in range(self.size * self.size)]
        self._fill()
        self._randomize(params)

    def __str__(self) -> str:
        """
        Convert grid values to headerless CSV text.

        Returns:
            Printable CSV string representation of grid values.
        """

        return "\n".join(
            ",".join(str(self[x, y]) for x in range(self.size))
            for y in range(self.size - 1, -1, -1)
        )

    def __getitem__(self, key: tuple[int, int]) -> Any:
        """
        Get grid element.

        Args:
            key: (x, y) coordinates.

        Returns:
            Value at that location.

        Raises:
            ValueError: If either coordinate out of range.
        """

        self._validate_coords(key)
        x, y = key
        return self.cells[x * self.size + y]

    def __setitem__(self, key: tuple[int, int], value: float):
        """
        Set grid element.

        Args:
            key: (x, y) coordinates.
            value: new value.

        Raises:
            ValueError: If either coordinate out of range.
        """

        x, y = key
        self.cells[x * self.size + y] = value

    @classmethod
    def make(cls, params: Parameters) -> list["Grid"]:
        """
        Construct multiple grids.

        Args:
            params: Parameters object.

        Returns:
            List of grids.
        """

        origins = cls._make_origins(params)
        return [
            Grid(
                size=params.grid_size,
                spacing=params.grid_spacing,
                lat0=origin[0],
                lon0=origin[1],
                params=params,
            )
            for origin in origins
        ]

    @classmethod
    def save_csv(cls, outdir: Path | str, objects: list):
        """
        Save grids as CSV. Scalar properties of all grids are saved in
        one file; grid cell values are pivoted to long form and saved
        in a separate file.

        Args:
            outdir: Output directory.
            objects: `Grid` objects to save.
        """

        super().save_csv(outdir, objects)

        with open(Path(outdir, "grid_cells.csv"), "w", newline="") as stream:
            pivoted = cls._grid_cells(objects)
            writer = cls._csv_dict_writer(stream, list(pivoted[0].keys()))
            for obj in pivoted:
                writer.writerow(obj)

    @classmethod
    def save_db(cls, db: Database, objects: list):
        """
        Save grids to database. Scalar properties of all grids are
        saved in one table; grid cell values are pivoted to long form
        and saved in a separate table.

        Args:
            db: Database connector.
            objects: `Grid` objects to save.
        """

        super().save_db(db, objects)

        table = db["grid_cells"]
        table.insert_all(  # type: ignore[possibly-missing-attribute]
            cls._grid_cells(objects),
            pk=("grid_id", "lat", "lon"),
            foreign_keys=[("grid_id", "grid", "ident")],
        )

    @classmethod
    def table_name(cls) -> str:
        """Database table name."""

        return "grid"

    @classmethod
    def _grid_cells(cls, grids):
        """
        Pivot grid cell values to long format for persistence.

        Args:
            grids: `Grid` objects to pivot.
        """

        return [
            {"grid_id": g.ident, **g.lat_lon(x, y, True), "value": g[x, y]}
            for g in grids
            for x in range(g.size)
            for y in range(g.size)
        ]

    @classmethod
    def _make_origins(cls, params):
        """
        Construct grid origins.

        Args:
            params: Parameters object.

        Returns:
            List of `params.num_grids` (lat, lon) origins.
        """

        possible = list(
            itertools.product(range(params.num_grids), range(params.num_grids))
        )
        actual = random.sample(possible, k=params.num_grids)
        dim = params.grid_size * params.grid_spacing * params.grid_separation
        return [lat_lon(params.lat0, params.lon0, x * dim, y * dim) for x, y in actual]

    def as_image(self, scale: float) -> Image.Image:
        """
        Convert grid to image.

        Args:
            scale: Scaling factor for grid values to ensure largest is black.

        Returns:
            `Image` object.
        """

        scale = scale or self.min_max()[1] or 1.0
        img_size = (self.size * CELL_SIZE) + ((self.size + 1) * BORDER_WIDTH)
        array = np.full((img_size, img_size), WHITE, dtype=np.uint8)
        spacing = CELL_SIZE + BORDER_WIDTH
        for ix, x in enumerate(range(BORDER_WIDTH, img_size, spacing)):
            for iy, y in enumerate(range(BORDER_WIDTH, img_size, spacing)):
                color = WHITE - math.floor(WHITE * self[ix, iy] / scale)
                array[y : y + CELL_SIZE + 1, x : x + CELL_SIZE + 1] = color

        return Image.fromarray(array)

    def lat_lon(
        self, x: int, y: int, as_dict: bool = False
    ) -> tuple[float, float] | dict[str, float]:
        """
        Calculate latitude and longitude of grid cell.

        Args:
            x: Grid X coordinate.
            y: Grid Y coordinate.
            as_dict: Return result as dict instead of pair.

        Returns:
            `(lat, lon)` pair or `{"lat": lat, "lon": lon}` dictionary.

        Raises:
            ValueError: if either coordinate out of range.
        """

        self._validate_coords((x, y))
        lat, lon = lat_lon(self.lat0, self.lon0, x * self.spacing, y * self.spacing)
        return {"lat": lat, "lon": lon} if as_dict else (lat, lon)

    def min_max(self) -> tuple[float, float]:
        """
        Find smallest and largest values in grid.

        Returns:
            `(min, max)` pair.
        """

        return min(self.cells), max(self.cells)

    def _fill(self):
        """Fill in grid values using random walk."""

        center = self.size // 2
        size_1 = self.size - 1
        x, y = center, center

        while (x != 0) and (y != 0) and (x != size_1) and (y != size_1):
            self[x, y] += 1
            m = random.choice(MOVES)
            x += m[0]
            y += m[1]

    def _randomize(self, params: Parameters | None):
        """
        Randomize values in grid after filling.

        Args:
            params: Parameters object.
        """

        assert params is not None
        for i, val in enumerate(self.cells):
            if val > 0.0:
                self.cells[i] = round(
                    abs(random.normalvariate(self.cells[i], params.grid_std_dev)),
                    GRID_PRECISION,
                )
            else:
                self.cells[i] = 0.0

    def _validate_coords(self, key: tuple[int, int]):
        """
        Validate (x, y) coordinate pair.

        Raises:
            ValueError: If either coordinate is out of range.
        """
        validate(0 <= key[0] < self.size, "invalid X coordinate {key[0]}")
        validate(0 <= key[1] < self.size, "invalid Y coordinate {key[1]}")

__post_init__(params)

Validate fields, generate unique identifier, and fill in cells.

Parameters:

Name Type Description Default
params Parameters | None

Parameters object.

 required

Raises:

Type Description
ValueError

If validation fails.
Source code in src/snailz/grid.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def __post_init__(self, params: Parameters | None):
    """
    Validate fields, generate unique identifier, and fill in cells.

    Args:
        params: Parameters object.

    Raises:
        ValueError: If validation fails.
    """

    validate(params is not None, "params required to initialize grid")
    validate(self.ident == "", "grid ID cannot be set externally")
    validate(self.size > 0, f"grid size must be positive not {self.size}")
    validate(
        self.spacing > 0.0, f"grid spacing must be positive not {self.spacing}"
    )
    validate_lat_lon("grid", self.lat0, self.lon0)
    validate(params is not None, "params required for initializing grid")

    self.ident = next(self._next_id)
    self.cells = [0.0 for _ in range(self.size * self.size)]
    self._fill()
    self._randomize(params)

__str__()

Convert grid values to headerless CSV text.

Returns:

Type Description
str

Printable CSV string representation of grid values.
Source code in src/snailz/grid.py 
88
89
90
91
92
93
94
95
96
97
98
99
def __str__(self) -> str:
    """
    Convert grid values to headerless CSV text.

    Returns:
        Printable CSV string representation of grid values.
    """

    return "\n".join(
        ",".join(str(self[x, y]) for x in range(self.size))
        for y in range(self.size - 1, -1, -1)
    )

__getitem__(key)

Get grid element.

Parameters:

Name Type Description Default
key tuple[int, int]

(x, y) coordinates.

 required

Returns:

Type Description
Any

Value at that location.

Raises:

Type Description
ValueError

If either coordinate out of range.
Source code in src/snailz/grid.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def __getitem__(self, key: tuple[int, int]) -> Any:
    """
    Get grid element.

    Args:
        key: (x, y) coordinates.

    Returns:
        Value at that location.

    Raises:
        ValueError: If either coordinate out of range.
    """

    self._validate_coords(key)
    x, y = key
    return self.cells[x * self.size + y]

__setitem__(key, value)

Set grid element.

Parameters:

Name Type Description Default
key tuple[int, int]

(x, y) coordinates.

 required
value float

new value.

 required

Raises:

Type Description
ValueError

If either coordinate out of range.
Source code in src/snailz/grid.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def __setitem__(self, key: tuple[int, int], value: float):
    """
    Set grid element.

    Args:
        key: (x, y) coordinates.
        value: new value.

    Raises:
        ValueError: If either coordinate out of range.
    """

    x, y = key
    self.cells[x * self.size + y] = value

make(params) classmethod

Construct multiple grids.

Parameters:

Name Type Description Default
params Parameters

Parameters object.

 required

Returns:

Type Description
list[Grid]

List of grids.
Source code in src/snailz/grid.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
@classmethod
def make(cls, params: Parameters) -> list["Grid"]:
    """
    Construct multiple grids.

    Args:
        params: Parameters object.

    Returns:
        List of grids.
    """

    origins = cls._make_origins(params)
    return [
        Grid(
            size=params.grid_size,
            spacing=params.grid_spacing,
            lat0=origin[0],
            lon0=origin[1],
            params=params,
        )
        for origin in origins
    ]

save_csv(outdir, objects) classmethod

Save grids as CSV. Scalar properties of all grids are saved in one file; grid cell values are pivoted to long form and saved in a separate file.

Parameters:

Name Type Description Default
outdir Path | str

Output directory.

 required
objects list

Grid objects to save.

 required
Source code in src/snailz/grid.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@classmethod
def save_csv(cls, outdir: Path | str, objects: list):
    """
    Save grids as CSV. Scalar properties of all grids are saved in
    one file; grid cell values are pivoted to long form and saved
    in a separate file.

    Args:
        outdir: Output directory.
        objects: `Grid` objects to save.
    """

    super().save_csv(outdir, objects)

    with open(Path(outdir, "grid_cells.csv"), "w", newline="") as stream:
        pivoted = cls._grid_cells(objects)
        writer = cls._csv_dict_writer(stream, list(pivoted[0].keys()))
        for obj in pivoted:
            writer.writerow(obj)

save_db(db, objects) classmethod

Save grids to database. Scalar properties of all grids are saved in one table; grid cell values are pivoted to long form and saved in a separate table.

Parameters:

Name Type Description Default
db Database

Database connector.

 required
objects list

Grid objects to save.

 required
Source code in src/snailz/grid.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@classmethod
def save_db(cls, db: Database, objects: list):
    """
    Save grids to database. Scalar properties of all grids are
    saved in one table; grid cell values are pivoted to long form
    and saved in a separate table.

    Args:
        db: Database connector.
        objects: `Grid` objects to save.
    """

    super().save_db(db, objects)

    table = db["grid_cells"]
    table.insert_all(  # type: ignore[possibly-missing-attribute]
        cls._grid_cells(objects),
        pk=("grid_id", "lat", "lon"),
        foreign_keys=[("grid_id", "grid", "ident")],
    )

table_name() classmethod

Database table name.

Source code in src/snailz/grid.py 
199
200
201
202
203
@classmethod
def table_name(cls) -> str:
    """Database table name."""

    return "grid"

as_image(scale)

Convert grid to image.

Parameters:

Name Type Description Default
scale float

Scaling factor for grid values to ensure largest is black.

 required

Returns:

Type Description
Image

Image object.
Source code in src/snailz/grid.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def as_image(self, scale: float) -> Image.Image:
    """
    Convert grid to image.

    Args:
        scale: Scaling factor for grid values to ensure largest is black.

    Returns:
        `Image` object.
    """

    scale = scale or self.min_max()[1] or 1.0
    img_size = (self.size * CELL_SIZE) + ((self.size + 1) * BORDER_WIDTH)
    array = np.full((img_size, img_size), WHITE, dtype=np.uint8)
    spacing = CELL_SIZE + BORDER_WIDTH
    for ix, x in enumerate(range(BORDER_WIDTH, img_size, spacing)):
        for iy, y in enumerate(range(BORDER_WIDTH, img_size, spacing)):
            color = WHITE - math.floor(WHITE * self[ix, iy] / scale)
            array[y : y + CELL_SIZE + 1, x : x + CELL_SIZE + 1] = color

    return Image.fromarray(array)

lat_lon(x, y, as_dict=False)

Calculate latitude and longitude of grid cell.

Parameters:

Name Type Description Default
x int

Grid X coordinate.

 required
y int

Grid Y coordinate.

 required
as_dict bool

Return result as dict instead of pair.

 False

Returns:

Type Description
tuple[float, float] | dict[str, float]

(lat, lon) pair or {"lat": lat, "lon": lon} dictionary.

Raises:

Type Description
ValueError

if either coordinate out of range.
Source code in src/snailz/grid.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def lat_lon(
    self, x: int, y: int, as_dict: bool = False
) -> tuple[float, float] | dict[str, float]:
    """
    Calculate latitude and longitude of grid cell.

    Args:
        x: Grid X coordinate.
        y: Grid Y coordinate.
        as_dict: Return result as dict instead of pair.

    Returns:
        `(lat, lon)` pair or `{"lat": lat, "lon": lon}` dictionary.

    Raises:
        ValueError: if either coordinate out of range.
    """

    self._validate_coords((x, y))
    lat, lon = lat_lon(self.lat0, self.lon0, x * self.spacing, y * self.spacing)
    return {"lat": lat, "lon": lon} if as_dict else (lat, lon)

min_max()

Find smallest and largest values in grid.

Returns:

Type Description
tuple[float, float]

(min, max) pair.
Source code in src/snailz/grid.py 
284
285
286
287
288
289
290
291
292
def min_max(self) -> tuple[float, float]:
    """
    Find smallest and largest values in grid.

    Returns:
        `(min, max)` pair.
    """

    return min(self.cells), max(self.cells)