Phase doesn't work inside a Dask Array

[theXYZT]

Note: I am just curious here and not suggesting this MUST be a feature of baseband-tasks. But I am wondering if something simple and obvious in the definition of Phase is preventing this from happening.

Here's something I tried:

In [8]: x = da.array(np.ones(8) * u.Hz)

In [9]: x
Out[9]: dask.array<array, shape=(8,), dtype=float64, chunksize=(8,), chunktype=astropy.Quantity>

In [10]: x * 2
Out[10]: dask.array<mul, shape=(8,), dtype=float64, chunksize=(8,), chunktype=astropy.Quantity>

In [11]: (x * 2).compute()
Out[11]: <Quantity [2., 2., 2., 2., 2., 2., 2., 2.] Hz>

In [12]: x = da.array(Phase([1.5, 2.5]))

In [13]: x
Out[13]: dask.array<array, shape=(2,), dtype=[('int', '<f8'), ('frac', '<f8')], chunksize=(2,), chunktype=Phase>

In [14]: x.compute()
Out[14]: Phase([2. 2.] cycle, [-0.5  0.5] cycle)

In [15]: x * 2
---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-15-80b83cb135a7> in <module>
----> 1 x * 2

TypeError: unsupported operand type(s) for *: 'Array' and 'int'

As you can see, Quantity arrays work perfectly fine inside dask arrays, and even successfully work when multiplying with a number, so the np.mulitply ufunc is being applied and propagated through dask. The same is true for the astropy.coordinates.Angle class. But this doesn't work for Phase!

Phase subclasses from Angle and seems to be sensible in its definition of __array_ufunc__ at first glance. It would be neat to figure out what's creating this problem and to fix it!

[mhvk]

Interesting that Quantity even works! But, given that, puzzling indeed that Phase does not work. It seems the problem arises here: https://github.com/dask/dask/blob/c00e01b8ba414d4fa645f74933f83b16387ef4e4/dask/array/core.py#L4128-L4131

[mhvk]

So, the problem is that dask does not take into account that the phase class may be able to deal with the structured dtype even though a plan ndarray cannot. So, can one avoid that test? Looking closer at the code, it seems one can, by passing in a dtype keyword argument:

np.multiply(x, 2, dtype=x.dtype).compute()
# Phase([3. 5.] cycle, [0. 0.] cycle)

In this case at least, that also works if x is just a plain Phase instance...