Merge pull request #814 from danielballan/bluesky-tiled-plugins

Factor out `bluesky-tiled-plugins` package

.git_archival.txt

@@ -0,0 +1,4 @@
+node: $Format:%H$
+node-date: $Format:%cI$
+describe-name: $Format:%(describe:tags=true,match=*[0-9]*)$
+ref-names: $Format:%D$

.gitattributes

@@ -1 +1,2 @@
 databroker/_version.py export-subst
+.git_archival.txt  export-subst

.github/workflows/publish-pypi.yml

@@ -1,33 +1,47 @@
-# This workflows will upload a Python Package using flit when a release is created
-# For more information see: https://help.github.com/en/actions/language-and-framework-guides/using-python-with-github-actions#publishing-to-package-registries
-
-name: Upload Python Package
+name: CD
 
 on:
+  workflow_dispatch:
+  pull_request:
+  push:
+    branches:
+      - main
   release:
-    types: [created]
+    types:
+      - published
 
 jobs:
-  deploy:
+  dist:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      # Build two packages: databroker and bluesky-tiled-plugins.
 
+      - uses: hynek/build-and-inspect-python-package@v2
+        with:
+          path: .
+          upload-name-suffix: "-databroker"
+
+      - uses: hynek/build-and-inspect-python-package@v2
+        with:
+          path: bluesky-tiled-plugins/
+          upload-name-suffix: "-bluesky-tiled-plugins"
+
+  publish:
+    needs: [dist]
+    environment: pypi
+    permissions:
+      id-token: write
     runs-on: ubuntu-latest
+    if: github.event_name == 'release' && github.event.action == 'published'
 
     steps:
-    - uses: actions/checkout@v2
-    - name: Set up Python
-      uses: actions/setup-python@v2
-      with:
-        python-version: '3.x'
-    - name: Install dependencies
-      run: |
-        python -m pip install --upgrade pip
-        pip install wheel twine setuptools
-    - name: Build and publish
-      env:
-        TWINE_USERNAME: __token__
-        # The PYPI_PASSWORD must be a pypi token with the "pypi-" prefix with sufficient permissions to upload this package
-        # https://pypi.org/help/#apitoken
-        TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
-      run: |
-        python setup.py sdist bdist_wheel
-        twine upload dist/*
+      - uses: actions/download-artifact@v4
+        with:
+          name: Packages
+          path: dist
+
+      - uses: pypa/gh-action-pypi-publish@release/v1

bluesky-tiled-plugins/README.md

@@ -0,0 +1,10 @@
+# bluesky-tiled-plugins
+
+This is a separate Python package, `bluesky-tiled-plugins`, that is
+developed in the databroker repository.
+
+For a user wishing to connect to a running Tiled server and access Bluesky data,
+this package, along with its dependency `tiled[client]`, is all they need.
+
+The databroker package is only required if the user wants to use the legacy
+`databroker.Broker` API.

bluesky-tiled-plugins/bluesky_tiled_plugins/__init__.py

@@ -0,0 +1,3 @@
+from .catalog_of_bluesky_runs import CatalogOfBlueskyRuns  # noqa: F401
+from .bluesky_event_stream import BlueskyEventStream  # noqa: F401
+from .bluesky_run import BlueskyRun  # noqa: F401

bluesky-tiled-plugins/bluesky_tiled_plugins/_common.py

@@ -0,0 +1,6 @@
+# There are methods that IPython will try to call.
+# We special-case them because we want to avoid the getattr
+# resulting in an unnecessary network hit just to raise
+# AttributeError.
+
+IPYTHON_METHODS = {"_ipython_canary_method_should_not_exist_", "_repr_mimebundle_"}

bluesky-tiled-plugins/bluesky_tiled_plugins/bluesky_event_stream.py

@@ -0,0 +1,83 @@
+import keyword
+import warnings
+
+from tiled.client.container import DEFAULT_STRUCTURE_CLIENT_DISPATCH, Container
+
+from ._common import IPYTHON_METHODS
+
+
+class BlueskyEventStream(Container):
+    """
+    This encapsulates the data and metadata for one 'stream' in a Bluesky 'run'.
+
+    This adds for bluesky-specific conveniences to the standard client Container.
+    """
+
+    def __repr__(self):
+        return f"<{type(self).__name__} {set(self)!r} stream_name={self.metadata['stream_name']!r}>"
+
+    @property
+    def descriptors(self):
+        return self.metadata["descriptors"]
+
+    @property
+    def _descriptors(self):
+        # For backward-compatibility.
+        # We do not normally worry about backward-compatibility of _ methods, but
+        # for a time databroker.v2 *only* have _descriptors and not descriptros,
+        # and I know there is useer code that relies on that.
+        warnings.warn("Use .descriptors instead of ._descriptors.", stacklevel=2)
+        return self.descriptors
+
+    def __getattr__(self, key):
+        """
+        Let run.X be a synonym for run['X'] unless run.X already exists.
+
+        This behavior is the same as with pandas.DataFrame.
+        """
+        # The wisdom of this kind of "magic" is arguable, but we
+        # need to support it for backward-compatibility reasons.
+        if key in IPYTHON_METHODS:
+            raise AttributeError(key)
+        if key in self:
+            return self[key]
+        raise AttributeError(key)
+
+    def __dir__(self):
+        # Build a list of entries that are valid attribute names
+        # and add them to __dir__ so that they tab-complete.
+        tab_completable_entries = [
+            entry
+            for entry in self
+            if (entry.isidentifier() and (not keyword.iskeyword(entry)))
+        ]
+        return super().__dir__() + tab_completable_entries
+
+    def read(self, *args, **kwargs):
+        """
+        Shortcut for reading the 'data' (as opposed to timestamps or config).
+
+        That is:
+
+        >>> stream.read(...)
+
+        is equivalent to
+
+        >>> stream["data"].read(...)
+        """
+        return self["data"].read(*args, **kwargs)
+
+    def to_dask(self):
+        warnings.warn(
+            """Do not use this method.
+Instead, set dask or when first creating the client, as in
+
+    >>> catalog = from_uri("...", "dask")
+
+and then read() will return dask objects.""",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        return self.new_variation(
+            structure_clients=DEFAULT_STRUCTURE_CLIENT_DISPATCH["dask"]
+        ).read()

bluesky-tiled-plugins/bluesky_tiled_plugins/bluesky_run.py

@@ -0,0 +1,147 @@
+import json
+import keyword
+import warnings
+from datetime import datetime
+
+from tiled.client.container import Container
+from tiled.client.utils import handle_error
+
+from ._common import IPYTHON_METHODS
+from .document import Start, Stop, Descriptor, EventPage, DatumPage, Resource
+
+
+_document_types = {
+    "start": Start,
+    "stop": Stop,
+    "descriptor": Descriptor,
+    "event_page": EventPage,
+    "datum_page": DatumPage,
+    "resource": Resource,
+}
+
+
+class BlueskyRun(Container):
+    """
+    This encapsulates the data and metadata for one Bluesky 'run'.
+
+    This adds for bluesky-specific conveniences to the standard client Container.
+    """
+
+    def __repr__(self):
+        metadata = self.metadata
+        datetime_ = datetime.fromtimestamp(metadata["start"]["time"])
+        return (
+            f"<{type(self).__name__} "
+            f"{set(self)!r} "
+            f"scan_id={metadata['start'].get('scan_id', 'UNSET')!s} "  # (scan_id is optional in the schema)
+            f"uid={metadata['start']['uid'][:8]!r} "  # truncated uid
+            f"{datetime_.isoformat(sep=' ', timespec='minutes')}"
+            ">"
+        )
+
+    @property
+    def start(self):
+        """
+        The Run Start document. A convenience alias:
+
+        >>> run.start is run.metadata["start"]
+        True
+        """
+        return self.metadata["start"]
+
+    @property
+    def stop(self):
+        """
+        The Run Stop document. A convenience alias:
+
+        >>> run.stop is run.metadata["stop"]
+        True
+        """
+        return self.metadata["stop"]
+
+    @property
+    def v2(self):
+        return self
+
+    def documents(self, fill=False):
+        # For back-compat with v2:
+        if fill == "yes":
+            fill = True
+        elif fill == "no":
+            fill = False
+        elif fill == "delayed":
+            raise NotImplementedError("fill='delayed' is not supported")
+        else:
+            fill = bool(fill)
+        link = self.item["links"]["self"].replace("/metadata", "/documents", 1)
+        with self.context.http_client.stream(
+            "GET",
+            link,
+            params={"fill": fill},
+            headers={"Accept": "application/json-seq"},
+        ) as response:
+            if response.is_error:
+                response.read()
+                handle_error(response)
+            tail = ""
+            for chunk in response.iter_bytes():
+                for line in chunk.decode().splitlines(keepends=True):
+                    if line[-1] == "\n":
+                        item = json.loads(tail + line)
+                        yield (item["name"], _document_types[item["name"]](item["doc"]))
+                        tail = ""
+                    else:
+                        tail += line
+            if tail:
+                item = json.loads(tail)
+                yield (item["name"], _document_types[item["name"]](item["doc"]))
+
+    def __getattr__(self, key):
+        """
+        Let run.X be a synonym for run['X'] unless run.X already exists.
+
+        This behavior is the same as with pandas.DataFrame.
+        """
+        # The wisdom of this kind of "magic" is arguable, but we
+        # need to support it for backward-compatibility reasons.
+        if key in IPYTHON_METHODS:
+            raise AttributeError(key)
+        if key in self:
+            return self[key]
+        raise AttributeError(key)
+
+    def __dir__(self):
+        # Build a list of entries that are valid attribute names
+        # and add them to __dir__ so that they tab-complete.
+        tab_completable_entries = [
+            entry
+            for entry in self
+            if (entry.isidentifier() and (not keyword.iskeyword(entry)))
+        ]
+        return super().__dir__() + tab_completable_entries
+
+    def describe(self):
+        "For back-compat with intake-based BlueskyRun"
+        warnings.warn(
+            "This will be removed. Use .metadata directly instead of describe()['metadata'].",
+            DeprecationWarning,
+        )
+        return {"metadata": self.metadata}
+
+    def __call__(self):
+        warnings.warn(
+            "Do not call a BlueskyRun. For now this returns self, for "
+            "backward-compatibility. but it will be removed in a future "
+            "release.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        return self
+
+    def read(self):
+        raise NotImplementedError(
+            "Reading any entire run is not supported. "
+            "Access a stream in this run and read that."
+        )
+
+    to_dask = read