Skip to content

Reference Manual

Main Pipeline

CLI tool for estimating solar irradiance on buildings from digital surface models.

detect_grass_base()

Auto-detect GRASS GIS installation path based on operating system.

Source code in src/pipeline.py
40
41
42
43
44
45
46
47
def detect_grass_base():
    """Auto-detect GRASS GIS installation path based on operating system."""
    if platform.system() == "Darwin":
        return "/Applications/GRASS-8.4.app/Contents/Resources"
    elif platform.system() == "Linux":
        return "/usr/lib/grass84"
    else:
        return None

Utility Modules

Building Outlines

Building outlines utilities for GRASS GIS.

This module contains helper functions to import building footprint vectors into a GRASS mapset, apply building masks so raster operations affect only building areas, extract masked raster values into building-specific rasters, and export final multi-band GeoTIFFs that include computed rasters (for example, solar irradiance) together with slope and aspect bands.

apply_building_mask(building_vector, grass_module)

Apply a raster mask based on building outlines.

The mask created by r.mask restricts subsequent raster operations to the area covered by the given vector. Typical workflows call this beforehand to copy or compute values only for buildings.

Parameters:

Name Type Description Default
building_vector str

Name of the building footprint vector in GRASS.

required
output_name

Friendly name returned by the function (no effect on mask).

required
grass_module Any

The GRASS Python scripting Module class.

required

Returns:

Type Description
None

None

Source code in src/utils/building_outlines.py
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def apply_building_mask(building_vector: str, grass_module: Any) -> None:
    """Apply a raster mask based on building outlines.

    The mask created by `r.mask` restricts subsequent raster operations to the
    area covered by the given vector. Typical workflows call this beforehand to copy
    or compute values only for buildings.

    Args:
        building_vector: Name of the building footprint vector in GRASS.
        output_name: Friendly name returned by the function (no effect on mask).
        grass_module: The GRASS Python scripting Module class.

    Returns:
        None
    """
    # Create a raster mask from the vector; overwrite any existing mask
    r_mask = grass_module("r.mask", vector=building_vector, overwrite=True)
    r_mask.run()

calculate_outline_raster(solar_irradiance_raster, building_vector, output_name, grass_module)

Create a raster containing values only for building outlines.

Parameters:

Name Type Description Default
solar_irradiance_raster str

Name of the solar irradiance raster to be masked.

required
building_vector str

Name of the building footprint vector to use for masking.

required
output_name str

Name to assign to the resulting building-only raster.

required
grass_module Any

The GRASS Python scripting Module class.

required

Returns:

Type Description
str

The GRASS raster name.

Source code in src/utils/building_outlines.py
 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
def calculate_outline_raster(
    solar_irradiance_raster: str,
    building_vector: str,
    output_name: str,
    grass_module: Any,
) -> str:
    """Create a raster containing values only for building outlines.

    Args:
        solar_irradiance_raster: Name of the solar irradiance raster to be masked.
        building_vector: Name of the building footprint vector to use for masking.
        output_name: Name to assign to the resulting building-only raster.
        grass_module: The GRASS Python scripting Module class.

    Returns:
        The GRASS raster name.
    """
    # Apply mask using the building vector
    apply_building_mask(building_vector, grass_module=grass_module)

    # Copy values from the source raster into the masked output raster
    r_mapcalc = grass_module(
        "r.mapcalc",
        expression=f"{output_name} = {solar_irradiance_raster}",
        overwrite=True,
    )
    r_mapcalc.run()

    return output_name

export_final_raster(raster_name, slope, aspect, output_tif, grass_module, output_dir=None)

Export a multi-band GeoTIFF containing raster, slope, and aspect.

This function
  • Creates an imagery group containing the three rasters using i.group.
  • Calls r.out.gdal to export the group as a multi-band GeoTIFF.
  • Removes the temporary imagery group.

Parameters:

Name Type Description Default
raster_name str

Name of the primary raster to export (e.g., building-only GHI).

required
slope str

Name of the slope raster to include as a band.

required
aspect str

Name of the aspect raster to include as a band.

required
output_tif str

Filename (or path) for the output GeoTIFF file to create. If output_dir is provided, the file is written inside that directory.

required
grass_module Any

The GRASS Python scripting Module class.

required
output_dir Optional[Path]

Optional directory in which to write the output file. When provided, output_tif is treated as a filename and joined with output_dir to form the full path.

None

Returns:

Type Description
str

The full output_tif path for convenience.

Source code in src/utils/building_outlines.py
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
def export_final_raster(
    raster_name: str,
    slope: str,
    aspect: str,
    output_tif: str,
    grass_module: Any,
    output_dir: Optional[Path] = None,
) -> str:
    """Export a multi-band GeoTIFF containing raster, slope, and aspect.

    This function:
      - Creates an imagery group containing the three rasters using `i.group`.
      - Calls `r.out.gdal` to export the group as a multi-band GeoTIFF.
      - Removes the temporary imagery group.

    Args:
        raster_name: Name of the primary raster to export (e.g., building-only GHI).
        slope: Name of the slope raster to include as a band.
        aspect: Name of the aspect raster to include as a band.
        output_tif: Filename (or path) for the output GeoTIFF file to create.
            If output_dir is provided, the file is written inside that directory.
        grass_module: The GRASS Python scripting Module class.
        output_dir: Optional directory in which to write the output file.
            When provided, output_tif is treated as a filename and joined with
            output_dir to form the full path.

    Returns:
        The full `output_tif` path for convenience.
    """
    # Create a temporary imagery group containing the three rasters. This groups
    # the rasters into a single multi-band dataset.
    group_name = f"{raster_name}_group"
    i_group = grass_module(
        "i.group",
        group=group_name,
        input=f"{raster_name},{slope},{aspect}",
    )
    i_group.run()

    # Resolve the output path, writing into output_dir when provided.
    if output_dir is not None:
        output_path = str(Path(output_dir) / output_tif)
    else:
        output_path = output_tif

    # TFW = World File containing georeferencing info
    # LZW = Lempel-Ziv-Welch lossless compression algorithm
    r_out_multiband = grass_module(
        "r.out.gdal",
        input=group_name,
        output=output_path,
        format="GTiff",
        createopt="TFW=YES,COMPRESS=LZW",
        overwrite=True,
    )
    r_out_multiband.run()

    # Clean up the temporary group to avoid leaving workspace state behind.
    g_remove = grass_module(
        "g.remove",
        type="group",
        name=group_name,
        flags="f",  # force removal without extra prompts
    )
    g_remove.run()

    return output_path

load_building_outlines(shapefile, output_name, grass_module)

Import building outlines (vector) from a shapefile into GRASS.

This uses the v.in.ogr GRASS module to import a vector dataset from a shapefile into the current GRASS mapset.

Parameters:

Name Type Description Default
shapefile str

Path to the directory containing a building footprints shapefile.

required
output_name str

Name to assign to the vector map inside GRASS.

required
grass_module Any

The GRASS Python scripting Module class.

required

Returns:

Type Description
str

The GRASS vector name.

Source code in src/utils/building_outlines.py
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def load_building_outlines(shapefile: str, output_name: str, grass_module: Any) -> str:
    """Import building outlines (vector) from a shapefile into GRASS.

    This uses the `v.in.ogr` GRASS module to import a vector dataset from a
    shapefile into the current GRASS mapset.

    Args:
        shapefile: Path to the directory containing a building footprints shapefile.
        output_name: Name to assign to the vector map inside GRASS.
        grass_module: The GRASS Python scripting Module class.

    Returns:
        The GRASS vector name.
    """
    # Import vector into GRASS; overwrite if a vector with the same name exists
    v_in = grass_module("v.in.ogr", input=shapefile, output=output_name, overwrite=True)
    v_in.run()

    return output_name

remove_masks(grass_module)

Remove any active raster mask(s) in the current GRASS session.

This helper attempts to reset raster masking by calling r.mask -r.

Parameters:

Name Type Description Default
grass_module Any

The GRASS Python scripting Module class.

required

Returns:

Type Description
None

None

Raises:

Type Description
Exception

if removing the mask encounters an error.

Source code in src/utils/building_outlines.py
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def remove_masks(grass_module: Any) -> None:
    """Remove any active raster mask(s) in the current GRASS session.

    This helper attempts to reset raster masking by calling `r.mask -r`.

    Args:
        grass_module: The GRASS Python scripting Module class.

    Returns:
        None

    Raises:
        Exception: if removing the mask encounters an error.
    """
    try:
        # The 'r' flag removes the active mask
        r_mask = grass_module("r.mask", flags="r")
        r_mask.run()
    except Exception as e:  # Don't break the workflow if it can't remove the mask
        logger.warning("Error removing GRASS mask: %s", e)

DSM Utilities

Digital Surface Model (DSM) utilities.

This module contains helper functions for working with DSM rasters, using GDAL and GRASS GIS.

High-level responsibilities: - Attaching virtual rasters (VRT) to GRASS as external rasters. - Merging tiled DSM GeoTIFFs into a single VRT (optionally can be translated to a GeoTIFF). - Calculating slope and aspect rasters from a DSM. - Filtering rasters.

calculate_slope_aspect_rasters(dsm, grass_module)

Compute slope and aspect rasters from a DSM using GRASS r.slope.aspect.

Parameters:

Name Type Description Default
dsm str

Name of the DSM raster in the GRASS mapset.

required
grass_module Any

The GRASS Python scripting Module class.

required

Returns:

Type Description
Tuple[str, str]

A tuple (aspect_raster_name, slope_raster_name).

Source code in src/utils/dsm.py
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def calculate_slope_aspect_rasters(dsm: str, grass_module: Any) -> Tuple[str, str]:
    """Compute slope and aspect rasters from a DSM using GRASS `r.slope.aspect`.

    Args:
        dsm: Name of the DSM raster in the GRASS mapset.
        grass_module: The GRASS Python scripting Module class.

    Returns:
        A tuple `(aspect_raster_name, slope_raster_name)`.
    """
    r_slope_aspect = grass_module(
        "r.slope.aspect",
        elevation=dsm,
        slope=f"{dsm}_slope",
        aspect=f"{dsm}_aspect",
        format="degrees",
        precision="FCELL",
        a=True,  # compute aspect
        nprocs=16,
        overwrite=True,
    )
    r_slope_aspect.run()

    return f"{dsm}_aspect", f"{dsm}_slope"

filter_raster_by_slope(input_raster, slope_raster, max_slope_degrees, output_name, grass_module)

Filter input_raster to only keep pixels where slope <= max_slope_degrees.

This function uses a r.mapcalc expression to produce a masked raster where any pixel with slope greater than max_slope_degrees is set to NULL (GRASS NULL) and valid pixels retain their original value.

Example expression used

output = if(slope_raster <= max_slope_degrees, input_raster, null())

Parameters:

Name Type Description Default
input_raster str

GRASS raster name containing the values to be filtered.

required
slope_raster str

GRASS raster name containing slope in degrees.

required
max_slope_degrees float

Maximum allowed slope (inclusive). Pixels with slope greater than this value will be masked to NULL.

required
output_name str

Name for the output raster.

required
grass_module Any

The GRASS Python scripting Module class.

required

Returns:

Type Description
str

The name of the output raster.

Source code in src/utils/dsm.py
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
def filter_raster_by_slope(
    input_raster: str,
    slope_raster: str,
    max_slope_degrees: float,
    output_name: str,
    grass_module: Any,
) -> str:
    """Filter `input_raster` to only keep pixels where slope <= max_slope_degrees.

    This function uses a `r.mapcalc` expression to produce a masked
    raster where any pixel with slope greater than `max_slope_degrees` is set
    to NULL (GRASS NULL) and valid pixels retain their original value.

    Example expression used:
        output = if(slope_raster <= max_slope_degrees, input_raster, null())

    Args:
        input_raster: GRASS raster name containing the values to be filtered.
        slope_raster: GRASS raster name containing slope in degrees.
        max_slope_degrees: Maximum allowed slope (inclusive). Pixels with slope
            greater than this value will be masked to NULL.
        output_name: Name for the output raster.
        grass_module: The GRASS Python scripting Module class.

    Returns:
        The name of the output raster.
    """
    # Build and run the r.mapcalc expression to mask out steep slopes
    expression = (
        f"{output_name} = if({slope_raster} <= {max_slope_degrees}, "
        f"{input_raster}, null())"
    )
    r_mapcalc = grass_module("r.mapcalc", expression=expression, overwrite=True)
    r_mapcalc.run()

    return output_name

load_virtual_raster_into_grass(input_vrt, output_name, grass_module)

Attach a VRT (virtual raster) to GRASS using r.external and set region.

Using r.external avoids copying data into the GRASS database; the VRT is referenced externally which is faster and uses no additional disk space.

Parameters:

Name Type Description Default
input_vrt str

Path to the VRT file on disk.

required
output_name str

The raster name to expose inside GRASS.

required
grass_module Any

The GRASS Python scripting Module class.

required

Returns:

Type Description
str

The GRASS raster name.

Source code in src/utils/dsm.py
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
def load_virtual_raster_into_grass(
    input_vrt: str, output_name: str, grass_module: Any
) -> str:
    """Attach a VRT (virtual raster) to GRASS using `r.external` and set region.

    Using `r.external` avoids copying data into the GRASS database; the VRT is
    referenced externally which is faster and uses no additional disk space.

    Args:
        input_vrt: Path to the VRT file on disk.
        output_name: The raster name to expose inside GRASS.
        grass_module: The GRASS Python scripting Module class.

    Returns:
        The GRASS raster name.
    """
    # Register the VRT as an external raster
    r_external = grass_module(
        "r.external", input=input_vrt, output=output_name, band=1, overwrite=True
    )
    r_external.run()

    # Print and set the region to match the attached raster
    g_region = grass_module("g.region", raster=output_name, flags="p")
    g_region.run()

    return output_name

merge_rasters(dsm_file_glob, area_name, output_dir)

Merge tiled DSM files into a single VRT using GDAL.

This function discovers DSM tiles using a glob pattern, builds a GDAL VRT (virtual raster) that mosaics them together, and returns the VRT filename. The function intentionally leaves the VRT on disk instead of translating to a final GeoTIFF so callers can decide on translation parameters (compression, data type, nodata handling) or feed the VRT directly to GRASS via r.external.

Parameters:

Name Type Description Default
dsm_file_glob str

Glob pattern matching input DSM tiles.

required
area_name str

Prefix to use for the generated VRT filename.

required

Returns:

Type Description
str

The path to the generated VRT file.

Raises:

Type Description
FileNotFoundError

If the glob pattern matches no files.

RuntimeError

If GDAL fails to create the VRT for any reason.

Source code in src/utils/dsm.py
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def merge_rasters(dsm_file_glob: str, area_name: str, output_dir: Path) -> str:
    """Merge tiled DSM files into a single VRT using GDAL.

    This function discovers DSM tiles using a glob pattern, builds a GDAL VRT
    (virtual raster) that mosaics them together, and returns the VRT filename.
    The function intentionally leaves the VRT on disk instead of translating to
    a final GeoTIFF so callers can decide on translation parameters (compression,
    data type, nodata handling) or feed the VRT directly to GRASS via `r.external`.

    Args:
        dsm_file_glob: Glob pattern matching input DSM tiles.
        area_name: Prefix to use for the generated VRT filename.

    Returns:
        The path to the generated VRT file.

    Raises:
        FileNotFoundError: If the glob pattern matches no files.
        RuntimeError: If GDAL fails to create the VRT for any reason.
    """
    # Find input files using the glob pattern
    dsm_files = glob.glob(dsm_file_glob)
    if not dsm_files:
        raise FileNotFoundError(f"🚫 No files found for pattern: {dsm_file_glob}")

    # Build a Virtual Raster (VRT). Use nearest-neighbor resampling by default
    try:
        vrt_options = gdal.BuildVRTOptions(resampleAlg=gdal.GRA_NearestNeighbour)
        vrt_path = f"{str(output_dir)}/{area_name}_merged.vrt"
        gdal.BuildVRT(vrt_path, dsm_files, options=vrt_options)
    except Exception as e:
        # Propagate any errors
        raise RuntimeError(
            f"🚫 Failed to build VRT from {len(dsm_files)} files: {e}"
        ) from e

    # Return the VRT path
    return vrt_path

Solar Irradiance

Solar irradiance calculation utilities for GRASS GIS.

This module provides functions for calculating solar irradiance using the GRASS GIS r.sun module. It supports both single-day calculations and interpolated multi-day calculations for seasonal analysis, as well as creating normalized coefficient rasters for WRF data adjustment.

The main workflow (if WRF data is provided) involves: 1. Calculate solar irradiance for key sample days using r.sun 2. Interpolate between key days to generate daily irradiance maps 3. Normalize to create coefficient rasters to adjust WRF data

calculate_solar_coefficients(day_irradiance_rasters, dsm, grass_module)

Calculate percent-of-max solar coefficients for each day's irradiance.

Converts irradiance values (Wh/m²) to relative coefficients (0-1) using percent-of-max normalization.

The coefficients are then used to adjust WRF irradiance data so it accounts for roof shape, shading, etc.

Parameters:

Name Type Description Default
day_irradiance_rasters dict[int, str]

Dict mapping day-of-year to irradiance raster names, as returned by calculate_solar_irradiance_interpolated().

required
dsm str

Name of the DSM raster, used as a prefix for output names.

required
grass_module

The GRASS Python scripting Module class.

required

Returns:

Type Description
dict[int, str]

Dict mapping day-of-year (int) to coefficient raster name (str).

Source code in src/utils/solar_irradiance.py
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
def calculate_solar_coefficients(
    day_irradiance_rasters: dict[int, str],
    dsm: str,
    grass_module,
) -> dict[int, str]:
    """Calculate percent-of-max solar coefficients for each day's irradiance.

    Converts irradiance values (Wh/m²) to relative coefficients (0-1)
    using percent-of-max normalization.

    The coefficients are then used to adjust WRF irradiance data so it
    accounts for roof shape, shading, etc.

    Args:
        day_irradiance_rasters: Dict mapping day-of-year to irradiance raster
            names, as returned by calculate_solar_irradiance_interpolated().
        dsm: Name of the DSM raster, used as a prefix for output names.
        grass_module: The GRASS Python scripting Module class.

    Returns:
        Dict mapping day-of-year (int) to coefficient raster name (str).
    """
    day_coefficient_rasters = {}

    for day, irradiance_raster in day_irradiance_rasters.items():
        coefficient_raster = f"{dsm}_solar_percentmax_day{day}"
        _percent_of_max_raster(irradiance_raster, coefficient_raster, grass_module)
        day_coefficient_rasters[day] = coefficient_raster

    return day_coefficient_rasters

calculate_solar_irradiance(dsm, grass_output, aspect, slope, day, step, grass_module)

Calculate solar irradiance for a single day using the GRASS r.sun module.

This function wraps the r.sun solar radiation model to compute global horizontal irradiance (GHI) under clear-sky conditions. The Linke turbidity factor is automatically interpolated based on the day of year.

Parameters:

Name Type Description Default
dsm str

Name of the input Digital Surface Model (elevation) raster in GRASS.

required
grass_output str

Name for the output global radiation raster.

required
aspect

Name of the aspect raster (direction of slope) in GRASS.

required
slope

Name of the slope raster (steepness) in GRASS.

required
day int

Day of year (1-365) for the irradiance calculation.

required
step float

Time step in hours for the r.sun calculation. Smaller values (e.g., 0.5) give more accurate results but take longer.

required
grass_module

The GRASS Python scripting Module class for running GRASS commands.

required

Returns:

Type Description
str

The name of the output global radiation raster (same as grass_output).

Note

This function assumes the GRASS computational region is already set appropriately for the DSM. The output units are Wh/m²/day.

Source code in src/utils/solar_irradiance.py
 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
def calculate_solar_irradiance(
    dsm: str,
    grass_output: str,
    aspect,
    slope,
    day: int,
    step: float,
    grass_module,
) -> str:
    """Calculate solar irradiance for a single day using the GRASS r.sun module.

    This function wraps the r.sun solar radiation model to compute global
    horizontal irradiance (GHI) under clear-sky conditions. The Linke turbidity
    factor is automatically interpolated based on the day of year.

    Args:
        dsm: Name of the input Digital Surface Model (elevation) raster in GRASS.
        grass_output: Name for the output global radiation raster.
        aspect: Name of the aspect raster (direction of slope) in GRASS.
        slope: Name of the slope raster (steepness) in GRASS.
        day: Day of year (1-365) for the irradiance calculation.
        step: Time step in hours for the r.sun calculation.
            Smaller values (e.g., 0.5) give more accurate results but take longer.
        grass_module: The GRASS Python scripting Module class for running
            GRASS commands.

    Returns:
        The name of the output global radiation raster (same as grass_output).

    Note:
        This function assumes the GRASS computational region is already set
        appropriately for the DSM. The output units are Wh/m²/day.
    """
    grass_module(
        "r.sun",
        elevation=dsm,
        aspect=aspect,
        slope=slope,
        day=day,
        step=step,
        linke_value=linke_by_day(day),
        nprocs=16,
        glob_rad=grass_output,
        overwrite=True,
    ).run()

    return grass_output

calculate_solar_irradiance_interpolated(dsm, aspect, slope, key_days, step, grass_module, export=False, output_dir=None)

Calculate interpolated solar irradiance between key sample days.

This function computes solar irradiance for a set of key days, then uses linear interpolation (via r.series.interp) to estimate irradiance for all days in the range.

Workflow
  1. Run r.sun for each key day to get irradiance values
  2. Interpolate between key days to fill in the intermediate days
  3. Sum all daily rasters to get total irradiance over the period

Parameters:

Name Type Description Default
dsm str

Name of the Digital Surface Model raster in GRASS.

required
aspect

Name of the aspect raster in GRASS.

required
slope

Name of the slope raster in GRASS.

required
key_days list[int]

List of day-of-year values (1-365) to estimate irradiance.

required
step float

Time step in hours for the r.sun calculation. Smaller values (e.g., 0.5) give more accurate results but take longer.

required
grass_module

The GRASS Python scripting Module class.

required
export bool

If True, export the summed irradiance raster as a GeoTIFF. Defaults to False.

False
output_dir Optional[Path]

Optional directory in which to write the exported GeoTIFF. Only used when export is True. When None, the file is written to the current working directory.

None

Returns:

Type Description
tuple[dict[int, str], str]

A tuple containing: - day_irradiance_rasters: Dict mapping day-of-year (int) to the irradiance raster name (str) for each day in the range from min(key_days) to max(key_days). This includes both the exact r.sun calculations for key_days and interpolated values for days in between. The caller is responsible for cleaning up these rasters when no longer needed. - summed_irradiance: Name of the raster containing the sum of all daily irradiance values (total Wh/m² over the period).

Source code in src/utils/solar_irradiance.py
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
def calculate_solar_irradiance_interpolated(
    dsm: str,
    aspect,
    slope,
    key_days: list[int],
    step: float,
    grass_module,
    export: bool = False,
    output_dir: Optional[Path] = None,
) -> tuple[dict[int, str], str]:
    """Calculate interpolated solar irradiance between key sample days.

    This function computes solar irradiance for a set of key days, then uses
    linear interpolation (via r.series.interp) to estimate irradiance for all
    days in the range.

    Workflow:
        1. Run r.sun for each key day to get irradiance values
        2. Interpolate between key days to fill in the intermediate days
        3. Sum all daily rasters to get total irradiance over the period

    Args:
        dsm: Name of the Digital Surface Model raster in GRASS.
        aspect: Name of the aspect raster in GRASS.
        slope: Name of the slope raster in GRASS.
        key_days: List of day-of-year values (1-365) to estimate irradiance.
        step: Time step in hours for the r.sun calculation.
            Smaller values (e.g., 0.5) give more accurate results but take longer.
        grass_module: The GRASS Python scripting Module class.
        export: If True, export the summed irradiance raster as a GeoTIFF.
            Defaults to False.
        output_dir: Optional directory in which to write the exported GeoTIFF.
            Only used when export is True.  When None, the file is written to
            the current working directory.

    Returns:
        A tuple containing:
            - day_irradiance_rasters: Dict mapping day-of-year (int) to the
              irradiance raster name (str) for each day in the range from
              min(key_days) to max(key_days). This includes both the exact
              r.sun calculations for key_days and interpolated values for
              days in between. The caller is responsible for cleaning up
              these rasters when no longer needed.
            - summed_irradiance: Name of the raster containing the sum of
              all daily irradiance values (total Wh/m² over the period).
    """
    # Step 1: Calculate irradiance for each key day
    key_day_rasters = []
    for day in key_days:
        day_map = calculate_solar_irradiance(
            dsm=dsm,
            grass_output=f"{dsm}_solar_irradiance_day{day}",
            aspect=aspect,
            slope=slope,
            day=day,
            step=step,
            grass_module=grass_module,
        )
        key_day_rasters.append(day_map)

    # Step 2: Interpolate to days between the key days (excluding key days themselves)
    all_days = list(range(min(key_days), max(key_days) + 1))
    key_days_set = set(key_days)
    interp_only_days = [day for day in all_days if day not in key_days_set]

    # Build the output raster mapping, starting with the key day rasters we already have
    day_irradiance_rasters = {day: key_day_rasters[i] for i, day in enumerate(key_days)}

    # Only run interpolation if there are days between the key days
    if interp_only_days:
        interp_rasters = [
            f"{dsm}_solar_irradiance_interp_day{day}" for day in interp_only_days
        ]

        # datapos = key_days positions correspond to input rasters
        # samplingpos = interp_only_days positions for output rasters
        grass_module(
            "r.series.interp",
            input=",".join(key_day_rasters),
            datapos=key_days,
            output=",".join(interp_rasters),
            samplingpos=interp_only_days,
            method="linear",
            overwrite=True,
        ).run()

        # Add interpolated rasters to the mapping
        for i, day in enumerate(interp_only_days):
            day_irradiance_rasters[day] = interp_rasters[i]

    # Step 3: Sum all rasters (key days + interpolated) to get total irradiance
    all_rasters = list(day_irradiance_rasters.values())
    summed_irradiance = f"{dsm}_solar_irradiance_interp"
    grass_module(
        "r.series",
        input=",".join(all_rasters),
        output=summed_irradiance,
        method="sum",
        overwrite=True,
    ).run()

    # Optionally export the summed raster as a GeoTIFF
    if export:
        output_filename = f"{summed_irradiance}.tif"
        if output_dir is not None:
            output_path = str(Path(output_dir) / output_filename)
        else:
            output_path = output_filename
        grass_module(
            "r.out.gdal",
            input=summed_irradiance,
            output=output_path,
            format="GTiff",
            createopt="TFW=YES,COMPRESS=LZW",
            overwrite=True,
        ).run()

    return day_irradiance_rasters, summed_irradiance

Statistics

Statistics helpers for rooftop solar calculations in GRASS GIS.

This module provides helpers to compute per-building statistics by sampling rasters (clear-sky irradiance and optional WRF-adjusted irradiance) using GRASS vector/raster database functions. The workflow implemented here is:

  1. Use v.rast.stats to compute aggregated raster statistics (sum, count) for each building polygon.
  2. Create and update attribute columns (kWh, MWh, usable sqm) using v.db.addcolumn and v.db.update.
  3. Optionally compute WRF-derived statistics and a percent loss comparison between the calculated clear-sky values and WRF measured values.
  4. Export results to a GeoPackage and optionally a CSV.

create_stats(area, building_outlines, output_dir, rooftop_raster, grass_module, wrf_raster=None, output_csv=True)

High-level workflow to produce building-level rooftop irradiance statistics.

Parameters:

Name Type Description Default
area str

Descriptive name for the area used as a prefix for output files.

required
building_outlines str

GRASS vector name containing building polygons.

required
rooftop_raster str

GRASS raster name containing rooftop irradiance (Wh).

required
grass_module Any

The GRASS Python scripting Module class.

required
wrf_raster Optional[str]

Optional GRASS raster name for WRF-adjusted irradiance.

None
output_csv bool

If True, also export a CSV summary. Defaults to True.

True

Returns:

Type Description
str

Path to the generated GeoPackage file containing building statistics.

Source code in src/utils/stats.py
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
def create_stats(
    area: str,
    building_outlines: str,
    output_dir: Path,
    rooftop_raster: str,
    grass_module: Any,
    wrf_raster: Optional[str] = None,
    output_csv: bool = True,
) -> str:
    """High-level workflow to produce building-level rooftop irradiance statistics.

    Args:
        area: Descriptive name for the area used as a prefix for output files.
        building_outlines: GRASS vector name containing building polygons.
        rooftop_raster: GRASS raster name containing rooftop irradiance (Wh).
        grass_module: The GRASS Python scripting Module class.
        wrf_raster: Optional GRASS raster name for WRF-adjusted irradiance.
        output_csv: If True, also export a CSV summary. Defaults to True.

    Returns:
        Path to the generated GeoPackage file containing building statistics.
    """
    # Compute clear-sky (calculated) stats and update vector attributes
    building_outlines = _calculate_clear_sky_stats(
        building_outlines, rooftop_raster, grass_module
    )

    # Optionally compute WRF-derived stats
    has_wrf = wrf_raster is not None
    if has_wrf:
        building_outlines = _calculate_wrf_stats(
            building_outlines, wrf_raster, grass_module
        )

    # Export combined stats to GeoPackage and optionally CSV
    gpkg_file = _export_combined_stats(
        area, building_outlines, output_dir, output_csv, grass_module, has_wrf=has_wrf
    )

    return gpkg_file

Weather (WRF)

WRF (Weather Research and Forecasting) functionality for GRASS GIS workflows.

This module contains utilities to load, reproject, clip, import and summarise WRF-derived solar variables (surface downward shortwave radiation).

High level responsibilities: - Read WRF NetCDF files via xarray and manage the CRS, including reprojections, and creating per-day WRF rasters. - Helpers to multiply WRF rasters by normalized coefficient rasters, sum per-day adjusted rasters, and produce a summed WRF raster for comparison against clear-sky modeled values.

calculate_wrf_adjusted_per_day(wrf_day_rasters, coefficient_rasters, grass_module, output_prefix='wrf_adjusted')

Multiply per-day WRF rasters by corresponding coefficient rasters.

This creates a new GRASS raster for each day named "_day{doy}".

Parameters:

Name Type Description Default
wrf_day_rasters Dict[int, str]

Mapping from day-of-year (int) to GRASS raster name containing that day's WRF data.

required
coefficient_rasters Dict[int, str]

Mapping from day-of-year (int) to GRASS raster name containing the normalized coefficient (0-1) for that day.

required
grass_module Any

The GRASS Python scripting Module class.

required
output_prefix str

Prefix to use when naming the adjusted rasters.

'wrf_adjusted'

Returns:

Type Description
Dict[int, str]

Mapping from day-of-year to the created adjusted raster names.

Source code in src/utils/wrf.py
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
def calculate_wrf_adjusted_per_day(
    wrf_day_rasters: Dict[int, str],
    coefficient_rasters: Dict[int, str],
    grass_module: Any,
    output_prefix: str = "wrf_adjusted",
) -> Dict[int, str]:
    """Multiply per-day WRF rasters by corresponding coefficient rasters.

    This creates a new GRASS raster for each day named "<output_prefix>_day{doy}".

    Args:
        wrf_day_rasters: Mapping from day-of-year (int) to GRASS raster name
            containing that day's WRF data.
        coefficient_rasters: Mapping from day-of-year (int) to GRASS raster
            name containing the normalized coefficient (0-1) for that day.
        grass_module: The GRASS Python scripting Module class.
        output_prefix: Prefix to use when naming the adjusted rasters.

    Returns:
        Mapping from day-of-year to the created adjusted raster names.
    """
    adjusted_rasters: Dict[int, str] = {}

    for day, wrf_raster in wrf_day_rasters.items():
        # If coefficients are missing for a day, skip with a warning
        if day not in coefficient_rasters:
            logger.warning("No coefficient raster for day %s, skipping", day)
            continue

        coeff_raster = coefficient_rasters[day]
        output_name = f"{output_prefix}_day{day}"

        # Use r.mapcalc to multiply rasters in GRASS
        mapcalc_expr = f"{output_name} = {wrf_raster} * {coeff_raster}"
        grass_module(
            "r.mapcalc",
            expression=mapcalc_expr,
            overwrite=True,
        ).run()

        adjusted_rasters[day] = output_name

    return adjusted_rasters

calculate_wrf_on_buildings(wrf_summed_raster, building_vector, output_name, grass_module)

Apply a building mask and create a building-only WRF raster.

Parameters:

Name Type Description Default
wrf_summed_raster str

Name of the summed WRF raster in GRASS.

required
building_vector str

Name of the building footprints vector in GRASS.

required
output_name str

Name to create for the building-only WRF raster.

required
grass_module Any

GRASS Module-like callable.

required

Returns:

Type Description
str

The name of the created building-only raster (output_name).

Source code in src/utils/wrf.py
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
def calculate_wrf_on_buildings(
    wrf_summed_raster: str, building_vector: str, output_name: str, grass_module: Any
) -> str:
    """Apply a building mask and create a building-only WRF raster.

    Args:
        wrf_summed_raster: Name of the summed WRF raster in GRASS.
        building_vector: Name of the building footprints vector in GRASS.
        output_name: Name to create for the building-only WRF raster.
        grass_module: GRASS Module-like callable.

    Returns:
        The name of the created building-only raster (`output_name`).
    """
    # Apply building mask to restrict operations to building footprints
    apply_building_mask(building_vector, grass_module=grass_module)

    r_mapcalc = grass_module(
        "r.mapcalc",
        expression=f"{output_name} = {wrf_summed_raster}",
        overwrite=True,
    )
    r_mapcalc.run()

    remove_masks(grass_module)

    return output_name

cleanup_wrf_intermediates(day_rasters, summed_raster, grass_module)

Remove intermediate WRF rasters from the GRASS mapset.

Parameters:

Name Type Description Default
day_rasters Union[Dict[int, str], Iterable[str]]

Dict or iterable containing the names of per-day rasters.

required
summed_raster Optional[str]

Optional name of the summed raster to remove as well.

required
grass_module Any

The GRASS Python scripting Module class.

required

Returns:

Type Description
None

None

Source code in src/utils/wrf.py
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
def cleanup_wrf_intermediates(
    day_rasters: Union[Dict[int, str], Iterable[str]],
    summed_raster: Optional[str],
    grass_module: Any,
) -> None:
    """Remove intermediate WRF rasters from the GRASS mapset.

    Args:
        day_rasters: Dict or iterable containing the names of per-day rasters.
        summed_raster: Optional name of the summed raster to remove as well.
        grass_module: The GRASS Python scripting Module class.

    Returns:
        None
    """
    if isinstance(day_rasters, dict):
        raster_list = list(day_rasters.values())
    else:
        raster_list = list(day_rasters)

    for raster in raster_list:
        grass_module(
            "g.remove", type="raster", name=raster, flags="f", quiet=True
        ).run()

    if summed_raster:
        grass_module(
            "g.remove", type="raster", name=summed_raster, flags="f", quiet=True
        ).run()

process_wrf_for_grass(nc_file_path, output_prefix, grass_module, source_crs='EPSG:4326', target_crs=None, days=None, clip_to_raster=None, print_diagnostics=False)

Load WRF NetCDF, (optionally) reproject, import per-day rasters into GRASS and sum.

Parameters:

Name Type Description Default
nc_file_path str

Path to the WRF NetCDF file.

required
output_prefix str

Prefix to use when naming imported rasters in GRASS.

required
grass_module Any

The GRASS Python scripting Module class.

required
source_crs str

CRS to attach to the raw WRF dataset (default EPSG:4326).

'EPSG:4326'
target_crs Optional[str]

If provided, reproject the dataset to this CRS before import.

None
days Optional[Iterable[int]]

Iterable of day-of-year integers to import. If None, the function imports the full range present in the dataset.

None
clip_to_raster Optional[str]

If provided, set the GRASS region to this raster and clip imported rasters to that region.

None
print_diagnostics bool

If True, call the diagnostics helper to print dataset metadata.

False

Returns:

Type Description
Dict[int, str]

A tuple (imported_rasters, summed_raster_name) where imported_rasters is

str

a dict mapping day-of-year to GRASS raster name and summed_raster_name

Tuple[Dict[int, str], str]

is the name of the combined total raster produced.

Source code in src/utils/wrf.py
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
def process_wrf_for_grass(
    nc_file_path: str,
    output_prefix: str,
    grass_module: Any,
    source_crs: str = "EPSG:4326",
    target_crs: Optional[str] = None,
    days: Optional[Iterable[int]] = None,
    clip_to_raster: Optional[str] = None,
    print_diagnostics: bool = False,
) -> Tuple[Dict[int, str], str]:
    """Load WRF NetCDF, (optionally) reproject, import per-day rasters into GRASS and sum.

    Args:
        nc_file_path: Path to the WRF NetCDF file.
        output_prefix: Prefix to use when naming imported rasters in GRASS.
        grass_module: The GRASS Python scripting Module class.
        source_crs: CRS to attach to the raw WRF dataset (default EPSG:4326).
        target_crs: If provided, reproject the dataset to this CRS before import.
        days: Iterable of day-of-year integers to import. If None, the function
            imports the full range present in the dataset.
        clip_to_raster: If provided, set the GRASS region to this raster and
            clip imported rasters to that region.
        print_diagnostics: If True, call the diagnostics helper to print dataset
            metadata.

    Returns:
        A tuple (imported_rasters, summed_raster_name) where `imported_rasters` is
        a dict mapping day-of-year to GRASS raster name and `summed_raster_name`
        is the name of the combined total raster produced.
    """
    # Optionally print diagnostics
    if print_diagnostics:
        from .diagnostics import print_wrf_diagnostics

        wrf_ds = print_wrf_diagnostics(nc_file_path)

        # Ensure expected coordinate names and metadata for subsequent operations
        wrf_ds = wrf_ds.rename({"lon": "x", "lat": "y"})
        wrf_ds.rio.write_crs(source_crs, inplace=True)
        wrf_ds.rio.set_spatial_dims(x_dim="x", y_dim="y", inplace=True)
    else:
        wrf_ds = _load_wrf_with_crs(nc_file_path, crs=source_crs)

    # Reproject to the target CRS if requested
    if target_crs:
        wrf_ds = wrf_ds.rio.reproject(target_crs)

    # If days is not provided infer the full range from dataset dayofyear values
    if days is None:
        # Convert to a sorted list of unique dayofyear integers present in dataset
        days = sorted(int(d) for d in xr.DataArray(wrf_ds["dayofyear"]).values)

    imported_rasters = _import_wrf_to_grass(
        wrf_ds, output_prefix, grass_module, days, clip_to_raster
    )

    # Sum imported daily rasters into a single total raster
    summed_raster_name = f"{output_prefix}_total"
    summed_raster = _sum_wrf_rasters(imported_rasters, summed_raster_name, grass_module)

    return imported_rasters, summed_raster

sum_adjusted_rasters(adjusted_rasters, output_name, grass_module, cleanup=True)

Sum a collection of per-day adjusted WRF rasters into a single raster.

This uses r.series (method=sum) when provided a list of raster names; if a dict of day->raster is supplied the dict values are used.

Parameters:

Name Type Description Default
adjusted_rasters Union[Dict[int, str], Iterable[str]]

Mapping or iterable of raster names to sum.

required
output_name str

Name for the summed raster in GRASS.

required
grass_module Any

The GRASS Python scripting Module class.

required
cleanup bool

If True, remove the input adjusted rasters from the mapset after the sum is created.

True

Returns:

Type Description
str

The name of the summed raster (output_name).

Source code in src/utils/wrf.py
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
def sum_adjusted_rasters(
    adjusted_rasters: Union[Dict[int, str], Iterable[str]],
    output_name: str,
    grass_module: Any,
    cleanup: bool = True,
) -> str:
    """Sum a collection of per-day adjusted WRF rasters into a single raster.

    This uses `r.series` (method=sum) when provided a list of raster names; if
    a dict of day->raster is supplied the dict values are used.

    Args:
        adjusted_rasters: Mapping or iterable of raster names to sum.
        output_name: Name for the summed raster in GRASS.
        grass_module: The GRASS Python scripting Module class.
        cleanup: If True, remove the input adjusted rasters from the mapset
            after the sum is created.

    Returns:
        The name of the summed raster (`output_name`).
    """
    if isinstance(adjusted_rasters, dict):
        raster_list = list(adjusted_rasters.values())
    else:
        raster_list = list(adjusted_rasters)

    grass_module(
        "r.series",
        input=",".join(raster_list),
        output=output_name,
        method="sum",
        overwrite=True,
    ).run()

    # Optionally remove intermediate rasters to keep the GRASS mapset tidy
    if cleanup and raster_list:
        grass_module(
            "g.remove",
            type="raster",
            name=",".join(raster_list),
            flags="f",
        ).run()

    return output_name

Environment & GRASS Setup

GRASS GIS environment setup helper.

This module provides a single convenience function, setup_grass, which enables the programmatic usage of GRASS GIS.

setup_grass(gisbase, grassdata_dir='grassdata', location='solar_estimates', mapset='PERMANENT')

Prepare the GRASS Python bindings and initialize a session.

This function
  • uses GISBASE to locate the GRASS installation
  • appends relevant GRASS directories to PATH,
  • ensures grassdata_dir exists and creates it if missing
  • calls gscript.setup.init(grassdata_dir, location, mapset)

Parameters:

Name Type Description Default
gisbase str

Filesystem path to the GRASS installation root (contains bin/, scripts/, and etc/ directories).

required
grassdata_dir str

Directory to host GRASS locations (created if missing).

'grassdata'
location str

Name of the location under grassdata_dir to use or create.

'solar_estimates'
mapset str

GRASS mapset to initialize inside the Location.

'PERMANENT'

Returns:

Type Description
object

A tuple (gscript, Module) where gscript is the imported grass.script

type

module and Module is the class from grass.pygrass.modules. These are

Tuple[object, type]

used for running GRASS's modules.

Raises:

Type Description
ImportError

If GRASS Python modules cannot be imported after modifying sys.path.

CalledProcessError

If the attempt to create a new GRASS Location fails.

Source code in src/utils/grass_utils.py
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 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
def setup_grass(
    gisbase: str,
    grassdata_dir: str = "grassdata",
    location: str = "solar_estimates",
    mapset: str = "PERMANENT",
) -> Tuple[object, type]:
    """Prepare the GRASS Python bindings and initialize a session.

    This function:
      - uses GISBASE to locate the GRASS installation
      - appends relevant GRASS directories to PATH,
      - ensures `grassdata_dir` exists and creates it if missing
      - calls `gscript.setup.init(grassdata_dir, location, mapset)`

    Args:
        gisbase: Filesystem path to the GRASS installation root (contains `bin/`, `scripts/`,
            and `etc/` directories).
        grassdata_dir: Directory to host GRASS locations (created if missing).
        location: Name of the location under `grassdata_dir` to use or create.
        mapset: GRASS mapset to initialize inside the Location.

    Returns:
        A tuple `(gscript, Module)` where `gscript` is the imported `grass.script`
        module and `Module` is the class from `grass.pygrass.modules`. These are
        used for running GRASS's modules.

    Raises:
        ImportError: If GRASS Python modules cannot be imported after modifying `sys.path`.
        subprocess.CalledProcessError: If the attempt to create a new GRASS Location fails.
    """
    # Set the GISBASE environment variable and locate GRASS dirs
    os.environ["GISBASE"] = gisbase
    grass_bin = os.path.join(os.environ["GISBASE"], "bin")
    grass_scripts = os.path.join(os.environ["GISBASE"], "scripts")
    grass_python = os.path.join(os.environ["GISBASE"], "etc", "python")

    # Ensure GRASS executables and scripts can be found by subprocesses
    os.environ["PATH"] += os.pathsep + grass_bin + os.pathsep + grass_scripts

    # Ensure Python can import GRASS packages
    sys.path.insert(0, grass_python)

    # Import GRASS scripting interfaces
    try:
        import grass.script as gscript  # type: ignore
        from grass.pygrass.modules import Module  # type: ignore
    except ImportError as e:
        # Provide diagnostic context before re-raising
        logger.error("Error importing GRASS Python modules. Check if dependencies are installed correctly.")
        logger.error("GISBASE used: %s", gisbase)
        raise

    # Ensure the grassdata directory exists
    os.makedirs(grassdata_dir, exist_ok=True)

    # Create the requested Location if it does not exist.
    # TODO: take EPSG as argument?
    location_path = os.path.join(grassdata_dir, location)
    if not os.path.exists(location_path):
        cmd = ["grass", "--text", "-c", "EPSG:2193", location_path]
        logger.debug("Attempting to create GRASS Location: %s", " ".join(cmd))

        proc = subprocess.Popen(
            cmd,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
        # Provide an "exit" in case interactive prompts appear
        out, err = proc.communicate("exit\n")

        if proc.returncode != 0:
            # Log diagnostics and raise an error that includes output
            logger.error("GRASS LOCATION CREATION FAILED")
            logger.error("Command: %s", " ".join(cmd))
            logger.error("Return Code: %s", proc.returncode)
            logger.error("GRASS STDOUT: %s", out)
            logger.error("GRASS STDERR: %s", err)

            raise subprocess.CalledProcessError(
                proc.returncode,
                cmd,
                output=out,
                stderr=err,
            )

    # Initialize a GRASS session in this process
    gscript.setup.init(grassdata_dir, location, mapset)

    # Return the scripting interface and Module class for running GRASS modules
    return gscript, Module

Diagnostics

print_wrf_diagnostics(nc_file_path)

Load a WRF NetCDF file and print comprehensive diagnostic information.

Source code in src/utils/diagnostics.py
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
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
def print_wrf_diagnostics(nc_file_path):
    """Load a WRF NetCDF file and print comprehensive diagnostic information."""

    print("=" * 80)
    print("WRF NetCDF FILE DIAGNOSTICS")
    print("=" * 80)
    print(f"\nFile: {nc_file_path}\n")

    ds = xr.open_dataset(nc_file_path, engine="h5netcdf")

    print("-" * 80)
    print("DATASET OVERVIEW")
    print("-" * 80)
    print(ds)
    print()

    print("-" * 80)
    print("DATASET INFO")
    print("-" * 80)
    print(ds.info())
    print()

    print("-" * 80)
    print("DIMENSIONS")
    print("-" * 80)
    for dim_name, dim_size in ds.sizes.items():
        print(f"  {dim_name}: {dim_size}")
    print()

    print("-" * 80)
    print("COORDINATES")
    print("-" * 80)
    for coord_name, coord_data in ds.coords.items():
        print(f"\n  {coord_name}:")
        print(f"    Shape: {coord_data.shape}")
        print(f"    Dtype: {coord_data.dtype}")
        print(f"    Min: {coord_data.min().values}, Max: {coord_data.max().values}")
        if coord_data.attrs:
            print(f"    Attributes: {dict(coord_data.attrs)}")
    print()

    print("-" * 80)
    print("DATA VARIABLES")
    print("-" * 80)
    for var_name, var_data in ds.data_vars.items():
        print(f"\n  {var_name}:")
        print(f"    Shape: {var_data.shape}")
        print(f"    Dtype: {var_data.dtype}")
        print(f"    Dimensions: {var_data.dims}")
        try:
            print(f"    Min: {var_data.min().values}, Max: {var_data.max().values}")
            print(f"    Mean: {var_data.mean().values}")
        except Exception as e:
            print(f"    (Could not compute statistics: {e})")
        if var_data.attrs:
            print("    Attributes:")
            for attr_key, attr_val in var_data.attrs.items():
                print(f"      {attr_key}: {attr_val}")
    print()

    print("-" * 80)
    print("GLOBAL ATTRIBUTES")
    print("-" * 80)
    if ds.attrs:
        for attr_key, attr_val in ds.attrs.items():
            print(f"  {attr_key}: {attr_val}")
    else:
        print("  No global attributes found")
    print()

    print("-" * 80)
    print("COORDINATE REFERENCE SYSTEM INFO")
    print("-" * 80)
    try:
        if hasattr(ds, "rio"):
            print(f"  CRS: {ds.rio.crs}")
            print(f"  Transform: {ds.rio.transform()}")
            print(f"  Bounds: {ds.rio.bounds()}")
        else:
            print("  No rioxarray spatial metadata found")
    except Exception as e:
        print(f"  Could not retrieve CRS info: {e}")
    print()

    print("=" * 80)
    print("END OF DIAGNOSTICS")
    print("=" * 80)
    print()

    return ds

Linke Turbidity

Logging Configuration

Logging configuration for the solar estimates pipeline.

This module provides structured logging with elapsed time since script start.

ElapsedTimeFormatter

Bases: Formatter

Formatter that prefixes each log record with elapsed time since script start.

Source code in src/utils/logging_config.py
39
40
41
42
43
44
45
class ElapsedTimeFormatter(logging.Formatter):
    """Formatter that prefixes each log record with elapsed time since script start."""

    def format(self, record: logging.LogRecord) -> str:
        elapsed = int(record.created - _start_time)
        record.elapsed_str = _format_elapsed(elapsed)
        return super().format(record)

get_logger(name=None)

Get a logger instance for a module.

Parameters:

Name Type Description Default
name str | None

Logger name, typically __name__.

None

Returns:

Type Description
Logger

A child logger under the solar_pipeline hierarchy.

Source code in src/utils/logging_config.py
69
70
71
72
73
74
75
76
77
78
def get_logger(name: str | None = None) -> logging.Logger:
    """Get a logger instance for a module.

    Args:
        name: Logger name, typically ``__name__``.

    Returns:
        A child logger under the ``solar_pipeline`` hierarchy.
    """
    return logging.getLogger("solar_pipeline" + ("." + name if name else ""))

setup_logging(log_level=logging.INFO)

Configure logging with elapsed time formatting.

Parameters:

Name Type Description Default
log_level int

Logging level (default: INFO).

INFO

Returns:

Type Description
Logger

The configured root pipeline logger.

Source code in src/utils/logging_config.py
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def setup_logging(log_level: int = logging.INFO) -> logging.Logger:
    """Configure logging with elapsed time formatting.

    Args:
        log_level: Logging level (default: INFO).

    Returns:
        The configured root pipeline logger.
    """
    logger = logging.getLogger("solar_pipeline")
    logger.setLevel(log_level)
    logger.handlers.clear()

    formatter = ElapsedTimeFormatter("%(elapsed_str)s : %(levelname)s - %(message)s")
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(formatter)
    logger.addHandler(handler)

    return logger