Skip to content
Snippets Groups Projects

Draft: Resolve "Daskify LISA Instrument"

Open Olaf Hartwig requested to merge 158-daskify-lisa-instrument into master
All threads resolved!
1 file
+ 84
0
Compare changes
  • Side-by-side
  • Inline
+ 84
0
"""Functions for applying FIR filters to dask arrays.
To create a fir filter operating on dask arrays, use the `make_filter_fir_dask` function.
"""
from __future__ import annotations
from typing import TypeAlias, Callable
import numpy as np
import dask
import dask.array as da
from lisainstrument.fir_filters_numpy import (
SemiLocalMapType,
EdgeHandling,
DefFilterFIR,
FIRCoreOp,
)
def semilocal_map_dask(
op: SemiLocalMapType,
bound_left: EdgeHandling,
bound_right: EdgeHandling,
data: dask.array,
) -> dask.array:
"""Apply a semi local map to dask array and employ boundary conditions.
Arguments:
op: the semi local mapping
bound_left: Boundary treatment on left side
bound_right: Boundary treatment on right side
data: the 1D array to be mapped
Returns:
The mapped data. The size is the same as the input if both boundary
conditions are ZEROPAD. A boundary condition VALID reduces the output
size by the corresponding margin size of the semilocal map.
"""
margin = max(op.margin_left, op.margin_right)
def ext_op(d: np.ndarray) -> np.ndarray:
r = op(d)
if op.margin_left < margin:
r = r[margin - op.margin_left :]
if op.margin_right < margin:
r = r[: -(margin - op.margin_right)]
return r
ext = da.map_overlap(ext_op, data, depth={0: margin}, boundary=0, trim=False)
if bound_left == EdgeHandling.VALID:
ext = ext[op.margin_left :]
if bound_right == EdgeHandling.VALID:
ext = ext[: -op.margin_right]
return ext
FilterFirDaskType: TypeAlias = Callable[[dask.array], dask.array]
def make_filter_fir_dask(
fdef: DefFilterFIR, bound_left: EdgeHandling, bound_right: EdgeHandling
) -> FilterFirDaskType:
"""Create a function that applies a given FIR filter to dask arrays,
employing the specified boundary treatment.
Arguments:
fdef: The definition of the FIR filter
bound_left: Boundary treatment on left side
bound_right: Boundary treatment on right side
Returns:
Function which accepts a single 1D dask array as input and returns
the filtered dask array.
"""
fmap = FIRCoreOp(fdef)
def op(data):
return semilocal_map_dask(fmap, bound_left, bound_right, data)
return op
Loading