|
3 | 3 | from __future__ import annotations |
4 | 4 |
|
5 | 5 | import textwrap |
6 | | -from typing import TYPE_CHECKING, Any, cast |
| 6 | +from dataclasses import dataclass |
| 7 | +from typing import TYPE_CHECKING, Any, Literal, cast |
7 | 8 |
|
8 | 9 | import numpy as np |
9 | 10 | import xarray as xr |
10 | 11 |
|
| 12 | +from parcels._python import isinstance_noimport |
| 13 | + |
11 | 14 | if TYPE_CHECKING: |
12 | 15 | from parcels import Field, FieldSet, ParticleSet |
13 | 16 | from parcels._core.field import VectorField |
| 17 | + from parcels._core.model import ModelData |
| 18 | + from parcels._core.utils.time import TimeInterval |
14 | 19 |
|
15 | 20 |
|
16 | 21 | def fieldset_repr(fieldset: FieldSet) -> str: |
@@ -177,3 +182,81 @@ def _format_list_items_multiline(items: list[str] | dict, level: int = 1, with_b |
177 | 182 |
|
178 | 183 | def is_builtin_object(obj): |
179 | 184 | return obj.__class__.__module__ == "builtins" |
| 185 | + |
| 186 | + |
| 187 | +@dataclass |
| 188 | +class _FieldSetDescriptionRow: |
| 189 | + type_: Literal["Field", "VectorField", "Context"] |
| 190 | + model_id: int | None |
| 191 | + name: str |
| 192 | + interp_method_or_value: str |
| 193 | + |
| 194 | + def to_dict(self) -> dict[str, str]: |
| 195 | + return { |
| 196 | + "Name": self.name, |
| 197 | + "Type": self.type_, |
| 198 | + "Grid number": str(self.model_id) if self.model_id is not None else "-", |
| 199 | + "Interp method / value": self.interp_method_or_value, |
| 200 | + } |
| 201 | + |
| 202 | + |
| 203 | +def _print_table(rows: list[_FieldSetDescriptionRow]) -> str: |
| 204 | + import pandas as pd |
| 205 | + |
| 206 | + dicts = [r.to_dict() for r in rows] |
| 207 | + return pd.DataFrame(dicts).sort_values(["Grid number", "Type", "Name"]).to_markdown(index=False) |
| 208 | + |
| 209 | + |
| 210 | +def _print_time_interval(time_interval: TimeInterval | None) -> str: |
| 211 | + if time_interval is None: |
| 212 | + return repr(time_interval) |
| 213 | + return repr((time_interval.left, time_interval.right)) |
| 214 | + |
| 215 | + |
| 216 | +def fieldset_describe(fieldset: FieldSet) -> str: |
| 217 | + rows: list[_FieldSetDescriptionRow] = [] |
| 218 | + models: dict[int, int] = {} # mapping of memory ID to a human readable ID |
| 219 | + |
| 220 | + assert fieldset._fields is not None |
| 221 | + |
| 222 | + for field in fieldset._fields.values(): |
| 223 | + model_id: int |
| 224 | + |
| 225 | + # Set human readable model ID |
| 226 | + parent_id = id(_get_parent_model(field)) |
| 227 | + models[parent_id] = models.get(parent_id, len(models)) |
| 228 | + model_id = models[parent_id] |
| 229 | + |
| 230 | + type_ = cast(Literal["Field", "VectorField", "Context"], field.__class__.__name__) |
| 231 | + |
| 232 | + rows.append( |
| 233 | + _FieldSetDescriptionRow( |
| 234 | + type_=type_, |
| 235 | + model_id=model_id, |
| 236 | + name=field.name, |
| 237 | + interp_method_or_value=repr(field.interp_method), |
| 238 | + ) |
| 239 | + ) |
| 240 | + for k, v in fieldset.context.items(): |
| 241 | + rows.append( |
| 242 | + _FieldSetDescriptionRow( |
| 243 | + type_="Context", |
| 244 | + model_id=None, |
| 245 | + name=k, |
| 246 | + interp_method_or_value=repr(v), |
| 247 | + ) |
| 248 | + ) |
| 249 | + return ( |
| 250 | + _print_table(rows) |
| 251 | + + f"""\ |
| 252 | +
|
| 253 | +
|
| 254 | +mesh: {fieldset.models[0].grid._mesh} |
| 255 | +time interval: {_print_time_interval(fieldset.time_interval)}""" |
| 256 | + ) |
| 257 | + |
| 258 | + |
| 259 | +def _get_parent_model(field: Field | VectorField) -> ModelData: |
| 260 | + if isinstance_noimport(field, "Field"): |
| 261 | + return field.model # type:ignore[union-attr] |
| 262 | + return field.U.model # type:ignore[union-attr] |
0 commit comments