bcg-x-official / fluxus Goto Github PK

.. Begin-Badges

|pypi| |conda| |python_versions| |code_style| |made_with_sphinx_doc| |License_badge|GHA|TestCov|Contributor_Convenant

.. End-Badges

.. Begin-Badges

.. |conda| image:: https://anaconda.org/bcg_gamma/gamma-facet/badges/version.svg
    :target: https://anaconda.org/BCG_Gamma/fluxus

.. |pypi| image:: https://badge.fury.io/py/artkit.svg
    :target: https://pypi.org/project/fluxus/

.. |python_versions| image:: https://img.shields.io/badge/python-3.10|3.11|3.12-blue.svg
   :target: https://www.python.org/downloads/release/python-3100/

.. |code_style| image:: https://img.shields.io/badge/code%20style-black-000000.svg
   :target: https://github.com/psf/black

.. |made_with_sphinx_doc| image:: https://img.shields.io/badge/Made%20with-Sphinx-1f425f.svg
   :target: https://bcg-x-official.github.io/facet/index.html

.. |license_badge| image:: https://img.shields.io/badge/License-Apache%202.0-olivegreen.svg
   :target: https://opensource.org/licenses/Apache-2.0

.. End-Badges

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
Cell In[154], line 11
      4     return [{"input": i} for i in range(3)]
      7 pipeline = ak.chain(
      8     ak.step("toy", toy),
      9     ak.step("toy", toy),
     10 )
---> 11 ak.run(pipeline, input={"input": 1}).to_frame()

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/fluxus/functional/_functions.py:753, in run(steps, input, timestamps)
    750     return arun((steps >> consumer).arun())
    752 elif isinstance(steps, BaseTransformer):
--> 753     return arun(
    754         (
    755             DictProducer(lambda: {} if input is None else input, name="input")
    756             >> steps
    757             >> consumer
    758         ).arun()
    759     )
    761 else:
    762     message = (
    763         f"arg steps must be a step or composition of steps, but got a "
    764         f"{type(steps).__qualname__}"
    765     )

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/pytools/asyncio/_asyncio.py:252, in arun(coroutine)
    247 try:
    248     # Use ThreadPoolExecutor to run the event loop in a new thread
    249     with ThreadPoolExecutor() as executor:
    250         # Schedule the execution of the coroutine in the new event loop,
    251         # and return the result once the coroutine has completed
--> 252         return executor.submit(_run_event_loop, new_loop, coroutine).result()
    253 except ExceptionGroup as e:
    254     # One or more coroutines raised exceptions. These exceptions are combined
    255     # into a (possibly nested) ExceptionGroup, which we catch here.
    256     # Traverse the nested exception groups to find the first atomic
    257     # sub-exception.
    258     exception_group = e

File ~/.pyenv/versions/3.11.9/lib/python3.11/concurrent/futures/_base.py:456, in Future.result(self, timeout)
    454     raise CancelledError()
    455 elif self._state == FINISHED:
--> 456     return self.__get_result()
    457 else:
    458     raise TimeoutError()

File ~/.pyenv/versions/3.11.9/lib/python3.11/concurrent/futures/_base.py:401, in Future.__get_result(self)
    399 if self._exception:
    400     try:
--> 401         raise self._exception
    402     finally:
    403         # Break a reference cycle with the exception in self._exception
    404         self = None

File ~/.pyenv/versions/3.11.9/lib/python3.11/concurrent/futures/thread.py:58, in _WorkItem.run(self)
     55     return
     57 try:
---> 58     result = self.fn(*self.args, **self.kwargs)
     59 except BaseException as exc:
     60     self.future.set_exception(exc)

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/pytools/asyncio/_asyncio.py:220, in arun.<locals>._run_event_loop(loop, awaitable)
    218 def _run_event_loop(loop: asyncio.AbstractEventLoop, awaitable: Awaitable[T]) -> T:
    219     asyncio.set_event_loop(loop)
--> 220     return loop.run_until_complete(awaitable)

File ~/.pyenv/versions/3.11.9/lib/python3.11/asyncio/base_events.py:654, in BaseEventLoop.run_until_complete(self, future)
    651 if not future.done():
    652     raise RuntimeError('Event loop stopped before Future completed.')
--> 654 return future.result()

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/fluxus/core/producer/_chained_.py:126, in _ProducerFlow.arun(self)
    123 """[see superclass]"""
    125 # noinspection PyProtectedMember
--> 126 return await _aconsume(producer=self._producer, consumer=self._consumer)

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/fluxus/core/producer/_chained_.py:274, in _aconsume(producer, consumer)
    271         yield (producer_index, product)
    273 # noinspection PyTypeChecker
--> 274 return await consumer.aconsume(
    275     async_flatten(
    276         _annotate(producer_index, producer)
    277         async for producer_index, producer in (
    278             aenumerate(producer.aiter_concurrent_conduits())
    279         )
    280     )
    281 )

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/fluxus/functional/conduit/_consumer.py:92, in DictConsumer.aconsume(self, products)
     89 run_start = time.perf_counter()
     90 cumulative_time = 0.0
---> 92 async for group, end_product in products:
     93     # Get the lineage attributes for the end product, which is a dictionary
     94     # of dictionaries. The outer dictionary maps product names to their
     95     # attributes, and the inner dictionaries map attribute names to their
     96     # values.
     97     attributes = end_product.get_lineage_attributes()
     99     # Iterate over products and their attribute dicts

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/pytools/asyncio/_asyncio.py:166, in _async_flatten_310(async_iter_of_iters)
    163     yield value
    165 # Wait for the producer task to complete
--> 166 await producer_task

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/pytools/asyncio/_asyncio.py:145, in _async_flatten_310.<locals>._producer()
    141 async def _producer() -> None:
    142     try:
    143 
    144         # Wait for all tasks to complete, propagating any exceptions.
--> 145         await asyncio.gather(
    146             *[
    147                 asyncio.create_task(_process_nested(async_iter))
    148                 async for async_iter in async_iter_of_iters
    149             ]
    150         )
    151     finally:
    152         # Signal the end of processing by putting a sentinel value in the queue.
    153         await queue.put(cast(T, _END))

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/pytools/asyncio/_asyncio.py:138, in _async_flatten_310.<locals>._process_nested(async_iter)
    137 async def _process_nested(async_iter: AsyncIterable[T]) -> None:
--> 138     async for _value in async_iter:
    139         await queue.put(_value)

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/fluxus/core/producer/_chained_.py:270, in _aconsume.<locals>._annotate(producer_index, producer_)
    267 async def _annotate(
    268     producer_index: int, producer_: SerialProducer[T_SourceProduct_arg]
    269 ) -> AsyncIterator[tuple[int, T_SourceProduct_arg]]:
--> 270     async for product in producer_:
    271         yield (producer_index, product)

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/pytools/asyncio/_asyncio.py:166, in _async_flatten_310(async_iter_of_iters)
    163     yield value
    165 # Wait for the producer task to complete
--> 166 await producer_task

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/pytools/asyncio/_asyncio.py:145, in _async_flatten_310.<locals>._producer()
    141 async def _producer() -> None:
    142     try:
    143 
    144         # Wait for all tasks to complete, propagating any exceptions.
--> 145         await asyncio.gather(
    146             *[
    147                 asyncio.create_task(_process_nested(async_iter))
    148                 async for async_iter in async_iter_of_iters
    149             ]
    150         )
    151     finally:
    152         # Signal the end of processing by putting a sentinel value in the queue.
    153         await queue.put(cast(T, _END))

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/pytools/asyncio/_asyncio.py:138, in _async_flatten_310.<locals>._process_nested(async_iter)
    137 async def _process_nested(async_iter: AsyncIterable[T]) -> None:
--> 138     async for _value in async_iter:
    139         await queue.put(_value)

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/pytools/asyncio/_asyncio.py:166, in _async_flatten_310(async_iter_of_iters)
    163     yield value
    165 # Wait for the producer task to complete
--> 166 await producer_task

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/pytools/asyncio/_asyncio.py:146, in _async_flatten_310.<locals>._producer()
    141 async def _producer() -> None:
    142     try:
    143 
    144         # Wait for all tasks to complete, propagating any exceptions.
    145         await asyncio.gather(
--> 146             *[
    147                 asyncio.create_task(_process_nested(async_iter))
    148                 async for async_iter in async_iter_of_iters
    149             ]
    150         )
    151     finally:
    152         # Signal the end of processing by putting a sentinel value in the queue.
    153         await queue.put(cast(T, _END))

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/pytools/asyncio/_asyncio.py:146, in <listcomp>(.0)
    141 async def _producer() -> None:
    142     try:
    143 
    144         # Wait for all tasks to complete, propagating any exceptions.
    145         await asyncio.gather(
--> 146             *[
    147                 asyncio.create_task(_process_nested(async_iter))
    148                 async for async_iter in async_iter_of_iters
    149             ]
    150         )
    151     finally:
    152         # Signal the end of processing by putting a sentinel value in the queue.
    153         await queue.put(cast(T, _END))

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/fluxus/core/transformer/_chained_.py:196, in <genexpr>(.0)
    193 """[see superclass]"""
    195 # noinspection PyTypeChecker
--> 196 return async_flatten(
    197     self.second.atransform(tx_first)
    198     async for tx_first in self.first.atransform(source_product)
    199 )

File ~/.pyenv/versions/3.11.9/envs/rai_red_ft/lib/python3.11/site-packages/fluxus/functional/conduit/_step.py:270, in Step.atransform(self, source_product)
    267 end_time = time.perf_counter()
    269 if not isinstance(attributes, Mapping):
--> 270     raise TypeError(
    271         f"Expected function {self._function.__name__}() of step "
    272         f"{self.name!r} to return a Mapping or dict, but got: "
    273         f"{attributes!r}"
    274     )
    276 log.debug(
    277     f"Completed step {self.name!r} in {end_time - start_time:g} "
    278     f"seconds:\n"
   (...)
    285     )
    286 )
    288 yield DictProduct(
    289     name=self.name,
    290     product_attributes=attributes,
   (...)
    293     end_time=end_time,
    294 )

TypeError: Expected function toy() of step 'toy' to return a Mapping or dict, but got: [{'input': 0}, {'input': 1}, {'input': 2}]