Produce SDMX structural metadata for transport

- Add ExogenousDataFile.generate_dfd().
- Add tests.

message_ix_models/model/transport/files.py

@@ -3,11 +3,14 @@
 
 from genno import Key
 
+from message_ix_models.util import minimum_version
+
 from .key import pdt_cap
 
 if TYPE_CHECKING:
     import genno
     from genno.core.key import KeyLike
+    from sdmx.model.common import BaseDataflowDefinition
 
     from message_ix_models import Context
 
@@ -86,6 +89,54 @@ def add_tasks(
         c.add("load_file", path, key=self.key, dims=dims, name=self.key.name)
         return (self.key,)
 
+    @minimum_version("ixmp 3.8")
+    def generate_dfd(self) -> "BaseDataflowDefinition":
+        from importlib.metadata import version
+
+        from packaging.version import parse
+        from sdmx.model.common import ConceptScheme
+        from sdmx.model.v21 import DataflowDefinition, DataStructureDefinition
+
+        from message_ix_models.util.ixmp import get_reversed_rename_dims
+        from message_ix_models.util.sdmx import read
+
+        # Read the existing agency scheme
+        ece = read("IIASA_ECE:AGENCIES")["IIASA_ECE"]
+        name_for_id = self.key.name.upper().replace(" ", "_")
+        version = parse(version("message_ix_models")).base_version
+        ma_kwargs = dict(maintainer=ece, version=version)
+
+        # Create a shared concept scheme
+        cs = ConceptScheme(id="CS_MESSAGE_TRANSPORT", **ma_kwargs)
+
+        # Create a data structure definition
+        dsd = DataStructureDefinition(
+            id=f"DS_{name_for_id}", **ma_kwargs, name=self.doc
+        )
+
+        # Add dimensions
+        dims = get_reversed_rename_dims()
+        for dim in self.key.dims:
+            # Symbol ('n') → Dimension ID ('node') → upper case
+            dim_id = dims.get(dim, dim).upper()
+
+            concept = cs.setdefault(id="dim_id")
+            dsd.dimensions.getdefault(id=dim_id, concept_identity=concept)
+
+        dfd = DataflowDefinition(
+            id=f"DF_{name_for_id}", **ma_kwargs, name=self.doc, structure=dsd
+        )
+        # TODO Add annotations: preferred file name
+
+        # TODO Generate a CSV template file
+        # 1. In the current format.abs
+        # 2. In SDMX-CSV.
+        # dm = DataMessage()
+        # dm.data.append(DataSet(structure))
+        # template =
+
+        return dfd
+
 
 ExogenousDataFile(
     "pdt-cap-ref",

message_ix_models/tests/model/transport/test_files.py

@@ -43,3 +43,8 @@ def test_configure_build(
 
         # Dimensions are as expected
         assert set(Key(result).dims) == set(file.key.dims)
+
+    @pytest.mark.parametrize("file", FILES, ids=lambda f: "-".join(f.parts))
+    def test_generate_dfd(self, file) -> None:
+        dfd = file.generate_dfd()
+        del dfd

message_ix_models/util/ixmp.py

@@ -2,6 +2,7 @@
 
 try:
     # ixmp 3.8.0 and later
+    from ixmp.report.util import get_reversed_rename_dims
     from ixmp.util import (
         discard_on_error,
         maybe_check_out,
@@ -13,7 +14,10 @@
     # ixmp <= 3.7.0
     from contextlib import nullcontext
 
-    from ixmp.utils import (  # type: ignore [import-not-found,no-redef]  # noqa: F401
+    from ixmp.reporting.util import (  # type: ignore [import-not-found,no-redef]
+        get_reversed_rename_dims,
+    )
+    from ixmp.utils import (  # type: ignore [import-not-found,no-redef]
         maybe_check_out,
         maybe_commit,
         parse_url,
@@ -24,6 +28,16 @@ def discard_on_error(*args):
         return nullcontext()
 
 
+__all__ = [
+    "get_reversed_rename_dims",
+    "maybe_check_out",
+    "maybe_commit",
+    "parse_url",
+    "rename_dims",
+    "show_versions",
+]
+
+
 def rename_dims() -> Dict[str, str]:
     """Access :data:`.ixmp.report.common.RENAME_DIMS`.