weather

Weather file creation, reading, and conversion for EMOD simulations.

Main features:

• Create EMOD weather files from CSV, DataFrame, or dictionary data.
• Read existing EMOD weather files into Python objects.
• Convert between EMOD binary weather format and tabular formats.
• Configure EMOD climate model settings.

DataFrameInfo

Column name configuration for weather DataFrames.

Source code in emodpy_malaria/weather/weather_data.py

class DataFrameInfo:
    """Column name configuration for weather DataFrames."""

    _variable_values = [str(v.value).lower() for v in WeatherVariable.list()]
    _default_column_candidates = {
        "node": ["nodes", "node", "node_id", "node_ids", "nodeid", "id", "ids"],
        "step": ["steps", "step", "time"],
        "value": ["values", "value", "series", "data"] + _variable_values,
    }

    def __init__(self,
                 node_column: str = None,
                 step_column: str = None,
                 value_column: str = None,
                 only_unique_series: bool = False):
        self._node_column = node_column
        self._step_column = step_column
        self._value_column = value_column
        self.only_unique_series = only_unique_series
        self._set_defaults()

    def __eq__(self, other):
        if not isinstance(other, DataFrameInfo):
            return NotImplemented
        return (self._node_column == other.node_column
                and self._step_column == other.step_column
                and self._value_column == other.value_column
                and self.only_unique_series == other.only_unique_series)

    @property
    def node_column(self):
        return self._node_column

    @property
    def step_column(self):
        return self._step_column

    @property
    def value_column(self):
        return self._value_column

    def _set_defaults(self) -> None:
        self._node_column = self._node_column or self._default_column_candidates["node"][0]
        self._step_column = self._step_column or self._default_column_candidates["step"][0]
        self._value_column = self._value_column or self._default_column_candidates["value"][0]

    @classmethod
    def detect_columns(cls, df: pd.DataFrame,
                       column_candidates: dict[str, list[str]] = None) -> "DataFrameInfo":
        """Auto-detect node, step, and value column names from a DataFrame."""
        column_candidates = column_candidates or cls._default_column_candidates
        column_types = ["node", "step", "value"]
        columns = [cls._detect_column(df, column_candidates[name]) for name in column_types]
        not_found = [name for name, col in zip(column_types, columns) if col is None]
        if not_found:
            raise NameError(f"Unable to detect columns: {not_found}")
        return DataFrameInfo(*columns)

    @staticmethod
    def _detect_column(df: pd.DataFrame, column_candidates: list[str]) -> str | None:
        cols = [c for c in df.columns if str(c).strip().lower() in column_candidates]
        return cols[0] if cols else None

detect_columns(df, column_candidates=None) classmethod

Auto-detect node, step, and value column names from a DataFrame.

Source code in emodpy_malaria/weather/weather_data.py

@classmethod
def detect_columns(cls, df: pd.DataFrame,
                   column_candidates: dict[str, list[str]] = None) -> "DataFrameInfo":
    """Auto-detect node, step, and value column names from a DataFrame."""
    column_candidates = column_candidates or cls._default_column_candidates
    column_types = ["node", "step", "value"]
    columns = [cls._detect_column(df, column_candidates[name]) for name in column_types]
    not_found = [name for name, col in zip(column_types, columns) if col is None]
    if not_found:
        raise NameError(f"Unable to detect columns: {not_found}")
    return DataFrameInfo(*columns)

WeatherAttributes

Metadata attributes for EMOD weather files.

Manages the key/value pairs stored in the "Metadata" section of a .bin.json file. Provides sensible defaults when no explicit values are given.

Source code in emodpy_malaria/weather/weather_metadata.py

class WeatherAttributes:
    """Metadata attributes for EMOD weather files.

    Manages the key/value pairs stored in the ``"Metadata"`` section of a
    ``.bin.json`` file.  Provides sensible defaults when no explicit values
    are given.
    """

    def __init__(self,
                 attributes_dict: dict[str, Union[str, int, float]] = None,
                 reference: str = None,
                 resolution: str = None,
                 provenance: str = None,
                 update_freq: str = None,
                 start_year: int = None,
                 end_year: int = None,
                 start_doy: int = None,
                 lat_min: float = None,
                 lat_max: float = None,
                 lon_min: float = None,
                 lon_max: float = None,
                 tool: str = None,
                 author: str = None,
                 schema_version: str = None,
                 notes: str = None):
        self._attributes_dict: dict[str, Union[str, int, float]] = (
            attributes_dict or self.metadata_defaults_dict()
        )

        date_years = f"{start_year}-{end_year}" if start_year is not None and end_year is not None else None

        metadata_args = {
            _META_ID_REFERENCE: reference,
            _META_SPATIAL_RESOLUTION: resolution,
            _META_PROVENANCE: provenance,
            _META_UPDATE_FREQUENCY: update_freq,
            _META_DATA_YEARS: date_years,
            _META_START_DOY: start_doy,
            _META_LAT_MIN: lat_min,
            _META_LAT_MAX: lat_max,
            _META_LON_MIN: lon_min,
            _META_LON_MAX: lon_max,
            _META_TOOL: tool,
            _META_AUTHOR: author,
            _META_WEATHER_SCHEMA_V: schema_version,
            _META_NOTES: notes,
        }

        metadata_args = {k: v for k, v in metadata_args.items() if v is not None}
        self._attributes_dict.update(metadata_args)

        missing = self.required_metadata_defaults_dict(exclude_keys=list(self._attributes_dict.keys()))
        self._attributes_dict.update(missing)

    def __eq__(self, other):
        if not isinstance(other, WeatherAttributes):
            return NotImplemented
        return self.attributes_dict == other.attributes_dict

    @property
    def attributes_dict(self) -> dict[str, Union[str, int, float]]:
        return self._attributes_dict

    @property
    def tool(self) -> str | None:
        return self._attributes_dict.get(_META_TOOL)

    @tool.setter
    def tool(self, value: str) -> None:
        validate_str_value(value)
        self._attributes_dict[_META_TOOL] = value

    @property
    def date_created(self) -> str | None:
        return self._attributes_dict.get(_META_DATE_CREATED)

    @date_created.setter
    def date_created(self, value: str) -> None:
        validate_str_value(str(value))
        self._attributes_dict[_META_DATE_CREATED] = str(value)

    @property
    def author(self) -> str | None:
        return self._attributes_dict.get(_META_AUTHOR)

    @author.setter
    def author(self, value: str) -> None:
        validate_str_value(value)
        self._attributes_dict[_META_AUTHOR] = value

    @property
    def id_reference(self) -> str | None:
        return self._attributes_dict.get(_META_ID_REFERENCE)

    @id_reference.setter
    def id_reference(self, value: str) -> None:
        validate_str_value(value)
        self._attributes_dict[_META_ID_REFERENCE] = value

    @property
    def update_resolution(self) -> str | None:
        return self._attributes_dict.get(_META_UPDATE_FREQUENCY)

    @update_resolution.setter
    def update_resolution(self, value: str) -> None:
        validate_str_value(value)
        self._attributes_dict[_META_UPDATE_FREQUENCY] = value

    @property
    def data_years(self) -> str | None:
        return self._attributes_dict.get(_META_DATA_YEARS)

    @data_years.setter
    def data_years(self, value: str) -> None:
        validate_str_value(value)
        import re
        if not re.match(r"20[0-3][0-9]-20[0-3][0-9]", value):
            raise ValueError("Years range format must be 20YY-20YY")
        self._attributes_dict[_META_DATA_YEARS] = value

    @property
    def provenance(self) -> str | None:
        return self._attributes_dict.get(_META_PROVENANCE)

    @provenance.setter
    def provenance(self, value: str) -> None:
        validate_str_value(value)
        self._attributes_dict[_META_PROVENANCE] = value

    @property
    def spatial_resolution(self) -> str | None:
        return self._attributes_dict.get(_META_SPATIAL_RESOLUTION)

    @spatial_resolution.setter
    def spatial_resolution(self, value: str) -> None:
        validate_str_value(value)
        self._attributes_dict[_META_SPATIAL_RESOLUTION] = value

    @property
    def notes(self) -> str | None:
        return self._attributes_dict.get(_META_NOTES)

    @notes.setter
    def notes(self, value: str) -> None:
        validate_str_value(value)
        self._attributes_dict[_META_NOTES] = value

    @classmethod
    def format_create_date(cls, created: datetime) -> str:
        return created.strftime("%Y-%m-%d")

    @classmethod
    def metadata_defaults_dict(cls) -> dict[str, Union[str, int, float]]:
        created = datetime.now()
        return {
            _META_DATE_CREATED: cls.format_create_date(created),
            _META_ID_REFERENCE: _META_DEFAULT_ID_REFERENCE,
            _META_SPATIAL_RESOLUTION: _META_DEFAULT_UNSPECIFIED,
            _META_PROVENANCE: _META_DEFAULT_UNSPECIFIED,
            _META_UPDATE_FREQUENCY: _META_DEFAULT_UNSPECIFIED,
            _META_DATA_YEARS: f"{created.year}-{created.year}",
            _META_START_DOY: _META_DEFAULT_START_DOY,
            _META_TOOL: _META_DEFAULT_TOOL,
            _META_AUTHOR: _META_DEFAULT_AUTHOR,
            _META_WEATHER_SCHEMA_V: _META_DEFAULT_WEATHER_SCHEMA_V,
        }

    @classmethod
    def required_metadata_defaults_dict(cls, exclude_keys: list[str] = None) -> dict[str, Union[str, int, float]]:
        exclude_keys = exclude_keys or []
        return {
            k: v for k, v in cls.metadata_defaults_dict().items()
            if k in _META_REQUIRED_ARGS and k not in exclude_keys
        }

    def update(self, value: dict[str, Union[int, str]]) -> None:
        if not isinstance(value, dict):
            raise TypeError("Metadata must be a dictionary.")
        self._attributes_dict.update(value)

    def validate(self) -> None:
        for a in _META_REQUIRED_ARGS + _META_REQUIRED_CALC:
            val = self._attributes_dict.get(a)
            if val is None or len(str(val).strip()) == 0:
                raise ValueError(f"Required metadata attribute {a!r} is not set.")

WeatherData

Binary weather data and its metadata for a single weather variable.

Source code in emodpy_malaria/weather/weather_data.py

class WeatherData:
    """Binary weather data and its metadata for a single weather variable."""

    def __init__(self, data: np.ndarray, metadata: WeatherMetadata = None):
        """Create from a NumPy array of unique time series.

        Args:
            data (np.ndarray): float32 array. Shape ``(n_unique_series, series_len)`` or
                  a flat 1-D array that will be reshaped using ***metadata***.
            metadata (WeatherMetadata): If omitted, auto-generated with node IDs 1..N.
        """
        data = self._ensure_data_type(data)
        self._data: np.ndarray = data

        if metadata is not None:
            self._metadata = metadata
            expected = self._expected_shape()
            if data.shape != expected:
                self._data = data.reshape(expected)
        else:
            self._metadata = WeatherMetadata(
                node_ids=list(range(1, data.shape[0] + 1)),
                series_len=data.shape[1],
            )

        self.validate()

    def __eq__(self, other):
        if not isinstance(other, WeatherData):
            return NotImplemented
        return self.metadata == other.metadata and np.array_equal(self.data, other.data)

    def _expected_shape(self) -> tuple[int, int]:
        return self.metadata.series_unique_count, self.metadata.series_len

    def validate(self) -> None:
        expected = self._expected_shape()
        if self._data.shape != expected:
            raise ValueError(
                f"Data shape {self._data.shape} doesn't match metadata "
                f"(expected {expected})."
            )

    @property
    def metadata(self) -> WeatherMetadata:
        return self._metadata

    @property
    def data(self) -> np.ndarray:
        return self._data

    def to_base_weather(self) -> BaseWeather:
        """Create an [Weather](https://emod.idmod.org/emod-api/autoapi/emod_api/weather/weather/) instance.

        Useful for interoperability with code that expects the emod-api
        ``Weather`` object.  Note: shared offsets are expanded — each node
        gets its own copy of the data in the returned object.
        """
        expanded = self.to_dict()
        node_ids = sorted(expanded.keys())
        data = np.array([expanded[n] for n in node_ids], dtype=np.float32)
        base_meta = self._metadata.to_base_metadata()
        return BaseWeather(
            node_ids=node_ids,
            datavalue_count=self._metadata.series_len,
            author=base_meta.author,
            provenance=base_meta.provenance,
            reference=base_meta.id_reference,
            data=data,
        )

    @classmethod
    def from_base_weather(cls, base: BaseWeather,
                          attributes: WeatherAttributes = None) -> "WeatherData":
        """Create from an [Weather](https://emod.idmod.org/emod-api/autoapi/emod_api/weather/weather/) instance."""
        node_series = {
            node_id: base.nodes[node_id].data
            for node_id in base.node_ids
        }
        return cls.from_dict(node_series=node_series, attributes=attributes)

    # ------------------------------------------------------------------ #
    # Import / Export
    # ------------------------------------------------------------------ #

    @classmethod
    def from_dict(cls,
                  node_series: dict[int, Union[np.ndarray, list[float]]],
                  same_nodes: dict[int, list[int]] = None,
                  attributes: WeatherAttributes = None) -> "WeatherData":
        """Create from a ``{node_id: time_series}`` dictionary.

        Identifies unique series and builds a compact binary representation.

        Args:
            node_series (dict[int, Union[np.ndarray, list[float]]]): Node ID to time series mapping.
            same_nodes (dict[int, list[int]]): Optional mapping of nodes in ***node_series*** to
                additional node IDs that share the same data.
            attributes (WeatherAttributes): Optional metadata attributes.
        """
        if not isinstance(node_series, dict) or len(node_series) == 0:
            exc = TypeError if not isinstance(node_series, dict) else ValueError
            raise exc("node_series must be a non-empty dictionary.")

        try:
            series_values = np.array(list(node_series.values()), dtype=np.float32)
        except (ValueError, TypeError):
            raise ValueError("Time series contains values that cannot be converted to float32.")

        if np.any(np.isinf(series_values)):
            raise ValueError("Time series contains infinite values.")
        if len(series_values.shape) != 2:
            raise ValueError("All time series must be non-empty lists of equal length.")
        if any(np.isnan(list(node_series))):
            raise ValueError("Node ID list contains NaN values.")
        if np.any(np.isnan(series_values)):
            raise ValueError("Time series contains NaN values.")

        same_nodes = same_nodes or {}

        node_series_hashes = {int(n): hash_series(s) for n, s in node_series.items()}
        unique_nodes = {h: nn[0] for h, nn in invert_dict(node_series_hashes).items()}
        unique_series = [node_series[n] for n in unique_nodes.values()]

        offset_increment = series_values.shape[1] * SERIES_BYTE_VALUE_SIZE
        node_offsets = {n: (i * offset_increment) for i, n in enumerate(unique_nodes.values())}
        node_offsets.update({n: node_offsets[unique_nodes[h]] for n, h in node_series_hashes.items()})

        same_inverted = invert_dict(same_nodes, single_value=True)
        node_offsets.update({same: node_offsets[unique] for same, unique in same_inverted.items()})
        node_offsets = dict(sorted(node_offsets.items()))

        data = np.array(unique_series, dtype=np.float32)
        wm = WeatherMetadata(node_ids=node_offsets, series_len=data.shape[1], attributes=attributes)
        return WeatherData(data=data, metadata=wm)

    def to_dict(self, only_unique_series: bool = False, copy_data: bool = True) -> dict[int, np.ndarray]:
        """Export as ``{node_id: series}`` dictionary."""
        data_dict = {}
        node_groups = self.metadata.offset_nodes.values()
        series_list = np.copy(self._data) if copy_data else self._data
        for ng, s in zip(node_groups, series_list):
            nodes = ng[:1] if only_unique_series else ng
            data_dict.update(dict(zip(nodes, [s] * len(nodes))))
        return dict(sorted(data_dict.items()))

    @classmethod
    def from_csv(cls, file_path: Union[str, Path],
                 info: "DataFrameInfo" = None,
                 attributes: WeatherAttributes = None) -> "WeatherData":
        """Load from a CSV with node, step, and value columns."""
        if not Path(file_path).is_file():
            raise FileNotFoundError(f"Weather CSV not found: {file_path}")
        df = pd.read_csv(file_path)
        return cls.from_dataframe(df, info=info, attributes=attributes)

    def to_csv(self, file_path: Union[str, Path], info: "DataFrameInfo" = None) -> pd.DataFrame:
        """Write to CSV and return the DataFrame."""
        make_path(Path(file_path).parent)
        df = self.to_dataframe(info=info)
        df.to_csv(file_path, index=False)
        return df

    @classmethod
    def from_dataframe(cls, df: pd.DataFrame,
                       info: "DataFrameInfo" = None,
                       attributes: WeatherAttributes = None) -> "WeatherData":
        """Create from a pandas DataFrame with node, step, and value columns."""
        if not isinstance(df, pd.DataFrame) or len(df) == 0:
            exc = TypeError if not isinstance(df, pd.DataFrame) else ValueError
            raise exc("df must be a non-empty pandas DataFrame.")

        info = info or DataFrameInfo.detect_columns(df=df)
        nc, sc, vc = info.node_column, info.step_column, info.value_column

        for c in [nc, sc, vc]:
            if df[c].hasnans:
                raise ValueError(f"Column {c!r} contains NaN values.")

        df = df[[nc, sc, vc]].sort_values(by=[nc, sc])
        df = df[[nc, vc]].set_index(nc)
        node_series = df.groupby(nc).apply(lambda r: r.to_dict("records"), include_groups=False).to_dict()
        node_series = {node: [list(d.values())[0] for d in rw] for node, rw in node_series.items()}

        return cls.from_dict(node_series=node_series, attributes=attributes)

    def to_dataframe(self, info: "DataFrameInfo" = None) -> pd.DataFrame:
        """Convert to a DataFrame with node, step, and value columns."""
        info = info or DataFrameInfo()
        data_dict = self.to_dict(only_unique_series=info.only_unique_series)

        actual_nodes = list(data_dict.keys())
        sl = self.metadata.series_len
        nodes = np.repeat(actual_nodes, sl)
        steps = list(range(1, sl + 1)) * len(actual_nodes)
        values = np.array(list(data_dict.values())).reshape(len(data_dict) * sl)

        df = pd.DataFrame({
            info.node_column: nodes,
            info.step_column: steps,
            info.value_column: values,
        })
        df[info.node_column] = df[info.node_column].astype(int)
        df[info.step_column] = df[info.step_column].astype(int)
        df[info.value_column] = df[info.value_column].astype(np.float32)
        df.sort_values(by=[info.node_column, info.step_column], inplace=True)
        return df

    @classmethod
    def from_file(cls, file_path: Union[str, Path]) -> "WeatherData":
        """Read from a ``.bin`` / ``.bin.json`` file pair."""
        file_path = str(file_path)
        wm = WeatherMetadata.from_file(f"{file_path}.json")
        if not Path(file_path).is_file():
            raise FileNotFoundError(f"Data file not found: {file_path}")
        data = np.fromfile(file_path, dtype=np.float32)
        if wm.total_value_count != len(data):
            raise ValueError(
                f"Data length {len(data)} doesn't match metadata "
                f"({wm.series_count} * {wm.series_len} = {wm.total_value_count})."
            )
        return WeatherData(data=data, metadata=wm)

    def to_file(self, file_path: Union[str, Path]) -> None:
        """Write ``.bin`` and ``.bin.json`` files."""
        file_path = str(file_path)
        self.validate()
        make_path(Path(file_path).parent)
        self._ensure_data_type(self._data)
        with open(file_path, "wb") as bf:
            self._data.reshape(self.metadata.total_value_count).tofile(bf)
        self._metadata.to_file(f"{file_path}.json")

    @classmethod
    def _ensure_data_type(cls, data) -> np.ndarray:
        if data is None or not hasattr(data, "__len__") or len(data) == 0:
            raise ValueError("Data must be a non-empty iterable.")
        return np.array(data, dtype=np.float32)

__init__(data, metadata=None)

Create from a NumPy array of unique time series.

Parameters:

Name	Type	Description	Default
data	ndarray	float32 array. Shape (n_unique_series, series_len) or a flat 1-D array that will be reshaped using metadata.	required
metadata	WeatherMetadata	If omitted, auto-generated with node IDs 1..N.	None

Source code in emodpy_malaria/weather/weather_data.py

def __init__(self, data: np.ndarray, metadata: WeatherMetadata = None):
    """Create from a NumPy array of unique time series.

    Args:
        data (np.ndarray): float32 array. Shape ``(n_unique_series, series_len)`` or
              a flat 1-D array that will be reshaped using ***metadata***.
        metadata (WeatherMetadata): If omitted, auto-generated with node IDs 1..N.
    """
    data = self._ensure_data_type(data)
    self._data: np.ndarray = data

    if metadata is not None:
        self._metadata = metadata
        expected = self._expected_shape()
        if data.shape != expected:
            self._data = data.reshape(expected)
    else:
        self._metadata = WeatherMetadata(
            node_ids=list(range(1, data.shape[0] + 1)),
            series_len=data.shape[1],
        )

    self.validate()

from_base_weather(base, attributes=None) classmethod

Create from an Weather instance.

Source code in emodpy_malaria/weather/weather_data.py

@classmethod
def from_base_weather(cls, base: BaseWeather,
                      attributes: WeatherAttributes = None) -> "WeatherData":
    """Create from an [Weather](https://emod.idmod.org/emod-api/autoapi/emod_api/weather/weather/) instance."""
    node_series = {
        node_id: base.nodes[node_id].data
        for node_id in base.node_ids
    }
    return cls.from_dict(node_series=node_series, attributes=attributes)

from_csv(file_path, info=None, attributes=None) classmethod

Load from a CSV with node, step, and value columns.

Source code in emodpy_malaria/weather/weather_data.py

@classmethod
def from_csv(cls, file_path: Union[str, Path],
             info: "DataFrameInfo" = None,
             attributes: WeatherAttributes = None) -> "WeatherData":
    """Load from a CSV with node, step, and value columns."""
    if not Path(file_path).is_file():
        raise FileNotFoundError(f"Weather CSV not found: {file_path}")
    df = pd.read_csv(file_path)
    return cls.from_dataframe(df, info=info, attributes=attributes)

from_dataframe(df, info=None, attributes=None) classmethod

Create from a pandas DataFrame with node, step, and value columns.

Source code in emodpy_malaria/weather/weather_data.py

@classmethod
def from_dataframe(cls, df: pd.DataFrame,
                   info: "DataFrameInfo" = None,
                   attributes: WeatherAttributes = None) -> "WeatherData":
    """Create from a pandas DataFrame with node, step, and value columns."""
    if not isinstance(df, pd.DataFrame) or len(df) == 0:
        exc = TypeError if not isinstance(df, pd.DataFrame) else ValueError
        raise exc("df must be a non-empty pandas DataFrame.")

    info = info or DataFrameInfo.detect_columns(df=df)
    nc, sc, vc = info.node_column, info.step_column, info.value_column

    for c in [nc, sc, vc]:
        if df[c].hasnans:
            raise ValueError(f"Column {c!r} contains NaN values.")

    df = df[[nc, sc, vc]].sort_values(by=[nc, sc])
    df = df[[nc, vc]].set_index(nc)
    node_series = df.groupby(nc).apply(lambda r: r.to_dict("records"), include_groups=False).to_dict()
    node_series = {node: [list(d.values())[0] for d in rw] for node, rw in node_series.items()}

    return cls.from_dict(node_series=node_series, attributes=attributes)

from_dict(node_series, same_nodes=None, attributes=None) classmethod

Create from a {node_id: time_series} dictionary.

Identifies unique series and builds a compact binary representation.

Parameters:

Name	Type	Description	Default
node_series	dict[int, Union[ndarray, list[float]]]	Node ID to time series mapping.	required
same_nodes	dict[int, list[int]]	Optional mapping of nodes in node_series to additional node IDs that share the same data.	None
attributes	WeatherAttributes	Optional metadata attributes.	None

Source code in emodpy_malaria/weather/weather_data.py

@classmethod
def from_dict(cls,
              node_series: dict[int, Union[np.ndarray, list[float]]],
              same_nodes: dict[int, list[int]] = None,
              attributes: WeatherAttributes = None) -> "WeatherData":
    """Create from a ``{node_id: time_series}`` dictionary.

    Identifies unique series and builds a compact binary representation.

    Args:
        node_series (dict[int, Union[np.ndarray, list[float]]]): Node ID to time series mapping.
        same_nodes (dict[int, list[int]]): Optional mapping of nodes in ***node_series*** to
            additional node IDs that share the same data.
        attributes (WeatherAttributes): Optional metadata attributes.
    """
    if not isinstance(node_series, dict) or len(node_series) == 0:
        exc = TypeError if not isinstance(node_series, dict) else ValueError
        raise exc("node_series must be a non-empty dictionary.")

    try:
        series_values = np.array(list(node_series.values()), dtype=np.float32)
    except (ValueError, TypeError):
        raise ValueError("Time series contains values that cannot be converted to float32.")

    if np.any(np.isinf(series_values)):
        raise ValueError("Time series contains infinite values.")
    if len(series_values.shape) != 2:
        raise ValueError("All time series must be non-empty lists of equal length.")
    if any(np.isnan(list(node_series))):
        raise ValueError("Node ID list contains NaN values.")
    if np.any(np.isnan(series_values)):
        raise ValueError("Time series contains NaN values.")

    same_nodes = same_nodes or {}

    node_series_hashes = {int(n): hash_series(s) for n, s in node_series.items()}
    unique_nodes = {h: nn[0] for h, nn in invert_dict(node_series_hashes).items()}
    unique_series = [node_series[n] for n in unique_nodes.values()]

    offset_increment = series_values.shape[1] * SERIES_BYTE_VALUE_SIZE
    node_offsets = {n: (i * offset_increment) for i, n in enumerate(unique_nodes.values())}
    node_offsets.update({n: node_offsets[unique_nodes[h]] for n, h in node_series_hashes.items()})

    same_inverted = invert_dict(same_nodes, single_value=True)
    node_offsets.update({same: node_offsets[unique] for same, unique in same_inverted.items()})
    node_offsets = dict(sorted(node_offsets.items()))

    data = np.array(unique_series, dtype=np.float32)
    wm = WeatherMetadata(node_ids=node_offsets, series_len=data.shape[1], attributes=attributes)
    return WeatherData(data=data, metadata=wm)

from_file(file_path) classmethod

Read from a .bin / .bin.json file pair.

Source code in emodpy_malaria/weather/weather_data.py

@classmethod
def from_file(cls, file_path: Union[str, Path]) -> "WeatherData":
    """Read from a ``.bin`` / ``.bin.json`` file pair."""
    file_path = str(file_path)
    wm = WeatherMetadata.from_file(f"{file_path}.json")
    if not Path(file_path).is_file():
        raise FileNotFoundError(f"Data file not found: {file_path}")
    data = np.fromfile(file_path, dtype=np.float32)
    if wm.total_value_count != len(data):
        raise ValueError(
            f"Data length {len(data)} doesn't match metadata "
            f"({wm.series_count} * {wm.series_len} = {wm.total_value_count})."
        )
    return WeatherData(data=data, metadata=wm)

to_base_weather()

Create an Weather instance.

Useful for interoperability with code that expects the emod-api Weather object. Note: shared offsets are expanded — each node gets its own copy of the data in the returned object.

Source code in emodpy_malaria/weather/weather_data.py

def to_base_weather(self) -> BaseWeather:
    """Create an [Weather](https://emod.idmod.org/emod-api/autoapi/emod_api/weather/weather/) instance.

    Useful for interoperability with code that expects the emod-api
    ``Weather`` object.  Note: shared offsets are expanded — each node
    gets its own copy of the data in the returned object.
    """
    expanded = self.to_dict()
    node_ids = sorted(expanded.keys())
    data = np.array([expanded[n] for n in node_ids], dtype=np.float32)
    base_meta = self._metadata.to_base_metadata()
    return BaseWeather(
        node_ids=node_ids,
        datavalue_count=self._metadata.series_len,
        author=base_meta.author,
        provenance=base_meta.provenance,
        reference=base_meta.id_reference,
        data=data,
    )

to_csv(file_path, info=None)

Write to CSV and return the DataFrame.

Source code in emodpy_malaria/weather/weather_data.py

def to_csv(self, file_path: Union[str, Path], info: "DataFrameInfo" = None) -> pd.DataFrame:
    """Write to CSV and return the DataFrame."""
    make_path(Path(file_path).parent)
    df = self.to_dataframe(info=info)
    df.to_csv(file_path, index=False)
    return df

to_dataframe(info=None)

Convert to a DataFrame with node, step, and value columns.

Source code in emodpy_malaria/weather/weather_data.py

def to_dataframe(self, info: "DataFrameInfo" = None) -> pd.DataFrame:
    """Convert to a DataFrame with node, step, and value columns."""
    info = info or DataFrameInfo()
    data_dict = self.to_dict(only_unique_series=info.only_unique_series)

    actual_nodes = list(data_dict.keys())
    sl = self.metadata.series_len
    nodes = np.repeat(actual_nodes, sl)
    steps = list(range(1, sl + 1)) * len(actual_nodes)
    values = np.array(list(data_dict.values())).reshape(len(data_dict) * sl)

    df = pd.DataFrame({
        info.node_column: nodes,
        info.step_column: steps,
        info.value_column: values,
    })
    df[info.node_column] = df[info.node_column].astype(int)
    df[info.step_column] = df[info.step_column].astype(int)
    df[info.value_column] = df[info.value_column].astype(np.float32)
    df.sort_values(by=[info.node_column, info.step_column], inplace=True)
    return df

to_dict(only_unique_series=False, copy_data=True)

Export as {node_id: series} dictionary.

Source code in emodpy_malaria/weather/weather_data.py

def to_dict(self, only_unique_series: bool = False, copy_data: bool = True) -> dict[int, np.ndarray]:
    """Export as ``{node_id: series}`` dictionary."""
    data_dict = {}
    node_groups = self.metadata.offset_nodes.values()
    series_list = np.copy(self._data) if copy_data else self._data
    for ng, s in zip(node_groups, series_list):
        nodes = ng[:1] if only_unique_series else ng
        data_dict.update(dict(zip(nodes, [s] * len(nodes))))
    return dict(sorted(data_dict.items()))

to_file(file_path)

Write .bin and .bin.json files.

Source code in emodpy_malaria/weather/weather_data.py

def to_file(self, file_path: Union[str, Path]) -> None:
    """Write ``.bin`` and ``.bin.json`` files."""
    file_path = str(file_path)
    self.validate()
    make_path(Path(file_path).parent)
    self._ensure_data_type(self._data)
    with open(file_path, "wb") as bf:
        self._data.reshape(self.metadata.total_value_count).tofile(bf)
    self._metadata.to_file(f"{file_path}.json")

WeatherMetadata

Bases: WeatherAttributes

Weather metadata with node offsets and count fields.

Wraps Metadata for core node-offset computation and basic .bin.json parsing, extending it with rich metadata attributes and shared-offset (deduplication) support.

Source code in emodpy_malaria/weather/weather_metadata.py

class WeatherMetadata(WeatherAttributes):
    """Weather metadata with node offsets and count fields.

    Wraps [Metadata](https://emod.idmod.org/emod-api/autoapi/emod_api/weather/weather/) for core node-offset
    computation and basic ``.bin.json`` parsing, extending it with rich
    metadata attributes and shared-offset (deduplication) support.
    """

    def __init__(self,
                 node_ids: Union[list[int], dict[int, int]],
                 series_len: int = None,
                 attributes: Union["WeatherMetadata", WeatherAttributes,
                                   dict[str, Union[str, int, float]]] = None):
        if isinstance(attributes, (WeatherMetadata, WeatherAttributes)):
            attributes_dict = attributes.attributes_dict
        else:
            attributes_dict = attributes

        super().__init__(attributes_dict=attributes_dict)

        if isinstance(node_ids, dict):
            # Shared-offset case — store the dict directly
            self._node_offsets = node_ids
            series_len = int(series_len or self._expected_series_len())
            self._validate_series_len(series_len)
        else:
            # Simple list — delegate offset calculation to emod-api Metadata
            self._validate_series_len(series_len)
            base = BaseMetadata(list(node_ids), series_len)
            self._node_offsets = dict(base.nodes)

        self._series_len = series_len
        self.update(self._metadata_count_dict)
        self.validate()

    def __eq__(self, other):
        if not isinstance(other, WeatherMetadata):
            return NotImplemented
        return (super().__eq__(other)
                and sorted(self.node_offsets) == sorted(other.node_offsets)
                and all(self.node_offsets[k] == other.node_offsets[k]
                        for k in self.node_offsets))

    def to_base_metadata(self) -> BaseMetadata:
        """Create an [Metadata](https://emod.idmod.org/emod-api/autoapi/emod_api/weather/weather/) instance.

        Useful for interoperability with code that expects the emod-api
        ``Metadata`` object.  Note: shared offsets are expanded — each
        node gets its own sequential offset in the returned object.
        """
        return BaseMetadata(
            node_ids=self.nodes,
            datavalue_count=self._series_len,
            author=self.author,
            provenance=self.provenance,
            reference=self.id_reference,
            frequency=self.update_resolution,
        )

    @property
    def _metadata_count_dict(self):
        node_count = len(self.nodes)
        return {
            _META_OFFSET_COUNT: len(self._node_offsets),
            _META_DTK_NODES_COUNT: node_count,
            _META_NODE_COUNT: node_count,
            _META_WEATHER_CELL_COUNT: node_count,
            _META_DATA_VALUE_COUNT: self._series_len,
            _META_DATA_CELL_VALUE_COUNT: self._series_len,
        }

    @classmethod
    def _validate_series_len(cls, series_len: int) -> None:
        if not isinstance(series_len, int) or series_len <= 0:
            raise ValueError("Weather time series length must be a positive integer.")

    def _expected_series_len(self) -> int:
        if self._node_offsets and len(self._node_offsets) > 0:
            offsets = sorted(set(self._node_offsets.values()))[:2]
            if len(offsets) > 1:
                return int((offsets[1] - offsets[0]) / SERIES_BYTE_VALUE_SIZE)
        return -1

    def validate(self) -> None:
        super().validate()

        if not self.nodes:
            raise ValueError("node_ids must not be empty.")
        if not all(isinstance(i, int) for i in self.nodes):
            raise TypeError("node_ids must be integers.")

        max_uint32 = 0xFFFFFFFF
        invalid_nodes = [n for n in self.nodes if not (0 < n <= max_uint32)]
        if invalid_nodes:
            raise ValueError(
                f"Node IDs must be in (0, {max_uint32}]. "
                f"Invalid: {invalid_nodes[:5]}"
            )

        invalid_offsets = [
            o for o in self.node_offsets.values()
            if not (0 <= o <= max_uint32)
        ]
        if invalid_offsets:
            raise ValueError(
                f"Offsets must be in [0, {max_uint32}]. "
                f"Invalid: {invalid_offsets[:5]}"
            )

        if len(set(self.nodes)) != len(self.nodes):
            raise ValueError("node_ids must be unique.")

        if len(self.node_offset_str) != self.node_count * 16:
            raise ValueError("node_offset_str length doesn't match node count.")

        self._validate_series_len(self._series_len)
        expected = self._expected_series_len()
        if 0 < expected != self._series_len:
            raise ValueError("Time series length doesn't match offset distances.")

    @property
    def attributes(self) -> WeatherAttributes:
        meta_keys = list(self._metadata_count_dict)
        attributes_dict = {
            k: v for k, v in self._attributes_dict.items()
            if k not in meta_keys
        }
        return WeatherAttributes(attributes_dict=attributes_dict)

    @property
    def datavalue_count(self) -> int:
        return self._series_len

    @property
    def series_len(self) -> int:
        return self._series_len

    @property
    def series_count(self) -> int:
        return len(set(self._node_offsets.values()))

    @property
    def series_unique_count(self) -> int:
        return len(self.offset_nodes)

    @property
    def total_value_count(self) -> int:
        return self.series_count * self.series_len

    @property
    def nodes(self) -> list[int]:
        return list(self._node_offsets)

    @property
    def node_count(self) -> int:
        return len(self.nodes)

    @property
    def node_offset_str(self) -> str:
        return self._convert_offset_dict_to_str(self._node_offsets)

    @property
    def node_offsets(self) -> dict[int, int]:
        return self._node_offsets

    @property
    def offset_nodes(self) -> dict[int, list[int]]:
        return invert_dict(self._node_offsets, sort=True)

    def to_file(self, file_path: Union[str, Path]) -> None:
        """Write the rich ``.bin.json`` metadata file.

        Produces a superset of the format written by
        [Metadata.write_file()](https://emod.idmod.org/emod-api/autoapi/emod_api/weather/weather/), including
        additional attributes like ``Tool``, ``WeatherSchemaVersion``,
        ``Resolution``, etc.
        """
        self.validate()
        make_path(Path(file_path).parent)
        offset_str = self._convert_offset_dict_to_str(self._node_offsets)
        content = dict(Metadata=self.attributes_dict, NodeOffsets=offset_str)
        save_json(content=content, file_path=file_path)

    @classmethod
    def from_file(cls, file_path: Union[str, Path]) -> "WeatherMetadata":
        """Read a ``.bin.json`` file.

        Uses [Metadata.from_file()](https://emod.idmod.org/emod-api/autoapi/emod_api/weather/weather/) for core
        parsing (node offsets, datavalue count), then augments with any
        additional attributes present in the file.
        """
        base = BaseMetadata.from_file(str(file_path))

        with open(str(file_path), "rb") as f:
            content = json.load(f)
        raw_meta = content.get("Metadata", {})

        return WeatherMetadata(
            node_ids=dict(base.nodes),
            series_len=base.datavalue_count,
            attributes=raw_meta,
        )

    @staticmethod
    def _convert_offset_str_to_dict(offset_str: str) -> dict[int, int]:
        entry_count = len(offset_str) // 16
        node_offsets = {}
        for i in range(entry_count):
            idx = i * 16
            entry = offset_str[idx: idx + 16]
            node_offsets[int(entry[:8], 16)] = int(entry[8:16], 16)
        return node_offsets

    @staticmethod
    def _convert_offset_dict_to_str(node_offsets: dict[int, int]) -> str:
        return "".join(f"{node_id:08x}{offset:08x}"
                       for node_id, offset in node_offsets.items())

from_file(file_path) classmethod

Read a .bin.json file.

Uses Metadata.from_file() for core parsing (node offsets, datavalue count), then augments with any additional attributes present in the file.

Source code in emodpy_malaria/weather/weather_metadata.py

@classmethod
def from_file(cls, file_path: Union[str, Path]) -> "WeatherMetadata":
    """Read a ``.bin.json`` file.

    Uses [Metadata.from_file()](https://emod.idmod.org/emod-api/autoapi/emod_api/weather/weather/) for core
    parsing (node offsets, datavalue count), then augments with any
    additional attributes present in the file.
    """
    base = BaseMetadata.from_file(str(file_path))

    with open(str(file_path), "rb") as f:
        content = json.load(f)
    raw_meta = content.get("Metadata", {})

    return WeatherMetadata(
        node_ids=dict(base.nodes),
        series_len=base.datavalue_count,
        attributes=raw_meta,
    )

to_base_metadata()

Create an Metadata instance.

Useful for interoperability with code that expects the emod-api Metadata object. Note: shared offsets are expanded — each node gets its own sequential offset in the returned object.

Source code in emodpy_malaria/weather/weather_metadata.py

def to_base_metadata(self) -> BaseMetadata:
    """Create an [Metadata](https://emod.idmod.org/emod-api/autoapi/emod_api/weather/weather/) instance.

    Useful for interoperability with code that expects the emod-api
    ``Metadata`` object.  Note: shared offsets are expanded — each
    node gets its own sequential offset in the returned object.
    """
    return BaseMetadata(
        node_ids=self.nodes,
        datavalue_count=self._series_len,
        author=self.author,
        provenance=self.provenance,
        reference=self.id_reference,
        frequency=self.update_resolution,
    )

to_file(file_path)

Write the rich .bin.json metadata file.

Produces a superset of the format written by Metadata.write_file(), including additional attributes like Tool, WeatherSchemaVersion, Resolution, etc.

Source code in emodpy_malaria/weather/weather_metadata.py

def to_file(self, file_path: Union[str, Path]) -> None:
    """Write the rich ``.bin.json`` metadata file.

    Produces a superset of the format written by
    [Metadata.write_file()](https://emod.idmod.org/emod-api/autoapi/emod_api/weather/weather/), including
    additional attributes like ``Tool``, ``WeatherSchemaVersion``,
    ``Resolution``, etc.
    """
    self.validate()
    make_path(Path(file_path).parent)
    offset_str = self._convert_offset_dict_to_str(self._node_offsets)
    content = dict(Metadata=self.attributes_dict, NodeOffsets=offset_str)
    save_json(content=content, file_path=file_path)

WeatherSet

A set of weather files for all (or a subset of) EMOD weather variables.

Source code in emodpy_malaria/weather/weather_set.py

class WeatherSet:
    """A set of weather files for all (or a subset of) EMOD weather variables."""

    def __init__(self,
                 dir_path: Union[str, Path] = None,
                 file_names: dict[WeatherVariable, str] = None,
                 weather_columns: dict[WeatherVariable, str] = None):
        self._dir_path: Union[str, Path] = dir_path
        self._file_names: dict[WeatherVariable, str] = file_names or {}
        self._weather_columns: dict[WeatherVariable, str] = weather_columns or {}
        self._weather_dict: dict[WeatherVariable, WeatherData] = {}

    def __getitem__(self, weather_variable: WeatherVariable):
        return self._weather_dict[weather_variable]

    def __setitem__(self, weather_variable: WeatherVariable, weather_object: WeatherData):
        self._weather_dict[weather_variable] = weather_object

    def __len__(self):
        return len(self._weather_dict)

    def __str__(self):
        return str(self.weather_variables)

    def __eq__(self, other):
        if not isinstance(other, WeatherSet):
            return NotImplemented
        if self.weather_variables != other.weather_variables:
            return False
        return all(self[v] == other[v] for v in self.weather_variables)

    def keys(self):
        return self._weather_dict.keys()

    def values(self) -> list[WeatherData]:
        return list(self._weather_dict.values())

    def items(self):
        return self._weather_dict.items()

    @property
    def dir_path(self) -> str:
        return str(self._dir_path)

    @property
    def file_names(self) -> dict[WeatherVariable, str]:
        return self._file_names

    @property
    def attributes(self) -> WeatherAttributes | None:
        if self.weather_variables:
            return self.values()[0].metadata.attributes
        return None

    @property
    def node_ids(self) -> list[int]:
        if self.weather_variables:
            return self.values()[0].metadata.nodes
        return []

    @property
    def id_reference(self) -> str | None:
        if self.weather_variables:
            return self.values()[0].metadata.id_reference
        return None

    @id_reference.setter
    def id_reference(self, value: str) -> None:
        for wd in self.values():
            wd.metadata.id_reference = value

    @property
    def update_resolution(self) -> str | None:
        if self.weather_variables:
            return self.values()[0].metadata.update_resolution
        return None

    @update_resolution.setter
    def update_resolution(self, value: str) -> None:
        for wd in self.values():
            wd.metadata.update_resolution = value

    @property
    def notes(self) -> str | None:
        if self.weather_variables:
            return self.values()[0].metadata.notes
        return None

    @notes.setter
    def notes(self, value: str) -> None:
        for wd in self.values():
            wd.metadata.notes = value

    @property
    def weather_variables(self) -> list[WeatherVariable]:
        return list(self._weather_dict)

    @property
    def weather_columns(self) -> dict[WeatherVariable, str]:
        return self._weather_columns

    # ------------------------------------------------------------------ #
    # CSV / DataFrame I/O
    # ------------------------------------------------------------------ #

    @classmethod
    def from_dataframe(cls, df: pd.DataFrame,
                       node_column: str = None,
                       step_column: str = None,
                       weather_columns: dict[WeatherVariable, str] = None,
                       attributes: WeatherAttributes = None,
                       notes: str = None) -> "WeatherSet":
        """Create from a DataFrame containing all weather variables as columns.

        Args:
            df (pd.DataFrame): DataFrame with node, step, and weather variable columns.
            node_column (str): Column name for node IDs.
            step_column (str): Column name for time steps.
            weather_columns (dict[WeatherVariable, str]): ``{WeatherVariable: column_name}`` mapping.
            attributes (WeatherAttributes): Optional metadata attributes.
            notes (str): Free-text note stored in the weather file metadata.
                Use this to record where the original data came from and
                how it was processed.
        """
        if not isinstance(df, pd.DataFrame):
            raise TypeError(f"Expected DataFrame, got {type(df)}.")
        return cls._from_csv_data(
            data_csv=df, node_column=node_column, step_column=step_column,
            weather_columns=weather_columns, attributes=attributes, notes=notes,
        )

    @classmethod
    def from_csv(cls, file_path: Union[str, Path],
                 node_column: str = None,
                 step_column: str = None,
                 weather_columns: dict[WeatherVariable, str] = None,
                 attributes: WeatherAttributes = None,
                 notes: str = None) -> "WeatherSet":
        """Create from a CSV file containing all weather variables.

        Args:
            file_path (Union[str, Path]): Path to the CSV file.
            node_column (str): Column name for node IDs.
            step_column (str): Column name for time steps.
            weather_columns (dict[WeatherVariable, str]): ``{WeatherVariable: column_name}`` mapping.
            attributes (WeatherAttributes): Optional metadata attributes.
            notes (str): Free-text note stored in the weather file metadata.
                Use this to record where the original data came from and
                how it was processed.
        """
        if not Path(file_path).is_file():
            raise FileNotFoundError(f"CSV file not found: {file_path}")
        return cls._from_csv_data(
            data_csv=str(file_path), node_column=node_column, step_column=step_column,
            weather_columns=weather_columns, attributes=attributes, notes=notes,
        )

    @classmethod
    def _from_csv_data(cls,
                       data_csv: Union[str, pd.DataFrame],
                       node_column: str = None,
                       step_column: str = None,
                       weather_columns: dict[WeatherVariable, str] = None,
                       attributes: WeatherAttributes = None,
                       notes: str = None) -> "WeatherSet":
        infos, weather_columns = cls._init_dataframe_info_dict(node_column, step_column, weather_columns)
        attributes = attributes or WeatherAttributes()
        if notes is not None:
            attributes.notes = notes
        ws = WeatherSet(weather_columns=weather_columns)
        for v, info in infos.items():
            if isinstance(data_csv, str):
                ws[v] = WeatherData.from_csv(file_path=data_csv, info=info, attributes=attributes)
            elif isinstance(data_csv, pd.DataFrame):
                ws[v] = WeatherData.from_dataframe(df=data_csv, info=info, attributes=attributes)
            else:
                raise TypeError(f"Unsupported data type {type(data_csv)}.")
        ws.validate()
        return ws

    def to_dataframe(self,
                     node_column: str = None,
                     step_column: str = None,
                     weather_columns: dict[WeatherVariable, str] = None) -> pd.DataFrame:
        """Export all variables to a single DataFrame."""
        weather_columns = weather_columns or {v: None for v in self.weather_variables}
        not_available = [v for v in weather_columns if v not in self.weather_variables]
        if not_available:
            raise ValueError(f"Requested unavailable weather variables: {not_available}")

        infos, weather_columns = self._init_dataframe_info_dict(node_column, step_column, weather_columns)
        self._weather_columns = weather_columns
        df = None
        for v in infos:
            df2 = self[v].to_dataframe(infos[v])
            if df is None:
                df = df2
            else:
                df[infos[v].value_column] = df2[infos[v].value_column]
        return df

    def to_csv(self, file_path: Union[str, Path],
               node_column: str = None,
               step_column: str = None,
               weather_columns: dict[WeatherVariable, str] = None) -> pd.DataFrame:
        """Export all variables to a single CSV."""
        df = self.to_dataframe(node_column, step_column, weather_columns)
        df.to_csv(file_path, index=False)
        return df

    # ------------------------------------------------------------------ #
    # Binary file I/O
    # ------------------------------------------------------------------ #

    def _load(self) -> "WeatherSet":
        if not self.dir_path or not Path(self.dir_path).is_dir():
            raise ValueError("A valid directory is required.")
        if not self.file_names:
            raise ValueError("File names dictionary is required.")
        for v, n in self.file_names.items():
            bin_path = self._weather_file_path(n)
            self[v] = WeatherData.from_file(bin_path)
        self.validate()
        return self

    def _save(self) -> None:
        if not self._dir_path:
            raise ValueError("Directory is required.")
        if not self._file_names:
            raise ValueError("File names are required.")
        make_path(self._dir_path)
        for v, wd in self._weather_dict.items():
            bin_path = self._weather_file_path(self._file_names[v])
            wd.to_file(bin_path)

    @classmethod
    def from_files(cls, dir_path: Union[str, Path],
                   prefix: str = "",
                   file_names: dict[WeatherVariable, str] = None) -> "WeatherSet":
        """Load from existing ``.bin`` / ``.bin.json`` file pairs in a directory."""
        WeatherVariable.validate_types(file_names, [str, Path])
        file_names = file_names or cls.select_weather_files(dir_path=dir_path, prefix=prefix)
        ws = WeatherSet(dir_path=dir_path, file_names=file_names)
        ws._load()
        return ws

    def to_files(self, dir_path: Union[str, Path],
                 file_names: dict[WeatherVariable, str] = None) -> None:
        """Write all ``.bin`` / ``.bin.json`` file pairs to a directory."""
        file_names = file_names or self.make_file_paths()
        self._dir_path = Path(dir_path)
        self._file_names = file_names
        self._save()

    # ------------------------------------------------------------------ #
    # Helpers
    # ------------------------------------------------------------------ #

    @classmethod
    def _init_weather_columns(cls, weather_columns: dict[WeatherVariable, str | None] = None
                              ) -> dict[WeatherVariable, str]:
        WeatherVariable.validate_types(weather_columns, [str, None])
        if weather_columns:
            weather_variables = list(weather_columns)
        else:
            weather_variables = WeatherVariable.list(exclude=WeatherVariable.LAND_TEMPERATURE)
        weather_columns = weather_columns or {}
        return {v: weather_columns.get(v) or v.value for v in weather_variables}

    @classmethod
    def _init_dataframe_info_dict(cls,
                                  node_column: str = None,
                                  step_column: str = None,
                                  weather_columns: dict[WeatherVariable, str] = None
                                  ) -> tuple[dict[WeatherVariable, DataFrameInfo], dict[WeatherVariable, str]]:
        weather_columns = cls._init_weather_columns(weather_columns)
        info_dict = {
            v: DataFrameInfo(node_column=node_column, step_column=step_column, value_column=weather_columns[v])
            for v in weather_columns
        }
        return info_dict, weather_columns

    @classmethod
    def _make_file_templates(cls,
                             prefix: str = "*",
                             suffix: str = "*{}*.bin",
                             weather_variables: list[WeatherVariable] = None,
                             weather_names: dict[WeatherVariable, str] = None) -> dict[WeatherVariable, str]:
        if prefix is None:
            raise ValueError("Prefix cannot be None.")
        if suffix is None:
            raise ValueError("Suffix cannot be None.")
        WeatherVariable.validate_types(weather_names, [str])

        if not suffix.endswith(".bin") and not suffix.endswith("*"):
            suffix += "*.bin"

        template = prefix + suffix
        template = template.replace("**", "*")

        if not weather_variables:
            weather_variables = WeatherVariable.list(exclude=WeatherVariable.LAND_TEMPERATURE)
        weather_names = weather_names or {v: v.value for v in weather_variables}

        return {v: template.format(weather_names[v]) for v in weather_names}

    @classmethod
    def make_file_paths(cls,
                        dir_path: Union[str, Path] = None,
                        prefix: str = "",
                        suffix: str = "{}.bin",
                        weather_variables: list[WeatherVariable] = None,
                        weather_names: dict[WeatherVariable, str] = None) -> dict[WeatherVariable, str]:
        """Generate conventional EMOD weather file paths."""
        names = cls._make_file_templates(
            prefix=prefix, suffix=suffix,
            weather_names=weather_names, weather_variables=weather_variables,
        )
        if dir_path is not None:
            names = {v: str(Path(dir_path) / n) for v, n in names.items()}
        return names

    @classmethod
    def select_weather_files(cls, dir_path: Union[str, Path],
                             prefix: str = "*",
                             suffix: str = "*{}*.bin",
                             weather_variables: list[WeatherVariable] = None,
                             weather_names: dict[WeatherVariable, str] = None) -> dict[WeatherVariable, str]:
        """Find weather files in a directory by name pattern."""
        if dir_path is None:
            raise ValueError("Directory path is required.")
        templates = cls._make_file_templates(
            prefix=prefix, suffix=suffix,
            weather_names=weather_names, weather_variables=weather_variables,
        )
        names = {}
        for v, pattern in templates.items():
            files = list(Path(dir_path).glob(pattern))
            if len(files) > 1:
                raise ValueError(f"Multiple files match pattern {pattern!r}")
            if len(files) == 1:
                names[v] = files[0].name
        return names

    def _weather_file_path(self, file_name: Union[str, Path]) -> Path:
        return Path(self.dir_path) / str(file_name)

    def validate(self) -> None:
        series_len0 = node_count0 = id_ref0 = resolution0 = years0 = None

        for v, wd in self._weather_dict.items():
            wd.validate()
            wd.metadata.validate()
            wm = wd.metadata

            series_len0 = series_len0 or wm.series_len
            node_count0 = node_count0 or wm.node_count
            id_ref0 = id_ref0 or wm.id_reference
            resolution0 = resolution0 or wm.spatial_resolution
            years0 = years0 or wm.data_years

            label = f" ({self.file_names[v]})" if v in self.file_names else ""
            if series_len0 != wm.series_len:
                raise ValueError(f"series_len mismatch for {v}{label}")
            if node_count0 != wm.node_count:
                raise ValueError(f"node_count mismatch for {v}{label}")
            if id_ref0 != wm.id_reference:
                raise ValueError(f"id_reference mismatch for {v}{label}")
            if resolution0 != wm.spatial_resolution:
                raise ValueError(f"spatial_resolution mismatch for {v}{label}")
            if years0 != wm.data_years:
                raise ValueError(f"data_years mismatch for {v}{label}")

        if self._weather_columns:
            for v in WeatherVariable.list():
                in_data = v in self._weather_dict
                in_cols = v in self._weather_columns
                if in_data != in_cols:
                    raise ValueError(
                        f"Weather variable {v} is in "
                        f"{'data' if in_data else 'columns'} but not "
                        f"{'columns' if in_data else 'data'}."
                    )

from_csv(file_path, node_column=None, step_column=None, weather_columns=None, attributes=None, notes=None) classmethod

Create from a CSV file containing all weather variables.

Parameters:

Name	Type	Description	Default
file_path	Union[str, Path]	Path to the CSV file.	required
node_column	str	Column name for node IDs.	None
step_column	str	Column name for time steps.	None
weather_columns	dict[WeatherVariable, str]	{WeatherVariable: column_name} mapping.	None
attributes	WeatherAttributes	Optional metadata attributes.	None
notes	str	Free-text note stored in the weather file metadata. Use this to record where the original data came from and how it was processed.	None

Source code in emodpy_malaria/weather/weather_set.py

@classmethod
def from_csv(cls, file_path: Union[str, Path],
             node_column: str = None,
             step_column: str = None,
             weather_columns: dict[WeatherVariable, str] = None,
             attributes: WeatherAttributes = None,
             notes: str = None) -> "WeatherSet":
    """Create from a CSV file containing all weather variables.

    Args:
        file_path (Union[str, Path]): Path to the CSV file.
        node_column (str): Column name for node IDs.
        step_column (str): Column name for time steps.
        weather_columns (dict[WeatherVariable, str]): ``{WeatherVariable: column_name}`` mapping.
        attributes (WeatherAttributes): Optional metadata attributes.
        notes (str): Free-text note stored in the weather file metadata.
            Use this to record where the original data came from and
            how it was processed.
    """
    if not Path(file_path).is_file():
        raise FileNotFoundError(f"CSV file not found: {file_path}")
    return cls._from_csv_data(
        data_csv=str(file_path), node_column=node_column, step_column=step_column,
        weather_columns=weather_columns, attributes=attributes, notes=notes,
    )

from_dataframe(df, node_column=None, step_column=None, weather_columns=None, attributes=None, notes=None) classmethod

Create from a DataFrame containing all weather variables as columns.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame with node, step, and weather variable columns.	required
node_column	str	Column name for node IDs.	None
step_column	str	Column name for time steps.	None
weather_columns	dict[WeatherVariable, str]	{WeatherVariable: column_name} mapping.	None
attributes	WeatherAttributes	Optional metadata attributes.	None
notes	str	Free-text note stored in the weather file metadata. Use this to record where the original data came from and how it was processed.	None

Source code in emodpy_malaria/weather/weather_set.py

@classmethod
def from_dataframe(cls, df: pd.DataFrame,
                   node_column: str = None,
                   step_column: str = None,
                   weather_columns: dict[WeatherVariable, str] = None,
                   attributes: WeatherAttributes = None,
                   notes: str = None) -> "WeatherSet":
    """Create from a DataFrame containing all weather variables as columns.

    Args:
        df (pd.DataFrame): DataFrame with node, step, and weather variable columns.
        node_column (str): Column name for node IDs.
        step_column (str): Column name for time steps.
        weather_columns (dict[WeatherVariable, str]): ``{WeatherVariable: column_name}`` mapping.
        attributes (WeatherAttributes): Optional metadata attributes.
        notes (str): Free-text note stored in the weather file metadata.
            Use this to record where the original data came from and
            how it was processed.
    """
    if not isinstance(df, pd.DataFrame):
        raise TypeError(f"Expected DataFrame, got {type(df)}.")
    return cls._from_csv_data(
        data_csv=df, node_column=node_column, step_column=step_column,
        weather_columns=weather_columns, attributes=attributes, notes=notes,
    )

from_files(dir_path, prefix='', file_names=None) classmethod

Load from existing .bin / .bin.json file pairs in a directory.

Source code in emodpy_malaria/weather/weather_set.py

@classmethod
def from_files(cls, dir_path: Union[str, Path],
               prefix: str = "",
               file_names: dict[WeatherVariable, str] = None) -> "WeatherSet":
    """Load from existing ``.bin`` / ``.bin.json`` file pairs in a directory."""
    WeatherVariable.validate_types(file_names, [str, Path])
    file_names = file_names or cls.select_weather_files(dir_path=dir_path, prefix=prefix)
    ws = WeatherSet(dir_path=dir_path, file_names=file_names)
    ws._load()
    return ws

make_file_paths(dir_path=None, prefix='', suffix='{}.bin', weather_variables=None, weather_names=None) classmethod

Generate conventional EMOD weather file paths.

Source code in emodpy_malaria/weather/weather_set.py

@classmethod
def make_file_paths(cls,
                    dir_path: Union[str, Path] = None,
                    prefix: str = "",
                    suffix: str = "{}.bin",
                    weather_variables: list[WeatherVariable] = None,
                    weather_names: dict[WeatherVariable, str] = None) -> dict[WeatherVariable, str]:
    """Generate conventional EMOD weather file paths."""
    names = cls._make_file_templates(
        prefix=prefix, suffix=suffix,
        weather_names=weather_names, weather_variables=weather_variables,
    )
    if dir_path is not None:
        names = {v: str(Path(dir_path) / n) for v, n in names.items()}
    return names

select_weather_files(dir_path, prefix='*', suffix='*{}*.bin', weather_variables=None, weather_names=None) classmethod

Find weather files in a directory by name pattern.

Source code in emodpy_malaria/weather/weather_set.py

@classmethod
def select_weather_files(cls, dir_path: Union[str, Path],
                         prefix: str = "*",
                         suffix: str = "*{}*.bin",
                         weather_variables: list[WeatherVariable] = None,
                         weather_names: dict[WeatherVariable, str] = None) -> dict[WeatherVariable, str]:
    """Find weather files in a directory by name pattern."""
    if dir_path is None:
        raise ValueError("Directory path is required.")
    templates = cls._make_file_templates(
        prefix=prefix, suffix=suffix,
        weather_names=weather_names, weather_variables=weather_variables,
    )
    names = {}
    for v, pattern in templates.items():
        files = list(Path(dir_path).glob(pattern))
        if len(files) > 1:
            raise ValueError(f"Multiple files match pattern {pattern!r}")
        if len(files) == 1:
            names[v] = files[0].name
    return names

to_csv(file_path, node_column=None, step_column=None, weather_columns=None)

Export all variables to a single CSV.

Source code in emodpy_malaria/weather/weather_set.py

def to_csv(self, file_path: Union[str, Path],
           node_column: str = None,
           step_column: str = None,
           weather_columns: dict[WeatherVariable, str] = None) -> pd.DataFrame:
    """Export all variables to a single CSV."""
    df = self.to_dataframe(node_column, step_column, weather_columns)
    df.to_csv(file_path, index=False)
    return df

to_dataframe(node_column=None, step_column=None, weather_columns=None)

Export all variables to a single DataFrame.

Source code in emodpy_malaria/weather/weather_set.py

def to_dataframe(self,
                 node_column: str = None,
                 step_column: str = None,
                 weather_columns: dict[WeatherVariable, str] = None) -> pd.DataFrame:
    """Export all variables to a single DataFrame."""
    weather_columns = weather_columns or {v: None for v in self.weather_variables}
    not_available = [v for v in weather_columns if v not in self.weather_variables]
    if not_available:
        raise ValueError(f"Requested unavailable weather variables: {not_available}")

    infos, weather_columns = self._init_dataframe_info_dict(node_column, step_column, weather_columns)
    self._weather_columns = weather_columns
    df = None
    for v in infos:
        df2 = self[v].to_dataframe(infos[v])
        if df is None:
            df = df2
        else:
            df[infos[v].value_column] = df2[infos[v].value_column]
    return df

to_files(dir_path, file_names=None)

Write all .bin / .bin.json file pairs to a directory.

Source code in emodpy_malaria/weather/weather_set.py

def to_files(self, dir_path: Union[str, Path],
             file_names: dict[WeatherVariable, str] = None) -> None:
    """Write all ``.bin`` / ``.bin.json`` file pairs to a directory."""
    file_names = file_names or self.make_file_paths()
    self._dir_path = Path(dir_path)
    self._file_names = file_names
    self._save()

WeatherVariable

Bases: Enum

Weather variables required by EMOD.

Each variable corresponds to a pair of binary (.bin) and metadata (.bin.json) files that EMOD reads when Climate_Model is set to CLIMATE_BY_DATA.

Source code in emodpy_malaria/weather/weather_variable.py

class WeatherVariable(Enum):
    """Weather variables required by EMOD.

    Each variable corresponds to a pair of binary (``.bin``) and metadata
    (``.bin.json``) files that EMOD reads when ``Climate_Model`` is set to
    ``CLIMATE_BY_DATA``.
    """
    AIR_TEMPERATURE = "airtemp"
    RELATIVE_HUMIDITY = "humidity"
    RAINFALL = "rainfall"
    LAND_TEMPERATURE = "landtemp"

    def __hash__(self):
        return hash(self.name + self.value)

    @classmethod
    def list(cls, exclude: "WeatherVariable | list[WeatherVariable] | None" = None) -> list["WeatherVariable"]:
        """Return all weather variables, optionally excluding some.

        Args:
            exclude ('WeatherVariable | list[WeatherVariable] | None'): Variable(s) to exclude from the list.
        """
        exclude = exclude or []
        if isinstance(exclude, WeatherVariable):
            exclude = [exclude]
        if not isinstance(exclude, list):
            raise TypeError("exclude must be a WeatherVariable or list of WeatherVariable.")
        return [v for v in cls if v not in exclude]

    @classmethod
    def validate_types(cls, value_dict: dict["WeatherVariable", object] | None,
                       value_types: type | list[type] | None = None) -> None:
        """Validate that dict keys are WeatherVariable and values match the given type(s)."""
        if value_dict is None:
            return
        if not isinstance(value_dict, dict):
            raise TypeError("Expected a dictionary.")
        for variable, item in value_dict.items():
            if not isinstance(variable, WeatherVariable):
                raise TypeError("Dictionary keys must be WeatherVariable instances.")
            if value_types is not None:
                types = value_types if isinstance(value_types, list) else [value_types]
                if not any((tp is None and item is None) or (tp is not None and isinstance(item, tp))
                           for tp in types):
                    raise TypeError(f"Dictionary values must be of type {types}, got {type(item)}.")

list(exclude=None) classmethod

Return all weather variables, optionally excluding some.

Parameters:

Name	Type	Description	Default
exclude	'WeatherVariable | list[WeatherVariable] | None'	Variable(s) to exclude from the list.	None

Source code in emodpy_malaria/weather/weather_variable.py

@classmethod
def list(cls, exclude: "WeatherVariable | list[WeatherVariable] | None" = None) -> list["WeatherVariable"]:
    """Return all weather variables, optionally excluding some.

    Args:
        exclude ('WeatherVariable | list[WeatherVariable] | None'): Variable(s) to exclude from the list.
    """
    exclude = exclude or []
    if isinstance(exclude, WeatherVariable):
        exclude = [exclude]
    if not isinstance(exclude, list):
        raise TypeError("exclude must be a WeatherVariable or list of WeatherVariable.")
    return [v for v in cls if v not in exclude]

validate_types(value_dict, value_types=None) classmethod

Validate that dict keys are WeatherVariable and values match the given type(s).

Source code in emodpy_malaria/weather/weather_variable.py

@classmethod
def validate_types(cls, value_dict: dict["WeatherVariable", object] | None,
                   value_types: type | list[type] | None = None) -> None:
    """Validate that dict keys are WeatherVariable and values match the given type(s)."""
    if value_dict is None:
        return
    if not isinstance(value_dict, dict):
        raise TypeError("Expected a dictionary.")
    for variable, item in value_dict.items():
        if not isinstance(variable, WeatherVariable):
            raise TypeError("Dictionary keys must be WeatherVariable instances.")
        if value_types is not None:
            types = value_types if isinstance(value_types, list) else [value_types]
            if not any((tp is None and item is None) or (tp is not None and isinstance(item, tp))
                       for tp in types):
                raise TypeError(f"Dictionary values must be of type {types}, got {type(item)}.")

csv_to_weather(csv_data, node_column='nodes', step_column='steps', weather_columns=None, attributes=None, weather_dir=None, weather_file_names=None)

Convert a CSV file or DataFrame to EMOD weather files.

Parameters:

Name	Type	Description	Default
csv_data	Union[str, Path, DataFrame]	Path to CSV file or a pandas DataFrame containing weather data with node, step, and weather variable columns.	required
node_column	str	Column name for node IDs.	'nodes'
step_column	str	Column name for time step indices.	'steps'
weather_columns	dict[WeatherVariable, str]	{WeatherVariable: column_name} mapping. If omitted, default column names from WeatherVariable values are used (airtemp, humidity, rainfall).	None
attributes	WeatherAttributes	Optional metadata attributes for the output files.	None
weather_dir	Union[str, Path]	If specified, write .bin/.bin.json files here.	None
weather_file_names	dict[WeatherVariable, str]	Optional {WeatherVariable: filename} mapping. If omitted, conventional names are generated.	None

Returns:

Type	Description
WeatherSet	WeatherSet containing the parsed weather data.

Source code in emodpy_malaria/weather/__init__.py

def csv_to_weather(csv_data: Union[str, Path, pd.DataFrame],
                   node_column: str = "nodes",
                   step_column: str = "steps",
                   weather_columns: dict[WeatherVariable, str] = None,
                   attributes: WeatherAttributes = None,
                   weather_dir: Union[str, Path] = None,
                   weather_file_names: dict[WeatherVariable, str] = None) -> WeatherSet:
    """Convert a CSV file or DataFrame to EMOD weather files.

    Args:
        csv_data (Union[str, Path, pd.DataFrame]): Path to CSV file or a pandas DataFrame containing weather
            data with node, step, and weather variable columns.
        node_column (str): Column name for node IDs.
        step_column (str): Column name for time step indices.
        weather_columns (dict[WeatherVariable, str]): ``{WeatherVariable: column_name}`` mapping. If
            omitted, default column names from [WeatherVariable](https://emod.idmod.org/emodpy-malaria/autoapi/emodpy_malaria/weather/weather_variable/)
            values are used (``airtemp``, ``humidity``, ``rainfall``).
        attributes (WeatherAttributes): Optional metadata attributes for the output files.
        weather_dir (Union[str, Path]): If specified, write ``.bin``/``.bin.json`` files here.
        weather_file_names (dict[WeatherVariable, str]): Optional ``{WeatherVariable: filename}``
            mapping. If omitted, conventional names are generated.

    Returns:
        [WeatherSet](https://emod.idmod.org/emodpy-malaria/autoapi/emodpy_malaria/weather/weather_set/) containing the parsed weather data.
    """
    if isinstance(csv_data, pd.DataFrame):
        ws = WeatherSet.from_dataframe(
            df=csv_data, node_column=node_column, step_column=step_column,
            weather_columns=weather_columns, attributes=attributes,
        )
    elif isinstance(csv_data, (str, Path)):
        ws = WeatherSet.from_csv(
            file_path=csv_data, node_column=node_column, step_column=step_column,
            weather_columns=weather_columns, attributes=attributes,
        )
    else:
        raise TypeError("csv_data must be a file path or a pandas DataFrame.")

    if weather_dir:
        ws.to_files(dir_path=weather_dir, file_names=weather_file_names)

    return ws

set_climate_by_data(config, *, air_temperature_filename, rainfall_filename, relative_humidity_filename, update_resolution=ClimateUpdateResolution.CLIMATE_UPDATE_DAY, air_temperature_offset=0.0, air_temperature_variance=0.0, rainfall_scale_factor=1.0, enable_rainfall_stochasticity=False, relative_humidity_scale_factor=1.0, relative_humidity_variance=0.0)

Configure CLIMATE_BY_DATA mode to read weather from binary files.

EMOD reads four .bin / .bin.json file pairs for air temperature, land temperature, rainfall, and relative humidity.

Enable_Climate_Stochasticity is set automatically when any variance is non-zero or rainfall stochasticity is enabled.

Parameters:

Name	Type	Description	Default
config	object	The EMOD config object (task.config).	required
air_temperature_filename	str	Path to air temperature .bin file.	required
rainfall_filename	str	Path to rainfall .bin file.	required
relative_humidity_filename	str	Path to relative humidity .bin file.	required
update_resolution	Union[ClimateUpdateResolution, str]	Climate update frequency.	CLIMATE_UPDATE_DAY
air_temperature_offset	float	Additive offset applied to all air temperature values (Celsius).	0.0
air_temperature_variance	float	Standard deviation (Celsius) for Gaussian noise on daily air temperature. If set to 0, relative humidity does not vary from the data. Set this to > 0 to enable stochasticity.	0.0
rainfall_scale_factor	float	Multiplicative factor applied to all rainfall values.	1.0
enable_rainfall_stochasticity	bool	When True, draw daily rainfall from an exponential distribution with mean equal to the data value.	False
relative_humidity_scale_factor	float	Multiplicative factor applied to all relative humidity values.	1.0
relative_humidity_variance	float	Standard deviation (fraction) for Gaussian noise on daily relative humidity. If set to 0, relative humidity does not vary from the data. Set this to > 0 to enable stochasticity.	0.0

Source code in emodpy_malaria/weather/weather_config.py

def set_climate_by_data(config: object, *,
                        air_temperature_filename: str,
                        rainfall_filename: str,
                        relative_humidity_filename: str,
                        update_resolution: Union[ClimateUpdateResolution, str] = (
                            ClimateUpdateResolution.CLIMATE_UPDATE_DAY),
                        air_temperature_offset: float = 0.0,
                        air_temperature_variance: float = 0.0,
                        rainfall_scale_factor: float = 1.0,
                        enable_rainfall_stochasticity: bool = False,
                        relative_humidity_scale_factor: float = 1.0,
                        relative_humidity_variance: float = 0.0):
    """Configure ``CLIMATE_BY_DATA`` mode to read weather from binary files.

    EMOD reads four ``.bin`` / ``.bin.json`` file pairs for air temperature,
    land temperature, rainfall, and relative humidity.

    ``Enable_Climate_Stochasticity`` is set automatically when any
    variance is non-zero or rainfall stochasticity is enabled.

    Args:
        config (object): The EMOD config object (``task.config``).
        air_temperature_filename (str): Path to air temperature ``.bin`` file.
        rainfall_filename (str): Path to rainfall ``.bin`` file.
        relative_humidity_filename (str): Path to relative humidity ``.bin`` file.
        update_resolution (Union[ClimateUpdateResolution, str]): Climate update frequency.
        air_temperature_offset (float): Additive offset applied to all air
            temperature values (Celsius).
        air_temperature_variance (float): Standard deviation (Celsius) for
            Gaussian noise on daily air temperature.  If set to 0, relative humidity
            does not vary from the data. Set this to > 0 to enable stochasticity.
        rainfall_scale_factor (float): Multiplicative factor applied to all
            rainfall values.
        enable_rainfall_stochasticity (bool): When True, draw daily rainfall from an
            exponential distribution with mean equal to the data value.
        relative_humidity_scale_factor (float): Multiplicative factor applied to
            all relative humidity values.
        relative_humidity_variance (float): Standard deviation (fraction) for
            Gaussian noise on daily relative humidity. If set to 0, relative humidity
            does not vary from the data.
            Set this to > 0 to enable stochasticity.
    """
    if not isinstance(update_resolution, ClimateUpdateResolution):
        try:
            update_resolution = ClimateUpdateResolution(update_resolution)
        except ValueError:
            raise ValueError(
                f"Invalid update_resolution {update_resolution!r}. "
                f"Valid options: {list(ClimateUpdateResolution)}"
            )

    stochastic = (air_temperature_variance != 0.0
                  or relative_humidity_variance != 0.0
                  or enable_rainfall_stochasticity)

    config.parameters.Climate_Model = ClimateModel.CLIMATE_BY_DATA
    config.parameters.Climate_Update_Resolution = update_resolution

    config.parameters.Air_Temperature_Filename = str(air_temperature_filename)
    config.parameters.Land_Temperature_Filename = str(air_temperature_filename)
    config.parameters.Rainfall_Filename = str(rainfall_filename)
    config.parameters.Relative_Humidity_Filename = str(relative_humidity_filename)

    config.parameters.Air_Temperature_Offset = air_temperature_offset
    config.parameters.Land_Temperature_Offset = 0.0
    config.parameters.Rainfall_Scale_Factor = rainfall_scale_factor
    config.parameters.Relative_Humidity_Scale_Factor = relative_humidity_scale_factor

    config.parameters.Enable_Climate_Stochasticity = int(stochastic)
    config.parameters.Air_Temperature_Variance = air_temperature_variance
    config.parameters.Relative_Humidity_Variance = relative_humidity_variance
    config.parameters.Enable_Rainfall_Stochasticity = int(enable_rainfall_stochasticity)

    return config

set_climate_constant(config, *, air_temperature=27.0, rainfall=10.0, relative_humidity=0.75, update_resolution=ClimateUpdateResolution.CLIMATE_UPDATE_DAY, air_temperature_variance=0.0, relative_humidity_variance=0.0, enable_rainfall_stochasticity=False)

Configure CLIMATE_CONSTANT mode with user-specified base values.

EMOD uses these constant values (plus optional stochastic noise) for every node on every time step, ignoring any weather files.

Base_Land_Temperature is set equal to air_temperature — EMOD requires it but does not use it for malaria simulations.

Enable_Climate_Stochasticity is set automatically when any variance is non-zero or rainfall stochasticity is enabled.

Parameters:

Name	Type	Description	Default
config	object	The EMOD config object (task.config).	required
air_temperature	float	Base air temperature in Celsius.	27.0
rainfall	float	Base rainfall in mm/update_resolution.	10.0
relative_humidity	float	Base relative humidity (0.0 -- 1.0).	0.75
update_resolution	Union[ClimateUpdateResolution, str]	Climate update frequency.	CLIMATE_UPDATE_DAY
air_temperature_variance	float	Standard deviation (Celsius) for Gaussian noise on daily air temperature.	0.0
relative_humidity_variance	float	Standard deviation (fraction) for Gaussian noise on daily relative humidity.	0.0
enable_rainfall_stochasticity	bool	Draw daily rainfall from an exponential distribution with mean equal to the base value.	False

Source code in emodpy_malaria/weather/weather_config.py

def set_climate_constant(config: object, *,
                         air_temperature: float = 27.0,
                         rainfall: float = 10.0,
                         relative_humidity: float = 0.75,
                         update_resolution: Union[ClimateUpdateResolution, str] = ClimateUpdateResolution.CLIMATE_UPDATE_DAY,
                         air_temperature_variance: float = 0.0,
                         relative_humidity_variance: float = 0.0,
                         enable_rainfall_stochasticity: bool = False):
    """Configure ``CLIMATE_CONSTANT`` mode with user-specified base values.

    EMOD uses these constant values (plus optional stochastic noise) for
    every node on every time step, ignoring any weather files.

    ``Base_Land_Temperature`` is set equal to *air_temperature* — EMOD
    requires it but does not use it for malaria simulations.

    ``Enable_Climate_Stochasticity`` is set automatically when any
    variance is non-zero or rainfall stochasticity is enabled.

    Args:
        config (object): The EMOD config object (``task.config``).
        air_temperature (float): Base air temperature in Celsius.
        rainfall (float): Base rainfall in mm/**update_resolution**.
        relative_humidity (float): Base relative humidity (0.0 -- 1.0).
        update_resolution (Union[ClimateUpdateResolution, str]): Climate update frequency.
        air_temperature_variance (float): Standard deviation (Celsius) for
            Gaussian noise on daily air temperature.
        relative_humidity_variance (float): Standard deviation (fraction) for
            Gaussian noise on daily relative humidity.
        enable_rainfall_stochasticity (bool): Draw daily **rainfall** from an
            exponential distribution with mean equal to the base value.
    """
    if not isinstance(update_resolution, ClimateUpdateResolution):
        try:
            update_resolution = ClimateUpdateResolution(update_resolution)
        except ValueError:
            raise ValueError(
                f"Invalid update_resolution {update_resolution!r}. "
                f"Valid options: {list(ClimateUpdateResolution)}"
            )

    stochastic = (air_temperature_variance != 0.0
                  or relative_humidity_variance != 0.0
                  or enable_rainfall_stochasticity)

    config.parameters.Climate_Model = ClimateModel.CLIMATE_CONSTANT
    config.parameters.Climate_Update_Resolution = update_resolution
    config.parameters.Base_Air_Temperature = air_temperature
    config.parameters.Base_Land_Temperature = air_temperature
    config.parameters.Base_Rainfall = rainfall
    config.parameters.Base_Relative_Humidity = relative_humidity
    config.parameters.Enable_Climate_Stochasticity = int(stochastic)
    config.parameters.Air_Temperature_Variance = air_temperature_variance
    config.parameters.Relative_Humidity_Variance = relative_humidity_variance
    config.parameters.Enable_Rainfall_Stochasticity = int(enable_rainfall_stochasticity)

    return config

weather_to_csv(weather_dir, weather_file_prefix='', weather_file_names=None, csv_file=None, node_column='nodes', step_column='steps', weather_columns=None)

Convert EMOD weather files to a CSV file or DataFrame.

Parameters:

Name	Type	Description	Default
weather_dir	Union[str, Path]	Directory containing .bin/.bin.json file pairs.	required
weather_file_prefix	str	File name prefix for auto-detection (e.g. "namawala_weather").	''
weather_file_names	dict[WeatherVariable, str]	Explicit {WeatherVariable: filename} mapping.	None
csv_file	Union[str, Path]	If specified, write the output DataFrame to this path.	None
node_column	str	Column name for node IDs in the output.	'nodes'
step_column	str	Column name for time step indices in the output.	'steps'
weather_columns	dict[WeatherVariable, str]	{WeatherVariable: column_name} mapping for the output.	None

Returns:

Type	Description
tuple[DataFrame, WeatherAttributes]	Tuple of (DataFrame, WeatherAttributes).

Source code in emodpy_malaria/weather/__init__.py

def weather_to_csv(weather_dir: Union[str, Path],
                   weather_file_prefix: str = "",
                   weather_file_names: dict[WeatherVariable, str] = None,
                   csv_file: Union[str, Path] = None,
                   node_column: str = "nodes",
                   step_column: str = "steps",
                   weather_columns: dict[WeatherVariable, str] = None
                   ) -> tuple[pd.DataFrame, WeatherAttributes]:
    """Convert EMOD weather files to a CSV file or DataFrame.

    Args:
        weather_dir (Union[str, Path]): Directory containing ``.bin``/``.bin.json`` file pairs.
        weather_file_prefix (str): File name prefix for auto-detection (e.g.
            ``"namawala_weather"``).
        weather_file_names (dict[WeatherVariable, str]): Explicit ``{WeatherVariable: filename}`` mapping.
        csv_file (Union[str, Path]): If specified, write the output DataFrame to this path.
        node_column (str): Column name for node IDs in the output.
        step_column (str): Column name for time step indices in the output.
        weather_columns (dict[WeatherVariable, str]): ``{WeatherVariable: column_name}`` mapping for
            the output.

    Returns:
        Tuple of (DataFrame, WeatherAttributes).
    """
    ws = WeatherSet.from_files(dir_path=weather_dir, prefix=weather_file_prefix, file_names=weather_file_names)
    if csv_file:
        df = ws.to_csv(
            file_path=csv_file, node_column=node_column,
            step_column=step_column, weather_columns=weather_columns,
        )
    else:
        df = ws.to_dataframe(
            node_column=node_column, step_column=step_column,
            weather_columns=weather_columns,
        )
    return df, ws.attributes