io

I/O helpers for weather data and parameter files.

load_weather_csv(path, columns=None, dtype=torch.float32)

Load a single-site daily weather CSV.

The CSV must contain the columns listed in WEATHER_CHANNELS (lowercase) or an explicit columns argument may be passed mapping file columns to the expected order.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the CSV file. The first line must be a header row.	required
columns	Sequence[str] | None	Optional override for the expected column names in the CSV, in the order used by WEATHER_CHANNELS. If None, WEATHER_CHANNELS is used verbatim.	None
dtype	torch.dtype	Floating-point dtype for the returned tensor.	torch.float32

Returns:

Type	Description
WeatherDriver	WeatherDriver of shape [1, T, C] where T is the number of data rows in the CSV.

Exceptions:

Type	Description
ValueError	If columns has a different length than WEATHER_CHANNELS.

Source code in torchcrop/utils/io.py

def load_weather_csv(
    path: str | Path,
    columns: Sequence[str] | None = None,
    dtype: torch.dtype = torch.float32,
) -> WeatherDriver:
    """Load a single-site daily weather CSV.

    The CSV must contain the columns listed in
    `WEATHER_CHANNELS` (lowercase) or an
    explicit ``columns`` argument may be passed mapping file columns to the
    expected order.

    Args:
        path: Path to the CSV file. The first line must be a header row.
        columns: Optional override for the expected column names in the
            CSV, in the order used by
            `WEATHER_CHANNELS`. If
            ``None``, ``WEATHER_CHANNELS`` is used verbatim.
        dtype: Floating-point dtype for the returned tensor.

    Returns:
        `WeatherDriver` of shape
        ``[1, T, C]`` where ``T`` is the number of data rows in the CSV.

    Raises:
        ValueError: If ``columns`` has a different length than
            `WEATHER_CHANNELS`.
    """
    path = Path(path)
    if columns is None:
        columns = WEATHER_CHANNELS
    if len(columns) != len(WEATHER_CHANNELS):
        raise ValueError(
            f"expected {len(WEATHER_CHANNELS)} columns, got {len(columns)}"
        )

    with path.open() as fh:
        header = fh.readline().strip().split(",")
    idx = [header.index(c) for c in columns]
    raw = np.loadtxt(path, delimiter=",", skiprows=1, usecols=idx)
    if raw.ndim == 1:
        raw = raw[None, :]
    data = torch.tensor(raw, dtype=dtype).unsqueeze(0)  # [1, T, C]
    return WeatherDriver(data)

make_constant_weather(batch_size, n_days, davtmp=15.0, tmin=10.0, tmax=20.0, irrad=15.0, rain=2.0, vp=1.2, wind=2.0, start_doy=1, dtype=torch.float32, device='cpu')

Build a synthetic constant-weather driver for quick tests and demos.

Parameters:

Name	Type	Description	Default
batch_size	int	Number of batch elements B in the output driver.	required
n_days	int	Simulation length T in days.	required
davtmp	float	Constant mean daily temperature [°C].	15.0
tmin	float	Constant minimum daily temperature [°C].	10.0
tmax	float	Constant maximum daily temperature [°C].	20.0
irrad	float	Constant solar radiation [MJ m⁻² d⁻¹].	15.0
rain	float	Constant daily precipitation [mm d⁻¹].	2.0
vp	float	Constant vapour pressure [kPa].	1.2
wind	float	Constant wind speed [m s⁻¹].	2.0
start_doy	int	Day of year for the first simulated day; doy increments by 1 and wraps modulo 365.	1
dtype	torch.dtype	Floating-point dtype for all tensors.	torch.float32
device	torch.device | str	Device on which to allocate the tensors.	'cpu'

Returns:

Type	Description
WeatherDriver	WeatherDriver of shape [B, T, C] with identical rows for every batch element.

Source code in torchcrop/utils/io.py

def make_constant_weather(
    batch_size: int,
    n_days: int,
    davtmp: float = 15.0,
    tmin: float = 10.0,
    tmax: float = 20.0,
    irrad: float = 15.0,
    rain: float = 2.0,
    vp: float = 1.2,
    wind: float = 2.0,
    start_doy: int = 1,
    dtype: torch.dtype = torch.float32,
    device: torch.device | str = "cpu",
) -> WeatherDriver:
    """Build a synthetic constant-weather driver for quick tests and demos.

    Args:
        batch_size: Number of batch elements ``B`` in the output driver.
        n_days: Simulation length ``T`` in days.
        davtmp: Constant mean daily temperature [°C].
        tmin: Constant minimum daily temperature [°C].
        tmax: Constant maximum daily temperature [°C].
        irrad: Constant solar radiation [MJ m⁻² d⁻¹].
        rain: Constant daily precipitation [mm d⁻¹].
        vp: Constant vapour pressure [kPa].
        wind: Constant wind speed [m s⁻¹].
        start_doy: Day of year for the first simulated day; ``doy``
            increments by 1 and wraps modulo 365.
        dtype: Floating-point dtype for all tensors.
        device: Device on which to allocate the tensors.

    Returns:
        `WeatherDriver` of shape
        ``[B, T, C]`` with identical rows for every batch element.
    """
    doy = torch.arange(n_days, dtype=dtype, device=device) + float(start_doy)
    doy = ((doy - 1) % 365) + 1
    channels = {
        "doy": doy,
        "davtmp": torch.full((n_days,), davtmp, dtype=dtype, device=device),
        "tmin": torch.full((n_days,), tmin, dtype=dtype, device=device),
        "tmax": torch.full((n_days,), tmax, dtype=dtype, device=device),
        "irrad": torch.full((n_days,), irrad, dtype=dtype, device=device),
        "rain": torch.full((n_days,), rain, dtype=dtype, device=device),
        "vp": torch.full((n_days,), vp, dtype=dtype, device=device),
        "wind": torch.full((n_days,), wind, dtype=dtype, device=device),
    }
    stacked = torch.stack([channels[name] for name in WEATHER_CHANNELS], dim=-1)
    data = stacked.unsqueeze(0).expand(batch_size, -1, -1).contiguous()
    return WeatherDriver(data)