feat(coordinator)!: plan and run reads through the list of sources

packages/timenet-connectors/src/timenet_connectors/datasets/chengsenwang/tsqa/tests/test_connector.py

@@ -8,7 +8,7 @@
 from timenet.dataset import TimeFDataset
 from timenet.errors import TimeFValidationError
 from timenet.reader import TimeFReader
-from timenet.registry import LocalRegistry
+from timenet.registry import LocalRegistry, open_local_version
 from timenet.types import AnswerTask
 from timenet.types.splits import Split
 from timenet.types.text import read_value
@@ -188,7 +188,7 @@ def test_splits_survive_storage_round_trip(tmp_path):
     original = TSQAConnector().convert(_refs(tmp_path, _balanced_rows()))
     registry = LocalRegistry(tmp_path / "registry")
     registry.put_dataset(original)
-    with TimeFReader(registry.open_version("chengsenwang/tsqa", "1.2.0")) as reader:
+    with TimeFReader(open_local_version(registry.root / "chengsenwang/tsqa/1.2.0")) as reader:
         restored = reader.read()
     assert {read_value(task.prompt): task.split for task in restored.tasks} == {
         read_value(task.prompt): task.split for task in original.tasks

packages/timenet-connectors/src/timenet_connectors/datasets/tsaqa/tsaqa_benchmark/tests/test_connector.py

@@ -13,7 +13,7 @@
 from timenet.errors import TimeFFormatError, TimeFValidationError
 from timenet.json import JsonValue
 from timenet.reader import TimeFReader
-from timenet.registry import LocalRegistry
+from timenet.registry import LocalRegistry, open_local_version
 from timenet.types import AnswerTask, ClassificationTask, Split
 from timenet_connectors.bases.huggingface import HubFileRef
 from timenet_connectors.datasets.tsaqa.tsaqa_benchmark import connector
@@ -127,7 +127,7 @@ def test_storage_round_trip_keeps_the_whole_conversion(tmp_path):
     registry = LocalRegistry(tmp_path / "registry")
     registry.put_dataset(original)
     published = _published()
-    with TimeFReader(registry.open_version("tsaqa/tsaqa-benchmark", "1.0.0")) as reader:
+    with TimeFReader(open_local_version(registry.root / "tsaqa/tsaqa-benchmark/1.0.0")) as reader:
         restored = reader.read()
         assert {a.key: a.value for a in restored.annotations if a.key not in {"task", "question_type"}} == {
             "source_repository": "TSAQA/TSAQA-Benchmark",

packages/timenet-connectors/tests/test_build_plan.py

@@ -20,7 +20,7 @@
     TimeNetMissingParentError,
 )
 from timenet.reader import TimeFReader
-from timenet.registry import LocalRegistry, WritableRegistry
+from timenet.registry import LocalRegistry, WritableRegistry, open_local_version
 from timenet.registry._paths import release_dir
 from timenet.sources import InstalledConnectors
 from timenet.testing import install_backends, make_dataset
@@ -176,7 +176,7 @@ def _child_of(registry: LocalRegistry, reference: DatasetRef, parent: DatasetRef
         update={"dataset_id": reference.dataset_id, "parents": {ParentAlias("parent"): parent}}
     )
     child = TimeFDataset(metadata=metadata)
-    with TimeFReader(registry.open_version(parent.dataset_id, str(parent.version))) as reader:
+    with TimeFReader(open_local_version(release_dir(registry.root, parent))) as reader:
         child.import_from(reader, records=[record_id])
         own = Signal.from_values(
             np.arange(16, dtype=np.float32), spec=_SPEC, name="own", time_axis=RegularAxis.from_rate_hz(16)
@@ -435,8 +435,8 @@ def get_manifest(self, dataset_id, version=None):
     def open_file(self, dataset_id, version, relpath):
         return self._local.open_file(dataset_id, version, relpath)
 
-    def open_version(self, dataset_id, version=None):
-        return self._local.open_version(dataset_id, version)
+    def open_release(self, reference, **options):
+        return self._local.open_release(reference, **options)
 
     def put_dataset(self, dataset, **options):
         return self._local.put_dataset(dataset, **options)

packages/timenet/src/timenet/client.py

@@ -17,6 +17,7 @@
 from timenet.builders import installed_builder
 from timenet.config import settings
 from timenet.coordinator.build import plan_release_build, run_plan
+from timenet.coordinator.read import read_release
 from timenet.dataset import TimeFDataset
 from timenet.errors import (
     TimeFValidationError,
@@ -256,7 +257,7 @@ def open(
                     "fetch_from copies releases into a local registry; a remote client reads from its own registry"
                 )
             self._reject_unhosted_access(dataset_id, version)
-            return TimeFReader(registry.open_version(dataset_id, version))
+            return TimeFReader(read_release(dataset_id, version, sources=[registry], store=None))
         source = open_registry(fetch_from) if isinstance(fetch_from, str) else fetch_from
         try:
             return TimeFReader(_prepare(registry, dataset_id, version, auto_build=auto_build, fetch_from=source))

packages/timenet/src/timenet/coordinator/build.py

@@ -4,7 +4,7 @@
 from typing import assert_never
 
 from timenet.builders import BuildOptions, declared
-from timenet.coordinator.plan import BuildStep, CopyStep, Plan, plan_steps
+from timenet.coordinator.plan import BuildStep, CopyStep, Plan, ReadStep, plan_steps
 from timenet.manifest import PayloadSource
 from timenet.progress import StepFinished, StepStarted
 from timenet.registry import copy_release
@@ -58,21 +58,26 @@ def plan_release_build(  # ruff: ignore[too-many-arguments]
 
 
 def run_plan(plan: Plan, store: LocalRegistry, *, options: BuildOptions | None = None) -> None:
-    """Run a plan's copy and build steps in order.
+    """Run the copy and build steps of a plan in order.
+
+    Skip read steps.
 
     Args:
-        plan: The steps to run in dependency order.
+        plan: The steps to run.
         store: The local registry to copy and build into.
-        options: Build settings, including the progress callback for copies and builds.
+        options: Build settings, including the progress callback.
 
     Raises:
-        TimeNetVersionConflictError: If ``store`` holds different files for a copied release.
-        TimeFFormatError: If a build fails to reproduce pinned files.
+        TimeNetVersionConflictError: If a copied release conflicts with the files in ``store``.
+        TimeFFormatError: If a build changes files pinned by other releases.
         TimeNetBuildError: If a build fails.
     """  # ruff: ignore[docstring-extraneous-exception] - raised by the copy and the backends
     options = options or BuildOptions()
     for step in plan.steps:
         match step:
+            case ReadStep():
+                # A read opens the release where it is, so it has nothing to run.
+                pass
             case CopyStep():
                 copy_release(
                     step.release,