Weather

This module handles weather data ingestion and processing for fire simulations. It supports two input sources: the Open-Meteo historical reanalysis API and RAWS-format weather station files (.wxs). The WeatherStream class builds a time-indexed sequence of weather observations, computes derived quantities (solar radiation, Growing Season Index, foliar moisture content, live fuel moisture), and provides the weather data that drives the simulation's fire behavior calculations.

Weather data ingestion and processing for fire simulation.

Fetch, parse, and package hourly weather observations into a stream of WeatherEntry records for use by the fire simulation. Supports two input sources:

• OpenMeteo: Historical reanalysis data fetched via the Open-Meteo API.
• File: RAWS-format weather station files (.wxs).

Also computes derived quantities: site-specific temperature/humidity corrections, local solar radiation, foliar moisture content (FMC), and the Growing Season Index (GSI) for estimating live fuel moisture.

Classes:

Name	Description
- WeatherStream	Build a weather stream from config parameters.

Functions:

Name	Description
- filter_hourly_data	Subset hourly data by datetime range.
- apply_site_specific_correction	Elevation-lapse adjustment for temperature and humidity.
- calc_local_solar_radiation	Slope- and canopy-adjusted irradiance.
- datetime_to_julian_date	Convert a datetime to Julian date.

CellData dataclass

Cell-level data for fire behavior calculations.

Attributes:

Name	Type	Description
fuel_type	Fuel	Fuel model for this cell.
elevation	float	Elevation in meters.
aspect	float	Aspect in degrees (0=N, 90=E).
slope_deg	float	Slope in degrees.
canopy_cover	float	Canopy cover as a percentage.
canopy_height	float	Canopy height in meters.
canopy_base_height	float	Canopy base height in meters.
canopy_bulk_density	float	Canopy bulk density in kg/m^3.
init_dead_mf	List[float]	Initial dead fuel moisture fractions [1hr, 10hr, 100hr].
live_h_mf	float	Live herbaceous moisture fraction.
live_w_mf	float	Live woody moisture fraction.

Source code in embrs/utilities/data_classes.py

@dataclass
class CellData:
    """Cell-level data for fire behavior calculations.

    Attributes:
        fuel_type (Fuel): Fuel model for this cell.
        elevation (float): Elevation in meters.
        aspect (float): Aspect in degrees (0=N, 90=E).
        slope_deg (float): Slope in degrees.
        canopy_cover (float): Canopy cover as a percentage.
        canopy_height (float): Canopy height in meters.
        canopy_base_height (float): Canopy base height in meters.
        canopy_bulk_density (float): Canopy bulk density in kg/m^3.
        init_dead_mf (List[float]): Initial dead fuel moisture fractions [1hr, 10hr, 100hr].
        live_h_mf (float): Live herbaceous moisture fraction.
        live_w_mf (float): Live woody moisture fraction.
    """

    fuel_type: Optional[Fuel] = None
    elevation: Optional[float] = None
    aspect: Optional[float] = None
    slope_deg: Optional[float] = None
    canopy_cover: Optional[float] = None
    canopy_height: Optional[float] = None
    canopy_base_height: Optional[float] = None
    canopy_bulk_density: Optional[float] = None
    init_dead_mf: Optional[List[float]] = field(default_factory=lambda: [0.06, 0.07, 0.08])
    live_h_mf: Optional[float] = 0.3
    live_w_mf: Optional[float] = 0.3

CellStatistics dataclass

Statistics for a single metric across ensemble members.

Attributes:

Name	Type	Description
mean	float	Mean value across ensemble members.
std	float	Standard deviation across ensemble members.
min	float	Minimum value across ensemble members.
max	float	Maximum value across ensemble members.
count	int	Number of ensemble members with data for this cell.

Source code in embrs/utilities/data_classes.py

@dataclass
class CellStatistics:
    """Statistics for a single metric across ensemble members.

    Attributes:
        mean (float): Mean value across ensemble members.
        std (float): Standard deviation across ensemble members.
        min (float): Minimum value across ensemble members.
        max (float): Maximum value across ensemble members.
        count (int): Number of ensemble members with data for this cell.
    """

    mean: float
    std: float
    min: float
    max: float
    count: int

DailySummary dataclass

One day of aggregated weather for GSI computation.

Attributes:

Name	Type	Description
date	datetime	Calendar date for this summary.
min_temp_F	float	Daily minimum temperature (Fahrenheit).
max_temp_F	float	Daily maximum temperature (Fahrenheit).
min_rh	float	Daily minimum relative humidity (percent, 0-100).
rain_cm	float	Total rainfall for this day (cm).

Source code in embrs/models/weather.py

@dataclass
class DailySummary:
    """One day of aggregated weather for GSI computation.

    Attributes:
        date: Calendar date for this summary.
        min_temp_F: Daily minimum temperature (Fahrenheit).
        max_temp_F: Daily maximum temperature (Fahrenheit).
        min_rh: Daily minimum relative humidity (percent, 0-100).
        rain_cm: Total rainfall for this day (cm).
    """
    date: datetime
    min_temp_F: float
    max_temp_F: float
    min_rh: float
    rain_cm: float

EnsemblePredictionOutput dataclass

Output from ensemble fire prediction runs.

Aggregates statistics across multiple prediction runs with varying parameters to quantify prediction uncertainty.

Attributes:

Name	Type	Description
n_ensemble	int	Number of ensemble members.
burn_probability	dict	Maps time (s) to dict of (x, y) to burn probability (0-1).
flame_len_m_stats	dict	Maps (x, y) to CellStatistics for flame length.
fli_kw_m_stats	dict	Maps (x, y) to CellStatistics for fireline intensity.
ros_ms_stats	dict	Maps (x, y) to CellStatistics for rate of spread.
spread_dir_stats	dict	Maps (x, y) to dict with 'mean_x' and 'mean_y' for circular mean spread direction.
crown_fire_probability	dict	Maps time in seconds to dict of (x, y) to crown fire probability (0-1).
hold_prob_stats	dict	Maps (x, y) to CellStatistics for hold probability.
breach_frequency	dict	Maps (x, y) to breach probability (0-1).
active_fire_probability	Optional[dict]	Maps time in seconds to dict of (x, y) to probability of active fire presence (0-1). Optional.
burnt_probability	Optional[dict]	Maps time in seconds to dict of (x, y) to probability of being burnt (0-1). Optional.
individual_predictions	List[PredictionOutput]	Individual prediction outputs for inspection. Optional.
forecast_indices	Optional[List[int]]	List of ints indicating the indices of the forecast_pool used for each of the predictions in the ensemble.

Source code in embrs/utilities/data_classes.py

@dataclass
class EnsemblePredictionOutput:
    """Output from ensemble fire prediction runs.

    Aggregates statistics across multiple prediction runs with varying
    parameters to quantify prediction uncertainty.

    Attributes:
        n_ensemble (int): Number of ensemble members.
        burn_probability (dict): Maps time (s) to dict of (x, y) to burn
            probability (0-1).
        flame_len_m_stats (dict): Maps (x, y) to CellStatistics for flame length.
        fli_kw_m_stats (dict): Maps (x, y) to CellStatistics for fireline intensity.
        ros_ms_stats (dict): Maps (x, y) to CellStatistics for rate of spread.
        spread_dir_stats (dict): Maps (x, y) to dict with 'mean_x' and 'mean_y'
            for circular mean spread direction.
        crown_fire_probability (dict): Maps time in seconds to dict of (x, y) to crown fire probability (0-1).
        hold_prob_stats (dict): Maps (x, y) to CellStatistics for hold probability.
        breach_frequency (dict): Maps (x, y) to breach probability (0-1).
        active_fire_probability (Optional[dict]): Maps time in seconds to dict of (x, y) to probability of active fire presence (0-1). Optional.
        burnt_probability (Optional[dict]): Maps time in seconds to dict of (x, y) to probability of being burnt (0-1). Optional.
        individual_predictions (List[PredictionOutput]): Individual prediction
            outputs for inspection. Optional.
        forecast_indices (Optional[List[int]]): List of ints indicating the indices of 
            the forecast_pool used for each of the predictions in the ensemble.
    """

    n_ensemble: int
    burn_probability: dict
    flame_len_m_stats: dict
    fli_kw_m_stats: dict
    ros_ms_stats: dict
    spread_dir_stats: dict
    crown_fire_probability: dict
    hold_prob_stats: dict
    breach_frequency: dict
    active_fire_probability: Optional[dict] = None
    burnt_probability: Optional[dict] = None
    individual_predictions: Optional[List[PredictionOutput]] = None
    forecast_indices: Optional[List[int]] = None

Fuel

Base fuel model for Rothermel fire spread calculations.

Encapsulate the physical properties of a fuel type and precompute derived constants used in the Rothermel (1972) equations. Non-burnable fuel types (e.g., water, urban) store only name and model number.

All internal units follow the Rothermel convention: loading in lb/ft², surface-area-to-volume ratio in 1/ft, fuel depth in ft, heat content in BTU/lb.

Attributes:

Name	Type	Description
name	str	Human-readable fuel model name.
model_num	int	Numeric fuel model identifier.
burnable	bool	Whether this fuel can sustain fire.
dynamic	bool	Whether herbaceous fuel transfer is applied.
load	ndarray	Fuel loading per class (tons/acre), shape (6,). Order: [1h, 10h, 100h, dead herb, live herb, live woody].
s	ndarray	Surface-area-to-volume ratio per class (1/ft), shape (6,).
sav_ratio	int	Characteristic SAV ratio (1/ft).
dead_mx	float	Dead fuel moisture of extinction (fraction).
fuel_depth_ft	float	Fuel bed depth (feet).
heat_content	float	Heat content (BTU/lb), default 8000.
rho_p	float	Particle density (lb/ft³), default 32.

Source code in embrs/models/fuel_models.py

class Fuel:
    """Base fuel model for Rothermel fire spread calculations.

    Encapsulate the physical properties of a fuel type and precompute
    derived constants used in the Rothermel (1972) equations. Non-burnable
    fuel types (e.g., water, urban) store only name and model number.

    All internal units follow the Rothermel convention:
    loading in lb/ft², surface-area-to-volume ratio in 1/ft, fuel depth in
    ft, heat content in BTU/lb.

    Attributes:
        name (str): Human-readable fuel model name.
        model_num (int): Numeric fuel model identifier.
        burnable (bool): Whether this fuel can sustain fire.
        dynamic (bool): Whether herbaceous fuel transfer is applied.
        load (np.ndarray): Fuel loading per class (tons/acre), shape (6,).
            Order: [1h, 10h, 100h, dead herb, live herb, live woody].
        s (np.ndarray): Surface-area-to-volume ratio per class (1/ft),
            shape (6,).
        sav_ratio (int): Characteristic SAV ratio (1/ft).
        dead_mx (float): Dead fuel moisture of extinction (fraction).
        fuel_depth_ft (float): Fuel bed depth (feet).
        heat_content (float): Heat content (BTU/lb), default 8000.
        rho_p (float): Particle density (lb/ft³), default 32.
    """

    def __init__(self, name: str, model_num: int, burnable: bool, dynamic: bool, w_0: np.ndarray,
                 s: np.ndarray, s_total: int, dead_mx: float, fuel_depth: float):
        """Initialize a fuel model.

        Args:
            name (str): Human-readable fuel model name.
            model_num (int): Numeric identifier for the fuel model.
            burnable (bool): Whether this fuel can sustain fire.
            dynamic (bool): Whether herbaceous transfer applies.
            w_0 (np.ndarray): Fuel loading per class (tons/acre), shape (6,).
                None for non-burnable models.
            s (np.ndarray): SAV ratio per class (1/ft), shape (6,).
                None for non-burnable models.
            s_total (int): Characteristic SAV ratio (1/ft).
            dead_mx (float): Dead fuel moisture of extinction (fraction).
            fuel_depth (float): Fuel bed depth (feet).
        """
        self.name = name
        self.model_num = model_num
        self.burnable = burnable
        self.dynamic = dynamic
        self.rel_indices = []

        if self.burnable:
            # Standard constants
            self.s_T = 0.055
            self.s_e = 0.010
            self.rho_p = 32
            self.heat_content = 8000 # btu/lb (used for live and dead)

            # Load data defining the fuel type
            self.load = w_0
            self.s = s
            self.sav_ratio = s_total
            self.fuel_depth_ft = fuel_depth
            self.dead_mx = dead_mx

            # Compute weighting factors
            self.compute_f_and_g_weights()

            # Compute f live and f dead
            self.f_dead_arr = self.f_ij[0, 0:4]
            self.f_live_arr = self.f_ij[1, 4:]

            # Compute g live and g dead
            self.g_dead_arr = self.g_ij[0, 0:4]
            self.g_live_arr = self.g_ij[1, 4:]

            # Compute the net fuel loading
            self.w_0 = TPA_to_Lbsft2(self.load)
            w_n = self.w_0 * (1 - self.s_T)
            self.set_fuel_loading(w_n)

            # Store nominal net dead loading
            self.w_n_dead_nominal = self.w_n_dead

            # Compute helpful constants for rothermel equations
            self.beta = np.sum(self.w_0) / 32 / self.fuel_depth_ft
            self.beta_op = 3.348 / (self.sav_ratio ** 0.8189)
            self.rat = self.beta / self.beta_op
            self.A = 133 * self.sav_ratio ** (-0.7913)
            self.gammax = (self.sav_ratio ** 1.5) / (495 + 0.0594 * self.sav_ratio ** 1.5)
            self.gamma = self.gammax * (self.rat ** self.A) * math.exp(self.A*(1-self.rat))
            self.rho_b = np.sum(self.w_0) / self.fuel_depth_ft
            self.flux_ratio = self.calc_flux_ratio()
            self.E, self.B, self.C = self.calc_E_B_C()
            self.W = self.calc_W(w_0)

            # Mark indices that are relevant for this fuel model
            for i in range(6):
                if self.w_0[i] > 0:
                    self.rel_indices.append(i)

            self.rel_indices = np.array(self.rel_indices)
            self.num_classes = len(self.rel_indices)

    def calc_flux_ratio(self) -> float:
        """Compute propagating flux ratio for the Rothermel equation.

        Returns:
            float: Propagating flux ratio (dimensionless).
        """
        packing_ratio = self.rho_b / self.rho_p
        flux_ratio = (192 + 0.2595*self.sav_ratio)**(-1) * math.exp((0.792 + 0.681*math.sqrt(self.sav_ratio))*(packing_ratio + 0.1))

        return flux_ratio

    def calc_E_B_C(self) -> tuple:
        """Compute wind factor coefficients E, B, and C.

        These coefficients parameterize the wind factor equation in the
        Rothermel model as a function of the characteristic SAV ratio.

        Returns:
            Tuple[float, float, float]: ``(E, B, C)`` wind factor
                coefficients (dimensionless).
        """

        sav_ratio = self.sav_ratio

        E = 0.715 * math.exp(-3.59e-4 * sav_ratio)
        B = 0.02526 * sav_ratio ** 0.54
        C = 7.47 * math.exp(-0.133 * sav_ratio**0.55)

        return E, B, C

    def compute_f_and_g_weights(self):
        """Compute fuel class weighting factors f_ij, g_ij, and category fractions f_i.

        Derive weighting arrays from fuel loading and SAV ratios. ``f_ij``
        gives fractional area weights within dead/live categories. ``g_ij``
        gives SAV-bin-based moisture weighting factors. ``f_i`` gives the
        dead vs. live category fractions.

        Side Effects:
            Sets ``self.f_ij`` (2×6), ``self.g_ij`` (2×6), and
            ``self.f_i`` (2,) arrays.
        """
        f_ij = np.zeros((2, 6))
        g_ij = np.zeros((2, 6))
        f_i = np.zeros(2)

        # ── Class grouping ─────────────────────────────────────────────
        dead_indices = [0, 1, 2, 3]  # 1h, 10h, 100h, dead herb
        live_indices = [4, 5]       # live herb, live woody

        # ── Compute a[i] = load[i] * SAV[i] / 32 ──────────────────────
        a_dead = np.array([
            self.load[i] * self.s[i] / 32.0 for i in dead_indices
        ])
        a_live = np.array([
            self.load[i] * self.s[i] / 32.0 for i in live_indices
        ])

        a_sum_dead = np.sum(a_dead)
        a_sum_live = np.sum(a_live)

        # ── f_ij (fractional weights) ─────────────────────────────────
        for idx in dead_indices:
            f_ij[0, idx] = a_dead[dead_indices.index(idx)] / a_sum_dead if a_sum_dead > 0 else 0.0
        for idx in live_indices:
            f_ij[1, idx] = a_live[live_indices.index(idx)] / a_sum_live if a_sum_live > 0 else 0.0

        # ── g_ij (SAV bin-based moisture weighting) ───────────────────
        def sav_bin(sav):
            if sav >= 1200.0: return 0
            elif sav >= 192.0: return 1
            elif sav >= 96.0: return 2
            elif sav >= 48.0: return 3
            elif sav >= 16.0: return 4
            else: return -1

        # Dead gx bin sums
        gx_dead = np.zeros(5)
        for i in dead_indices:
            bin_idx = sav_bin(self.s[i])
            if bin_idx >= 0:
                gx_dead[bin_idx] += f_ij[0, i]

        # Live gx bin sums
        gx_live = np.zeros(5)
        for i in live_indices:
            bin_idx = sav_bin(self.s[i])
            if bin_idx >= 0:
                gx_live[bin_idx] += f_ij[1, i]

        for i in dead_indices:
            bin_idx = sav_bin(self.s[i])
            g_ij[0, i] = gx_dead[bin_idx] if bin_idx >= 0 else 0.0

        for i in live_indices:
            bin_idx = sav_bin(self.s[i])
            g_ij[1, i] = gx_live[bin_idx] if bin_idx >= 0 else 0.0

        f_i[0] = a_sum_dead / (a_sum_dead + a_sum_live)
        f_i[1] = 1.0 - f_i[0]

        self.f_ij = f_ij
        self.g_ij = g_ij
        self.f_i = f_i

    def set_fuel_loading(self, w_n: np.ndarray):
        """Set net fuel loading and recompute weighted dead/live net loadings.

        Args:
            w_n (np.ndarray): Net fuel loading per class (lb/ft²), shape (6,).

        Side Effects:
            Updates ``self.w_n``, ``self.w_n_dead``, and ``self.w_n_live``.
        """
        self.w_n = w_n
        self.w_n_dead = np.dot(self.g_dead_arr, self.w_n[0:4])
        self.w_n_live = np.dot(self.g_live_arr, self.w_n[4:])

    def calc_W(self, w_0_tpa: np.ndarray) -> float:
        """Compute dead-to-live fuel loading ratio W.

        W is used to determine live fuel moisture of extinction. Returns
        ``np.inf`` when there is no live fuel loading (denominator is zero).

        Args:
            w_0_tpa (np.ndarray): Fuel loading per class (tons/acre),
                shape (6,).

        Returns:
            float: Dead-to-live loading ratio (dimensionless), or ``np.inf``
                if no live fuel is present.
        """
        w = w_0_tpa
        s = self.s

        num = 0

        for i in range(4):
            if s[i] != 0:
                num += w[i] * math.exp(-138/s[i])

        den = 0
        for i in range(4, 6):
            if s[i] != 0:
                den += w[i] * math.exp(-500/s[i])

        if den == 0:
            W = math.inf # Live moisture does not apply here

        else:
            W = num/den

        return W

__init__(name, model_num, burnable, dynamic, w_0, s, s_total, dead_mx, fuel_depth)

Initialize a fuel model.

Parameters:

Name	Type	Description	Default
name	str	Human-readable fuel model name.	required
model_num	int	Numeric identifier for the fuel model.	required
burnable	bool	Whether this fuel can sustain fire.	required
dynamic	bool	Whether herbaceous transfer applies.	required
w_0	ndarray	Fuel loading per class (tons/acre), shape (6,). None for non-burnable models.	required
s	ndarray	SAV ratio per class (1/ft), shape (6,). None for non-burnable models.	required
s_total	int	Characteristic SAV ratio (1/ft).	required
dead_mx	float	Dead fuel moisture of extinction (fraction).	required
fuel_depth	float	Fuel bed depth (feet).	required

Source code in embrs/models/fuel_models.py

def __init__(self, name: str, model_num: int, burnable: bool, dynamic: bool, w_0: np.ndarray,
             s: np.ndarray, s_total: int, dead_mx: float, fuel_depth: float):
    """Initialize a fuel model.

    Args:
        name (str): Human-readable fuel model name.
        model_num (int): Numeric identifier for the fuel model.
        burnable (bool): Whether this fuel can sustain fire.
        dynamic (bool): Whether herbaceous transfer applies.
        w_0 (np.ndarray): Fuel loading per class (tons/acre), shape (6,).
            None for non-burnable models.
        s (np.ndarray): SAV ratio per class (1/ft), shape (6,).
            None for non-burnable models.
        s_total (int): Characteristic SAV ratio (1/ft).
        dead_mx (float): Dead fuel moisture of extinction (fraction).
        fuel_depth (float): Fuel bed depth (feet).
    """
    self.name = name
    self.model_num = model_num
    self.burnable = burnable
    self.dynamic = dynamic
    self.rel_indices = []

    if self.burnable:
        # Standard constants
        self.s_T = 0.055
        self.s_e = 0.010
        self.rho_p = 32
        self.heat_content = 8000 # btu/lb (used for live and dead)

        # Load data defining the fuel type
        self.load = w_0
        self.s = s
        self.sav_ratio = s_total
        self.fuel_depth_ft = fuel_depth
        self.dead_mx = dead_mx

        # Compute weighting factors
        self.compute_f_and_g_weights()

        # Compute f live and f dead
        self.f_dead_arr = self.f_ij[0, 0:4]
        self.f_live_arr = self.f_ij[1, 4:]

        # Compute g live and g dead
        self.g_dead_arr = self.g_ij[0, 0:4]
        self.g_live_arr = self.g_ij[1, 4:]

        # Compute the net fuel loading
        self.w_0 = TPA_to_Lbsft2(self.load)
        w_n = self.w_0 * (1 - self.s_T)
        self.set_fuel_loading(w_n)

        # Store nominal net dead loading
        self.w_n_dead_nominal = self.w_n_dead

        # Compute helpful constants for rothermel equations
        self.beta = np.sum(self.w_0) / 32 / self.fuel_depth_ft
        self.beta_op = 3.348 / (self.sav_ratio ** 0.8189)
        self.rat = self.beta / self.beta_op
        self.A = 133 * self.sav_ratio ** (-0.7913)
        self.gammax = (self.sav_ratio ** 1.5) / (495 + 0.0594 * self.sav_ratio ** 1.5)
        self.gamma = self.gammax * (self.rat ** self.A) * math.exp(self.A*(1-self.rat))
        self.rho_b = np.sum(self.w_0) / self.fuel_depth_ft
        self.flux_ratio = self.calc_flux_ratio()
        self.E, self.B, self.C = self.calc_E_B_C()
        self.W = self.calc_W(w_0)

        # Mark indices that are relevant for this fuel model
        for i in range(6):
            if self.w_0[i] > 0:
                self.rel_indices.append(i)

        self.rel_indices = np.array(self.rel_indices)
        self.num_classes = len(self.rel_indices)

calc_E_B_C()

Compute wind factor coefficients E, B, and C.

These coefficients parameterize the wind factor equation in the Rothermel model as a function of the characteristic SAV ratio.

Returns:

Type	Description
tuple	Tuple[float, float, float]: (E, B, C) wind factor coefficients (dimensionless).

Source code in embrs/models/fuel_models.py

def calc_E_B_C(self) -> tuple:
    """Compute wind factor coefficients E, B, and C.

    These coefficients parameterize the wind factor equation in the
    Rothermel model as a function of the characteristic SAV ratio.

    Returns:
        Tuple[float, float, float]: ``(E, B, C)`` wind factor
            coefficients (dimensionless).
    """

    sav_ratio = self.sav_ratio

    E = 0.715 * math.exp(-3.59e-4 * sav_ratio)
    B = 0.02526 * sav_ratio ** 0.54
    C = 7.47 * math.exp(-0.133 * sav_ratio**0.55)

    return E, B, C

calc_W(w_0_tpa)

Compute dead-to-live fuel loading ratio W.

W is used to determine live fuel moisture of extinction. Returns np.inf when there is no live fuel loading (denominator is zero).

Parameters:

Name	Type	Description	Default
w_0_tpa	ndarray	Fuel loading per class (tons/acre), shape (6,).	required

Returns:

Name	Type	Description
float	float	Dead-to-live loading ratio (dimensionless), or np.inf if no live fuel is present.

Source code in embrs/models/fuel_models.py

def calc_W(self, w_0_tpa: np.ndarray) -> float:
    """Compute dead-to-live fuel loading ratio W.

    W is used to determine live fuel moisture of extinction. Returns
    ``np.inf`` when there is no live fuel loading (denominator is zero).

    Args:
        w_0_tpa (np.ndarray): Fuel loading per class (tons/acre),
            shape (6,).

    Returns:
        float: Dead-to-live loading ratio (dimensionless), or ``np.inf``
            if no live fuel is present.
    """
    w = w_0_tpa
    s = self.s

    num = 0

    for i in range(4):
        if s[i] != 0:
            num += w[i] * math.exp(-138/s[i])

    den = 0
    for i in range(4, 6):
        if s[i] != 0:
            den += w[i] * math.exp(-500/s[i])

    if den == 0:
        W = math.inf # Live moisture does not apply here

    else:
        W = num/den

    return W

calc_flux_ratio()

Compute propagating flux ratio for the Rothermel equation.

Returns:

Name	Type	Description
float	float	Propagating flux ratio (dimensionless).

Source code in embrs/models/fuel_models.py

def calc_flux_ratio(self) -> float:
    """Compute propagating flux ratio for the Rothermel equation.

    Returns:
        float: Propagating flux ratio (dimensionless).
    """
    packing_ratio = self.rho_b / self.rho_p
    flux_ratio = (192 + 0.2595*self.sav_ratio)**(-1) * math.exp((0.792 + 0.681*math.sqrt(self.sav_ratio))*(packing_ratio + 0.1))

    return flux_ratio

compute_f_and_g_weights()

Compute fuel class weighting factors f_ij, g_ij, and category fractions f_i.

Derive weighting arrays from fuel loading and SAV ratios. f_ij gives fractional area weights within dead/live categories. g_ij gives SAV-bin-based moisture weighting factors. f_i gives the dead vs. live category fractions.

Side Effects

Sets self.f_ij (2×6), self.g_ij (2×6), and self.f_i (2,) arrays.

Source code in embrs/models/fuel_models.py

def compute_f_and_g_weights(self):
    """Compute fuel class weighting factors f_ij, g_ij, and category fractions f_i.

    Derive weighting arrays from fuel loading and SAV ratios. ``f_ij``
    gives fractional area weights within dead/live categories. ``g_ij``
    gives SAV-bin-based moisture weighting factors. ``f_i`` gives the
    dead vs. live category fractions.

    Side Effects:
        Sets ``self.f_ij`` (2×6), ``self.g_ij`` (2×6), and
        ``self.f_i`` (2,) arrays.
    """
    f_ij = np.zeros((2, 6))
    g_ij = np.zeros((2, 6))
    f_i = np.zeros(2)

    # ── Class grouping ─────────────────────────────────────────────
    dead_indices = [0, 1, 2, 3]  # 1h, 10h, 100h, dead herb
    live_indices = [4, 5]       # live herb, live woody

    # ── Compute a[i] = load[i] * SAV[i] / 32 ──────────────────────
    a_dead = np.array([
        self.load[i] * self.s[i] / 32.0 for i in dead_indices
    ])
    a_live = np.array([
        self.load[i] * self.s[i] / 32.0 for i in live_indices
    ])

    a_sum_dead = np.sum(a_dead)
    a_sum_live = np.sum(a_live)

    # ── f_ij (fractional weights) ─────────────────────────────────
    for idx in dead_indices:
        f_ij[0, idx] = a_dead[dead_indices.index(idx)] / a_sum_dead if a_sum_dead > 0 else 0.0
    for idx in live_indices:
        f_ij[1, idx] = a_live[live_indices.index(idx)] / a_sum_live if a_sum_live > 0 else 0.0

    # ── g_ij (SAV bin-based moisture weighting) ───────────────────
    def sav_bin(sav):
        if sav >= 1200.0: return 0
        elif sav >= 192.0: return 1
        elif sav >= 96.0: return 2
        elif sav >= 48.0: return 3
        elif sav >= 16.0: return 4
        else: return -1

    # Dead gx bin sums
    gx_dead = np.zeros(5)
    for i in dead_indices:
        bin_idx = sav_bin(self.s[i])
        if bin_idx >= 0:
            gx_dead[bin_idx] += f_ij[0, i]

    # Live gx bin sums
    gx_live = np.zeros(5)
    for i in live_indices:
        bin_idx = sav_bin(self.s[i])
        if bin_idx >= 0:
            gx_live[bin_idx] += f_ij[1, i]

    for i in dead_indices:
        bin_idx = sav_bin(self.s[i])
        g_ij[0, i] = gx_dead[bin_idx] if bin_idx >= 0 else 0.0

    for i in live_indices:
        bin_idx = sav_bin(self.s[i])
        g_ij[1, i] = gx_live[bin_idx] if bin_idx >= 0 else 0.0

    f_i[0] = a_sum_dead / (a_sum_dead + a_sum_live)
    f_i[1] = 1.0 - f_i[0]

    self.f_ij = f_ij
    self.g_ij = g_ij
    self.f_i = f_i

set_fuel_loading(w_n)

Set net fuel loading and recompute weighted dead/live net loadings.

Parameters:

Name	Type	Description	Default
w_n	ndarray	Net fuel loading per class (lb/ft²), shape (6,).	required

Side Effects

Updates self.w_n, self.w_n_dead, and self.w_n_live.

Source code in embrs/models/fuel_models.py

def set_fuel_loading(self, w_n: np.ndarray):
    """Set net fuel loading and recompute weighted dead/live net loadings.

    Args:
        w_n (np.ndarray): Net fuel loading per class (lb/ft²), shape (6,).

    Side Effects:
        Updates ``self.w_n``, ``self.w_n_dead``, and ``self.w_n_live``.
    """
    self.w_n = w_n
    self.w_n_dead = np.dot(self.g_dead_arr, self.w_n[0:4])
    self.w_n_live = np.dot(self.g_live_arr, self.w_n[4:])

GSITracker

Track rolling weather data and recompute GSI during simulation.

Maintains a 56-day rolling buffer of :class:DailySummary records (seeded from pre-simulation data) and accumulates hourly weather entries into daily summaries as the simulation advances. Call :meth:compute_gsi after each day boundary to get an updated Growing Season Index.

Parameters:

Name	Type	Description	Default
geo	GeoInfo	:class:GeoInfo with center_lat, center_lon, and timezone for photoperiod calculations.	required
pre_sim_summaries	List[DailySummary]	Daily summaries from the pre-simulation window (up to 56 days). Oldest first.	required
initial_cum_rain	float	Cumulative rain value (cm) from the last pre-simulation weather entry, used to correctly compute rain deltas for the first simulation hour.	0.0

Source code in embrs/models/weather.py

class GSITracker:
    """Track rolling weather data and recompute GSI during simulation.

    Maintains a 56-day rolling buffer of :class:`DailySummary` records
    (seeded from pre-simulation data) and accumulates hourly weather
    entries into daily summaries as the simulation advances. Call
    :meth:`compute_gsi` after each day boundary to get an updated
    Growing Season Index.

    Args:
        geo: :class:`GeoInfo` with ``center_lat``, ``center_lon``, and
            ``timezone`` for photoperiod calculations.
        pre_sim_summaries: Daily summaries from the pre-simulation window
            (up to 56 days). Oldest first.
        initial_cum_rain: Cumulative rain value (cm) from the last
            pre-simulation weather entry, used to correctly compute
            rain deltas for the first simulation hour.
    """

    def __init__(self, geo: GeoInfo, pre_sim_summaries: List[DailySummary],
                 initial_cum_rain: float = 0.0):
        self._geo = geo
        self._daily_buffer: deque[DailySummary] = deque(maxlen=56)
        for s in pre_sim_summaries:
            self._daily_buffer.append(s)

        # Current-day accumulators
        self._hourly_temps: List[float] = []
        self._hourly_rhs: List[float] = []
        self._hourly_rain_cm: float = 0.0
        self._last_cum_rain: float = initial_cum_rain
        self._current_date: Optional[datetime] = None

    def ingest_hourly(self, entry: WeatherEntry, sim_datetime: datetime) -> bool:
        """Feed one hourly weather observation into the tracker.

        Accumulates temperature, humidity, and rain delta. When the
        calendar day changes, the previous day is finalized into a
        :class:`DailySummary` and appended to the rolling buffer.

        Args:
            entry: Current weather entry (temp in F, rel_humidity in
                percent, rain cumulative in cm).
            sim_datetime: Simulation datetime corresponding to this entry.

        Returns:
            True if a day boundary was crossed (a new daily summary was
            finalized), False otherwise.
        """
        entry_date = sim_datetime.date()
        day_changed = False

        if self._current_date is not None and entry_date != self._current_date:
            # Day boundary — finalize the previous day
            if self._hourly_temps:
                self._finalize_day()
                day_changed = True

            # Reset accumulators for the new day
            self._hourly_temps = []
            self._hourly_rhs = []
            self._hourly_rain_cm = 0.0

        self._current_date = entry_date

        # Accumulate hourly observations
        self._hourly_temps.append(entry.temp)
        self._hourly_rhs.append(entry.rel_humidity)

        # Rain delta from cumulative value
        if self._last_cum_rain is not None:
            delta = entry.rain - self._last_cum_rain
            if delta > 0:
                self._hourly_rain_cm += delta
        self._last_cum_rain = entry.rain

        return day_changed

    def _finalize_day(self):
        """Build a DailySummary from the current-day accumulators and
        append it to the rolling buffer."""
        summary = DailySummary(
            date=self._current_date,
            min_temp_F=min(self._hourly_temps),
            max_temp_F=max(self._hourly_temps),
            min_rh=min(self._hourly_rhs),
            rain_cm=self._hourly_rain_cm,
        )
        self._daily_buffer.append(summary)

    def compute_gsi(self) -> float:
        """Compute GSI from the rolling daily buffer.

        Uses the same sub-index formulas as
        :meth:`GSITracker.compute_gsi`: photoperiod, minimum temperature,
        vapor pressure deficit, and 28-day accumulated precipitation.

        Returns:
            GSI value in [0, 1], or -1 if fewer than 2 days are available.
        """
        buf = self._daily_buffer
        n = len(buf)
        if n < 2:
            return -1.0

        # Use last 28 entries for non-rain sub-indices
        start = max(0, n - 28)

        pv_loc = pvlib.location.Location(
            self._geo.center_lat, self._geo.center_lon, tz=self._geo.timezone
        )

        gsi = 0.0
        count = 0

        for i in range(start, n):
            day_summary = buf[i]

            # Photoperiod sub-index
            times = pd.date_range(
                pd.Timestamp(day_summary.date), periods=1, freq='D',
                tz=self._geo.timezone
            )
            solpos = pv_loc.get_sun_rise_set_transit(times)
            sunrise = solpos['sunrise'].iloc[0]
            sunset = solpos['sunset'].iloc[0]
            if sunset < sunrise:
                sunset += pd.Timedelta(days=1)
            day_len = (sunset - sunrise).total_seconds()
            iPhoto = max(0.0, min(1.0, (day_len - 36000) / (39600 - 36000)))

            # Min temperature sub-index (convert F → C)
            min_temp_C = F_to_C(day_summary.min_temp_F)
            iTmin = max(0.0, min(1.0, (min_temp_C + 2) / (5 + 2)))

            # VPD sub-index
            max_temp_C = F_to_C(day_summary.max_temp_F)
            min_rh = day_summary.min_rh
            vpd = (1 - min_rh / 100) * 0.6108 * math.exp(
                (17.27 * max_temp_C) / (max_temp_C + 237.3)
            )
            iVPD = max(0.0, min(1.0, (vpd - 0.9) / (4.1 - 0.9)))

            # Precipitation sub-index: 28-day accumulated rain ending at day i
            rain_start = max(0, i - 28)
            tot_rain_mm = sum(buf[j].rain_cm * 10 for j in range(rain_start, i))
            iPrcp = max(0.0, min(1.0, tot_rain_mm / 10))

            gsi += iTmin * iPhoto * iVPD * iPrcp
            count += 1

        return gsi / count

compute_gsi()

Compute GSI from the rolling daily buffer.

Uses the same sub-index formulas as :meth:GSITracker.compute_gsi: photoperiod, minimum temperature, vapor pressure deficit, and 28-day accumulated precipitation.

Returns:

Type	Description
float	GSI value in [0, 1], or -1 if fewer than 2 days are available.

Source code in embrs/models/weather.py

def compute_gsi(self) -> float:
    """Compute GSI from the rolling daily buffer.

    Uses the same sub-index formulas as
    :meth:`GSITracker.compute_gsi`: photoperiod, minimum temperature,
    vapor pressure deficit, and 28-day accumulated precipitation.

    Returns:
        GSI value in [0, 1], or -1 if fewer than 2 days are available.
    """
    buf = self._daily_buffer
    n = len(buf)
    if n < 2:
        return -1.0

    # Use last 28 entries for non-rain sub-indices
    start = max(0, n - 28)

    pv_loc = pvlib.location.Location(
        self._geo.center_lat, self._geo.center_lon, tz=self._geo.timezone
    )

    gsi = 0.0
    count = 0

    for i in range(start, n):
        day_summary = buf[i]

        # Photoperiod sub-index
        times = pd.date_range(
            pd.Timestamp(day_summary.date), periods=1, freq='D',
            tz=self._geo.timezone
        )
        solpos = pv_loc.get_sun_rise_set_transit(times)
        sunrise = solpos['sunrise'].iloc[0]
        sunset = solpos['sunset'].iloc[0]
        if sunset < sunrise:
            sunset += pd.Timedelta(days=1)
        day_len = (sunset - sunrise).total_seconds()
        iPhoto = max(0.0, min(1.0, (day_len - 36000) / (39600 - 36000)))

        # Min temperature sub-index (convert F → C)
        min_temp_C = F_to_C(day_summary.min_temp_F)
        iTmin = max(0.0, min(1.0, (min_temp_C + 2) / (5 + 2)))

        # VPD sub-index
        max_temp_C = F_to_C(day_summary.max_temp_F)
        min_rh = day_summary.min_rh
        vpd = (1 - min_rh / 100) * 0.6108 * math.exp(
            (17.27 * max_temp_C) / (max_temp_C + 237.3)
        )
        iVPD = max(0.0, min(1.0, (vpd - 0.9) / (4.1 - 0.9)))

        # Precipitation sub-index: 28-day accumulated rain ending at day i
        rain_start = max(0, i - 28)
        tot_rain_mm = sum(buf[j].rain_cm * 10 for j in range(rain_start, i))
        iPrcp = max(0.0, min(1.0, tot_rain_mm / 10))

        gsi += iTmin * iPhoto * iVPD * iPrcp
        count += 1

    return gsi / count

ingest_hourly(entry, sim_datetime)

Feed one hourly weather observation into the tracker.

Accumulates temperature, humidity, and rain delta. When the calendar day changes, the previous day is finalized into a :class:DailySummary and appended to the rolling buffer.

Parameters:

Name	Type	Description	Default
entry	WeatherEntry	Current weather entry (temp in F, rel_humidity in percent, rain cumulative in cm).	required
sim_datetime	datetime	Simulation datetime corresponding to this entry.	required

Returns:

Type	Description
bool	True if a day boundary was crossed (a new daily summary was
bool	finalized), False otherwise.

Source code in embrs/models/weather.py

def ingest_hourly(self, entry: WeatherEntry, sim_datetime: datetime) -> bool:
    """Feed one hourly weather observation into the tracker.

    Accumulates temperature, humidity, and rain delta. When the
    calendar day changes, the previous day is finalized into a
    :class:`DailySummary` and appended to the rolling buffer.

    Args:
        entry: Current weather entry (temp in F, rel_humidity in
            percent, rain cumulative in cm).
        sim_datetime: Simulation datetime corresponding to this entry.

    Returns:
        True if a day boundary was crossed (a new daily summary was
        finalized), False otherwise.
    """
    entry_date = sim_datetime.date()
    day_changed = False

    if self._current_date is not None and entry_date != self._current_date:
        # Day boundary — finalize the previous day
        if self._hourly_temps:
            self._finalize_day()
            day_changed = True

        # Reset accumulators for the new day
        self._hourly_temps = []
        self._hourly_rhs = []
        self._hourly_rain_cm = 0.0

    self._current_date = entry_date

    # Accumulate hourly observations
    self._hourly_temps.append(entry.temp)
    self._hourly_rhs.append(entry.rel_humidity)

    # Rain delta from cumulative value
    if self._last_cum_rain is not None:
        delta = entry.rain - self._last_cum_rain
        if delta > 0:
            self._hourly_rain_cm += delta
    self._last_cum_rain = entry.rain

    return day_changed

GeoInfo dataclass

Geographic information for a simulation area.

Attributes:

Name	Type	Description
bounds	BoundingBox	Spatial bounds from rasterio.
center_lat	float	Center latitude in degrees (WGS84).
center_lon	float	Center longitude in degrees (WGS84).
timezone	str	IANA timezone string (e.g., 'America/Denver').
north_angle_deg	float	Rotation from grid north to true north in degrees.

Source code in embrs/utilities/data_classes.py

@dataclass
class GeoInfo:
    """Geographic information for a simulation area.

    Attributes:
        bounds (BoundingBox): Spatial bounds from rasterio.
        center_lat (float): Center latitude in degrees (WGS84).
        center_lon (float): Center longitude in degrees (WGS84).
        timezone (str): IANA timezone string (e.g., 'America/Denver').
        north_angle_deg (float): Rotation from grid north to true north in degrees.
    """

    bounds: Optional[BoundingBox] = None
    center_lat: Optional[float] = None
    center_lon: Optional[float] = None
    timezone: Optional[str] = None
    north_angle_deg: Optional[float] = None

    def calc_center_coords(self, source_crs: 'pyproj.CRS') -> None:
        """Calculate center lat/lon from bounds and source CRS.

        Args:
            source_crs (pyproj.CRS): Coordinate reference system of the bounds.

        Raises:
            ValueError: If bounds is None.
        """
        if self.bounds is None:
            raise ValueError("Can't perform this function without bounds")

        mid_x = (self.bounds.left + self.bounds.right) / 2
        mid_y = (self.bounds.bottom + self.bounds.top) / 2

        transformer = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True)

        self.center_lon, self.center_lat = transformer.transform(mid_x, mid_y)

    def calc_time_zone(self) -> None:
        """Calculate timezone from center coordinates.

        Raises:
            ValueError: If center coordinates are not set.
        """
        if self.center_lat is None or self.center_lon is None:
            raise ValueError("Center coordinates must be set before computing the time zone")

        tf = TimezoneFinder()
        self.timezone = tf.timezone_at(lng=self.center_lon, lat=self.center_lat)

calc_center_coords(source_crs)

Calculate center lat/lon from bounds and source CRS.

Parameters:

Name	Type	Description	Default
source_crs	CRS	Coordinate reference system of the bounds.	required

Raises:

Type	Description
ValueError	If bounds is None.

Source code in embrs/utilities/data_classes.py

def calc_center_coords(self, source_crs: 'pyproj.CRS') -> None:
    """Calculate center lat/lon from bounds and source CRS.

    Args:
        source_crs (pyproj.CRS): Coordinate reference system of the bounds.

    Raises:
        ValueError: If bounds is None.
    """
    if self.bounds is None:
        raise ValueError("Can't perform this function without bounds")

    mid_x = (self.bounds.left + self.bounds.right) / 2
    mid_y = (self.bounds.bottom + self.bounds.top) / 2

    transformer = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True)

    self.center_lon, self.center_lat = transformer.transform(mid_x, mid_y)

calc_time_zone()

Calculate timezone from center coordinates.

Raises:

Type	Description
ValueError	If center coordinates are not set.

Source code in embrs/utilities/data_classes.py

def calc_time_zone(self) -> None:
    """Calculate timezone from center coordinates.

    Raises:
        ValueError: If center coordinates are not set.
    """
    if self.center_lat is None or self.center_lon is None:
        raise ValueError("Center coordinates must be set before computing the time zone")

    tf = TimezoneFinder()
    self.timezone = tf.timezone_at(lng=self.center_lon, lat=self.center_lat)

LandscapeData dataclass

Landscape raster data extracted from LCP files.

Attributes:

Name	Type	Description
elevation_map	ndarray	Elevation in meters.
slope_map	ndarray	Slope in degrees.
aspect_map	ndarray	Aspect in degrees (0=N, 90=E).
fuel_map	ndarray	Fuel model IDs.
canopy_cover_map	ndarray	Canopy cover as fraction.
canopy_height_map	ndarray	Canopy height in meters.
canopy_base_height_map	ndarray	Canopy base height in meters.
canopy_bulk_density_map	ndarray	Canopy bulk density in kg/m^3.
rows	int	Number of raster rows.
cols	int	Number of raster columns.
resolution	int	Raster resolution in meters.
width_m	float	Total width in meters.
height_m	float	Total height in meters.
transform	any	Rasterio affine transform.
crs	any	Coordinate reference system.

Source code in embrs/utilities/data_classes.py

@dataclass
class LandscapeData:
    """Landscape raster data extracted from LCP files.

    Attributes:
        elevation_map (np.ndarray): Elevation in meters.
        slope_map (np.ndarray): Slope in degrees.
        aspect_map (np.ndarray): Aspect in degrees (0=N, 90=E).
        fuel_map (np.ndarray): Fuel model IDs.
        canopy_cover_map (np.ndarray): Canopy cover as fraction.
        canopy_height_map (np.ndarray): Canopy height in meters.
        canopy_base_height_map (np.ndarray): Canopy base height in meters.
        canopy_bulk_density_map (np.ndarray): Canopy bulk density in kg/m^3.
        rows (int): Number of raster rows.
        cols (int): Number of raster columns.
        resolution (int): Raster resolution in meters.
        width_m (float): Total width in meters.
        height_m (float): Total height in meters.
        transform (any): Rasterio affine transform.
        crs (any): Coordinate reference system.
    """

    elevation_map: np.ndarray
    slope_map: np.ndarray
    aspect_map: np.ndarray
    fuel_map: np.ndarray
    canopy_cover_map: np.ndarray
    canopy_height_map: np.ndarray
    canopy_base_height_map: np.ndarray
    canopy_bulk_density_map: np.ndarray
    rows: int
    cols: int
    resolution: int
    width_m: float
    height_m: float
    transform: any
    crs: any

MapDrawerData dataclass

Data collected from the interactive map drawing interface.

Attributes:

Name	Type	Description
fire_breaks	Dict	Fire break geometries keyed by ID.
break_widths	List	Width in meters for each fire break.
break_ids	List	Identifiers for fire breaks.
initial_ign	List	Initial ignition point coordinates.

Source code in embrs/utilities/data_classes.py

@dataclass
class MapDrawerData:
    """Data collected from the interactive map drawing interface.

    Attributes:
        fire_breaks (Dict): Fire break geometries keyed by ID.
        break_widths (List): Width in meters for each fire break.
        break_ids (List): Identifiers for fire breaks.
        initial_ign (List): Initial ignition point coordinates.
    """

    fire_breaks: Optional[Dict] = field(default_factory=dict)
    break_widths: Optional[List] = field(default_factory=list)
    break_ids: Optional[List] = field(default_factory=list)
    initial_ign: Optional[List] = field(default_factory=list)

MapParams dataclass

Parameters for map configuration.

Attributes:

Name	Type	Description
folder	str	Path to the map data folder.
lcp_filepath	str	Path to the source LCP file.
cropped_lcp_path	str	Path to cropped LCP file if applicable.
import_roads	bool	Whether to import roads from OSM.
lcp_data	LandscapeData	Extracted landscape data.
roads	List	List of road geometries.
geo_info	GeoInfo	Geographic information.
scenario_data	MapDrawerData	User-drawn scenario elements.
fbfm_type	str	Fuel model type ('Anderson' or 'ScottBurgan').

Source code in embrs/utilities/data_classes.py

@dataclass
class MapParams:
    """Parameters for map configuration.

    Attributes:
        folder (str): Path to the map data folder.
        lcp_filepath (str): Path to the source LCP file.
        cropped_lcp_path (str): Path to cropped LCP file if applicable.
        import_roads (bool): Whether to import roads from OSM.
        lcp_data (LandscapeData): Extracted landscape data.
        roads (List): List of road geometries.
        geo_info (GeoInfo): Geographic information.
        scenario_data (MapDrawerData): User-drawn scenario elements.
        fbfm_type (str): Fuel model type ('Anderson' or 'ScottBurgan').
    """

    folder: Optional[str] = None
    lcp_filepath: Optional[str] = None
    cropped_lcp_path: Optional[str] = None
    import_roads: Optional[bool] = None
    lcp_data: Optional[LandscapeData] = None
    roads: Optional[List] = field(default_factory=list)
    geo_info: Optional[GeoInfo] = None
    scenario_data: Optional[MapDrawerData] = None
    fbfm_type: Optional[str] = "Anderson"

    def size(self) -> Tuple[float, float]:
        """Get the map size in meters.

        Returns:
            Tuple[float, float]: (width_m, height_m).
        """
        return (self.lcp_data.width_m, self.lcp_data.height_m)

    def shape(self, cell_size: int) -> Tuple[int, int]:
        """Calculate grid shape for a given cell size.

        Args:
            cell_size (int): Hexagon side length in meters.

        Returns:
            Tuple[int, int]: (rows, cols) for the hexagonal grid.
        """
        rows = int(np.floor(self.lcp_data.height_m/(1.5*cell_size))) + 1
        cols = int(np.floor(self.lcp_data.width_m/(np.sqrt(3)*cell_size))) + 1

        return (rows, cols)

shape(cell_size)

Calculate grid shape for a given cell size.

Parameters:

Name	Type	Description	Default
cell_size	int	Hexagon side length in meters.	required

Returns:

Type	Description
Tuple[int, int]	Tuple[int, int]: (rows, cols) for the hexagonal grid.

Source code in embrs/utilities/data_classes.py

def shape(self, cell_size: int) -> Tuple[int, int]:
    """Calculate grid shape for a given cell size.

    Args:
        cell_size (int): Hexagon side length in meters.

    Returns:
        Tuple[int, int]: (rows, cols) for the hexagonal grid.
    """
    rows = int(np.floor(self.lcp_data.height_m/(1.5*cell_size))) + 1
    cols = int(np.floor(self.lcp_data.width_m/(np.sqrt(3)*cell_size))) + 1

    return (rows, cols)

size()

Get the map size in meters.

Returns:

Type	Description
Tuple[float, float]	Tuple[float, float]: (width_m, height_m).

Source code in embrs/utilities/data_classes.py

def size(self) -> Tuple[float, float]:
    """Get the map size in meters.

    Returns:
        Tuple[float, float]: (width_m, height_m).
    """
    return (self.lcp_data.width_m, self.lcp_data.height_m)

PlaybackVisualizerParams dataclass

Parameters for playback visualization from log files.

Attributes:

Name	Type	Description
cell_file	str	Path to cell_logs.parquet file.
init_location	bool	Whether init_state.parquet was found.
save_video	bool	Whether to save visualization as video.
video_folder	str	Folder to save video output.
video_name	str	Video filename.
has_agents	bool	Whether agent logs are available.
has_actions	bool	Whether action logs are available.
has_predictions	bool	Whether prediction logs are available.
video_fps	int	Video frames per second.
agent_file	str	Path to agent_logs.parquet.
action_file	str	Path to action_logs.parquet.
prediction_file	str	Path to prediction_logs.parquet.
freq	float	Update frequency in seconds.
scale_km	float	Scale bar size in kilometers.
show_legend	bool	Whether to display fuel legend.
show_wind_cbar	bool	Whether to show wind colorbar.
show_wind_field	bool	Whether to show wind field.
show_weather_data	bool	Whether to show weather info.
show_compass	bool	Whether to show compass.
show_visualization	bool	Whether to render visualization.
show_temp_in_F	bool	Whether to display temperature in Fahrenheit.

Source code in embrs/utilities/data_classes.py

@dataclass
class PlaybackVisualizerParams:
    """Parameters for playback visualization from log files.

    Attributes:
        cell_file (str): Path to cell_logs.parquet file.
        init_location (bool): Whether init_state.parquet was found.
        save_video (bool): Whether to save visualization as video.
        video_folder (str): Folder to save video output.
        video_name (str): Video filename.
        has_agents (bool): Whether agent logs are available.
        has_actions (bool): Whether action logs are available.
        has_predictions (bool): Whether prediction logs are available.
        video_fps (int): Video frames per second.
        agent_file (str): Path to agent_logs.parquet.
        action_file (str): Path to action_logs.parquet.
        prediction_file (str): Path to prediction_logs.parquet.
        freq (float): Update frequency in seconds.
        scale_km (float): Scale bar size in kilometers.
        show_legend (bool): Whether to display fuel legend.
        show_wind_cbar (bool): Whether to show wind colorbar.
        show_wind_field (bool): Whether to show wind field.
        show_weather_data (bool): Whether to show weather info.
        show_compass (bool): Whether to show compass.
        show_visualization (bool): Whether to render visualization.
        show_temp_in_F (bool): Whether to display temperature in Fahrenheit.
    """

    cell_file: str
    init_location: bool
    save_video: bool
    video_folder: str
    video_name: str
    has_agents: bool
    has_actions: bool
    has_predictions: bool
    video_fps: Optional[int] = 10
    agent_file: Optional[str] = None
    action_file: Optional[str] = None
    prediction_file: Optional[str] = None

    freq: Optional[float] = 300
    scale_km: Optional[float] = 1.0
    show_legend: Optional[bool] = True
    show_wind_cbar: Optional[bool] = True
    show_wind_field: Optional[bool] = True
    show_weather_data: Optional[bool] = True
    show_compass: Optional[bool] = True
    show_visualization: Optional[bool] = True
    show_temp_in_F: Optional[bool] = True

PredictionOutput dataclass

Output from a single fire spread prediction.

Attributes:

Name	Type	Description
spread	dict	Maps time in seconds to list of (x, y) positions where fire is predicted to arrive at that time.
flame_len_m	dict	Maps (x, y) to flame length in meters.
fli_kw_m	dict	Maps (x, y) to fireline intensity in kW/m.
ros_ms	dict	Maps (x, y) to rate of spread in m/s.
spread_dir	dict	Maps (x, y) to spread direction in degrees.
all_crown_fire	dict	Maps (x, y) to time in seconds when crown fire is first predicted.
active_crown_fire	dict	Maps time in seconds to dict of ((x, y), crown_fire_status('active' or 'passive')).
end_active_crown	dict	Maps (x,y) to time in seconds when crown fire goes out.
breaches	dict	Maps (x, y) to breach status (bool).
active_fire_front	dict	Maps time in seconds to list of (x, y) positions where fire is predicted to be currently burning at that time.
burnt_spread	dict	Maps time in seconds to list of (x, y) positions where cells are predicted to already be fully burnt.
forecast_index	Optional[int]	Indicates the index of the forecast_pool that was used to generate the prediction.

Source code in embrs/utilities/data_classes.py

@dataclass
class PredictionOutput:
    """Output from a single fire spread prediction.

    Attributes:
        spread (dict): Maps time in seconds to list of (x, y) positions
            where fire is predicted to arrive at that time.
        flame_len_m (dict): Maps (x, y) to flame length in meters.
        fli_kw_m (dict): Maps (x, y) to fireline intensity in kW/m.
        ros_ms (dict): Maps (x, y) to rate of spread in m/s.
        spread_dir (dict): Maps (x, y) to spread direction in degrees.
        all_crown_fire (dict): Maps (x, y) to time in seconds when crown fire is first predicted.
        active_crown_fire (dict): Maps time in seconds to dict of ((x, y), crown_fire_status('active' or 'passive')).
        end_active_crown (dict): Maps (x,y) to time in seconds when crown fire goes out.
        breaches (dict): Maps (x, y) to breach status (bool).
        active_fire_front (dict): Maps time in seconds to list of (x, y) positions
            where fire is predicted to be currently burning at that time.
        burnt_spread (dict): Maps time in seconds to list of (x, y) positions where
            cells are predicted to already be fully burnt.
        forecast_index (Optional[int]): Indicates the index of the forecast_pool that was used to
            generate the prediction.
    """

    spread: dict
    flame_len_m: dict
    fli_kw_m: dict
    ros_ms: dict
    spread_dir: dict
    all_crown_fire: dict
    active_crown_fire: dict
    end_active_crown: dict
    hold_probs: dict
    breaches: dict
    active_fire_front: dict
    burnt_spread: dict
    forecast_index: Optional[int] = None

PredictorParams dataclass

Fire predictor configuration.

Attributes:

Name	Type	Description
time_horizon_hr	float	Prediction time horizon in hours.
time_step_s	int	Prediction time step in seconds.
cell_size_m	float	Cell size in meters.
dead_mf	float	Dead fuel moisture fraction.
live_mf	float	Live fuel moisture fraction.
model_spotting	bool	Whether to model spotting.
spot_delay_s	float	Spot ignition delay in seconds.
wind_speed_bias	float	Wind speed bias in m/s.
wind_dir_bias	float	Wind direction bias in degrees.
wind_uncertainty_factor	float	Wind uncertainty scaling factor.
ros_bias	float	Rate of spread bias factor.
max_wind_speed_bias	float	Maximum wind speed bias in m/s.
max_wind_dir_bias	float	Maximum wind direction bias in degrees.
base_wind_spd_std	float	Base wind speed std dev in m/s.
base_wind_dir_std	float	Base wind direction std dev in degrees.
max_beta	float	Maximum uncertainty beta value.

Source code in embrs/utilities/data_classes.py

@dataclass
class PredictorParams:
    """Fire predictor configuration.

    Attributes:
        time_horizon_hr (float): Prediction time horizon in hours.
        time_step_s (int): Prediction time step in seconds.
        cell_size_m (float): Cell size in meters.
        dead_mf (float): Dead fuel moisture fraction.
        live_mf (float): Live fuel moisture fraction.
        model_spotting (bool): Whether to model spotting.
        spot_delay_s (float): Spot ignition delay in seconds.
        wind_speed_bias (float): Wind speed bias in m/s.
        wind_dir_bias (float): Wind direction bias in degrees.
        wind_uncertainty_factor (float): Wind uncertainty scaling factor.
        ros_bias (float): Rate of spread bias factor.
        max_wind_speed_bias (float): Maximum wind speed bias in m/s.
        max_wind_dir_bias (float): Maximum wind direction bias in degrees.
        base_wind_spd_std (float): Base wind speed std dev in m/s.
        base_wind_dir_std (float): Base wind direction std dev in degrees.
        max_beta (float): Maximum uncertainty beta value.
    """

    time_horizon_hr: float = 2.0
    time_step_s: int = 30
    cell_size_m: float = 30
    dead_mf: float = 0.08
    live_mf: float = 0.30
    model_spotting: bool = False
    spot_delay_s: float = 1200
    wind_speed_bias: float = 0
    wind_dir_bias: float = 0
    wind_uncertainty_factor: float = 0
    ros_bias: float = 0

    max_wind_speed_bias: float = 2.5
    max_wind_dir_bias: float = 20.0
    base_wind_spd_std: float = 1.0
    base_wind_dir_std: float = 5.0
    max_beta: float = 0.95

SimParams dataclass

Full simulation parameters.

Attributes:

Name	Type	Description
map_params	MapParams	Map configuration.
log_folder	str	Path to log output folder.
weather_input	WeatherParams	Weather configuration.
t_step_s	int	Simulation time step in seconds.
cell_size	int	Hexagon side length in meters.
init_mf	List[float]	Initial dead fuel moisture [1hr, 10hr, 100hr].
fuel_moisture_map	Dict	Per-fuel-model moisture values.
fms_has_live	bool	Whether FMS file includes live moisture.
live_h_mf	float	Live herbaceous moisture fraction.
live_w_mf	float	Live woody moisture fraction.
model_spotting	bool	Whether to model spotting.
canopy_species	int	Canopy species ID for spotting.
dbh_cm	float	Diameter at breast height in cm.
spot_ign_prob	float	Spot ignition probability (0-1).
min_spot_dist	float	Minimum spotting distance in meters.
spot_delay_s	float	Spot ignition delay in seconds.
duration_s	float	Simulation duration in seconds.
visualize	bool	Whether to show real-time visualization.
num_runs	int	Number of simulation iterations.
user_path	str	Path to user control module.
user_class	str	Name of user control class.
write_logs	bool	Whether to write log files.

Source code in embrs/utilities/data_classes.py

@dataclass
class SimParams:
    """Full simulation parameters.

    Attributes:
        map_params (MapParams): Map configuration.
        log_folder (str): Path to log output folder.
        weather_input (WeatherParams): Weather configuration.
        t_step_s (int): Simulation time step in seconds.
        cell_size (int): Hexagon side length in meters.
        init_mf (List[float]): Initial dead fuel moisture [1hr, 10hr, 100hr].
        fuel_moisture_map (Dict): Per-fuel-model moisture values.
        fms_has_live (bool): Whether FMS file includes live moisture.
        live_h_mf (float): Live herbaceous moisture fraction.
        live_w_mf (float): Live woody moisture fraction.
        model_spotting (bool): Whether to model spotting.
        canopy_species (int): Canopy species ID for spotting.
        dbh_cm (float): Diameter at breast height in cm.
        spot_ign_prob (float): Spot ignition probability (0-1).
        min_spot_dist (float): Minimum spotting distance in meters.
        spot_delay_s (float): Spot ignition delay in seconds.
        duration_s (float): Simulation duration in seconds.
        visualize (bool): Whether to show real-time visualization.
        num_runs (int): Number of simulation iterations.
        user_path (str): Path to user control module.
        user_class (str): Name of user control class.
        write_logs (bool): Whether to write log files.
    """

    map_params: Optional[MapParams] = None
    log_folder: Optional[str] = None
    weather_input: Optional[WeatherParams] = None
    t_step_s: Optional[int] = None
    cell_size: Optional[int] = None
    init_mf: Optional[List[float]] = field(default_factory=lambda: [0.06, 0.07, 0.08])
    fuel_moisture_map: Dict[int, List[float]] = field(default_factory=dict)
    fms_has_live: bool = False
    live_h_mf: Optional[float] = None
    live_w_mf: Optional[float] = None
    model_spotting: Optional[bool] = False
    canopy_species: Optional[int] = 5
    dbh_cm: Optional[float] = 20.0
    spot_ign_prob: Optional[float] = 0.05
    min_spot_dist: Optional[float] = 50
    spot_delay_s: Optional[float] = 1200
    duration_s: Optional[float] = None
    visualize: Optional[bool] = None
    num_runs: Optional[int] = None
    user_path: Optional[str] = None
    user_class: Optional[str] = None
    burn_area_threshold: Optional[float] = 0.75
    write_logs: Optional[bool] = None

StateEstimate dataclass

Estimated fire state from observations.

Attributes:

Name	Type	Description
burnt_polys	List[Polygon]	Polygons of burnt area.
burning_polys	List[Polygon]	Polygons of actively burning area.
start_time_s	Optional[float]	Start time in seconds from simulation start. If None, uses current fire simulation time.
scheduled_ignitions	Optional[Dict[float, List[Polygon]]]	Future ignitions keyed by absolute simulation time (seconds). Used for staggered strip firing rollouts. Polygons are resolved to cells inside the worker process via get_cells_at_geometry().

Source code in embrs/utilities/data_classes.py

@dataclass
class StateEstimate:
    """Estimated fire state from observations.

    Attributes:
        burnt_polys (List[Polygon]): Polygons of burnt area.
        burning_polys (List[Polygon]): Polygons of actively burning area.
        start_time_s (Optional[float]): Start time in seconds from simulation start.
            If None, uses current fire simulation time.
        scheduled_ignitions (Optional[Dict[float, List[Polygon]]]): Future ignitions
            keyed by absolute simulation time (seconds). Used for staggered strip
            firing rollouts. Polygons are resolved to cells inside the worker
            process via get_cells_at_geometry().
    """

    burnt_polys: List[Polygon]
    burning_polys: List[Polygon]
    start_time_s: Optional[float] = None
    scheduled_ignitions: Optional[Dict[float, List[Polygon]]] = None

VisualizerInputs dataclass

Input data for the real-time visualizer.

Attributes:

Name	Type	Description
cell_size	float	Hexagon side length in meters.
sim_shape	Tuple[int, int]	Grid shape (rows, cols).
sim_size	Tuple[float, float]	Map size (width_m, height_m).
start_datetime	datetime	Simulation start time.
north_dir_deg	float	Rotation to true north in degrees.
wind_forecast	ndarray	Wind field data array.
wind_resolution	float	Wind mesh resolution in meters.
wind_t_step	float	Wind time step in seconds.
wind_xpad	float	Wind field x padding in meters.
wind_ypad	float	Wind field y padding in meters.
temp_forecast	ndarray	Temperature forecast values.
rh_forecast	ndarray	Relative humidity forecast values.
forecast_t_step	float	Forecast time step in seconds.
elevation	ndarray	Coarse elevation data for display.
roads	list	Road geometries for display.
fire_breaks	list	Fire break geometries for display.
init_entries	list	Initial cell state entries.
scale_bar_km	float	Scale bar size in kilometers.
show_legend	bool	Whether to show fuel legend.
show_wind_cbar	bool	Whether to show wind colorbar.
show_wind_field	bool	Whether to show wind field.
show_weather_data	bool	Whether to show weather info.
show_temp_in_F	bool	Whether to show temperature in Fahrenheit.
show_compass	bool	Whether to show compass.

Source code in embrs/utilities/data_classes.py

@dataclass
class VisualizerInputs:
    """Input data for the real-time visualizer.

    Attributes:
        cell_size (float): Hexagon side length in meters.
        sim_shape (Tuple[int, int]): Grid shape (rows, cols).
        sim_size (Tuple[float, float]): Map size (width_m, height_m).
        start_datetime (datetime): Simulation start time.
        north_dir_deg (float): Rotation to true north in degrees.
        wind_forecast (np.ndarray): Wind field data array.
        wind_resolution (float): Wind mesh resolution in meters.
        wind_t_step (float): Wind time step in seconds.
        wind_xpad (float): Wind field x padding in meters.
        wind_ypad (float): Wind field y padding in meters.
        temp_forecast (np.ndarray): Temperature forecast values.
        rh_forecast (np.ndarray): Relative humidity forecast values.
        forecast_t_step (float): Forecast time step in seconds.
        elevation (np.ndarray): Coarse elevation data for display.
        roads (list): Road geometries for display.
        fire_breaks (list): Fire break geometries for display.
        init_entries (list): Initial cell state entries.
        scale_bar_km (float): Scale bar size in kilometers.
        show_legend (bool): Whether to show fuel legend.
        show_wind_cbar (bool): Whether to show wind colorbar.
        show_wind_field (bool): Whether to show wind field.
        show_weather_data (bool): Whether to show weather info.
        show_temp_in_F (bool): Whether to show temperature in Fahrenheit.
        show_compass (bool): Whether to show compass.
    """

    cell_size: float
    sim_shape: Tuple[int, int]
    sim_size: Tuple[float, float]
    start_datetime: datetime
    north_dir_deg: float
    wind_forecast: np.ndarray
    wind_resolution: float
    wind_t_step: float
    wind_xpad: float
    wind_ypad: float
    temp_forecast: np.ndarray
    rh_forecast: np.ndarray
    forecast_t_step: float
    elevation: np.ndarray
    roads: list
    fire_breaks: list
    init_entries: list

    scale_bar_km: Optional[float] = 1.0
    show_legend: Optional[bool] = True
    show_wind_cbar: Optional[bool] = True
    show_wind_field: Optional[bool] = True
    show_weather_data: Optional[bool] = True
    show_temp_in_F: Optional[bool] = True
    show_compass: Optional[bool] = True

WeatherEntry dataclass

Single weather observation at a point in time.

Attributes:

Name	Type	Description
wind_speed	float	Wind speed in m/s.
wind_dir_deg	float	Wind direction in degrees (meteorological).
temp	float	Temperature in Celsius.
rel_humidity	float	Relative humidity as fraction (0-1).
cloud_cover	float	Cloud cover as fraction (0-1).
rain	float	Rainfall in mm.
dni	float	Direct normal irradiance in W/m^2.
dhi	float	Diffuse horizontal irradiance in W/m^2.
ghi	float	Global horizontal irradiance in W/m^2.
solar_zenith	float	Solar zenith angle in degrees.
solar_azimuth	float	Solar azimuth angle in degrees.

Source code in embrs/utilities/data_classes.py

@dataclass
class WeatherEntry:
    """Single weather observation at a point in time.

    Attributes:
        wind_speed (float): Wind speed in m/s.
        wind_dir_deg (float): Wind direction in degrees (meteorological).
        temp (float): Temperature in Celsius.
        rel_humidity (float): Relative humidity as fraction (0-1).
        cloud_cover (float): Cloud cover as fraction (0-1).
        rain (float): Rainfall in mm.
        dni (float): Direct normal irradiance in W/m^2.
        dhi (float): Diffuse horizontal irradiance in W/m^2.
        ghi (float): Global horizontal irradiance in W/m^2.
        solar_zenith (float): Solar zenith angle in degrees.
        solar_azimuth (float): Solar azimuth angle in degrees.
    """

    wind_speed: float
    wind_dir_deg: float
    temp: float
    rel_humidity: float
    cloud_cover: float
    rain: float
    dni: float
    dhi: float
    ghi: float
    solar_zenith: float
    solar_azimuth: float

WeatherParams dataclass

Weather input configuration.

Attributes:

Name	Type	Description
input_type	str	Weather source type ('OpenMeteo' or 'File').
file	str	Path to weather file if using file input.
mesh_resolution	int	WindNinja mesh resolution in meters.
conditioning_start	datetime	Start time for fuel moisture conditioning.
start_datetime	datetime	Simulation start time.
end_datetime	datetime	Simulation end time.

Source code in embrs/utilities/data_classes.py

@dataclass
class WeatherParams:
    """Weather input configuration.

    Attributes:
        input_type (str): Weather source type ('OpenMeteo' or 'File').
        file (str): Path to weather file if using file input.
        mesh_resolution (int): WindNinja mesh resolution in meters.
        conditioning_start (datetime): Start time for fuel moisture conditioning.
        start_datetime (datetime): Simulation start time.
        end_datetime (datetime): Simulation end time.
    """

    input_type: Optional[str] = None
    file: Optional[str] = ""
    mesh_resolution: Optional[int] = None
    conditioning_start: Optional[datetime] = None
    start_datetime: Optional[datetime] = None
    end_datetime: Optional[datetime] = None

WeatherStream

Build and manage a weather stream for fire simulation.

Ingest weather data from either the Open-Meteo API or a RAWS-format .wxs file, compute derived quantities (solar geometry, GSI, live fuel moisture, foliar moisture content), and produce a list of WeatherEntry records indexed by time.

Attributes:

Name	Type	Description
stream	list[WeatherEntry]	Ordered weather entries for the simulation period (including conditioning lead-in).
stream_times	DatetimeIndex	Timestamps corresponding to each entry in stream.
sim_start_idx	int	Index into stream where the actual simulation begins (after conditioning period).
fmc	float	Foliar moisture content (percent).
live_h_mf	float | None	Live herbaceous fuel moisture (fraction), or None if GSI is disabled.
live_w_mf	float | None	Live woody fuel moisture (fraction), or None if GSI is disabled.
time_step	int	Weather observation interval (minutes).
ref_elev	float	Reference elevation of the weather source (meters).

Source code in embrs/models/weather.py

class WeatherStream:
    """Build and manage a weather stream for fire simulation.

    Ingest weather data from either the Open-Meteo API or a RAWS-format
    ``.wxs`` file, compute derived quantities (solar geometry, GSI, live
    fuel moisture, foliar moisture content), and produce a list of
    ``WeatherEntry`` records indexed by time.

    Attributes:
        stream (list[WeatherEntry]): Ordered weather entries for the
            simulation period (including conditioning lead-in).
        stream_times (pd.DatetimeIndex): Timestamps corresponding to
            each entry in ``stream``.
        sim_start_idx (int): Index into ``stream`` where the actual
            simulation begins (after conditioning period).
        fmc (float): Foliar moisture content (percent).
        live_h_mf (float | None): Live herbaceous fuel moisture (fraction),
            or None if GSI is disabled.
        live_w_mf (float | None): Live woody fuel moisture (fraction),
            or None if GSI is disabled.
        time_step (int): Weather observation interval (minutes).
        ref_elev (float): Reference elevation of the weather source (meters).
    """

    def __init__(self, params: WeatherParams, geo: GeoInfo, use_gsi: bool = True):
        """Initialize a weather stream from configuration parameters.

        Args:
            params (WeatherParams): Weather configuration specifying input
                type, date range, and optional file path.
            geo (GeoInfo): Geographic information (lat, lon, timezone,
                center coordinates).
            use_gsi (bool): Whether to compute the Growing Season Index for
                live fuel moisture estimation. Defaults to True.

        Raises:
            ValueError: If ``params.input_type`` is not 'OpenMeteo' or
                'File', or if 'File' is used without ``geo``.
        """
        self.params = params
        self.geo = geo
        self.use_gsi = use_gsi
        input_type = params.input_type

        if input_type == "OpenMeteo":
            self.get_stream_from_openmeteo()
        elif input_type == "File":
            if geo is not None:
                self.get_stream_from_wxs()
            else:
                raise ValueError("GeoInfo must be provided when using 'File' weather input_type")
        else:
            raise ValueError("Invalid weather input_type, must be either 'OpenMeteo' or 'File'")

    def get_stream_from_openmeteo(self):
        """Fetch weather data from the Open-Meteo historical archive API.

        Retrieves hourly wind, temperature, humidity, cloud cover, solar
        radiation, and precipitation. Computes GSI-based live fuel moisture
        if enabled, then builds the weather stream with a conditioning
        lead-in period.

        Side Effects:
            Sets ``self.stream``, ``self.stream_times``, ``self.sim_start_idx``,
            ``self.fmc``, ``self.live_h_mf``, ``self.live_w_mf``,
            ``self.ref_elev``, ``self.time_step``, and input unit attributes.
        """
        # Setup the Open-Meteo API client with cache and retry on error
        cache_session = requests_cache.CachedSession('.cache', expire_after = -1)
        retry_session = retry(cache_session, retries = 5, backoff_factor = 0.2)
        openmeteo = openmeteo_requests.Client(session = retry_session)
        local_tz = pytz.timezone(self.geo.timezone)

        # Buffer times and format for OpenMeteo
        conditioning_start = self.params.conditioning_start
        conditioning_start = local_tz.localize(conditioning_start)
        start_datetime = self.params.start_datetime
        start_datetime = local_tz.localize(start_datetime)
        buffered_start = start_datetime - timedelta(days=56)
        start_datetime_utc = buffered_start.astimezone(pytz.utc)

        end_datetime = self.params.end_datetime
        end_datetime = local_tz.localize(end_datetime)
        buffered_end = end_datetime + timedelta(days=1)
        end_datetime_utc = buffered_end.astimezone(pytz.utc)

        url = "https://archive-api.open-meteo.com/v1/archive"
        api_input = {
            "latitude": self.geo.center_lat,
            "longitude": self.geo.center_lon,
            "start_date": start_datetime_utc.date().strftime("%Y-%m-%d"),
            "end_date": end_datetime_utc.date().strftime("%Y-%m-%d"),
            "hourly": ["wind_speed_10m", "wind_direction_10m", "temperature_2m", "relative_humidity_2m", "cloud_cover", "shortwave_radiation", "diffuse_radiation", "direct_normal_irradiance", "rain"],
            "wind_speed_unit": "ms",
            "temperature_unit": "fahrenheit"
        }
        responses = openmeteo.weather_api(url, params=api_input)
        response = responses[0]

        self.ref_elev = response.Elevation()
        hourly = response.Hourly()
        hourly_wind_speed_10m = hourly.Variables(0).ValuesAsNumpy()
        hourly_wind_direction_10m = hourly.Variables(1).ValuesAsNumpy()
        hourly_temperature_2m = hourly.Variables(2).ValuesAsNumpy()
        hourly_rel_humidity_2m = hourly.Variables(3).ValuesAsNumpy()
        hourly_cloud_cover = hourly.Variables(4).ValuesAsNumpy()
        hourly_ghi = hourly.Variables(5).ValuesAsNumpy()
        hourly_dhi = hourly.Variables(6).ValuesAsNumpy()
        hourly_dni = hourly.Variables(7).ValuesAsNumpy()
        hourly_rain_mm = hourly.Variables(8).ValuesAsNumpy()

        hourly_data = {}

        hourly_data["date"] = pd.date_range(
            start=pd.to_datetime(hourly.Time(), unit="s"),
            end=pd.to_datetime(hourly.TimeEnd(), unit="s"),
            freq=pd.Timedelta(seconds=hourly.Interval()),
            inclusive="left"
        ).tz_localize('UTC').tz_convert(local_tz)

        self.times = pd.date_range(hourly_data["date"][0], hourly_data["date"][-1], freq='h', tz=local_tz)

        hourly_data["wind_speed"] = hourly_wind_speed_10m
        hourly_data["wind_direction"] = hourly_wind_direction_10m
        hourly_data["temperature"] = hourly_temperature_2m
        hourly_data["rel_humidity"] = hourly_rel_humidity_2m
        hourly_data["cloud_cover"] = hourly_cloud_cover
        hourly_data["ghi"] = hourly_ghi
        hourly_data["dhi"] = hourly_dhi
        hourly_data["dni"] = hourly_dni
        hourly_data["rain"] = hourly_rain_mm * 0.1 # convert to cm

        solpos = pvlib.solarposition.get_solarposition(self.times, self.geo.center_lat, self.geo.center_lon)

        hourly_data["solar_zenith"] = solpos["zenith"].values
        hourly_data["solar_azimuth"] = solpos["azimuth"].values

        if self.use_gsi:
            pre_summaries = self._build_pre_sim_summaries(
                hourly_data, buffered_start, start_datetime
            )
            self.gsi_tracker = GSITracker(self.geo, pre_summaries)
            gsi = self.gsi_tracker.compute_gsi()
            self.live_h_mf, self.live_w_mf = self.set_live_moistures(gsi)
        else:
            self.live_h_mf = None
            self.live_w_mf = None
            self.gsi_tracker = None

        hourly = filter_hourly_data(hourly_data, conditioning_start, end_datetime)
        self.stream_times = pd.DatetimeIndex(hourly["date"])
        self.stream = list(self.generate_stream(hourly))

        try:
            self.sim_start_idx = self.stream_times.get_loc(start_datetime)
        except KeyError:
            self.sim_start_idx = int(self.stream_times.searchsorted(start_datetime, side="left"))

        if self.use_gsi:
            # Build GSI tracker seeded with pre-simulation daily summaries
            pre_summaries = self._build_pre_sim_summaries(
                hourly_data, buffered_start, start_datetime
            )
            init_rain = self.stream[self.sim_start_idx].rain
            self.gsi_tracker = GSITracker(self.geo, pre_summaries, init_rain)
        else:
            self.gsi_tracker = None
        if self.gsi_tracker is not None:
            self.gsi_tracker._last_cum_rain = self.stream[self.sim_start_idx].rain

        # Calculate foliar moisture content
        self.fmc = self.calc_fmc()

        # Set units and time step based on OpenMeteo params
        self.time_step = 60
        self.input_wind_ht = 10
        self.input_wind_ht_units = "m"
        self.input_wind_vel_units = "mps"
        self.input_temp_units = "F"

    def get_stream_from_wxs(self):
        """Parse a RAWS-format ``.wxs`` weather file and build the stream.

        Read weather observations from the file, apply unit conversions
        (English or Metric to internal units), fetch solar irradiance from
        Open-Meteo to supplement the file data, compute GSI, and assemble
        the final weather stream.

        Side Effects:
            Sets ``self.stream``, ``self.stream_times``, ``self.sim_start_idx``,
            ``self.fmc``, ``self.live_h_mf``, ``self.live_w_mf``,
            ``self.ref_elev``, ``self.time_step``, and input unit attributes.

        Raises:
            ValueError: If the file has insufficient data or unknown units.
        """
        file = self.params.file

        weather_data = {
            "datetime": [],
            "temperature": [],
            "rel_humidity": [],
            "rain": [],
            "wind_speed": [],
            "wind_direction": [],
            "cloud_cover": [],
        }

        local_tz = pytz.timezone(self.geo.timezone)
        units = "english"

        # ── Step 1: Parse WXS line-by-line ───────────────────────────
        with open(file, "r") as f:
            header_found = False
            for line in f:
                line = line.strip()
                if not line:
                    continue
                if line.startswith("RAWS_UNITS:"):
                    units = line.split(":")[1].strip().lower()
                    continue
                elif line.startswith("RAWS_ELEVATION:"):
                    self.ref_elev = float(line.split(":")[1].strip().lower())
                    continue
                elif line.startswith("RAWS:"):
                    continue
                elif line.startswith("Year") and not header_found:
                    header_found = True
                    continue
                if not header_found:
                    continue

                parts = line.split()
                if len(parts) != 10:
                    continue
                try:
                    year, month, day = int(parts[0]), int(parts[1]), int(parts[2])
                    hour = int(parts[3].zfill(4)[:2])
                    dt = local_tz.localize(datetime(year, month, day, hour))
                    weather_data["datetime"].append(dt)
                    weather_data["temperature"].append(float(parts[4]))
                    weather_data["rel_humidity"].append(float(parts[5]))
                    weather_data["rain"].append(float(parts[6]))
                    weather_data["wind_speed"].append(float(parts[7]))
                    weather_data["wind_direction"].append(float(parts[8]))
                    weather_data["cloud_cover"].append(float(parts[9]))
                except Exception as e:
                    print(f"Skipping malformed line: {line} ({e})")
                    continue

        df = pd.DataFrame(weather_data).set_index("datetime")

        if len(df.index) < 2:
            raise ValueError("WXS file does not contain enough data to determine time step.")

        # ── Step 2: Infer time step and apply unit conversions ───────
        time_step_min = int((df.index[1] - df.index[0]).total_seconds() / 60)
        if units == "english":
            df["rain"] *= 2.54
            df["wind_speed"] *= 0.44704
            self.ref_elev = ft_to_m(self.ref_elev)
        elif units == "metric":
            df["temperature"] = df["temperature"] * 9 / 5 + 32
            df["rain"] /= 10
        else:
            raise ValueError(f"Unknown units: {units}")

        # ── Step 3: Fetch irradiance from Open-Meteo ─────────────────
        # Use buffer around sim dates
        conditioning_start = local_tz.localize(self.params.conditioning_start)
        start_datetime = local_tz.localize(self.params.start_datetime)
        end_datetime = local_tz.localize(self.params.end_datetime)

        # Check bounds
        wxs_start = df.index.min()
        wxs_end = df.index.max()

        if start_datetime < wxs_start:
            raise ValueError(f"Start datetime {start_datetime} is before WXS data begins at {wxs_start}.")

        if end_datetime > wxs_end:
            raise ValueError(f"End datetime {end_datetime} is after WXS data ends at {wxs_end}.")

        buffered_start = start_datetime - timedelta(days=56)
        buffered_end = end_datetime + timedelta(days=1)

        cache_session = requests_cache.CachedSession('.cache', expire_after=-1)
        retry_session = retry(cache_session, retries=5, backoff_factor=0.2)
        openmeteo = openmeteo_requests.Client(session=retry_session)

        url = "https://archive-api.open-meteo.com/v1/archive"
        api_input = {
            "latitude": self.geo.center_lat,
            "longitude": self.geo.center_lon,
            "start_date": buffered_start.astimezone(pytz.utc).date().strftime("%Y-%m-%d"),
            "end_date": buffered_end.astimezone(pytz.utc).date().strftime("%Y-%m-%d"),
            "hourly": ["shortwave_radiation", "diffuse_radiation", "direct_normal_irradiance"],
            "timezone": "auto"
        }

        responses = openmeteo.weather_api(url, params=api_input)
        response = responses[0]

        hourly = response.Hourly()
        hourly_data = {
            "ghi": hourly.Variables(0).ValuesAsNumpy(),
            "dhi": hourly.Variables(1).ValuesAsNumpy(),
            "dni": hourly.Variables(2).ValuesAsNumpy(),
            "date": pd.date_range(
                start=pd.to_datetime(hourly.Time(), unit="s"),
                end=pd.to_datetime(hourly.TimeEnd(), unit="s"),
                freq=pd.Timedelta(seconds=hourly.Interval()),
                inclusive="left"
            ).tz_localize(local_tz)
        }

        irradiance_df = pd.DataFrame(hourly_data).set_index("date")

        # ── Step 4: Resample Open-Meteo irradiance to WXS resolution ─
        target_freq = f"{time_step_min}T"
        if time_step_min < 60:
            irradiance_df = irradiance_df.resample(target_freq).interpolate(method="linear")
        elif time_step_min > 60:
            irradiance_df = irradiance_df.resample(target_freq).mean()

        # ── Step 5: Align WXS and Open-Meteo data on datetime index ──
        df = df.loc[(df.index >= buffered_start) & (df.index <= end_datetime)]
        irradiance_df = irradiance_df.loc[df.index]

        df["ghi"] = irradiance_df["ghi"].values
        df["dhi"] = irradiance_df["dhi"].values
        df["dni"] = irradiance_df["dni"].values

        # ── Step 6: Add solar geometry ───────────────────────────────
        solpos = pvlib.solarposition.get_solarposition(df.index, self.geo.center_lat, self.geo.center_lon)
        df["solar_zenith"] = solpos["zenith"].values
        df["solar_azimuth"] = solpos["azimuth"].values

        if self.use_gsi:
            # Determine how far back we can go
            min_date_in_wxs = df.index.min()
            desired_start = start_datetime - timedelta(days=56)
            data_start = max(min_date_in_wxs, desired_start)

            wxs_hourly = {
                "date": df.index,
                "temperature": df["temperature"].values,
                "rel_humidity": df["rel_humidity"].values,
                "rain": df["rain"].values,
            }
            pre_summaries = self._build_pre_sim_summaries(
                wxs_hourly, data_start, start_datetime
            )
            self.gsi_tracker = GSITracker(self.geo, pre_summaries)
            gsi = self.gsi_tracker.compute_gsi()
            self.live_h_mf, self.live_w_mf = self.set_live_moistures(gsi)
        else:
            self.live_h_mf = None
            self.live_w_mf = None
            self.gsi_tracker = None

        # Calculate foliar moisture content
        self.fmc = self.calc_fmc()

        # ── Step 8: Package final stream ─────────────────────────────
        df["date"] = df.index
        hourly_data = filter_hourly_data(df, conditioning_start, end_datetime)
        self.stream_times = pd.DatetimeIndex(hourly_data["date"])

        try:
            self.sim_start_idx = self.stream_times.get_loc(start_datetime)
        except KeyError:
            self.sim_start_idx = int(self.stream_times.searchsorted(start_datetime, side="left"))

        self.stream = list(self.generate_stream(hourly_data))

        if self.use_gsi:
            # Build GSI tracker seeded with pre-simulation daily summaries
            wxs_hourly = {
                "date": df.index,
                "temperature": df["temperature"].values,
                "rel_humidity": df["rel_humidity"].values,
                "rain": df["rain"].values,
            }
            pre_summaries = self._build_pre_sim_summaries(
                wxs_hourly, data_start, start_datetime
            )
            init_rain = self.stream[self.sim_start_idx].rain
            self.gsi_tracker = GSITracker(self.geo, pre_summaries, init_rain)
        else:
            self.gsi_tracker = None
        if self.gsi_tracker is not None:
            self.gsi_tracker._last_cum_rain = self.stream[self.sim_start_idx].rain

        # ── Step 9: Set metadata attributes ──────────────────────────
        self.time_step = time_step_min
        self.input_wind_ht = 6.1
        self.input_wind_ht_units = "m"
        self.input_wind_vel_units = "mps"
        self.input_temp_units = "F"

    def generate_stream(self, hourly_data: dict) -> Iterator[WeatherEntry]:
        """Yield full WeatherEntry records from hourly data arrays.

        Rainfall is accumulated across entries (cumulative sum).

        Args:
            hourly_data (dict): Dictionary with keys 'wind_speed',
                'wind_direction', 'temperature', 'rel_humidity',
                'cloud_cover', 'ghi', 'dhi', 'dni', 'rain',
                'solar_zenith', 'solar_azimuth'.

        Yields:
            WeatherEntry: Fully populated weather entry with cumulative
                rainfall.
        """
        cum_rain = 0
        for wind_speed, wind_dir, temp, rel_humidity, cloud_cover, ghi, dhi, dni, rain, solar_zenith, solar_azimuth in zip(
            hourly_data["wind_speed"],
            hourly_data["wind_direction"],
            hourly_data["temperature"],
            hourly_data["rel_humidity"],
            hourly_data["cloud_cover"],
            hourly_data["ghi"],
            hourly_data["dhi"],
            hourly_data["dni"],
            hourly_data["rain"],
            hourly_data["solar_zenith"],
            hourly_data["solar_azimuth"]
        ):
            cum_rain += rain
            yield WeatherEntry(
                wind_speed=wind_speed,
                wind_dir_deg=wind_dir,
                temp=temp,
                rel_humidity=rel_humidity,
                cloud_cover=cloud_cover,
                rain = cum_rain,
                dni=dni,
                dhi=dhi,
                ghi=ghi,
                solar_zenith=solar_zenith,
                solar_azimuth=solar_azimuth
            )

    def set_live_moistures(self, gsi: float) -> tuple:
        """Compute live fuel moisture fractions from the Growing Season Index.

        Map GSI to live herbaceous and woody moisture using linear
        interpolation between dormant and green-up values. Below the
        green-up threshold (GSI < 0.2), dormant values are returned.

        Args:
            gsi (float): Growing Season Index in [0, 1].

        Returns:
            Tuple[float, float]: ``(live_h_mf, live_w_mf)`` live herbaceous
                and live woody moisture content (fractions).
        """
        # Dormant values
        h_min = 0.3
        w_min = 0.6

        # Green-up threshold
        gu = 0.2

        # Max values
        h_max = 2.5
        w_max = 2.0

        if gsi < gu:
            return h_min, w_min

        else:
            m_h = (h_max - h_min) / (1.0 - gu)
            m_w = (w_max - w_min) / (1.0 - gu)

        live_h_mf = m_h * gsi + (h_max - m_h)
        live_w_mf = m_w * gsi + (w_max - m_w)

        return live_h_mf, live_w_mf

    def _build_pre_sim_summaries(self, hourly_data: dict,
                                 data_start, sim_start) -> List[DailySummary]:
        """Convert pre-simulation hourly data into daily summaries for GSI tracking.

        Aggregates hourly observations into per-day min/max temperature,
        min humidity, and total rainfall, matching the aggregation used
        by :meth:`GSITracker.compute_gsi`.

        Args:
            hourly_data: Hourly weather dictionary with keys ``'date'``,
                ``'temperature'`` (Fahrenheit), ``'rel_humidity'`` (percent),
                ``'rain'`` (cm, non-cumulative per-hour increments).
            data_start: Start of the pre-simulation data window.
            sim_start: Simulation start datetime.

        Returns:
            List of :class:`DailySummary`, oldest first, covering up to
            56 days before *sim_start*.
        """
        filtered = filter_hourly_data(hourly_data, data_start, sim_start)
        df = pd.DataFrame(filtered).set_index("date")

        if df.empty:
            return []

        daily_min_temp = df["temperature"].resample('D').min()
        daily_max_temp = df["temperature"].resample('D').max()
        daily_min_rh = df["rel_humidity"].resample('D').min()
        daily_rain = df["rain"].resample('D').sum()

        summaries = []
        for date in daily_min_temp.index:
            summaries.append(DailySummary(
                date=date.date() if hasattr(date, 'date') else date,
                min_temp_F=daily_min_temp[date],
                max_temp_F=daily_max_temp[date],
                min_rh=daily_min_rh[date],
                rain_cm=daily_rain[date],
            ))
        return summaries

    def _build_pre_sim_summaries(self, hourly_data: dict,
                                 data_start, sim_start) -> List[DailySummary]:
        """Convert pre-simulation hourly data into daily summaries for GSI tracking.

        Aggregates hourly observations into per-day min/max temperature,
        min humidity, and total rainfall, matching the aggregation used
        by :meth:`calc_GSI`.

        Args:
            hourly_data: Hourly weather dictionary with keys ``'date'``,
                ``'temperature'`` (Fahrenheit), ``'rel_humidity'`` (percent),
                ``'rain'`` (cm, non-cumulative per-hour increments).
            data_start: Start of the pre-simulation data window.
            sim_start: Simulation start datetime.

        Returns:
            List of :class:`DailySummary`, oldest first, covering up to
            56 days before *sim_start*.
        """
        filtered = filter_hourly_data(hourly_data, data_start, sim_start)
        df = pd.DataFrame(filtered).set_index("date")

        if df.empty:
            return []

        daily_min_temp = df["temperature"].resample('D').min()
        daily_max_temp = df["temperature"].resample('D').max()
        daily_min_rh = df["rel_humidity"].resample('D').min()
        daily_rain = df["rain"].resample('D').sum()

        summaries = []
        for date in daily_min_temp.index:
            summaries.append(DailySummary(
                date=date.date() if hasattr(date, 'date') else date,
                min_temp_F=daily_min_temp[date],
                max_temp_F=daily_max_temp[date],
                min_rh=daily_min_rh[date],
                rain_cm=daily_rain[date],
            ))
        return summaries

    def calc_fmc(self) -> float:
        """Compute foliar moisture content based on latitude and date.

        Uses the Forestry Canada Fire Danger Group (1992) method, which
        estimates the day of minimum FMC from latitude, longitude, and
        elevation, then applies a polynomial fit.

        Returns:
            float: Foliar moisture content (percent).
        """
        lat = self.geo.center_lat
        lon = -self.geo.center_lon

        # Convert start date to julian date
        date = self.params.start_datetime
        d_j = date.timetuple().tm_yday

        latn = 43 + 33.7 * np.exp(-0.0351 * (150 - lon))

        # Estimate day of the year where the min FMC occurs
        d_0 = 151 * (lat / latn) + 0.0172 * self.ref_elev

        # Estimate the number of days away current date is from d_0
        nd = min(abs(d_j - d_0), 365 - abs(d_j - d_0))

        # Calculate fmc based on nd threshold
        if nd < 30:
            fmc = 85 + 0.0189 * nd**2

        elif 30 <= nd < 50:
            fmc = 32.9 + 3.17 * nd - 0.0288 * nd**2

        else:
            fmc = 120

        return fmc

__init__(params, geo, use_gsi=True)

Initialize a weather stream from configuration parameters.

Parameters:

Name	Type	Description	Default
params	WeatherParams	Weather configuration specifying input type, date range, and optional file path.	required
geo	GeoInfo	Geographic information (lat, lon, timezone, center coordinates).	required
use_gsi	bool	Whether to compute the Growing Season Index for live fuel moisture estimation. Defaults to True.	True

Raises:

Type	Description
ValueError	If params.input_type is not 'OpenMeteo' or 'File', or if 'File' is used without geo.

Source code in embrs/models/weather.py

def __init__(self, params: WeatherParams, geo: GeoInfo, use_gsi: bool = True):
    """Initialize a weather stream from configuration parameters.

    Args:
        params (WeatherParams): Weather configuration specifying input
            type, date range, and optional file path.
        geo (GeoInfo): Geographic information (lat, lon, timezone,
            center coordinates).
        use_gsi (bool): Whether to compute the Growing Season Index for
            live fuel moisture estimation. Defaults to True.

    Raises:
        ValueError: If ``params.input_type`` is not 'OpenMeteo' or
            'File', or if 'File' is used without ``geo``.
    """
    self.params = params
    self.geo = geo
    self.use_gsi = use_gsi
    input_type = params.input_type

    if input_type == "OpenMeteo":
        self.get_stream_from_openmeteo()
    elif input_type == "File":
        if geo is not None:
            self.get_stream_from_wxs()
        else:
            raise ValueError("GeoInfo must be provided when using 'File' weather input_type")
    else:
        raise ValueError("Invalid weather input_type, must be either 'OpenMeteo' or 'File'")

calc_fmc()

Compute foliar moisture content based on latitude and date.

Uses the Forestry Canada Fire Danger Group (1992) method, which estimates the day of minimum FMC from latitude, longitude, and elevation, then applies a polynomial fit.

Returns:

Name	Type	Description
float	float	Foliar moisture content (percent).

Source code in embrs/models/weather.py

def calc_fmc(self) -> float:
    """Compute foliar moisture content based on latitude and date.

    Uses the Forestry Canada Fire Danger Group (1992) method, which
    estimates the day of minimum FMC from latitude, longitude, and
    elevation, then applies a polynomial fit.

    Returns:
        float: Foliar moisture content (percent).
    """
    lat = self.geo.center_lat
    lon = -self.geo.center_lon

    # Convert start date to julian date
    date = self.params.start_datetime
    d_j = date.timetuple().tm_yday

    latn = 43 + 33.7 * np.exp(-0.0351 * (150 - lon))

    # Estimate day of the year where the min FMC occurs
    d_0 = 151 * (lat / latn) + 0.0172 * self.ref_elev

    # Estimate the number of days away current date is from d_0
    nd = min(abs(d_j - d_0), 365 - abs(d_j - d_0))

    # Calculate fmc based on nd threshold
    if nd < 30:
        fmc = 85 + 0.0189 * nd**2

    elif 30 <= nd < 50:
        fmc = 32.9 + 3.17 * nd - 0.0288 * nd**2

    else:
        fmc = 120

    return fmc

generate_stream(hourly_data)

Yield full WeatherEntry records from hourly data arrays.

Rainfall is accumulated across entries (cumulative sum).

Parameters:

Name	Type	Description	Default
hourly_data	dict	Dictionary with keys 'wind_speed', 'wind_direction', 'temperature', 'rel_humidity', 'cloud_cover', 'ghi', 'dhi', 'dni', 'rain', 'solar_zenith', 'solar_azimuth'.	required

Yields:

Name	Type	Description
WeatherEntry	WeatherEntry	Fully populated weather entry with cumulative rainfall.

Source code in embrs/models/weather.py

def generate_stream(self, hourly_data: dict) -> Iterator[WeatherEntry]:
    """Yield full WeatherEntry records from hourly data arrays.

    Rainfall is accumulated across entries (cumulative sum).

    Args:
        hourly_data (dict): Dictionary with keys 'wind_speed',
            'wind_direction', 'temperature', 'rel_humidity',
            'cloud_cover', 'ghi', 'dhi', 'dni', 'rain',
            'solar_zenith', 'solar_azimuth'.

    Yields:
        WeatherEntry: Fully populated weather entry with cumulative
            rainfall.
    """
    cum_rain = 0
    for wind_speed, wind_dir, temp, rel_humidity, cloud_cover, ghi, dhi, dni, rain, solar_zenith, solar_azimuth in zip(
        hourly_data["wind_speed"],
        hourly_data["wind_direction"],
        hourly_data["temperature"],
        hourly_data["rel_humidity"],
        hourly_data["cloud_cover"],
        hourly_data["ghi"],
        hourly_data["dhi"],
        hourly_data["dni"],
        hourly_data["rain"],
        hourly_data["solar_zenith"],
        hourly_data["solar_azimuth"]
    ):
        cum_rain += rain
        yield WeatherEntry(
            wind_speed=wind_speed,
            wind_dir_deg=wind_dir,
            temp=temp,
            rel_humidity=rel_humidity,
            cloud_cover=cloud_cover,
            rain = cum_rain,
            dni=dni,
            dhi=dhi,
            ghi=ghi,
            solar_zenith=solar_zenith,
            solar_azimuth=solar_azimuth
        )

get_stream_from_openmeteo()

Fetch weather data from the Open-Meteo historical archive API.

Retrieves hourly wind, temperature, humidity, cloud cover, solar radiation, and precipitation. Computes GSI-based live fuel moisture if enabled, then builds the weather stream with a conditioning lead-in period.

Side Effects

Sets self.stream, self.stream_times, self.sim_start_idx, self.fmc, self.live_h_mf, self.live_w_mf, self.ref_elev, self.time_step, and input unit attributes.

Source code in embrs/models/weather.py

def get_stream_from_openmeteo(self):
    """Fetch weather data from the Open-Meteo historical archive API.

    Retrieves hourly wind, temperature, humidity, cloud cover, solar
    radiation, and precipitation. Computes GSI-based live fuel moisture
    if enabled, then builds the weather stream with a conditioning
    lead-in period.

    Side Effects:
        Sets ``self.stream``, ``self.stream_times``, ``self.sim_start_idx``,
        ``self.fmc``, ``self.live_h_mf``, ``self.live_w_mf``,
        ``self.ref_elev``, ``self.time_step``, and input unit attributes.
    """
    # Setup the Open-Meteo API client with cache and retry on error
    cache_session = requests_cache.CachedSession('.cache', expire_after = -1)
    retry_session = retry(cache_session, retries = 5, backoff_factor = 0.2)
    openmeteo = openmeteo_requests.Client(session = retry_session)
    local_tz = pytz.timezone(self.geo.timezone)

    # Buffer times and format for OpenMeteo
    conditioning_start = self.params.conditioning_start
    conditioning_start = local_tz.localize(conditioning_start)
    start_datetime = self.params.start_datetime
    start_datetime = local_tz.localize(start_datetime)
    buffered_start = start_datetime - timedelta(days=56)
    start_datetime_utc = buffered_start.astimezone(pytz.utc)

    end_datetime = self.params.end_datetime
    end_datetime = local_tz.localize(end_datetime)
    buffered_end = end_datetime + timedelta(days=1)
    end_datetime_utc = buffered_end.astimezone(pytz.utc)

    url = "https://archive-api.open-meteo.com/v1/archive"
    api_input = {
        "latitude": self.geo.center_lat,
        "longitude": self.geo.center_lon,
        "start_date": start_datetime_utc.date().strftime("%Y-%m-%d"),
        "end_date": end_datetime_utc.date().strftime("%Y-%m-%d"),
        "hourly": ["wind_speed_10m", "wind_direction_10m", "temperature_2m", "relative_humidity_2m", "cloud_cover", "shortwave_radiation", "diffuse_radiation", "direct_normal_irradiance", "rain"],
        "wind_speed_unit": "ms",
        "temperature_unit": "fahrenheit"
    }
    responses = openmeteo.weather_api(url, params=api_input)
    response = responses[0]

    self.ref_elev = response.Elevation()
    hourly = response.Hourly()
    hourly_wind_speed_10m = hourly.Variables(0).ValuesAsNumpy()
    hourly_wind_direction_10m = hourly.Variables(1).ValuesAsNumpy()
    hourly_temperature_2m = hourly.Variables(2).ValuesAsNumpy()
    hourly_rel_humidity_2m = hourly.Variables(3).ValuesAsNumpy()
    hourly_cloud_cover = hourly.Variables(4).ValuesAsNumpy()
    hourly_ghi = hourly.Variables(5).ValuesAsNumpy()
    hourly_dhi = hourly.Variables(6).ValuesAsNumpy()
    hourly_dni = hourly.Variables(7).ValuesAsNumpy()
    hourly_rain_mm = hourly.Variables(8).ValuesAsNumpy()

    hourly_data = {}

    hourly_data["date"] = pd.date_range(
        start=pd.to_datetime(hourly.Time(), unit="s"),
        end=pd.to_datetime(hourly.TimeEnd(), unit="s"),
        freq=pd.Timedelta(seconds=hourly.Interval()),
        inclusive="left"
    ).tz_localize('UTC').tz_convert(local_tz)

    self.times = pd.date_range(hourly_data["date"][0], hourly_data["date"][-1], freq='h', tz=local_tz)

    hourly_data["wind_speed"] = hourly_wind_speed_10m
    hourly_data["wind_direction"] = hourly_wind_direction_10m
    hourly_data["temperature"] = hourly_temperature_2m
    hourly_data["rel_humidity"] = hourly_rel_humidity_2m
    hourly_data["cloud_cover"] = hourly_cloud_cover
    hourly_data["ghi"] = hourly_ghi
    hourly_data["dhi"] = hourly_dhi
    hourly_data["dni"] = hourly_dni
    hourly_data["rain"] = hourly_rain_mm * 0.1 # convert to cm

    solpos = pvlib.solarposition.get_solarposition(self.times, self.geo.center_lat, self.geo.center_lon)

    hourly_data["solar_zenith"] = solpos["zenith"].values
    hourly_data["solar_azimuth"] = solpos["azimuth"].values

    if self.use_gsi:
        pre_summaries = self._build_pre_sim_summaries(
            hourly_data, buffered_start, start_datetime
        )
        self.gsi_tracker = GSITracker(self.geo, pre_summaries)
        gsi = self.gsi_tracker.compute_gsi()
        self.live_h_mf, self.live_w_mf = self.set_live_moistures(gsi)
    else:
        self.live_h_mf = None
        self.live_w_mf = None
        self.gsi_tracker = None

    hourly = filter_hourly_data(hourly_data, conditioning_start, end_datetime)
    self.stream_times = pd.DatetimeIndex(hourly["date"])
    self.stream = list(self.generate_stream(hourly))

    try:
        self.sim_start_idx = self.stream_times.get_loc(start_datetime)
    except KeyError:
        self.sim_start_idx = int(self.stream_times.searchsorted(start_datetime, side="left"))

    if self.use_gsi:
        # Build GSI tracker seeded with pre-simulation daily summaries
        pre_summaries = self._build_pre_sim_summaries(
            hourly_data, buffered_start, start_datetime
        )
        init_rain = self.stream[self.sim_start_idx].rain
        self.gsi_tracker = GSITracker(self.geo, pre_summaries, init_rain)
    else:
        self.gsi_tracker = None
    if self.gsi_tracker is not None:
        self.gsi_tracker._last_cum_rain = self.stream[self.sim_start_idx].rain

    # Calculate foliar moisture content
    self.fmc = self.calc_fmc()

    # Set units and time step based on OpenMeteo params
    self.time_step = 60
    self.input_wind_ht = 10
    self.input_wind_ht_units = "m"
    self.input_wind_vel_units = "mps"
    self.input_temp_units = "F"

get_stream_from_wxs()

Parse a RAWS-format .wxs weather file and build the stream.

Read weather observations from the file, apply unit conversions (English or Metric to internal units), fetch solar irradiance from Open-Meteo to supplement the file data, compute GSI, and assemble the final weather stream.

Side Effects

Sets self.stream, self.stream_times, self.sim_start_idx, self.fmc, self.live_h_mf, self.live_w_mf, self.ref_elev, self.time_step, and input unit attributes.

Raises:

Type	Description
ValueError	If the file has insufficient data or unknown units.

Source code in embrs/models/weather.py

def get_stream_from_wxs(self):
    """Parse a RAWS-format ``.wxs`` weather file and build the stream.

    Read weather observations from the file, apply unit conversions
    (English or Metric to internal units), fetch solar irradiance from
    Open-Meteo to supplement the file data, compute GSI, and assemble
    the final weather stream.

    Side Effects:
        Sets ``self.stream``, ``self.stream_times``, ``self.sim_start_idx``,
        ``self.fmc``, ``self.live_h_mf``, ``self.live_w_mf``,
        ``self.ref_elev``, ``self.time_step``, and input unit attributes.

    Raises:
        ValueError: If the file has insufficient data or unknown units.
    """
    file = self.params.file

    weather_data = {
        "datetime": [],
        "temperature": [],
        "rel_humidity": [],
        "rain": [],
        "wind_speed": [],
        "wind_direction": [],
        "cloud_cover": [],
    }

    local_tz = pytz.timezone(self.geo.timezone)
    units = "english"

    # ── Step 1: Parse WXS line-by-line ───────────────────────────
    with open(file, "r") as f:
        header_found = False
        for line in f:
            line = line.strip()
            if not line:
                continue
            if line.startswith("RAWS_UNITS:"):
                units = line.split(":")[1].strip().lower()
                continue
            elif line.startswith("RAWS_ELEVATION:"):
                self.ref_elev = float(line.split(":")[1].strip().lower())
                continue
            elif line.startswith("RAWS:"):
                continue
            elif line.startswith("Year") and not header_found:
                header_found = True
                continue
            if not header_found:
                continue

            parts = line.split()
            if len(parts) != 10:
                continue
            try:
                year, month, day = int(parts[0]), int(parts[1]), int(parts[2])
                hour = int(parts[3].zfill(4)[:2])
                dt = local_tz.localize(datetime(year, month, day, hour))
                weather_data["datetime"].append(dt)
                weather_data["temperature"].append(float(parts[4]))
                weather_data["rel_humidity"].append(float(parts[5]))
                weather_data["rain"].append(float(parts[6]))
                weather_data["wind_speed"].append(float(parts[7]))
                weather_data["wind_direction"].append(float(parts[8]))
                weather_data["cloud_cover"].append(float(parts[9]))
            except Exception as e:
                print(f"Skipping malformed line: {line} ({e})")
                continue

    df = pd.DataFrame(weather_data).set_index("datetime")

    if len(df.index) < 2:
        raise ValueError("WXS file does not contain enough data to determine time step.")

    # ── Step 2: Infer time step and apply unit conversions ───────
    time_step_min = int((df.index[1] - df.index[0]).total_seconds() / 60)
    if units == "english":
        df["rain"] *= 2.54
        df["wind_speed"] *= 0.44704
        self.ref_elev = ft_to_m(self.ref_elev)
    elif units == "metric":
        df["temperature"] = df["temperature"] * 9 / 5 + 32
        df["rain"] /= 10
    else:
        raise ValueError(f"Unknown units: {units}")

    # ── Step 3: Fetch irradiance from Open-Meteo ─────────────────
    # Use buffer around sim dates
    conditioning_start = local_tz.localize(self.params.conditioning_start)
    start_datetime = local_tz.localize(self.params.start_datetime)
    end_datetime = local_tz.localize(self.params.end_datetime)

    # Check bounds
    wxs_start = df.index.min()
    wxs_end = df.index.max()

    if start_datetime < wxs_start:
        raise ValueError(f"Start datetime {start_datetime} is before WXS data begins at {wxs_start}.")

    if end_datetime > wxs_end:
        raise ValueError(f"End datetime {end_datetime} is after WXS data ends at {wxs_end}.")

    buffered_start = start_datetime - timedelta(days=56)
    buffered_end = end_datetime + timedelta(days=1)

    cache_session = requests_cache.CachedSession('.cache', expire_after=-1)
    retry_session = retry(cache_session, retries=5, backoff_factor=0.2)
    openmeteo = openmeteo_requests.Client(session=retry_session)

    url = "https://archive-api.open-meteo.com/v1/archive"
    api_input = {
        "latitude": self.geo.center_lat,
        "longitude": self.geo.center_lon,
        "start_date": buffered_start.astimezone(pytz.utc).date().strftime("%Y-%m-%d"),
        "end_date": buffered_end.astimezone(pytz.utc).date().strftime("%Y-%m-%d"),
        "hourly": ["shortwave_radiation", "diffuse_radiation", "direct_normal_irradiance"],
        "timezone": "auto"
    }

    responses = openmeteo.weather_api(url, params=api_input)
    response = responses[0]

    hourly = response.Hourly()
    hourly_data = {
        "ghi": hourly.Variables(0).ValuesAsNumpy(),
        "dhi": hourly.Variables(1).ValuesAsNumpy(),
        "dni": hourly.Variables(2).ValuesAsNumpy(),
        "date": pd.date_range(
            start=pd.to_datetime(hourly.Time(), unit="s"),
            end=pd.to_datetime(hourly.TimeEnd(), unit="s"),
            freq=pd.Timedelta(seconds=hourly.Interval()),
            inclusive="left"
        ).tz_localize(local_tz)
    }

    irradiance_df = pd.DataFrame(hourly_data).set_index("date")

    # ── Step 4: Resample Open-Meteo irradiance to WXS resolution ─
    target_freq = f"{time_step_min}T"
    if time_step_min < 60:
        irradiance_df = irradiance_df.resample(target_freq).interpolate(method="linear")
    elif time_step_min > 60:
        irradiance_df = irradiance_df.resample(target_freq).mean()

    # ── Step 5: Align WXS and Open-Meteo data on datetime index ──
    df = df.loc[(df.index >= buffered_start) & (df.index <= end_datetime)]
    irradiance_df = irradiance_df.loc[df.index]

    df["ghi"] = irradiance_df["ghi"].values
    df["dhi"] = irradiance_df["dhi"].values
    df["dni"] = irradiance_df["dni"].values

    # ── Step 6: Add solar geometry ───────────────────────────────
    solpos = pvlib.solarposition.get_solarposition(df.index, self.geo.center_lat, self.geo.center_lon)
    df["solar_zenith"] = solpos["zenith"].values
    df["solar_azimuth"] = solpos["azimuth"].values

    if self.use_gsi:
        # Determine how far back we can go
        min_date_in_wxs = df.index.min()
        desired_start = start_datetime - timedelta(days=56)
        data_start = max(min_date_in_wxs, desired_start)

        wxs_hourly = {
            "date": df.index,
            "temperature": df["temperature"].values,
            "rel_humidity": df["rel_humidity"].values,
            "rain": df["rain"].values,
        }
        pre_summaries = self._build_pre_sim_summaries(
            wxs_hourly, data_start, start_datetime
        )
        self.gsi_tracker = GSITracker(self.geo, pre_summaries)
        gsi = self.gsi_tracker.compute_gsi()
        self.live_h_mf, self.live_w_mf = self.set_live_moistures(gsi)
    else:
        self.live_h_mf = None
        self.live_w_mf = None
        self.gsi_tracker = None

    # Calculate foliar moisture content
    self.fmc = self.calc_fmc()

    # ── Step 8: Package final stream ─────────────────────────────
    df["date"] = df.index
    hourly_data = filter_hourly_data(df, conditioning_start, end_datetime)
    self.stream_times = pd.DatetimeIndex(hourly_data["date"])

    try:
        self.sim_start_idx = self.stream_times.get_loc(start_datetime)
    except KeyError:
        self.sim_start_idx = int(self.stream_times.searchsorted(start_datetime, side="left"))

    self.stream = list(self.generate_stream(hourly_data))

    if self.use_gsi:
        # Build GSI tracker seeded with pre-simulation daily summaries
        wxs_hourly = {
            "date": df.index,
            "temperature": df["temperature"].values,
            "rel_humidity": df["rel_humidity"].values,
            "rain": df["rain"].values,
        }
        pre_summaries = self._build_pre_sim_summaries(
            wxs_hourly, data_start, start_datetime
        )
        init_rain = self.stream[self.sim_start_idx].rain
        self.gsi_tracker = GSITracker(self.geo, pre_summaries, init_rain)
    else:
        self.gsi_tracker = None
    if self.gsi_tracker is not None:
        self.gsi_tracker._last_cum_rain = self.stream[self.sim_start_idx].rain

    # ── Step 9: Set metadata attributes ──────────────────────────
    self.time_step = time_step_min
    self.input_wind_ht = 6.1
    self.input_wind_ht_units = "m"
    self.input_wind_vel_units = "mps"
    self.input_temp_units = "F"

set_live_moistures(gsi)

Compute live fuel moisture fractions from the Growing Season Index.

Map GSI to live herbaceous and woody moisture using linear interpolation between dormant and green-up values. Below the green-up threshold (GSI < 0.2), dormant values are returned.

Parameters:

Name	Type	Description	Default
gsi	float	Growing Season Index in [0, 1].	required

Returns:

Type	Description
tuple	Tuple[float, float]: (live_h_mf, live_w_mf) live herbaceous and live woody moisture content (fractions).

Source code in embrs/models/weather.py

def set_live_moistures(self, gsi: float) -> tuple:
    """Compute live fuel moisture fractions from the Growing Season Index.

    Map GSI to live herbaceous and woody moisture using linear
    interpolation between dormant and green-up values. Below the
    green-up threshold (GSI < 0.2), dormant values are returned.

    Args:
        gsi (float): Growing Season Index in [0, 1].

    Returns:
        Tuple[float, float]: ``(live_h_mf, live_w_mf)`` live herbaceous
            and live woody moisture content (fractions).
    """
    # Dormant values
    h_min = 0.3
    w_min = 0.6

    # Green-up threshold
    gu = 0.2

    # Max values
    h_max = 2.5
    w_max = 2.0

    if gsi < gu:
        return h_min, w_min

    else:
        m_h = (h_max - h_min) / (1.0 - gu)
        m_w = (w_max - w_min) / (1.0 - gu)

    live_h_mf = m_h * gsi + (h_max - m_h)
    live_w_mf = m_w * gsi + (w_max - m_w)

    return live_h_mf, live_w_mf

WindNinjaTask dataclass

Parameters for a WindNinja processing task.

Attributes:

Name	Type	Description
index	int	Task index in the processing queue.
time_step	float	Time step for this wind field.
entry	WeatherEntry	Weather data for this time step.
elevation_path	str	Path to elevation raster file.
timezone	str	IANA timezone string.
north_angle	float	Rotation to true north in degrees.
mesh_resolution	float	WindNinja mesh resolution in meters.
temp_file_path	str	Path for temporary output files.
cli_path	str	Path to WindNinja CLI executable.
start_datetime	timedelta	Time offset from simulation start.
wind_height	float	Wind measurement height.
wind_height_units	str	Units for wind height.
input_speed_units	str	Units for input wind speed.
temperature_units	str	Units for temperature.

Source code in embrs/utilities/data_classes.py

@dataclass
class WindNinjaTask:
    """Parameters for a WindNinja processing task.

    Attributes:
        index (int): Task index in the processing queue.
        time_step (float): Time step for this wind field.
        entry (WeatherEntry): Weather data for this time step.
        elevation_path (str): Path to elevation raster file.
        timezone (str): IANA timezone string.
        north_angle (float): Rotation to true north in degrees.
        mesh_resolution (float): WindNinja mesh resolution in meters.
        temp_file_path (str): Path for temporary output files.
        cli_path (str): Path to WindNinja CLI executable.
        start_datetime (timedelta): Time offset from simulation start.
        wind_height (float): Wind measurement height.
        wind_height_units (str): Units for wind height.
        input_speed_units (str): Units for input wind speed.
        temperature_units (str): Units for temperature.
    """

    index: int
    time_step: float
    entry: WeatherEntry
    elevation_path: str
    timezone: str
    north_angle: float
    mesh_resolution: float
    temp_file_path: str
    cli_path: str
    start_datetime: timedelta
    wind_height: float
    wind_height_units: str
    input_speed_units: str
    temperature_units: str

BTU_ft2_min_to_kW_m2(f_btu_ft2_min)

Convert heat flux from BTU/(ft^2*min) to kW/m^2.

Parameters:

Name	Type	Description	Default
f_btu_ft2_min	float	Heat flux in BTU/(ft^2*min).	required

Returns:

Name	Type	Description
float	float	Heat flux in kW/m^2.

Source code in embrs/utilities/unit_conversions.py

def BTU_ft2_min_to_kW_m2(f_btu_ft2_min: float) -> float:
    """Convert heat flux from BTU/(ft^2*min) to kW/m^2.

    Args:
        f_btu_ft2_min (float): Heat flux in BTU/(ft^2*min).

    Returns:
        float: Heat flux in kW/m^2.
    """
    return f_btu_ft2_min * _BTU_FT2_MIN_TO_KW_M2

BTU_ft_min_to_kW_m(f_btu_ft_min)

Convert fireline intensity from BTU/(ft*min) to kW/m.

Parameters:

Name	Type	Description	Default
f_btu_ft_min	float	Fireline intensity in BTU/(ft*min).	required

Returns:

Name	Type	Description
float	float	Fireline intensity in kW/m.

Source code in embrs/utilities/unit_conversions.py

def BTU_ft_min_to_kW_m(f_btu_ft_min: float) -> float:
    """Convert fireline intensity from BTU/(ft*min) to kW/m.

    Args:
        f_btu_ft_min (float): Fireline intensity in BTU/(ft*min).

    Returns:
        float: Fireline intensity in kW/m.
    """
    return f_btu_ft_min * _BTU_FT_MIN_TO_KW_M

BTU_ft_min_to_kcal_s_m(f_btu_ft_min)

Convert fireline intensity from BTU/(ftmin) to kcal/(sm).

Parameters:

Name	Type	Description	Default
f_btu_ft_min	float	Fireline intensity in BTU/(ft*min).	required

Returns:

Name	Type	Description
float	float	Fireline intensity in kcal/(s*m).

Source code in embrs/utilities/unit_conversions.py

def BTU_ft_min_to_kcal_s_m(f_btu_ft_min: float) -> float:
    """Convert fireline intensity from BTU/(ft*min) to kcal/(s*m).

    Args:
        f_btu_ft_min (float): Fireline intensity in BTU/(ft*min).

    Returns:
        float: Fireline intensity in kcal/(s*m).
    """
    return f_btu_ft_min * _BTU_FT_MIN_TO_KCAL_S_M

BTU_lb_to_cal_g(f_btu_lb)

Convert heat content from BTU/lb to cal/g.

Parameters:

Name	Type	Description	Default
f_btu_lb	float	Heat content in BTU/lb.	required

Returns:

Name	Type	Description
float	float	Heat content in cal/g.

Source code in embrs/utilities/unit_conversions.py

def BTU_lb_to_cal_g(f_btu_lb: float) -> float:
    """Convert heat content from BTU/lb to cal/g.

    Args:
        f_btu_lb (float): Heat content in BTU/lb.

    Returns:
        float: Heat content in cal/g.
    """
    return f_btu_lb * _BTU_LB_TO_CAL_G

F_to_C(f_f)

Convert temperature from Fahrenheit to Celsius.

Parameters:

Name	Type	Description	Default
f_f	float	Temperature in degrees Fahrenheit.	required

Returns:

Name	Type	Description
float	float	Temperature in degrees Celsius.

Source code in embrs/utilities/unit_conversions.py

def F_to_C(f_f: float) -> float:
    """Convert temperature from Fahrenheit to Celsius.

    Args:
        f_f (float): Temperature in degrees Fahrenheit.

    Returns:
        float: Temperature in degrees Celsius.
    """
    return (5.0 / 9.0) * (f_f - 32.0)

KiSq_to_Lbsft2(f_kisq)

Convert fuel loading from kg/m^2 to lb/ft^2.

Parameters:

Name	Type	Description	Default
f_kisq	float	Fuel loading in kg/m^2.	required

Returns:

Name	Type	Description
float	float	Fuel loading in lb/ft^2.

Source code in embrs/utilities/unit_conversions.py

def KiSq_to_Lbsft2(f_kisq: float) -> float:
    """Convert fuel loading from kg/m^2 to lb/ft^2.

    Args:
        f_kisq (float): Fuel loading in kg/m^2.

    Returns:
        float: Fuel loading in lb/ft^2.
    """
    return f_kisq * _KISQ_TO_LBSFT2

KiSq_to_TPA(f_kisq)

Convert fuel loading from kg/m^2 to tons per acre.

Parameters:

Name	Type	Description	Default
f_kisq	float	Fuel loading in kg/m^2.	required

Returns:

Name	Type	Description
float	float	Fuel loading in tons per acre.

Source code in embrs/utilities/unit_conversions.py

def KiSq_to_TPA(f_kisq: float) -> float:
    """Convert fuel loading from kg/m^2 to tons per acre.

    Args:
        f_kisq (float): Fuel loading in kg/m^2.

    Returns:
        float: Fuel loading in tons per acre.
    """
    return f_kisq * _KISQ_TO_TPA

Lbsft2_to_KiSq(f_libsft2)

Convert fuel loading from lb/ft^2 to kg/m^2.

Parameters:

Name	Type	Description	Default
f_libsft2	float	Fuel loading in lb/ft^2.	required

Returns:

Name	Type	Description
float	float	Fuel loading in kg/m^2.

Source code in embrs/utilities/unit_conversions.py

def Lbsft2_to_KiSq(f_libsft2: float) -> float:
    """Convert fuel loading from lb/ft^2 to kg/m^2.

    Args:
        f_libsft2 (float): Fuel loading in lb/ft^2.

    Returns:
        float: Fuel loading in kg/m^2.
    """
    return f_libsft2 * _LBSFT2_TO_KISQ

Lbsft2_to_TPA(f_lbsft2)

Convert fuel loading from lb/ft^2 to tons per acre.

Parameters:

Name	Type	Description	Default
f_lbsft2	float	Fuel loading in lb/ft^2.	required

Returns:

Name	Type	Description
float	float	Fuel loading in tons per acre.

Source code in embrs/utilities/unit_conversions.py

def Lbsft2_to_TPA(f_lbsft2: float) -> float:
    """Convert fuel loading from lb/ft^2 to tons per acre.

    Args:
        f_lbsft2 (float): Fuel loading in lb/ft^2.

    Returns:
        float: Fuel loading in tons per acre.
    """
    return f_lbsft2 * _LBSFT2_TO_TPA

TPA_to_KiSq(f_tpa)

Convert fuel loading from tons per acre to kg/m^2.

Parameters:

Name	Type	Description	Default
f_tpa	float	Fuel loading in tons per acre.	required

Returns:

Name	Type	Description
float	float	Fuel loading in kg/m^2.

Source code in embrs/utilities/unit_conversions.py

def TPA_to_KiSq(f_tpa: float) -> float:
    """Convert fuel loading from tons per acre to kg/m^2.

    Args:
        f_tpa (float): Fuel loading in tons per acre.

    Returns:
        float: Fuel loading in kg/m^2.
    """
    return f_tpa / _TPA_TO_KISQ_DIVISOR

TPA_to_Lbsft2(f_tpa)

Convert fuel loading from tons per acre to lb/ft^2.

Parameters:

Name	Type	Description	Default
f_tpa	float	Fuel loading in tons per acre.	required

Returns:

Name	Type	Description
float	float	Fuel loading in lb/ft^2.

Source code in embrs/utilities/unit_conversions.py

def TPA_to_Lbsft2(f_tpa: float) -> float:
    """Convert fuel loading from tons per acre to lb/ft^2.

    Args:
        f_tpa (float): Fuel loading in tons per acre.

    Returns:
        float: Fuel loading in lb/ft^2.
    """
    return f_tpa * _TPA_TO_LBSFT2

apply_site_specific_correction(cell, elev_ref, curr_weather)

Apply elevation lapse-rate correction for temperature and humidity.

Adjust the reference-station temperature and relative humidity to the cell's elevation using standard lapse rates (Stephenson 1988).

Parameters:

Name	Type	Description	Default
cell	Cell	Cell with elevation_m attribute (meters).	required
elev_ref	float	Reference weather station elevation (meters).	required
curr_weather	WeatherEntry	Current weather entry with temp (Fahrenheit) and rel_humidity (percent).	required

Returns:

Type	Description
Tuple[float, float]	Tuple[float, float]: (temp_c, rh) where temp_c is the corrected temperature (Celsius) and rh is the corrected relative humidity (fraction, capped at 0.99).

Source code in embrs/models/weather.py

def apply_site_specific_correction(cell: Cell, elev_ref: float,
                                   curr_weather: WeatherEntry) -> Tuple[float, float]:
    """Apply elevation lapse-rate correction for temperature and humidity.

    Adjust the reference-station temperature and relative humidity to the
    cell's elevation using standard lapse rates (Stephenson 1988).

    Args:
        cell: Cell with ``elevation_m`` attribute (meters).
        elev_ref (float): Reference weather station elevation (meters).
        curr_weather (WeatherEntry): Current weather entry with ``temp``
            (Fahrenheit) and ``rel_humidity`` (percent).

    Returns:
        Tuple[float, float]: ``(temp_c, rh)`` where ``temp_c`` is the
            corrected temperature (Celsius) and ``rh`` is the corrected
            relative humidity (fraction, capped at 0.99).
    """
    elev_diff = elev_ref - cell.elevation_m 
    elev_diff *= 3.2808 # convert to ft

    dewptref = -398.0-7469.0 / (np.log(curr_weather.rel_humidity/100.0)-7469.0/(curr_weather.temp+398.0))

    temp = curr_weather.temp + elev_diff/1000.0*5.5 # Stephenson 1988 found summer adiabat at 3.07 F/1000ft
    dewpt = dewptref + elev_diff/1000.0*1.1 # new humidity, new dewpt, and new humidity
    rh = 7469.0*(1.0/(temp+398.0)-1.0/(dewpt+398.0))
    rh = np.exp(rh)*100.0 #convert from ln units
    if (rh > 99.0):
        rh=99.0

    # Convert temp to celsius
    temp = F_to_C(temp)

    # Return humidity as a decimal
    rh /= 100.0

    return temp, rh

cal_g_to_BTU_lb(f_cal_g)

Convert heat content from cal/g to BTU/lb.

Parameters:

Name	Type	Description	Default
f_cal_g	float	Heat content in cal/g.	required

Returns:

Name	Type	Description
float	float	Heat content in BTU/lb.

Source code in embrs/utilities/unit_conversions.py

def cal_g_to_BTU_lb(f_cal_g: float) -> float:
    """Convert heat content from cal/g to BTU/lb.

    Args:
        f_cal_g (float): Heat content in cal/g.

    Returns:
        float: Heat content in BTU/lb.
    """
    return f_cal_g * _CAL_G_TO_BTU_LB

calc_local_solar_radiation(cell, curr_weather)

Compute slope- and canopy-adjusted solar irradiance at a cell.

Use pvlib to compute plane-of-array irradiance for the cell's slope and aspect, then reduce by canopy transmittance.

Parameters:

Name	Type	Description	Default
cell	Cell	Cell with slope_deg, aspect, canopy_cover (percent).	required
curr_weather	WeatherEntry	Entry with solar_zenith, solar_azimuth, dni, ghi, dhi.	required

Returns:

Name	Type	Description
float	float	Total irradiance at the cell surface (W/m²).

Source code in embrs/models/weather.py

def calc_local_solar_radiation(cell: Cell, curr_weather: WeatherEntry) -> float:
    """Compute slope- and canopy-adjusted solar irradiance at a cell.

    Use pvlib to compute plane-of-array irradiance for the cell's slope
    and aspect, then reduce by canopy transmittance.

    Args:
        cell: Cell with ``slope_deg``, ``aspect``, ``canopy_cover`` (percent).
        curr_weather (WeatherEntry): Entry with ``solar_zenith``,
            ``solar_azimuth``, ``dni``, ``ghi``, ``dhi``.

    Returns:
        float: Total irradiance at the cell surface (W/m²).
    """
    # Calculate total irradiance using pvlib
    total_irradiance = pvlib.irradiance.get_total_irradiance(
        surface_tilt=cell.slope_deg,
        surface_azimuth=cell.aspect,
        solar_zenith=curr_weather.solar_zenith,
        solar_azimuth=curr_weather.solar_azimuth,
        dni=curr_weather.dni,
        ghi=curr_weather.ghi,
        dhi=curr_weather.dhi,
        model='isotropic'
    )

    # Adjust for canopy transmittance (Only thing not modelled in pvlib)
    canopy_transmittance = 1 - (cell.canopy_cover/100)
    I = total_irradiance['poa_global'] * canopy_transmittance

    return I

datetime_to_julian_date(dt)

Convert a datetime to Julian date.

Parameters:

Name	Type	Description	Default
dt	datetime	A datetime-like object with year, month, day, hour, minute, and second attributes.	required

Returns:

Name	Type	Description
float	float	Julian date (fractional day number).

Source code in embrs/models/weather.py

def datetime_to_julian_date(dt: datetime) -> float:
    """Convert a datetime to Julian date.

    Args:
        dt: A datetime-like object with year, month, day, hour, minute,
            and second attributes.

    Returns:
        float: Julian date (fractional day number).
    """
    year = dt.year
    month = dt.month
    day = dt.day
    hour = dt.hour + dt.minute / 60 + dt.second / 3600

    if month <= 2:
        year -= 1
        month += 12

    A = np.floor(year / 100)
    B = 2 - A + np.floor(A / 4)

    jd_day = np.floor(365.25 * (year + 4716)) + \
            np.floor(30.6001 * (month + 1)) + \
            day + B - 1524.5

    jd = jd_day + hour / 24
    return jd

filter_hourly_data(hourly_data, start_datetime, end_datetime)

Filter hourly weather data to a datetime range (inclusive).

Parameters:

Name	Type	Description	Default
hourly_data	dict	Dictionary with a 'date' key and parallel value arrays.	required
start_datetime	datetime	Start of desired range (datetime-like).	required
end_datetime	datetime	End of desired range (datetime-like).	required

Returns:

Name	Type	Description
dict	dict	Filtered copy with the same keys, values trimmed to the matching datetime window.

Source code in embrs/models/weather.py

def filter_hourly_data(hourly_data: dict, start_datetime: datetime, end_datetime: datetime) -> dict:
    """Filter hourly weather data to a datetime range (inclusive).

    Args:
        hourly_data (dict): Dictionary with a 'date' key and parallel
            value arrays.
        start_datetime: Start of desired range (datetime-like).
        end_datetime: End of desired range (datetime-like).

    Returns:
        dict: Filtered copy with the same keys, values trimmed to the
            matching datetime window.
    """
    hourly_data["date"] = pd.to_datetime(hourly_data["date"])

    mask = (hourly_data["date"] >= start_datetime) & (hourly_data["date"] <= end_datetime)
    filtered_data = {key: np.array(value)[mask] for key, value in hourly_data.items()}
    return filtered_data

ft_min_to_m_s(f_ft_min)

Convert speed from feet per minute to meters per second.

Parameters:

Name	Type	Description	Default
f_ft_min	float	Speed in ft/min.	required

Returns:

Name	Type	Description
float	float	Speed in m/s.

Source code in embrs/utilities/unit_conversions.py

def ft_min_to_m_s(f_ft_min: float) -> float:
    """Convert speed from feet per minute to meters per second.

    Args:
        f_ft_min (float): Speed in ft/min.

    Returns:
        float: Speed in m/s.
    """
    return f_ft_min * _FT_MIN_TO_M_S

ft_min_to_mph(f_ft_min)

Convert speed from feet per minute to miles per hour.

Parameters:

Name	Type	Description	Default
f_ft_min	float	Speed in ft/min.	required

Returns:

Name	Type	Description
float	float	Speed in mph.

Source code in embrs/utilities/unit_conversions.py

def ft_min_to_mph(f_ft_min: float) -> float:
    """Convert speed from feet per minute to miles per hour.

    Args:
        f_ft_min (float): Speed in ft/min.

    Returns:
        float: Speed in mph.
    """
    return f_ft_min * _FT_MIN_TO_MPH

ft_to_m(f_ft)

Convert length from feet to meters.

Parameters:

Name	Type	Description	Default
f_ft	float	Length in feet.	required

Returns:

Name	Type	Description
float	float	Length in meters.

Source code in embrs/utilities/unit_conversions.py

def ft_to_m(f_ft: float) -> float:
    """Convert length from feet to meters.

    Args:
        f_ft (float): Length in feet.

    Returns:
        float: Length in meters.
    """
    return f_ft * _FT_TO_M

m_s_to_ft_min(m_s)

Convert speed from meters per second to feet per minute.

Parameters:

Name	Type	Description	Default
m_s	float	Speed in m/s.	required

Returns:

Name	Type	Description
float	float	Speed in ft/min.

Source code in embrs/utilities/unit_conversions.py

def m_s_to_ft_min(m_s: float) -> float:
    """Convert speed from meters per second to feet per minute.

    Args:
        m_s (float): Speed in m/s.

    Returns:
        float: Speed in ft/min.
    """
    return m_s * _M_S_TO_FT_MIN

m_to_ft(f_m)

Convert length from meters to feet.

Parameters:

Name	Type	Description	Default
f_m	float	Length in meters.	required

Returns:

Name	Type	Description
float	float	Length in feet.

Source code in embrs/utilities/unit_conversions.py

def m_to_ft(f_m: float) -> float:
    """Convert length from meters to feet.

    Args:
        f_m (float): Length in meters.

    Returns:
        float: Length in feet.
    """
    return f_m * _M_TO_FT

mph_to_ft_min(f_mph)

Convert speed from miles per hour to feet per minute.

Parameters:

Name	Type	Description	Default
f_mph	float	Speed in mph.	required

Returns:

Name	Type	Description
float	float	Speed in ft/min.

Source code in embrs/utilities/unit_conversions.py

def mph_to_ft_min(f_mph: float) -> float:
    """Convert speed from miles per hour to feet per minute.

    Args:
        f_mph (float): Speed in mph.

    Returns:
        float: Speed in ft/min.
    """
    return f_mph * _MPH_TO_FT_MIN