Coverage for src/efts_io/conventions.py: 30.70%

1"""Naming conventions for the EFTS netCDF file format."""

3from datetime import datetime # noqa: I001

4from typing import Any, Dict, Iterable, List, Optional, Union

6import numpy as np

7import pandas as pd

8import xarray as xr

9from enum import Enum

11# It may be important to import this AFTER xarray...

12import netCDF4 as nc # noqa: N813

14ConvertibleToTimestamp = Union[str, datetime, np.datetime64, pd.Timestamp]

15TYPES_CONVERTIBLE_TO_TIMESTAMP = [str, datetime, np.datetime64, pd.Timestamp]

16"""Definition of a 'type' for type hints.

17"""

20TIME_DIMNAME = "time"

21STATION_DIMNAME = "station"

22ENS_MEMBER_DIMNAME = "ens_member"

23LEAD_TIME_DIMNAME = "lead_time"

24STR_LEN_DIMNAME = "strLen"

26# New names for in-memory representation in an xarray way

27# https://github.com/csiro-hydroinformatics/efts-io/issues/2

28STATION_ID_DIMNAME = "station_id"

29REALISATION_DIMNAME = "realisation"

31# int station_id[station]

32STATION_ID_VARNAME = "station_id"

33# char station_name[str_len,station]

34STATION_NAME_VARNAME = "station_name"

35# float lat[station]

36LAT_VARNAME = "lat"

37# float lon[station]

38LON_VARNAME = "lon"

39# float x[station]

40X_VARNAME = "x"

41# float y[station]

42Y_VARNAME = "y"

43# float area[station]

44AREA_VARNAME = "area"

45# float elevation[station]

46ELEVATION_VARNAME = "elevation"

48conventional_varnames_mandatory = [

49 STATION_DIMNAME,

50 LEAD_TIME_DIMNAME,

51 TIME_DIMNAME,

52 ENS_MEMBER_DIMNAME,

53 STR_LEN_DIMNAME,

54 STATION_ID_VARNAME,

55 STATION_NAME_VARNAME,

56 LAT_VARNAME,

57 LON_VARNAME,

58]

60conventional_varnames_optional = [

61 X_VARNAME,

62 Y_VARNAME,

63 AREA_VARNAME,

64 ELEVATION_VARNAME,

65]

67conventional_varnames = conventional_varnames_mandatory + conventional_varnames_optional

69hydro_varnames = ("rain", "pet", "q", "swe", "tmin", "tmax")

70var_type = ("obs", "sim")

71obs_hydro_varnames = tuple(f"{var}_{var_type[0]}" for var in hydro_varnames)

72sim_hydro_varnames = tuple(f"{var}_{var_type[1]}" for var in hydro_varnames)

73obs_hydro_varnames_qul = tuple(f"{x}_qul" for x in obs_hydro_varnames)

74sim_hydro_varnames_qul = tuple(f"{x}_qul" for x in sim_hydro_varnames)

75known_hydro_varnames = obs_hydro_varnames + sim_hydro_varnames + obs_hydro_varnames_qul + sim_hydro_varnames_qul

77# TODO: perhaps deal with the state variable names. But, is it used in practice?

79TITLE_ATTR_KEY = "title"

80INSTITUTION_ATTR_KEY = "institution"

81SOURCE_ATTR_KEY = "source"

82CATCHMENT_ATTR_KEY = "catchment"

83STF_CONVENTION_VERSION_ATTR_KEY = "STF_convention_version"

84STF_NC_SPEC_ATTR_KEY = "STF_nc_spec"

85COMMENT_ATTR_KEY = "comment"

86HISTORY_ATTR_KEY = "history"

88TIME_STANDARD_ATTR_KEY = "time_standard"

89STANDARD_NAME_ATTR_KEY = "standard_name"

90LONG_NAME_ATTR_KEY = "long_name"

91AXIS_ATTR_KEY = "axis"

92UNITS_ATTR_KEY = "units"

94FILLVALUE_ATTR_KEY = "_FillValue"

95TYPE_ATTR_KEY = "type"

96TYPE_DESCRIPTION_ATTR_KEY = "type_description"

97DAT_TYPE_DESCRIPTION_ATTR_KEY = "dat_type_description"

98DAT_TYPE_ATTR_KEY = "dat_type"

99LOCATION_TYPE_ATTR_KEY = "location_type"

100

101# We use a URL at a specific commit point, to be used as a file attribute.

102# STF_2_0_URL = "https://github.com/csiro-hydroinformatics/efts/blob/d7d43a995fb5e459bcb894e09b7bb89de03e285c/docs/netcdf_for_water_forecasting.md"

103# July 2025, set a new location/commit point:

104STF_2_0_URL = "https://github.com/csiro-hydroinformatics/efts-io/blob/42ee35f0f019e9bad48b94914429476a7e8278dc/docs/netcdf_for_water_forecasting.md"

105

106

107mandatory_global_attributes = [

108 TITLE_ATTR_KEY,

109 INSTITUTION_ATTR_KEY,

110 SOURCE_ATTR_KEY,

111 CATCHMENT_ATTR_KEY,

112 STF_CONVENTION_VERSION_ATTR_KEY,

113 STF_NC_SPEC_ATTR_KEY,

114 COMMENT_ATTR_KEY,

115 HISTORY_ATTR_KEY,

116]

117

118mandatory_netcdf_dimensions = [TIME_DIMNAME, STATION_DIMNAME, LEAD_TIME_DIMNAME, STR_LEN_DIMNAME, ENS_MEMBER_DIMNAME]

119mandatory_xarray_dimensions = [TIME_DIMNAME, STATION_ID_DIMNAME, LEAD_TIME_DIMNAME, REALISATION_DIMNAME]

120

121mandatory_varnames_xr = [

122 TIME_DIMNAME,

123 LEAD_TIME_DIMNAME,

124 STATION_ID_VARNAME,

125 STATION_NAME_VARNAME,

126 REALISATION_DIMNAME,

127 LAT_VARNAME,

128 LON_VARNAME,

129]

130

131

132class AttributesErrorLevel(Enum):

133 """Controls the behavior of variable attribute checking functions."""

134

135 NONE = 1

136 ERROR = 2

137 # WARNING = 3

138

139

140def get_default_dim_order() -> List[str]:

141 """Default order of dimensions in the netCDF file.

142

143 Returns:

144 List[str]: dimension names: [lead_time, stations, ensemble_member, time]

145 """

146 return [

147 LEAD_TIME_DIMNAME,

148 STATION_DIMNAME,

149 ENS_MEMBER_DIMNAME,

150 TIME_DIMNAME,

151 ]

152

153

154def check_index_found(

155 index_id: Optional[int],

156 identifier: str,

157 dimension_id: str,

158) -> None:

159 """Helper function to check that a value (index) was is indeed found in the dimension."""

160 # return isinstance(index_id, np.int64)

161 if index_id is None:

162 raise ValueError(

163 f"identifier '{identifier}' not found in the dimension '{dimension_id}'",

164 )

165

166

167# MdDatasetsType = Union[nc.Dataset, xr.Dataset, xr.DataArray]

168MdDatasetsType = Union[xr.Dataset, xr.DataArray]

169

170

171def _is_nc_dataset(d: Any) -> bool:

172 return isinstance(d, nc.Dataset)

173

174def _is_nc_variable(d: Any) -> bool:

175 return isinstance(d, nc.Variable)

176

177def _is_ncdf4_withattrs(d: Any) -> bool:

178 return _is_nc_dataset(d) or _is_nc_variable(d)

179

180

181def _has_required_dimensions(

182 d: MdDatasetsType,

183 mandatory_dimensions: Iterable[str],

184) -> bool:

185 if _is_nc_dataset(d):

186 return set(d.dimensions.keys()) == set(mandatory_dimensions)

187 import warnings

188

189 with warnings.catch_warnings():

190 warnings.simplefilter(action="ignore", category=FutureWarning)

191 # FutureWarning: The return type of `Dataset.dims` will be changed

192 # to return a set of dimension names in future, in order to be more

193 # consistent with `DataArray.dims`.

194 dims = d.dims

195 # work around legacy discrepancy between data arrays and datasets: list and dict.

196 kk = set([k for k in dims]) # noqa: C403, C416

197 return kk == set(mandatory_dimensions)

198

199

200def has_required_stf2_dimensions(d: MdDatasetsType, mandatory_dimensions: Optional[Iterable[str]] = None) -> bool:

201 """Has the dataset the required dimensions for STF conventions.

202

203 Args:

204 d (MdDatasetsType): data object to check

205

206 Returns:

207 bool: Has it the minimum STF dimentions

208 """

209 mandatory_dimensions = mandatory_dimensions or mandatory_netcdf_dimensions

210 return _has_required_dimensions(d, mandatory_dimensions)

211

212

213def has_required_xarray_dimensions(d: MdDatasetsType) -> bool:

214 """Has the dataset the required dimensions for an in memory xarray representation."""

215 return _has_required_dimensions(d, mandatory_xarray_dimensions)

216

217

218def _has_all_members(tested: Iterable[str], reference: Iterable[str]) -> bool:

219 """Tests whether all the expected members are present in the tested set."""

220 r = set(reference)

221 return set(tested).intersection(r) == r

222

223

224def has_required_global_attributes(d: MdDatasetsType) -> bool:

225 """has_required_global_attributes."""

226 if _is_nc_dataset(d):

227 a = d.ncattrs()

228 tested = set(a)

229 else:

230 a = d.attrs.keys()

231 tested = set(a)

232 return _has_all_members(tested, mandatory_global_attributes)

233

234

235def has_required_variables_xr(d: MdDatasetsType) -> bool:

236 """has_required_variables."""

237 a = d.variables.keys()

238 tested = set(a)

239 # Note: even if xarray, we do not need to check for the 'data_vars' attribute here.

240 # a = d.data_vars.keys()

241 # tested = set(a)

242 return _has_all_members(tested, mandatory_varnames_xr)

243

244

245def has_variable(d: MdDatasetsType, varname: str) -> bool:

246 """has_variable."""

247 a = d.variables.keys()

248 tested = set(a)

249 return varname in tested

250

251

252def check_stf_compliance(file_path: str) -> Dict[str, List[str]]:

253 """Checks the compliance of a netCDF file with the STF convention.

254

255 Args:

256 file_path (str): The path to the netCDF file.

257

258 Returns:

259 Dict[str, List[str]]: A dictionary with keys "INFO", "WARNING", "ERROR" and values as lists of strings describing compliance issues.

260 """

261 try:

262 dataset = nc.Dataset(file_path, mode="r")

263 results = {"INFO": [], "WARNING": [], "ERROR": []}

264

265 # Check for required dimensions

266 required_dims = [TIME_DIMNAME, STATION_DIMNAME, LEAD_TIME_DIMNAME, ENS_MEMBER_DIMNAME, STR_LEN_DIMNAME]

267 available_dims = dataset.dimensions.keys()

268

269 for dim in required_dims:

270 if dim in available_dims:

271 results["INFO"].append(f"Dimension '{dim}' is present.")

272 else:

273 results["ERROR"].append(f"Missing required dimension '{dim}'.")

274

275 # Check global attributes

276 required_global_attributes = [

277 TITLE_ATTR_KEY,

278 INSTITUTION_ATTR_KEY,

279 SOURCE_ATTR_KEY,

280 CATCHMENT_ATTR_KEY,

281 STF_CONVENTION_VERSION_ATTR_KEY,

282 STF_NC_SPEC_ATTR_KEY,

283 COMMENT_ATTR_KEY,

284 HISTORY_ATTR_KEY,

285 ]

286 available_global_attributes = dataset.ncattrs()

287

288 for attr in required_global_attributes:

289 if attr in available_global_attributes:

290 results["INFO"].append(f"Global attribute '{attr}' is present.")

291 else:

292 results["WARNING"].append(f"Missing global attribute '{attr}'.")

293

294 # Check mandatory variables and their attributes

295 mandatory_variables = [

296 TIME_DIMNAME,

297 STATION_ID_VARNAME,

298 STATION_NAME_VARNAME,

299 ENS_MEMBER_DIMNAME,

300 LEAD_TIME_DIMNAME,

301 LAT_VARNAME,

302 LON_VARNAME,

303 ]

304 variable_attributes = {

305 TIME_DIMNAME: [

306 STANDARD_NAME_ATTR_KEY,

307 LONG_NAME_ATTR_KEY,

308 UNITS_ATTR_KEY,

309 TIME_STANDARD_ATTR_KEY,

310 AXIS_ATTR_KEY,

311 ],

312 STATION_ID_VARNAME: [LONG_NAME_ATTR_KEY],

313 STATION_NAME_VARNAME: [LONG_NAME_ATTR_KEY],

314 ENS_MEMBER_DIMNAME: [STANDARD_NAME_ATTR_KEY, LONG_NAME_ATTR_KEY, UNITS_ATTR_KEY, AXIS_ATTR_KEY],

315 LEAD_TIME_DIMNAME: [STANDARD_NAME_ATTR_KEY, LONG_NAME_ATTR_KEY, UNITS_ATTR_KEY, AXIS_ATTR_KEY],

316 LAT_VARNAME: [LONG_NAME_ATTR_KEY, UNITS_ATTR_KEY, AXIS_ATTR_KEY],

317 LON_VARNAME: [LONG_NAME_ATTR_KEY, UNITS_ATTR_KEY, AXIS_ATTR_KEY],

318 }

319

320 for var in mandatory_variables:

321 if var in dataset.variables:

322 results["INFO"].append(f"Mandatory variable '{var}' is present.")

323 # Check attributes

324 for attr, required_attrs in variable_attributes.items():

325 if var == attr:

326 for req_attr in required_attrs:

327 if req_attr in dataset.variables[var].ncattrs():

328 results["INFO"].append(f"Attribute '{req_attr}' for variable '{var}' is present.")

329 else:

330 results["WARNING"].append(

331 f"Missing required attribute '{req_attr}' for variable '{var}'.",

332 )

333 else:

334 results["ERROR"].append(f"Missing mandatory variable '{var}'.")

335

336 dataset.close()

337 return results # noqa: TRY300

338

339 except Exception as e: # noqa: BLE001

340 return {"ERROR": [f"Error opening file '{file_path}': {e!s}"]}

341

342

343def _is_structural_varname(name: str) -> bool:

344 return name in conventional_varnames

345

346

347def _is_known_hydro_varname(name: str) -> bool:

348 """Checks if the variable name is a known hydrologic variable."""

349 # TODO: perhaps deal with state variable conventional names.

350 return name in known_hydro_varnames

351

352

353def _is_observation_variable(name: str) -> bool:

354 return name in obs_hydro_varnames

355

356

357def _is_simulation_variable(name: str) -> bool:

358 return name in sim_hydro_varnames

359

360

361def _is_quality_variable(name: str) -> bool:

362 return name in obs_hydro_varnames_qul or name in sim_hydro_varnames_qul

363

364

365def _extract_var_type(variable: Any) -> str:

366 if _is_observation_variable(variable):

367 return "obs"

368 if _is_simulation_variable(variable):

369 return "sim"

370 if _is_quality_variable(variable):

371 return "qul"

372 return None

373

374

375def _check_variable_attributes_obs(

376 variable: Any,

377 error_threshold: AttributesErrorLevel = AttributesErrorLevel.NONE,

378) -> List[str]:

379 """Checks if the attributes of the observed variable comply with the conventions."""

380 missing_attributes_messages = []

381 required_attributes = {

382 LONG_NAME_ATTR_KEY: str,

383 UNITS_ATTR_KEY: str,

384 FILLVALUE_ATTR_KEY: float,

385 TYPE_ATTR_KEY: int,

386 TYPE_DESCRIPTION_ATTR_KEY: str,

387 DAT_TYPE_ATTR_KEY: str,

388 LOCATION_TYPE_ATTR_KEY: str,

389 }

390 return _check_attrs(variable, required_attributes, missing_attributes_messages, error_threshold=error_threshold)

391

392

393def _check_variable_attributes_sim(

394 variable: Any,

395 error_threshold: AttributesErrorLevel = AttributesErrorLevel.NONE,

396) -> List[str]:

397 """Checks if the attributes of the simulated variable comply with the conventions."""

398 missing_attributes_messages = []

399 required_attributes = {

400 LONG_NAME_ATTR_KEY: str,

401 UNITS_ATTR_KEY: str,

402 FILLVALUE_ATTR_KEY: float,

403 TYPE_ATTR_KEY: int,

404 TYPE_DESCRIPTION_ATTR_KEY: str,

405 DAT_TYPE_ATTR_KEY: str,

406 LOCATION_TYPE_ATTR_KEY: str,

407 }

408 return _check_attrs(variable, required_attributes, missing_attributes_messages, error_threshold=error_threshold)

409

410

411def _check_variable_attributes_qul(

412 variable: Any,

413 error_threshold: AttributesErrorLevel = AttributesErrorLevel.NONE,

414) -> List[str]:

415 """Checks if the attributes of the data quality code variable comply with the conventions."""

416 missing_attributes_messages = []

417 required_attributes = {

418 LONG_NAME_ATTR_KEY: str,

419 UNITS_ATTR_KEY: str,

420 FILLVALUE_ATTR_KEY: int,

421 LOCATION_TYPE_ATTR_KEY: str,

422 TYPE_DESCRIPTION_ATTR_KEY: str,

423 DAT_TYPE_ATTR_KEY: str,

424 }

425 return _check_attrs(variable, required_attributes, missing_attributes_messages, error_threshold=error_threshold)

426

427def _check_attrs_ncdataset(

428 variable: Any,

429 required_attributes: Dict[str, type],

430 missing_attributes_messages: List[str],

431 error_threshold: AttributesErrorLevel = AttributesErrorLevel.NONE,

432) -> List[str]:

433 for attr, attr_type in required_attributes.items():

434 if attr not in variable.ncattrs():

435 missing_attributes_messages.append(f"Missing required attribute '{attr}' for variable '{variable.name}'.")

436 else:

437 actual_type = type(variable.getncattr(attr))

438 if actual_type != attr_type:

439 missing_attributes_messages.append(

440 f"Attribute '{attr}' for variable '{variable.name}' has an unexpected type '{actual_type.__name__}'. Expected type: '{attr_type.__name__}'.",

441 )

442 if error_threshold == AttributesErrorLevel.ERROR and missing_attributes_messages:

443 raise ValueError(

444 f"Variable '{variable.name}' has missing or incorrect attributes: {missing_attributes_messages}",

445 )

446 return missing_attributes_messages

447

448

449def _check_attrs_xr(

450 variable: MdDatasetsType,

451 required_attributes: Dict[str, type],

452 missing_attributes_messages: List[str],

453 error_threshold: AttributesErrorLevel = AttributesErrorLevel.NONE,

454) -> List[str]:

455 for attr, attr_type in required_attributes.items():

456 if attr not in variable.attrs:

457 missing_attributes_messages.append(f"Missing required attribute '{attr}' for variable '{variable.name}'.")

458 else:

459 actual_type = type(variable.attrs[attr])

460 if actual_type != attr_type:

461 missing_attributes_messages.append(

462 f"Attribute '{attr}' for variable '{variable.name}' has an unexpected type '{actual_type.__name__}'. Expected type: '{attr_type.__name__}'.",

463 )

464 if error_threshold == AttributesErrorLevel.ERROR and missing_attributes_messages:

465 raise ValueError(

466 f"Variable '{variable.name}' has missing or incorrect attributes: {missing_attributes_messages}",

467 )

468 return missing_attributes_messages

469

470

471def _check_attrs(

472 variable: Any,

473 required_attributes: Dict[str, type],

474 missing_attributes_messages: List[str],

475 error_threshold: AttributesErrorLevel = AttributesErrorLevel.NONE,

476) -> List[str]:

477 if _is_ncdf4_withattrs(variable):

478 return _check_attrs_ncdataset(variable, required_attributes, missing_attributes_messages, error_threshold)

479 else: # noqa: RET505

480 return _check_attrs_xr(variable, required_attributes, missing_attributes_messages, error_threshold)

481

482

483def _check_variable_attributes(variable: Any) -> List[str]:

484 """Checks if the attributes of a variable comply with the conventions depending on the type of variable.

485

486 Args:

487 variable (Any): The netCDF variable whose attributes are to be checked.

488

489 Returns:

490 List[str]: A list of messages describing any missing attributes.

491 """

492 var_type = _extract_var_type(variable.name)

493

494 if var_type == "obs":

495 return _check_variable_attributes_obs(variable)

496 if var_type == "sim":

497 return _check_variable_attributes_sim(variable)

498 if var_type == "qul":

499 return _check_variable_attributes_qul(variable)

500

501 return []

502

503

504def check_hydrologic_variables(file_path: str) -> Dict[str, List[str]]:

505 """Checks if the variable names and attributes in a netCDF file comply with the STF convention.

506

507 Args:

508 file_path (str): The path to the netCDF file.

509

510 Returns:

511 Dict[str, List[str]]: A dictionary with keys "INFO", "WARNING", "ERROR" and values as lists of strings describing compliance issues.

512 """

513 try:

514 dataset = None

515 dataset = nc.Dataset(file_path, mode="r")

516 results = {"INFO": [], "WARNING": [], "ERROR": []}

517

518 for var in dataset.variables:

519 if _is_structural_varname(var):

520 continue

521 if _is_known_hydro_varname(var):

522 results["INFO"].append(f"Hydrologic variable '{var}' follows the recommended naming convention.")

523

524 # Check attributes

525 for msg in _check_variable_attributes(dataset.variables[var]):

526 results["WARNING"].append(msg)

527 else:

528 results["WARNING"].append(

529 f"Hydrologic variable '{var}' does not follow the recommended naming convention.",

530 )

531

532 return results # noqa: TRY300

533

534 except Exception as e: # noqa: BLE001

535 return {"ERROR": [f"Error opening or reading file '{file_path}': {e!s}"]}

536

537 finally:

538 if dataset:

539 dataset.close()

540

541

542def check_optional_variable_attributes(

543 variable: Any,

544 error_threshold: AttributesErrorLevel = AttributesErrorLevel.NONE,

545) -> List[str]:

546 """Checks if the attributes of the observed variable comply with the conventions."""

547 missing_attributes_messages = []

548 required_attributes = {

549 STANDARD_NAME_ATTR_KEY: str,

550 LONG_NAME_ATTR_KEY: str,

551 UNITS_ATTR_KEY: str,

552 }

553 return _check_attrs(variable, required_attributes, missing_attributes_messages, error_threshold=error_threshold)

554

555

556def convert_to_datetime64_utc(x: ConvertibleToTimestamp) -> np.datetime64:

557 """Converts a known timestamp representation an np.datetime64."""

558 if isinstance(x, pd.Timestamp):

559 if x.tz is None:

560 x = x.tz_localize("UTC")

561 x = x.tz_convert("UTC")

562 elif isinstance(x, datetime):

563 x = pd.Timestamp(x, tz="UTC") if x.tzinfo is None else pd.Timestamp(x).tz_convert("UTC")

564 elif isinstance(x, str):

565 x_dt = pd.to_datetime(x)

566 x = pd.Timestamp(x_dt, tz="UTC") if x_dt.tzinfo is None else pd.Timestamp(x_dt).tz_convert("UTC")

567 elif isinstance(x, np.datetime64):

568 x = pd.Timestamp(x).tz_localize("UTC")

569 else:

570 raise TypeError(f"Cannot convert {type(x)} to np.datetime64 with UTC timezone.")

571

572 return x.to_datetime64()

573

574def exportable_to_stf2(data:MdDatasetsType) -> bool:

575 """Check if the dataset can be written to a netCDF file compliant with STF 2.0 specification.

576

577 This method checks if the underlying xarray dataset or dataarray has the required dimensions and global attributes as specified by the STF 2.0 convention.

578

579 Returns:

580 bool: True if the dataset can be written to a STF 2.0 compliant netCDF file, False otherwise.

581 """

582 from efts_io.conventions import has_required_stf2_dimensions, has_required_global_attributes, has_required_variables_xr, mandatory_xarray_dimensions # noqa: I001

583 required_stf2_dimensions = has_required_stf2_dimensions(data, mandatory_xarray_dimensions)

584 required_attributes = has_required_global_attributes(data)

585 required_variables = has_required_variables_xr(data)

586

587 return required_stf2_dimensions and required_attributes and required_variables

588