Merge pull request #519 from scipp/fix-conda-build

Fix conda build

Author: jl-wynen

.buildconfig/ci-linux.yml

@@ -41,4 +41,4 @@ dependencies:
   - sphinxcontrib-bibtex==2.6.2
 
   # docs and tests
-  - sciline==24.01.1
+  - sciline==24.04.1

.copier-answers.yml

@@ -1,5 +1,5 @@
 # Changes here will be overwritten by Copier; NEVER EDIT MANUALLY
-_commit: cac2bde
+_commit: 9bb9876
 _src_path: gh:scipp/copier_template
 description: Neutron scattering tools for Data Reduction
 max_python: '3.12'

conda/meta.yaml

@@ -39,6 +39,7 @@ test:
     - pythreejs==2.4.2
     - python-confluent-kafka==2.1.1 [linux64]
     - ess-streaming-data-types==v0.14.0 [linux64]
+    - sciline==24.04.1
   source_files:
     - pyproject.toml
     - tests/

docs/conf.py

@@ -1,10 +1,15 @@
 import doctest
 import os
 import sys
+from importlib.metadata import PackageNotFoundError
 from importlib.metadata import version as get_version
 
+from sphinx.util import logging
+
 sys.path.insert(0, os.path.abspath('.'))
 
+logger = logging.getLogger(__name__)
+
 # General information about the project.
 project = 'ScippNeutron'
 copyright = '2024 Scipp contributors'
@@ -33,6 +38,8 @@
     import sciline.sphinxext.domain_types  # noqa: F401
 
     extensions.append('sciline.sphinxext.domain_types')
+    # See https://github.com/tox-dev/sphinx-autodoc-typehints/issues/457
+    suppress_warnings = ["config.cache"]
 except ModuleNotFoundError:
     pass
 
@@ -119,8 +126,15 @@
 # built documents.
 #
 
-release = get_version("scippneutron")
-version = ".".join(release.split('.')[:3])  # CalVer
+try:
+    release = get_version("scippneutron")
+    version = ".".join(release.split('.')[:3])  # CalVer
+except PackageNotFoundError:
+    logger.info(
+        "Warning: determining version from package metadata failed, falling back to "
+        "a dummy version number."
+    )
+    release = version = "0.0.0-dev"
 
 warning_is_error = True
 

docs/developer/data-stream.rst

@@ -180,9 +180,9 @@ Unit Testing
 
 For unit tests it would be convenient to use a fake consumer object in place of ``KafkaConsumer``
 instances. However, any input arguments or variables passed via the queue to the ``mp.Process``
-must be pickleable or ``mp.Queue``. This makes dependency injection difficult. To get around
+must be pickleable or ``mp.queues.Queue``. This makes dependency injection difficult. To get around
 this an enum can be passed via ``data_stream`` to the ``data_consumption_manager`` to tell it
-to create instances of ``FakeConsumer`` instead of ``KafkaConsumer``, additionally an ``mp.Queue``
+to create instances of ``FakeConsumer`` instead of ``KafkaConsumer``, additionally an ``mp.queues.Queue``
 can be provided and is passed to the ``FakeConsumer``. The ``FakeConsumer`` simply polls for messages
 on the queue instead of Kafka, thus allowing the test to provide the messages. There is no other
 configuration of ``FakeConsumer`` possible or necessary.

pyproject.toml

@@ -102,7 +102,7 @@ extend-exclude = [
 
 [tool.ruff.lint]
 # See https://docs.astral.sh/ruff/rules/
-select = ["B", "C4", "DTZ", "E", "F", "G", "I", "PERF", "PGH", "PT", "PYI", "RUF", "S", "T20", "W"]
+select = ["B", "C4", "DTZ", "E", "F", "G", "I", "PERF", "PGH", "PT", "PYI", "RUF", "S", "T20", "UP", "W"]
 ignore = [
     # Conflict with ruff format, see
     # https://docs.astral.sh/ruff/formatter/#conflicting-lint-rules

src/scippneutron/data_streaming/_consumer.py

@@ -19,6 +19,7 @@
 """
 
 import multiprocessing as mp
+import multiprocessing.queues
 import threading
 from collections.abc import Callable
 from queue import Empty as QueueEmpty
@@ -43,7 +44,7 @@ class FakeConsumer:
     to avoid network io in unit tests
     """
 
-    def __init__(self, input_queue: mp.Queue | None):
+    def __init__(self, input_queue: mp.queues.Queue | None):
         if input_queue is None:
             raise RuntimeError(
                 "A multiprocessing queue for test messages "
@@ -237,7 +238,7 @@ def create_consumers(
     kafka_broker: str,
     consumer_type_enum: ConsumerType,  # so we can inject fake consumer
     callback: Callable,
-    test_message_queue: mp.Queue | None,
+    test_message_queue: mp.queues.Queue | None,
 ) -> list[KafkaConsumer]:
     """
     Creates one consumer per TopicPartition that start consuming

src/scippneutron/data_streaming/_data_buffer.py

@@ -1,6 +1,7 @@
 # SPDX-License-Identifier: BSD-3-Clause
 # Copyright (c) 2023 Scipp contributors (https://github.com/scipp)
 import multiprocessing as mp
+import multiprocessing.queues
 import threading
 from collections.abc import Callable
 from datetime import datetime
@@ -108,7 +109,9 @@ class _FastMetadataBuffer:
     rather than via EPICS and the Forwarder.
     """
 
-    def __init__(self, stream_info: StreamInfo, buffer_size: int, data_queue: mp.Queue):
+    def __init__(
+        self, stream_info: StreamInfo, buffer_size: int, data_queue: mp.queues.Queue
+    ):
         self._buffer_mutex = threading.Lock()
         self._buffer_size = buffer_size
         self._name = stream_info.source_name
@@ -202,7 +205,9 @@ class _ChopperMetadataBuffer:
     serialised according to the flatbuffer schema with id CHOPPER_FB_ID.
     """
 
-    def __init__(self, stream_info: StreamInfo, buffer_size: int, data_queue: mp.Queue):
+    def __init__(
+        self, stream_info: StreamInfo, buffer_size: int, data_queue: mp.queues.Queue
+    ):
         self._buffer_mutex = threading.Lock()
         self._buffer_size = buffer_size
         self._name = stream_info.source_name
@@ -270,7 +275,7 @@ class StreamedDataBuffer:
 
     def __init__(
         self,
-        queue: mp.Queue,
+        queue: mp.queues.Queue,
         event_buffer_size: int,
         slow_metadata_buffer_size: int,
         fast_metadata_buffer_size: int,

src/scippneutron/data_streaming/_data_consumption_manager.py

@@ -1,6 +1,7 @@
 # SPDX-License-Identifier: BSD-3-Clause
 # Copyright (c) 2023 Scipp contributors (https://github.com/scipp)
 import multiprocessing as mp
+import multiprocessing.queues
 from dataclasses import dataclass
 from enum import Enum
 from queue import Empty as QueueEmpty
@@ -40,15 +41,15 @@ def data_consumption_manager(
     slow_metadata_buffer_size: int,
     fast_metadata_buffer_size: int,
     chopper_buffer_size: int,
-    worker_instruction_queue: mp.Queue,
-    data_queue: mp.Queue,
-    test_message_queue: mp.Queue | None,
+    worker_instruction_queue: mp.queues.Queue,
+    data_queue: mp.queues.Queue,
+    test_message_queue: mp.queues.Queue | None,
 ):
     """
     Starts and stops buffers and data consumers which collect data and
     send them back to the main process via a queue.
 
-    All input args must be mp.Queue or pickleable as this function is launched
+    All input args must be mp.queues.Queue or pickleable as this function is launched
     as a multiprocessing.Process.
     """
     buffer = StreamedDataBuffer(

src/scippneutron/data_streaming/data_stream.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import multiprocessing as mp
+import multiprocessing.queues
 import time
 from collections.abc import Generator
 from enum import Enum
@@ -136,16 +137,16 @@ def validate_buffer_size_args(
             raise ValueError(f"{buffer_name} must be greater than zero")
 
 
-def _cleanup_queue(queue: mp.Queue | None):
+def _cleanup_queue(queue: mp.queues.Queue | None):
     if queue is not None:
         queue.cancel_join_thread()
         queue.close()
         queue.join_thread()
 
 
 async def _data_stream(
-    data_queue: mp.Queue,
-    worker_instruction_queue: mp.Queue,
+    data_queue: mp.queues.Queue,
+    worker_instruction_queue: mp.queues.Queue,
     kafka_broker: str,
     topics: list[str] | None,
     interval: sc.Variable,
@@ -160,7 +161,7 @@ async def _data_stream(
     consumer_type: ConsumerType = ConsumerType.REAL,
     halt_after_n_data_chunks: int = np.iinfo(np.int32).max,
     halt_after_n_warnings: int = np.iinfo(np.int32).max,
-    test_message_queue: mp.Queue | None = None,  # for tests
+    test_message_queue: mp.queues.Queue | None = None,  # for tests
     timeout: sc.Variable | None = None,  # for tests
 ) -> Generator[sc.DataArray, None, None]:
     """