Skip to content

climate_ref.models #

Declaration of the models used by the REF.

These models are used to represent the data that is stored in the database.

Base #

Bases: DeclarativeBase

Base class for all models

Source code in packages/climate-ref/src/climate_ref/models/base.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class Base(DeclarativeBase):
    """
    Base class for all models
    """

    type_annotation_map = {  # noqa: RUF012
        dict[str, Any]: JSON,
        list[float | int]: JSON,
        list[float | int | str]: JSON,
    }
    metadata = MetaData(
        # Enforce a common naming convention for constraints
        # https://alembic.sqlalchemy.org/en/latest/naming.html
        naming_convention={
            "ix": "ix_%(column_0_label)s",
            "uq": "uq_%(table_name)s_%(column_0_name)s",
            "ck": "ck_%(table_name)s_`%(constraint_name)s`",
            "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
            "pk": "pk_%(table_name)s",
        }
    )

Dataset #

Bases: Base

Represents a dataset

A dataset is a collection of data files, that is used as an input to the benchmarking process. Adding/removing or updating a dataset will trigger a new diagnostic calculation.

A polymorphic association is used to capture the different types of datasets as each dataset type may have different metadata fields. This enables the use of a single table to store all datasets, but still allows for querying specific metadata fields for each dataset type.

Source code in packages/climate-ref/src/climate_ref/models/dataset.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
class Dataset(Base):
    """
    Represents a dataset

    A dataset is a collection of data files, that is used as an input to the benchmarking process.
    Adding/removing or updating a dataset will trigger a new diagnostic calculation.

    A polymorphic association is used to capture the different types of datasets as each
    dataset type may have different metadata fields.
    This enables the use of a single table to store all datasets,
    but still allows for querying specific metadata fields for each dataset type.
    """

    __tablename__ = "dataset"

    id: Mapped[int] = mapped_column(primary_key=True)
    slug: Mapped[str] = mapped_column(unique=True)
    """
    Globally unique identifier for the dataset.

    In the case of CMIP6 datasets, this is the instance_id.
    """
    dataset_type: Mapped[SourceDatasetType] = mapped_column(nullable=False, index=True)
    """
    Type of dataset
    """
    created_at: Mapped[datetime.datetime] = mapped_column(server_default=func.now())
    """
    When the dataset was added to the database
    """
    updated_at: Mapped[datetime.datetime] = mapped_column(server_default=func.now(), onupdate=func.now())
    """
    When the dataset was updated.

    Updating a dataset will trigger a new diagnostic calculation.
    """

    # Universal finalisation flag for all dataset types
    # Only CMIP6 currently uses unfinalised datasets in practice; other types should be finalised on creation.
    finalised: Mapped[bool] = mapped_column(default=True, nullable=False)
    """
    Whether the complete set of metadata for the dataset has been finalised.

    For CMIP6, ingestion may initially create unfinalised datasets (False) until all metadata is extracted.
    For other dataset types (e.g., obs4MIPs, PMP climatology), this should be True upon creation.
    """

    version_key: Mapped[int] = mapped_column(BigInteger, default=-1, server_default="-1", nullable=False)
    """
    Numeric ordering key for the subclass's ``version`` column,
    computed by :func:`climate_ref_core.datasets.version_sort_key`.

    Kept in sync by the ``_sync_version_key`` mapper event.
    Rows with no ``version`` attribute (base-table-only inserts) keep the ``-1`` backstop.

    Lives on the base table (not a subclass) so the SQL latest-version window function
    (``select_datasets(..., latest_group_by=...)``) can read it off any polymorphic row while
    partitioning on the subclass's ``dataset_id_metadata`` columns.

    Core writes to ``version``
    (``session.execute(update(...))``, ``connection.execute(...)``, ``bulk_update_mappings``)
    bypass the event and leave ``version_key`` stale, silently corrupting latest-version dedup.
    ALWAYS mutate ``version`` through an ORM instance.
    """

    def __repr__(self) -> str:
        return f"<Dataset slug={self.slug} dataset_type={self.dataset_type} >"

    __mapper_args__: ClassVar[Any] = {"polymorphic_on": dataset_type}  # type: ignore

created_at = mapped_column(server_default=(func.now())) class-attribute instance-attribute #

When the dataset was added to the database

dataset_type = mapped_column(nullable=False, index=True) class-attribute instance-attribute #

Type of dataset

finalised = mapped_column(default=True, nullable=False) class-attribute instance-attribute #

Whether the complete set of metadata for the dataset has been finalised.

For CMIP6, ingestion may initially create unfinalised datasets (False) until all metadata is extracted. For other dataset types (e.g., obs4MIPs, PMP climatology), this should be True upon creation.

slug = mapped_column(unique=True) class-attribute instance-attribute #

Globally unique identifier for the dataset.

In the case of CMIP6 datasets, this is the instance_id.

updated_at = mapped_column(server_default=(func.now()), onupdate=(func.now())) class-attribute instance-attribute #

When the dataset was updated.

Updating a dataset will trigger a new diagnostic calculation.

version_key = mapped_column(BigInteger, default=(-1), server_default='-1', nullable=False) class-attribute instance-attribute #

Numeric ordering key for the subclass's version column, computed by :func:climate_ref_core.datasets.version_sort_key.

Kept in sync by the _sync_version_key mapper event. Rows with no version attribute (base-table-only inserts) keep the -1 backstop.

Lives on the base table (not a subclass) so the SQL latest-version window function (select_datasets(..., latest_group_by=...)) can read it off any polymorphic row while partitioning on the subclass's dataset_id_metadata columns.

Core writes to version (session.execute(update(...)), connection.execute(...), bulk_update_mappings) bypass the event and leave version_key stale, silently corrupting latest-version dedup. ALWAYS mutate version through an ORM instance.

Diagnostic #

Bases: CreatedUpdatedMixin, Base

Represents a diagnostic that can be calculated

Source code in packages/climate-ref/src/climate_ref/models/diagnostic.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class Diagnostic(CreatedUpdatedMixin, Base):
    """
    Represents a diagnostic that can be calculated
    """

    __tablename__ = "diagnostic"
    __table_args__ = (UniqueConstraint("provider_id", "slug", name="diagnostic_ident"),)

    id: Mapped[int] = mapped_column(primary_key=True)
    slug: Mapped[str] = mapped_column()
    """
    Unique identifier for the diagnostic

    This will be used to reference the diagnostic in the benchmarking process
    """

    name: Mapped[str] = mapped_column()
    """
    Long name of the diagnostic
    """

    provider_id: Mapped[int] = mapped_column(ForeignKey("provider.id"))
    """
    The provider that provides the diagnostic
    """

    enabled: Mapped[bool] = mapped_column(default=True)
    """
    Whether the diagnostic is enabled or not

    If a diagnostic is not enabled, it will not be used for any calculations.
    """

    promoted_version: Mapped[int] = mapped_column(default=1, server_default="1")
    """
    Currently promoted diagnostic version for default queries.

    Default query helpers filter ``ExecutionGroup.diagnostic_version == Diagnostic.promoted_version``
    so consumers see exactly one version's worth of results.
    Recomputed as ``max(ExecutionGroup.diagnostic_version)`` after a new group is inserted
    (see ``recompute_promoted_version``).
    """

    provider: Mapped["Provider"] = relationship(back_populates="diagnostics")
    execution_groups: Mapped[list["ExecutionGroup"]] = relationship(back_populates="diagnostic")

    def __repr__(self) -> str:
        return f"<Metric slug={self.slug}>"

    def full_slug(self) -> str:
        """
        Get the full slug of the diagnostic, including the provider slug

        Returns
        -------
        str
            Full slug of the diagnostic
        """
        return f"{self.provider.slug}/{self.slug}"

enabled = mapped_column(default=True) class-attribute instance-attribute #

Whether the diagnostic is enabled or not

If a diagnostic is not enabled, it will not be used for any calculations.

name = mapped_column() class-attribute instance-attribute #

Long name of the diagnostic

promoted_version = mapped_column(default=1, server_default='1') class-attribute instance-attribute #

Currently promoted diagnostic version for default queries.

Default query helpers filter ExecutionGroup.diagnostic_version == Diagnostic.promoted_version so consumers see exactly one version's worth of results. Recomputed as max(ExecutionGroup.diagnostic_version) after a new group is inserted (see recompute_promoted_version).

provider_id = mapped_column(ForeignKey('provider.id')) class-attribute instance-attribute #

The provider that provides the diagnostic

slug = mapped_column() class-attribute instance-attribute #

Unique identifier for the diagnostic

This will be used to reference the diagnostic in the benchmarking process

full_slug() #

Get the full slug of the diagnostic, including the provider slug

Returns:

Type Description
str

Full slug of the diagnostic
Source code in packages/climate-ref/src/climate_ref/models/diagnostic.py 
63
64
65
66
67
68
69
70
71
72
def full_slug(self) -> str:
    """
    Get the full slug of the diagnostic, including the provider slug

    Returns
    -------
    str
        Full slug of the diagnostic
    """
    return f"{self.provider.slug}/{self.slug}"

Execution #

Bases: CreatedUpdatedMixin, Base

Represents a single execution of a diagnostic

Each result is part of a group of executions that share similar input datasets.

An execution group might be run multiple times as new data becomes available, each run will create a Execution.

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
class Execution(CreatedUpdatedMixin, Base):
    """
    Represents a single execution of a diagnostic

    Each result is part of a group of executions that share similar input datasets.

    An execution group might be run multiple times as new data becomes available,
    each run will create a `Execution`.
    """

    __tablename__ = "execution"

    id: Mapped[int] = mapped_column(primary_key=True)

    output_fragment: Mapped[str] = mapped_column()
    """
    Relative directory to store the output of the execution.

    During execution this directory is relative to the temporary directory.
    If the diagnostic execution is successful, the executions will be moved to the final output directory
    and the temporary directory will be cleaned up.
    This directory may contain multiple input and output files.
    """

    execution_group_id: Mapped[int] = mapped_column(
        ForeignKey(
            "execution_group.id",
            name="fk_execution_id",
        ),
        index=True,
    )
    """
    The execution group that this execution belongs to
    """

    dataset_hash: Mapped[str] = mapped_column(index=True)
    """
    Hash of the datasets used to calculate the diagnostic

    This is used to verify if an existing diagnostic execution has been run with the same datasets.
    """

    successful: Mapped[bool | None] = mapped_column(nullable=True, index=True)
    """
    Was the run successful
    """

    path: Mapped[str] = mapped_column(nullable=True)
    """
    Path to the output bundle

    Relative to the diagnostic execution result output directory
    """

    retracted: Mapped[bool] = mapped_column(default=False)
    """
    Whether the diagnostic execution result has been retracted or not

    This may happen if a dataset has been retracted, or if the diagnostic execution was incorrect.
    Rather than delete the values, they are marked as retracted.
    These data may still be visible in the UI, but should be marked as retracted.
    """

    provider_version: Mapped[str | None] = mapped_column(nullable=True)
    """
    Provider version recorded by the worker at run time.

    Snapshot of the worker-installed ``provider.version`` when the execution ran.
    Purely informational for audit; not used for validation or recomputation triggers.
    Rows that predate the column stay NULL.
    """

    execution_group: Mapped["ExecutionGroup"] = relationship(back_populates="executions")
    outputs: Mapped[list["ExecutionOutput"]] = relationship(back_populates="execution")
    values: Mapped[list["MetricValue"]] = relationship(back_populates="execution")

    datasets: Mapped[list[Dataset]] = relationship(secondary=execution_datasets)
    """
    The datasets used in this execution
    """

    def register_datasets(self, db: "Database", execution_dataset: "ExecutionDatasetCollection") -> None:
        """
        Register the datasets used in the diagnostic calculation with the execution
        """
        for _, dataset in execution_dataset.items():
            db.session.execute(
                execution_datasets.insert(),
                [{"execution_id": self.id, "dataset_id": idx} for idx in dataset.index],
            )

    def mark_successful(self, path: pathlib.Path | str) -> None:
        """
        Mark the diagnostic execution as successful
        """
        # TODO: this needs to accept both a diagnostic and output bundle
        self.successful = True
        self.path = str(path)

    def mark_failed(self) -> None:
        """
        Mark the diagnostic execution as unsuccessful
        """
        self.successful = False

dataset_hash = mapped_column(index=True) class-attribute instance-attribute #

Hash of the datasets used to calculate the diagnostic

This is used to verify if an existing diagnostic execution has been run with the same datasets.

datasets = relationship(secondary=execution_datasets) class-attribute instance-attribute #

The datasets used in this execution

execution_group_id = mapped_column(ForeignKey('execution_group.id', name='fk_execution_id'), index=True) class-attribute instance-attribute #

The execution group that this execution belongs to

output_fragment = mapped_column() class-attribute instance-attribute #

Relative directory to store the output of the execution.

During execution this directory is relative to the temporary directory. If the diagnostic execution is successful, the executions will be moved to the final output directory and the temporary directory will be cleaned up. This directory may contain multiple input and output files.

path = mapped_column(nullable=True) class-attribute instance-attribute #

Path to the output bundle

Relative to the diagnostic execution result output directory

provider_version = mapped_column(nullable=True) class-attribute instance-attribute #

Provider version recorded by the worker at run time.

Snapshot of the worker-installed provider.version when the execution ran. Purely informational for audit; not used for validation or recomputation triggers. Rows that predate the column stay NULL.

retracted = mapped_column(default=False) class-attribute instance-attribute #

Whether the diagnostic execution result has been retracted or not

This may happen if a dataset has been retracted, or if the diagnostic execution was incorrect. Rather than delete the values, they are marked as retracted. These data may still be visible in the UI, but should be marked as retracted.

successful = mapped_column(nullable=True, index=True) class-attribute instance-attribute #

Was the run successful

mark_failed() #

Mark the diagnostic execution as unsuccessful

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
278
279
280
281
282
def mark_failed(self) -> None:
    """
    Mark the diagnostic execution as unsuccessful
    """
    self.successful = False

mark_successful(path) #

Mark the diagnostic execution as successful

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
270
271
272
273
274
275
276
def mark_successful(self, path: pathlib.Path | str) -> None:
    """
    Mark the diagnostic execution as successful
    """
    # TODO: this needs to accept both a diagnostic and output bundle
    self.successful = True
    self.path = str(path)

register_datasets(db, execution_dataset) #

Register the datasets used in the diagnostic calculation with the execution

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
260
261
262
263
264
265
266
267
268
def register_datasets(self, db: "Database", execution_dataset: "ExecutionDatasetCollection") -> None:
    """
    Register the datasets used in the diagnostic calculation with the execution
    """
    for _, dataset in execution_dataset.items():
        db.session.execute(
            execution_datasets.insert(),
            [{"execution_id": self.id, "dataset_id": idx} for idx in dataset.index],
        )

ExecutionGroup #

Bases: CreatedUpdatedMixin, Base

Represents a group of executions with a shared set of input datasets.

When solving, the ExecutionGroups are derived from the available datasets, the defined diagnostics and their data requirements. From the information in the group an execution can be triggered, which is an actual run of a diagnostic calculation with a specific set of input datasets.

When the ExecutionGroup is created, it is marked dirty, meaning there are no current executions available. When an Execution was run successfully for a ExecutionGroup, the dirty mark is removed. After ingesting new data and solving again and if new versions of the input datasets are available, the ExecutionGroup will be marked dirty again.

The diagnostic_id and key form a unique identifier for ExecutionGroups.

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
class ExecutionGroup(CreatedUpdatedMixin, Base):
    """
    Represents a group of executions with a shared set of input datasets.

    When solving, the `ExecutionGroup`s are derived from the available datasets,
    the defined diagnostics and their data requirements. From the information in the
    group an execution can be triggered, which is an actual run of a diagnostic calculation
    with a specific set of input datasets.

    When the `ExecutionGroup` is created, it is marked dirty, meaning there are no
    current executions available. When an Execution was run successfully for a
    ExecutionGroup, the dirty mark is removed. After ingesting new data and
    solving again and if new versions of the input datasets are available, the
    ExecutionGroup will be marked dirty again.

    The diagnostic_id and key form a unique identifier for `ExecutionGroup`s.
    """

    __tablename__ = "execution_group"
    __table_args__ = (UniqueConstraint("diagnostic_id", "key", "diagnostic_version", name="execution_ident"),)

    id: Mapped[int] = mapped_column(primary_key=True)

    diagnostic_id: Mapped[int] = mapped_column(ForeignKey("diagnostic.id"), index=True)
    """
    The diagnostic that this execution group belongs to
    """

    key: Mapped[str] = mapped_column(index=True)
    """
    Key for the datasets in this Execution group.
    """

    diagnostic_version: Mapped[int] = mapped_column(default=1, server_default="1")
    """
    Diagnostic version that produced this group.

    Read from the live ``Diagnostic.version`` class attribute at solve time.
    Combined with ``diagnostic_id`` and ``key`` to form the unique identifier,
    so v1 and v2 groups for the same key coexist as separate rows.
    """

    dirty: Mapped[bool] = mapped_column(default=False)
    """
    Whether the execution group should be rerun

    An execution group is dirty if the diagnostic or any of the input datasets has been
    updated since the last execution.
    """

    selectors: Mapped[dict[str, Any]] = mapped_column(default=dict)
    """
    Collection of selectors that define the group

    These selectors are the unique key, value pairs that were selected during the initial groupby
    operation.
    These are also used to define the dataset key.
    """

    diagnostic: Mapped["Diagnostic"] = relationship(back_populates="execution_groups")
    executions: Mapped[list["Execution"]] = relationship(
        back_populates="execution_group", order_by="Execution.created_at, Execution.id"
    )

    def should_run(
        self,
        dataset_hash: str,
        rerun_failed: bool = False,
        stale_cutoff: "datetime.datetime | None" = None,
    ) -> bool:
        """
        Check if the diagnostic execution group needs to be executed.

        The dirty flag is the primary signal for whether an execution group needs to be rerun.
        It is set when the group is created or when new data is available,
        and cleared when an execution completes (whether successful or not).
        Manual intervention (``flag-dirty``, ``fail-running``) can set it back to True.

        The execution group should be run if:

        * no executions have been performed ever
        * the dataset hash is different from the last run
        * the execution group is marked as dirty
        * ``rerun_failed=True`` is passed and the last execution failed

        The execution group should NOT be run if:

        * an execution with the same dataset hash is already in progress
        * the last execution failed and the group is not dirty
          (use ``rerun_failed=True`` or ``flag-dirty`` to retry)

        Parameters
        ----------
        dataset_hash
            Hash of the candidate datasets for this run.
        rerun_failed
            Re-run the group even if the last execution failed and the group is not dirty.
        stale_cutoff
            When provided,
            an in-progress execution created before this timestamp is treated as already failed.
            A real solve reaps such abandoned executions (via ``fail_stale_in_progress_executions``)
            before evaluating this method.
        """
        if not self.executions:
            logger.debug(f"Execution group {self.diagnostic.slug}/{self.key} was never executed")
            return True

        last_execution = self.executions[-1]

        if last_execution.dataset_hash != dataset_hash:
            logger.debug(
                f"Execution group {self.diagnostic.slug}/{self.key} hash mismatch:"
                f" {last_execution.dataset_hash} != {dataset_hash}"
            )
            return True

        treat_as_failed = (
            last_execution.successful is None
            and stale_cutoff is not None
            and last_execution.created_at < stale_cutoff
        )

        # Don't submit duplicate tasks for an execution that is already in progress
        # Stuck tasks can be cleaned up with the `fail-running` command
        if last_execution.successful is None and not treat_as_failed:
            logger.debug(
                f"Execution group {self.diagnostic.slug}/{self.key} "
                f"already has an in-progress execution with hash {dataset_hash}"
            )
            return False

        # Dirty flag is the primary signal for rerunning existing jobs
        if self.dirty:
            logger.debug(f"Execution group {self.diagnostic.slug}/{self.key} is dirty")
            return True

        # Re-run all failed executions if explicitly requested
        if (last_execution.successful is False or treat_as_failed) and rerun_failed:
            logger.debug(
                f"Execution group {self.diagnostic.slug}/{self.key} "
                f"last execution failed, rerunning (rerun_failed=True)"
            )
            return True

        return False

diagnostic_id = mapped_column(ForeignKey('diagnostic.id'), index=True) class-attribute instance-attribute #

The diagnostic that this execution group belongs to

diagnostic_version = mapped_column(default=1, server_default='1') class-attribute instance-attribute #

Diagnostic version that produced this group.

Read from the live Diagnostic.version class attribute at solve time. Combined with diagnostic_id and key to form the unique identifier, so v1 and v2 groups for the same key coexist as separate rows.

dirty = mapped_column(default=False) class-attribute instance-attribute #

Whether the execution group should be rerun

An execution group is dirty if the diagnostic or any of the input datasets has been updated since the last execution.

key = mapped_column(index=True) class-attribute instance-attribute #

Key for the datasets in this Execution group.

selectors = mapped_column(default=dict) class-attribute instance-attribute #

Collection of selectors that define the group

These selectors are the unique key, value pairs that were selected during the initial groupby operation. These are also used to define the dataset key.

should_run(dataset_hash, rerun_failed=False, stale_cutoff=None) #

Check if the diagnostic execution group needs to be executed.

The dirty flag is the primary signal for whether an execution group needs to be rerun. It is set when the group is created or when new data is available, and cleared when an execution completes (whether successful or not). Manual intervention (flag-dirty, fail-running) can set it back to True.

The execution group should be run if:

  • no executions have been performed ever
  • the dataset hash is different from the last run
  • the execution group is marked as dirty
  • rerun_failed=True is passed and the last execution failed

The execution group should NOT be run if:

  • an execution with the same dataset hash is already in progress
  • the last execution failed and the group is not dirty (use rerun_failed=True or flag-dirty to retry)

Parameters:

Name Type Description Default
dataset_hash str

Hash of the candidate datasets for this run.

 required
rerun_failed bool

Re-run the group even if the last execution failed and the group is not dirty.

 False
stale_cutoff datetime | None

When provided, an in-progress execution created before this timestamp is treated as already failed. A real solve reaps such abandoned executions (via fail_stale_in_progress_executions) before evaluating this method.

 None
Source code in packages/climate-ref/src/climate_ref/models/execution.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def should_run(
    self,
    dataset_hash: str,
    rerun_failed: bool = False,
    stale_cutoff: "datetime.datetime | None" = None,
) -> bool:
    """
    Check if the diagnostic execution group needs to be executed.

    The dirty flag is the primary signal for whether an execution group needs to be rerun.
    It is set when the group is created or when new data is available,
    and cleared when an execution completes (whether successful or not).
    Manual intervention (``flag-dirty``, ``fail-running``) can set it back to True.

    The execution group should be run if:

    * no executions have been performed ever
    * the dataset hash is different from the last run
    * the execution group is marked as dirty
    * ``rerun_failed=True`` is passed and the last execution failed

    The execution group should NOT be run if:

    * an execution with the same dataset hash is already in progress
    * the last execution failed and the group is not dirty
      (use ``rerun_failed=True`` or ``flag-dirty`` to retry)

    Parameters
    ----------
    dataset_hash
        Hash of the candidate datasets for this run.
    rerun_failed
        Re-run the group even if the last execution failed and the group is not dirty.
    stale_cutoff
        When provided,
        an in-progress execution created before this timestamp is treated as already failed.
        A real solve reaps such abandoned executions (via ``fail_stale_in_progress_executions``)
        before evaluating this method.
    """
    if not self.executions:
        logger.debug(f"Execution group {self.diagnostic.slug}/{self.key} was never executed")
        return True

    last_execution = self.executions[-1]

    if last_execution.dataset_hash != dataset_hash:
        logger.debug(
            f"Execution group {self.diagnostic.slug}/{self.key} hash mismatch:"
            f" {last_execution.dataset_hash} != {dataset_hash}"
        )
        return True

    treat_as_failed = (
        last_execution.successful is None
        and stale_cutoff is not None
        and last_execution.created_at < stale_cutoff
    )

    # Don't submit duplicate tasks for an execution that is already in progress
    # Stuck tasks can be cleaned up with the `fail-running` command
    if last_execution.successful is None and not treat_as_failed:
        logger.debug(
            f"Execution group {self.diagnostic.slug}/{self.key} "
            f"already has an in-progress execution with hash {dataset_hash}"
        )
        return False

    # Dirty flag is the primary signal for rerunning existing jobs
    if self.dirty:
        logger.debug(f"Execution group {self.diagnostic.slug}/{self.key} is dirty")
        return True

    # Re-run all failed executions if explicitly requested
    if (last_execution.successful is False or treat_as_failed) and rerun_failed:
        logger.debug(
            f"Execution group {self.diagnostic.slug}/{self.key} "
            f"last execution failed, rerunning (rerun_failed=True)"
        )
        return True

    return False

ExecutionOutput #

Bases: DimensionMixin, CreatedUpdatedMixin, Base

An output generated as part of an execution.

This output may be a plot, data file or HTML file. These outputs are defined in the CMEC output bundle.

Outputs can be tagged with dimensions from the controlled vocabulary to enable filtering and organization.

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
class ExecutionOutput(DimensionMixin, CreatedUpdatedMixin, Base):
    """
    An output generated as part of an execution.

    This output may be a plot, data file or HTML file.
    These outputs are defined in the CMEC output bundle.

    Outputs can be tagged with dimensions from the controlled vocabulary
    to enable filtering and organization.
    """

    __tablename__ = "execution_output"

    _cv_dimensions: ClassVar[list[str]] = []

    id: Mapped[int] = mapped_column(primary_key=True)

    execution_id: Mapped[int] = mapped_column(ForeignKey("execution.id"), index=True)

    output_type: Mapped[ResultOutputType] = mapped_column(index=True)
    """
    Type of the output

    This will determine how the output is displayed
    """

    filename: Mapped[str] = mapped_column(nullable=True)
    """
    Path to the output

    Relative to the diagnostic execution result output directory
    """

    short_name: Mapped[str] = mapped_column(nullable=True)
    """
    Short key of the output

    This is unique for a given result and output type
    """

    long_name: Mapped[str] = mapped_column(nullable=True)
    """
    Human readable name describing the plot
    """

    description: Mapped[str] = mapped_column(nullable=True)
    """
    Long description describing the plot
    """

    execution: Mapped["Execution"] = relationship(back_populates="outputs")

    @classmethod
    def build(  # noqa: PLR0913
        cls,
        *,
        execution_id: int,
        output_type: ResultOutputType,
        dimensions: dict[str, str],
        filename: str | None = None,
        short_name: str | None = None,
        long_name: str | None = None,
        description: str | None = None,
    ) -> "ExecutionOutput":
        """
        Build an ExecutionOutput from dimensions and metadata

        This is a helper method that validates the dimensions supplied.

        Parameters
        ----------
        execution_id
            Execution that created the output
        output_type
            Type of the output
        dimensions
            Dimensions that describe the output
        filename
            Path to the output
        short_name
            Short key of the output
        long_name
            Human readable name
        description
            Long description

        Raises
        ------
        KeyError
            If an unknown dimension was supplied.

            Dimensions must exist in the controlled vocabulary.

        Returns
        -------
            Newly created ExecutionOutput
        """
        for k in dimensions:
            if k not in cls._cv_dimensions:
                raise KeyError(f"Unknown dimension column '{k}'")

        return ExecutionOutput(
            execution_id=execution_id,
            output_type=output_type,
            filename=filename,
            short_name=short_name,
            long_name=long_name,
            description=description,
            **dimensions,
        )

description = mapped_column(nullable=True) class-attribute instance-attribute #

Long description describing the plot

filename = mapped_column(nullable=True) class-attribute instance-attribute #

Path to the output

Relative to the diagnostic execution result output directory

long_name = mapped_column(nullable=True) class-attribute instance-attribute #

Human readable name describing the plot

output_type = mapped_column(index=True) class-attribute instance-attribute #

Type of the output

This will determine how the output is displayed

short_name = mapped_column(nullable=True) class-attribute instance-attribute #

Short key of the output

This is unique for a given result and output type

build(*, execution_id, output_type, dimensions, filename=None, short_name=None, long_name=None, description=None) classmethod #

Build an ExecutionOutput from dimensions and metadata

This is a helper method that validates the dimensions supplied.

Parameters:

Name Type Description Default
execution_id int

Execution that created the output

 required
output_type ResultOutputType

Type of the output

 required
dimensions dict[str, str]

Dimensions that describe the output

 required
filename str | None

Path to the output

 None
short_name str | None

Short key of the output

 None
long_name str | None

Human readable name

 None
description str | None

Long description

 None

Raises:

Type Description
KeyError

If an unknown dimension was supplied.

Dimensions must exist in the controlled vocabulary.

Returns:

Type Description
Newly created ExecutionOutput
Source code in packages/climate-ref/src/climate_ref/models/execution.py 
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
@classmethod
def build(  # noqa: PLR0913
    cls,
    *,
    execution_id: int,
    output_type: ResultOutputType,
    dimensions: dict[str, str],
    filename: str | None = None,
    short_name: str | None = None,
    long_name: str | None = None,
    description: str | None = None,
) -> "ExecutionOutput":
    """
    Build an ExecutionOutput from dimensions and metadata

    This is a helper method that validates the dimensions supplied.

    Parameters
    ----------
    execution_id
        Execution that created the output
    output_type
        Type of the output
    dimensions
        Dimensions that describe the output
    filename
        Path to the output
    short_name
        Short key of the output
    long_name
        Human readable name
    description
        Long description

    Raises
    ------
    KeyError
        If an unknown dimension was supplied.

        Dimensions must exist in the controlled vocabulary.

    Returns
    -------
        Newly created ExecutionOutput
    """
    for k in dimensions:
        if k not in cls._cv_dimensions:
            raise KeyError(f"Unknown dimension column '{k}'")

    return ExecutionOutput(
        execution_id=execution_id,
        output_type=output_type,
        filename=filename,
        short_name=short_name,
        long_name=long_name,
        description=description,
        **dimensions,
    )

MetricValue #

Bases: DimensionMixin, CreatedUpdatedMixin, Base

Represents a single metric value

This is a base class for different types of metric values (e.g. scalar, series) which are stored in a single table using single table inheritance.

This value has a number of dimensions which are used to query the diagnostic values. These dimensions describe aspects such as the type of statistic being measured, the region of interest or the model from which the statistic is being measured.

The columns in this table are not known statically because the REF can track an arbitrary set of dimensions depending on the controlled vocabulary that will be used. A call to register_cv_dimensions must be made before using this class.

Source code in packages/climate-ref/src/climate_ref/models/metric_value.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class MetricValue(DimensionMixin, CreatedUpdatedMixin, Base):
    """
    Represents a single metric value

    This is a base class for different types of metric values (e.g. scalar, series) which
    are stored in a single table using single table inheritance.

    This value has a number of dimensions which are used to query the diagnostic values.
    These dimensions describe aspects such as the type of statistic being measured,
    the region of interest or the model from which the statistic is being measured.

    The columns in this table are not known statically because the REF can track an arbitrary
    set of dimensions depending on the controlled vocabulary that will be used.
    A call to `register_cv_dimensions` must be made before using this class.
    """

    __tablename__ = "metric_value"

    __mapper_args__: ClassVar[Mapping[str, str]] = {  # type: ignore
        "polymorphic_on": "type",
    }

    _cv_dimensions: ClassVar[list[str]] = []

    id: Mapped[int] = mapped_column(primary_key=True)
    execution_id: Mapped[int] = mapped_column(ForeignKey("execution.id"), index=True)

    attributes: Mapped[dict[str, Any]] = mapped_column()

    execution: Mapped["Execution"] = relationship(back_populates="values")

    type: Mapped[MetricValueType] = mapped_column(index=True)
    """
    Type of metric value

    This value is used to determine how the metric value should be interpreted.
    """

    def __repr__(self) -> str:
        return f"<MetricValue id={self.id} execution={self.execution} dimensions={self.dimensions}>"

type = mapped_column(index=True) class-attribute instance-attribute #

Type of metric value

This value is used to determine how the metric value should be interpreted.

Provider #

Bases: CreatedUpdatedMixin, Base

Represents a provider that can provide diagnostic calculations

Source code in packages/climate-ref/src/climate_ref/models/provider.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class Provider(CreatedUpdatedMixin, Base):
    """
    Represents a provider that can provide diagnostic calculations
    """

    __tablename__ = "provider"

    id: Mapped[int] = mapped_column(primary_key=True)
    slug: Mapped[str] = mapped_column(unique=True)
    """
    Globally unique identifier for the provider.
    """

    name: Mapped[str] = mapped_column()
    """
    Long name of the provider
    """

    version: Mapped[str] = mapped_column(nullable=False)
    """
    Version of the provider.

    This should map to the package version.
    """

    diagnostics: Mapped[list["Diagnostic"]] = relationship(back_populates="provider")

    def __repr__(self) -> str:
        return f"<Provider slug={self.slug} version={self.version}>"

name = mapped_column() class-attribute instance-attribute #

Long name of the provider

slug = mapped_column(unique=True) class-attribute instance-attribute #

Globally unique identifier for the provider.

version = mapped_column(nullable=False) class-attribute instance-attribute #

Version of the provider.

This should map to the package version.

ScalarMetricValue #

Bases: MetricValue

A scalar value with an associated dimensions

This is a subclass of MetricValue that is used to represent a scalar value.

Source code in packages/climate-ref/src/climate_ref/models/metric_value.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class ScalarMetricValue(MetricValue):
    """
    A scalar value with an associated dimensions

    This is a subclass of MetricValue that is used to represent a scalar value.
    """

    __mapper_args__: ClassVar[Mapping[str, Any]] = {  # type: ignore
        "polymorphic_identity": MetricValueType.SCALAR,
    }

    # This is a scalar value
    value: Mapped[float] = mapped_column(nullable=True)

    def __repr__(self) -> str:
        return (
            f"<ScalarMetricValue "
            f"id={self.id} execution={self.execution} dimensions={self.dimensions} value={self.value}>"
        )

    @classmethod
    def build(
        cls,
        *,
        execution_id: int,
        value: float,
        dimensions: dict[str, str],
        attributes: dict[str, Any] | None,
    ) -> "MetricValue":
        """
        Build a MetricValue from a collection of dimensions and a value

        This is a helper method that validates the dimensions supplied and provides an interface
        similar to [climate_ref_core.metric_values.ScalarMetricValue][].

        Parameters
        ----------
        execution_id
            Execution that created the diagnostic value
        value
            The value of the diagnostic
        dimensions
            Dimensions that describe the diagnostic execution result
        attributes
            Optional additional attributes to describe the value,
            but are not in the controlled vocabulary.

        Raises
        ------
        KeyError
            If an unknown dimension was supplied.

            Dimensions must exist in the controlled vocabulary.

        Returns
        -------
            Newly created MetricValue
        """
        for k in dimensions:
            if k not in cls._cv_dimensions:
                raise KeyError(f"Unknown dimension column '{k}'")

        return ScalarMetricValue(
            execution_id=execution_id,
            value=value,
            attributes=attributes,
            **dimensions,
        )

build(*, execution_id, value, dimensions, attributes) classmethod #

Build a MetricValue from a collection of dimensions and a value

This is a helper method that validates the dimensions supplied and provides an interface similar to climate_ref_core.metric_values.ScalarMetricValue.

Parameters:

Name Type Description Default
execution_id int

Execution that created the diagnostic value

 required
value float

The value of the diagnostic

 required
dimensions dict[str, str]

Dimensions that describe the diagnostic execution result

 required
attributes dict[str, Any] | None

Optional additional attributes to describe the value, but are not in the controlled vocabulary.

 required

Raises:

Type Description
KeyError

If an unknown dimension was supplied.

Dimensions must exist in the controlled vocabulary.

Returns:

Type Description
Newly created MetricValue
Source code in packages/climate-ref/src/climate_ref/models/metric_value.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@classmethod
def build(
    cls,
    *,
    execution_id: int,
    value: float,
    dimensions: dict[str, str],
    attributes: dict[str, Any] | None,
) -> "MetricValue":
    """
    Build a MetricValue from a collection of dimensions and a value

    This is a helper method that validates the dimensions supplied and provides an interface
    similar to [climate_ref_core.metric_values.ScalarMetricValue][].

    Parameters
    ----------
    execution_id
        Execution that created the diagnostic value
    value
        The value of the diagnostic
    dimensions
        Dimensions that describe the diagnostic execution result
    attributes
        Optional additional attributes to describe the value,
        but are not in the controlled vocabulary.

    Raises
    ------
    KeyError
        If an unknown dimension was supplied.

        Dimensions must exist in the controlled vocabulary.

    Returns
    -------
        Newly created MetricValue
    """
    for k in dimensions:
        if k not in cls._cv_dimensions:
            raise KeyError(f"Unknown dimension column '{k}'")

    return ScalarMetricValue(
        execution_id=execution_id,
        value=value,
        attributes=attributes,
        **dimensions,
    )

SeriesIndex #

Bases: Base

A shared 1-d index axis for series metric values

Many series share the same index (for example a common monthly time axis), so the index is stored once here and referenced by SeriesMetricValue.index_id rather than duplicated on every row. Axes are deduplicated by hash.

Source code in packages/climate-ref/src/climate_ref/models/metric_value.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
class SeriesIndex(Base):
    """
    A shared 1-d index axis for series metric values

    Many series share the same index (for example a common monthly time axis),
    so the index is stored once here and referenced by
    [SeriesMetricValue.index_id][climate_ref.models.metric_value.SeriesMetricValue]
    rather than duplicated on every row. Axes are deduplicated by ``hash``.
    """

    __tablename__ = "index_axis"

    id: Mapped[int] = mapped_column(primary_key=True)

    hash: Mapped[str] = mapped_column(unique=True, index=True)
    """
    Content hash of ``(name, values)``.

    The axes are deduplicated by this hash, so identical axes will share the same row and be referenced by id.
    """

    name: Mapped[str] = mapped_column(nullable=True)
    """Name of the index (e.g. ``"time"``). Used for presentation."""

    values: Mapped[list[float | int | str]] = mapped_column()
    """The 1-d array of index values."""

    length: Mapped[int] = mapped_column()
    """Number of points in the index; used to validate series lengths."""

    def __repr__(self) -> str:
        return f"<SeriesIndex id={self.id} name={self.name} length={self.length}>"

    @staticmethod
    def compute_hash(name: str | None, values: Sequence[float | int | str]) -> str:
        """
        Compute the content hash used to deduplicate identical axes.

        The hash covers both the name and the ordered values,
        so two axes are only shared when they are genuinely identical.
        """
        return _content_hash([name, list(values)])

    @classmethod
    def get_or_create(
        cls, session: Session, name: str | None, values: Sequence[float | int | str]
    ) -> "SeriesIndex":
        """
        Return the existing axis with this content, or create and flush a new one.

        Parameters
        ----------
        session
            Active database session.
        name
            Name of the index.
        values
            1-d array of index values.

        Returns
        -------
            The shared [SeriesIndex][climate_ref.models.metric_value.SeriesIndex] row.
        """
        digest = cls.compute_hash(name, values)
        existing = session.execute(select(cls).where(cls.hash == digest)).scalar_one_or_none()
        if existing is not None:
            return existing
        axis = cls(hash=digest, name=name, values=list(values), length=len(values))
        session.add(axis)
        session.flush()
        return axis

    @classmethod
    def bulk_get_or_create(
        cls,
        session: Session,
        axes_by_hash: Mapping[str, tuple[str | None, Sequence[float | int | str]]],
    ) -> dict[str, int]:
        """
        Resolve many axes at once, returning a ``{hash: id}`` map.

        Existing axes are fetched in a single query and any missing axes are bulk-inserted,
        so a batch of series values costs two queries rather than one ``get_or_create`` per row.

        Parameters
        ----------
        session
            Active database session.
        axes_by_hash
            Content hash (see [compute_hash][climate_ref.models.metric_value.SeriesIndex.compute_hash])
            mapped to the axis ``(name, values)`` it represents.

        Returns
        -------
            The shared axis id for every hash in ``axes_by_hash``.

        Raises
        ------
        ValueError
            If any key does not equal ``compute_hash(name, values)`` for the axis it maps to.
            A mismatched key would otherwise insert a row whose stored ``hash`` does not match its content,
            breaking deduplication.
        """
        if not axes_by_hash:
            return {}

        for digest, (name, values) in axes_by_hash.items():
            expected = cls.compute_hash(name, values)
            if digest != expected:
                raise ValueError(
                    f"axes_by_hash key {digest!r} does not match compute_hash of its axis "
                    f"(expected {expected!r})"
                )

        id_by_hash: dict[str, int] = {
            digest: axis_id
            for digest, axis_id in session.execute(select(cls.hash, cls.id).where(cls.hash.in_(axes_by_hash)))
        }
        missing = [digest for digest in axes_by_hash if digest not in id_by_hash]
        if missing:
            session.execute(
                insert(cls),
                [
                    {
                        "hash": digest,
                        "name": axes_by_hash[digest][0],
                        "values": list(axes_by_hash[digest][1]),
                        "length": len(axes_by_hash[digest][1]),
                    }
                    for digest in missing
                ],
            )
            id_by_hash.update(
                {
                    digest: axis_id
                    for digest, axis_id in session.execute(
                        select(cls.hash, cls.id).where(cls.hash.in_(missing))
                    )
                }
            )
        return id_by_hash

hash = mapped_column(unique=True, index=True) class-attribute instance-attribute #

Content hash of (name, values).

The axes are deduplicated by this hash, so identical axes will share the same row and be referenced by id.

length = mapped_column() class-attribute instance-attribute #

Number of points in the index; used to validate series lengths.

name = mapped_column(nullable=True) class-attribute instance-attribute #

Name of the index (e.g. "time"). Used for presentation.

values = mapped_column() class-attribute instance-attribute #

The 1-d array of index values.

bulk_get_or_create(session, axes_by_hash) classmethod #

Resolve many axes at once, returning a {hash: id} map.

Existing axes are fetched in a single query and any missing axes are bulk-inserted, so a batch of series values costs two queries rather than one get_or_create per row.

Parameters:

Name Type Description Default
session Session

Active database session.

 required
axes_by_hash Mapping[str, tuple[str | None, Sequence[float | int | str]]]

Content hash (see compute_hash) mapped to the axis (name, values) it represents.

 required

Returns:

Type Description
The shared axis id for every hash in ``axes_by_hash``.

Raises:

Type Description
ValueError

If any key does not equal compute_hash(name, values) for the axis it maps to. A mismatched key would otherwise insert a row whose stored hash does not match its content, breaking deduplication.
Source code in packages/climate-ref/src/climate_ref/models/metric_value.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
@classmethod
def bulk_get_or_create(
    cls,
    session: Session,
    axes_by_hash: Mapping[str, tuple[str | None, Sequence[float | int | str]]],
) -> dict[str, int]:
    """
    Resolve many axes at once, returning a ``{hash: id}`` map.

    Existing axes are fetched in a single query and any missing axes are bulk-inserted,
    so a batch of series values costs two queries rather than one ``get_or_create`` per row.

    Parameters
    ----------
    session
        Active database session.
    axes_by_hash
        Content hash (see [compute_hash][climate_ref.models.metric_value.SeriesIndex.compute_hash])
        mapped to the axis ``(name, values)`` it represents.

    Returns
    -------
        The shared axis id for every hash in ``axes_by_hash``.

    Raises
    ------
    ValueError
        If any key does not equal ``compute_hash(name, values)`` for the axis it maps to.
        A mismatched key would otherwise insert a row whose stored ``hash`` does not match its content,
        breaking deduplication.
    """
    if not axes_by_hash:
        return {}

    for digest, (name, values) in axes_by_hash.items():
        expected = cls.compute_hash(name, values)
        if digest != expected:
            raise ValueError(
                f"axes_by_hash key {digest!r} does not match compute_hash of its axis "
                f"(expected {expected!r})"
            )

    id_by_hash: dict[str, int] = {
        digest: axis_id
        for digest, axis_id in session.execute(select(cls.hash, cls.id).where(cls.hash.in_(axes_by_hash)))
    }
    missing = [digest for digest in axes_by_hash if digest not in id_by_hash]
    if missing:
        session.execute(
            insert(cls),
            [
                {
                    "hash": digest,
                    "name": axes_by_hash[digest][0],
                    "values": list(axes_by_hash[digest][1]),
                    "length": len(axes_by_hash[digest][1]),
                }
                for digest in missing
            ],
        )
        id_by_hash.update(
            {
                digest: axis_id
                for digest, axis_id in session.execute(
                    select(cls.hash, cls.id).where(cls.hash.in_(missing))
                )
            }
        )
    return id_by_hash

compute_hash(name, values) staticmethod #

Compute the content hash used to deduplicate identical axes.

The hash covers both the name and the ordered values, so two axes are only shared when they are genuinely identical.

Source code in packages/climate-ref/src/climate_ref/models/metric_value.py 
188
189
190
191
192
193
194
195
196
@staticmethod
def compute_hash(name: str | None, values: Sequence[float | int | str]) -> str:
    """
    Compute the content hash used to deduplicate identical axes.

    The hash covers both the name and the ordered values,
    so two axes are only shared when they are genuinely identical.
    """
    return _content_hash([name, list(values)])

get_or_create(session, name, values) classmethod #

Return the existing axis with this content, or create and flush a new one.

Parameters:

Name Type Description Default
session Session

Active database session.

 required
name str | None

Name of the index.

 required
values Sequence[float | int | str]

1-d array of index values.

 required

Returns:

Type Description
The shared [SeriesIndex][climate_ref.models.metric_value.SeriesIndex] row.
Source code in packages/climate-ref/src/climate_ref/models/metric_value.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
@classmethod
def get_or_create(
    cls, session: Session, name: str | None, values: Sequence[float | int | str]
) -> "SeriesIndex":
    """
    Return the existing axis with this content, or create and flush a new one.

    Parameters
    ----------
    session
        Active database session.
    name
        Name of the index.
    values
        1-d array of index values.

    Returns
    -------
        The shared [SeriesIndex][climate_ref.models.metric_value.SeriesIndex] row.
    """
    digest = cls.compute_hash(name, values)
    existing = session.execute(select(cls).where(cls.hash == digest)).scalar_one_or_none()
    if existing is not None:
        return existing
    axis = cls(hash=digest, name=name, values=list(values), length=len(values))
    session.add(axis)
    session.flush()
    return axis

SeriesMetricValue #

Bases: MetricValue

A 1d series with associated dimensions

This is a subclass of MetricValue that is used to represent a series. This can be used to represent time series, vertical profiles or other 1d data.

Source code in packages/climate-ref/src/climate_ref/models/metric_value.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
class SeriesMetricValue(MetricValue):
    """
    A 1d series with associated dimensions

    This is a subclass of MetricValue that is used to represent a series.
    This can be used to represent time series, vertical profiles or other 1d data.
    """

    __mapper_args__: ClassVar[Mapping[str, Any]] = {  # type: ignore
        "polymorphic_identity": MetricValueType.SERIES,
    }

    values: Mapped[list[float | int]] = mapped_column(nullable=True)

    index_id: Mapped[int | None] = mapped_column(ForeignKey("index_axis.id"), nullable=True, index=True)
    index_axis: Mapped["SeriesIndex | None"] = relationship(lazy="joined")

    reference_id: Mapped[str | None] = mapped_column(nullable=True, index=True)
    """
    Content hash of the reference payload, for reference (observation) series only.

    Two reference series with an identical payload share the same ``reference_id``,
    so observations can be deduplicated deterministically across executions.
    It is ``None`` for model series. See
    [compute_reference_id][climate_ref.models.metric_value.SeriesMetricValue.compute_reference_id].
    """

    @staticmethod
    def compute_reference_id(
        values: Sequence[float | int],
        index: Sequence[float | int | str] | None,
        reference_source_id: str | None,
    ) -> str:
        """
        Compute the content hash that deduplicates an identical reference payload.

        The hash covers the values, the index and the reference source,
        so two reference series are only treated as the same observation
        when their payloads are genuinely identical.
        Keep this payload stable: it is the deduplication key used downstream.
        """
        return _content_hash([list(values), list(index) if index is not None else None, reference_source_id])

    @property
    def index(self) -> list[float | int | str] | None:
        """The 1-d index values, resolved from the shared axis."""
        return self.index_axis.values if self.index_axis is not None else None

    @property
    def index_name(self) -> str | None:
        """The name of the index, resolved from the shared axis."""
        return self.index_axis.name if self.index_axis is not None else None

    def __repr__(self) -> str:
        return (
            f"<SeriesMetricValue id={self.id} execution={self.execution} "
            f"dimensions={self.dimensions} index_name={self.index_name}>"
        )

    @classmethod
    def build(
        cls,
        *,
        execution_id: int,
        values: list[float | int],
        index_axis: "SeriesIndex",
        dimensions: dict[str, str],
        attributes: dict[str, Any] | None,
    ) -> "MetricValue":
        """
        Build a database object from a series

        Parameters
        ----------
        execution_id
            Execution that created the diagnostic value
        values
            1-d array of values
        index_axis
            The shared index axis for this series, obtained via
            [SeriesIndex.get_or_create][climate_ref.models.metric_value.SeriesIndex.get_or_create]
        dimensions
            Dimensions that describe the diagnostic execution result
        attributes
            Optional additional attributes to describe the value,
            but are not in the controlled vocabulary.

        Raises
        ------
        KeyError
            If an unknown dimension was supplied.

            Dimensions must exist in the controlled vocabulary.
        ValueError
            If the length of values and index do not match.

        Returns
        -------
            Newly created MetricValue
        """
        for k in dimensions:
            if k not in cls._cv_dimensions:
                raise KeyError(f"Unknown dimension column '{k}'")

        if len(values) != index_axis.length:
            raise ValueError(f"Index length ({index_axis.length}) must match values length ({len(values)})")

        return SeriesMetricValue(
            execution_id=execution_id,
            values=values,
            index_axis=index_axis,
            attributes=attributes,
            **dimensions,
        )

index property #

The 1-d index values, resolved from the shared axis.

index_name property #

The name of the index, resolved from the shared axis.

reference_id = mapped_column(nullable=True, index=True) class-attribute instance-attribute #

Content hash of the reference payload, for reference (observation) series only.

Two reference series with an identical payload share the same reference_id, so observations can be deduplicated deterministically across executions. It is None for model series. See compute_reference_id.

build(*, execution_id, values, index_axis, dimensions, attributes) classmethod #

Build a database object from a series

Parameters:

Name Type Description Default
execution_id int

Execution that created the diagnostic value

 required
values list[float | int]

1-d array of values

 required
index_axis SeriesIndex

The shared index axis for this series, obtained via SeriesIndex.get_or_create

 required
dimensions dict[str, str]

Dimensions that describe the diagnostic execution result

 required
attributes dict[str, Any] | None

Optional additional attributes to describe the value, but are not in the controlled vocabulary.

 required

Raises:

Type Description
KeyError

If an unknown dimension was supplied.

Dimensions must exist in the controlled vocabulary.
ValueError

If the length of values and index do not match.

Returns:

Type Description
Newly created MetricValue
Source code in packages/climate-ref/src/climate_ref/models/metric_value.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
@classmethod
def build(
    cls,
    *,
    execution_id: int,
    values: list[float | int],
    index_axis: "SeriesIndex",
    dimensions: dict[str, str],
    attributes: dict[str, Any] | None,
) -> "MetricValue":
    """
    Build a database object from a series

    Parameters
    ----------
    execution_id
        Execution that created the diagnostic value
    values
        1-d array of values
    index_axis
        The shared index axis for this series, obtained via
        [SeriesIndex.get_or_create][climate_ref.models.metric_value.SeriesIndex.get_or_create]
    dimensions
        Dimensions that describe the diagnostic execution result
    attributes
        Optional additional attributes to describe the value,
        but are not in the controlled vocabulary.

    Raises
    ------
    KeyError
        If an unknown dimension was supplied.

        Dimensions must exist in the controlled vocabulary.
    ValueError
        If the length of values and index do not match.

    Returns
    -------
        Newly created MetricValue
    """
    for k in dimensions:
        if k not in cls._cv_dimensions:
            raise KeyError(f"Unknown dimension column '{k}'")

    if len(values) != index_axis.length:
        raise ValueError(f"Index length ({index_axis.length}) must match values length ({len(values)})")

    return SeriesMetricValue(
        execution_id=execution_id,
        values=values,
        index_axis=index_axis,
        attributes=attributes,
        **dimensions,
    )

compute_reference_id(values, index, reference_source_id) staticmethod #

Compute the content hash that deduplicates an identical reference payload.

The hash covers the values, the index and the reference source, so two reference series are only treated as the same observation when their payloads are genuinely identical. Keep this payload stable: it is the deduplication key used downstream.

Source code in packages/climate-ref/src/climate_ref/models/metric_value.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
@staticmethod
def compute_reference_id(
    values: Sequence[float | int],
    index: Sequence[float | int | str] | None,
    reference_source_id: str | None,
) -> str:
    """
    Compute the content hash that deduplicates an identical reference payload.

    The hash covers the values, the index and the reference source,
    so two reference series are only treated as the same observation
    when their payloads are genuinely identical.
    Keep this payload stable: it is the deduplication key used downstream.
    """
    return _content_hash([list(values), list(index) if index is not None else None, reference_source_id])

sub-packages#

Sub-package Description
base
dataset
dataset_query Query builder for the polymorphic Dataset hierarchy.
diagnostic
execution
metric_value
mixins Model mixins for shared functionality
provider