content: from __future__ import annotations
  import copy
  from enum import Enum
  from packaging.version import Version
  import numpy as np
  from datashader.datashape import dshape, isnumeric, Record, Option
  from datashader.datashape import coretypes as ct
  from toolz import concat, unique
  import xarray as xr
  
  from datashader.antialias import AntialiasCombination, AntialiasStage2
  from datashader.utils import isminus1, isnull
  from numba import cuda as nb_cuda
  
  try:
      from datashader.transfer_functions._cuda_utils import (
          cuda_atomic_nanmin, cuda_atomic_nanmax, cuda_args, cuda_row_min_in_place,
          cuda_nanmax_n_in_place_4d, cuda_nanmax_n_in_place_3d,
          cuda_nanmin_n_in_place_4d, cuda_nanmin_n_in_place_3d,
          cuda_row_max_n_in_place_4d, cuda_row_max_n_in_place_3d,
          cuda_row_min_n_in_place_4d, cuda_row_min_n_in_place_3d, cuda_shift_and_insert,
      )
  except ImportError:
      (cuda_atomic_nanmin, cuda_atomic_nanmax, cuda_args, cuda_row_min_in_place,
          cuda_nanmax_n_in_place_4d, cuda_nanmax_n_in_place_3d,
          cuda_nanmin_n_in_place_4d, cuda_nanmin_n_in_place_3d,
          cuda_row_max_n_in_place_4d, cuda_row_max_n_in_place_3d,
          cuda_row_min_n_in_place_4d, cuda_row_min_n_in_place_3d, cuda_shift_and_insert,
      ) = None, None, None, None, None, None, None, None, None, None, None, None, None
  
  try:
      import cudf
      import cupy as cp
  except Exception:
      cudf = cp = None
  
  from .utils import (
      Expr, ngjit, nansum_missing, nanmax_in_place, nansum_in_place, row_min_in_place,
      nanmax_n_in_place_4d, nanmax_n_in_place_3d, nanmin_n_in_place_4d, nanmin_n_in_place_3d,
      row_max_n_in_place_4d, row_max_n_in_place_3d, row_min_n_in_place_4d, row_min_n_in_place_3d,
      shift_and_insert,
  )
  
  
  class SpecialColumn(Enum):
      """
      Internally datashader identifies the columns required by the user's
      Reductions and extracts them from the supplied source (e.g. DataFrame) to
      pass through the dynamically-generated append function in compiler.py and
      end up as arguments to the Reduction._append* functions. Each column is
      a string name or a SpecialColumn. A column of None is used in Reduction
      classes to denote that no column is required.
      """
      RowIndex = 1
  
  
  class UsesCudaMutex(Enum):
      """
      Enum that encapsulates the need for a Reduction to use a CUDA mutex to
      operate correctly on a GPU. Possible values:
  
      No: the Reduction append_cuda function is atomic and no mutex is required.
      Local: Reduction append_cuda needs wrapping in a mutex.
      Global: the overall compiled append function needs wrapping in a mutex.
      """
      No = 0
      Local = 1
      Global = 2
  
  
  class Preprocess(Expr):
      """Base class for preprocessing steps."""
      def __init__(self, column: str | SpecialColumn | None):
          self.column = column
  
      @property
      def inputs(self):
          return (self.column,)
  
      @property
      def nan_check_column(self):
          return None
  
  
  class extract(Preprocess):
      """Extract a column from a dataframe as a numpy array of values."""
      def apply(self, df, cuda):
          if self.column is SpecialColumn.RowIndex:
              attr_name = "_datashader_row_offset"
              if isinstance(df, xr.Dataset):
                  row_offset = df.attrs[attr_name]
                  row_length = df.attrs["_datashader_row_length"]
              else:
                  attrs = getattr(df, "attrs", None)
                  row_offset = getattr(attrs or df, attr_name, 0)
                  row_length = len(df)
  
          if cudf and isinstance(df, cudf.DataFrame):
              if self.column is SpecialColumn.RowIndex:
                  return cp.arange(row_offset, row_offset + row_length, dtype=np.int64)
  
              if df[self.column].dtype.kind == 'f':
                  nullval = np.nan
              else:
                  nullval = 0
              if Version(cudf.__version__) >= Version("22.02"):
                  return df[self.column].to_cupy(na_value=nullval)
              return cp.array(df[self.column].to_gpu_array(fillna=nullval))
          elif self.column is SpecialColumn.RowIndex:
              if cuda:
                  return cp.arange(row_offset, row_offset + row_length, dtype=np.int64)
              else:
                  return np.arange(row_offset, row_offset + row_length, dtype=np.int64)
          elif isinstance(df, xr.Dataset):
              if cuda and not isinstance(df[self.column].data, cp.ndarray):
                  return cp.asarray(df[self.column])
              else:
                  return df[self.column].data
          else:
              return df[self.column].values
  
  
  class CategoryPreprocess(Preprocess):
      """Base class for categorizing preprocessors."""
      @property
      def cat_column(self):
          """Returns name of categorized column"""
          return self.column
  
      def categories(self, input_dshape):
          """Returns list of categories corresponding to input shape"""
          raise NotImplementedError("categories not implemented")
  
      def validate(self, in_dshape):
          """Validates input shape"""
          raise NotImplementedError("validate not implemented")
  
      def apply(self, df, cuda):
          """Applies preprocessor to DataFrame and returns array"""
          raise NotImplementedError("apply not implemented")
  
  
  class category_codes(CategoryPreprocess):
      """
      Extract just the category codes from a categorical column.
  
      To create a new type of categorizer, derive a subclass from this
      class or one of its subclasses, implementing ``__init__``,
      ``_hashable_inputs``, ``categories``, ``validate``, and ``apply``.
  
      See the implementation of ``category_modulo`` in ``reductions.py``
      for an example.
      """
      def categories(self, input_dshape):
          return input_dshape.measure[self.column].categories
  
      def validate(self, in_dshape):
          if self.column not in in_dshape.dict:
              raise ValueError("specified column not found")
          if not isinstance(in_dshape.measure[self.column], ct.Categorical):
              raise ValueError("input must be categorical")
  
      def apply(self, df, cuda):
          if cudf and isinstance(df, cudf.DataFrame):
              if Version(cudf.__version__) >= Version("22.02"):
                  return df[self.column].cat.codes.to_cupy()
              return df[self.column].cat.codes.to_gpu_array()
          else:
              return df[self.column].cat.codes.values
  
  class category_modulo(category_codes):
      """
      A variation on category_codes that assigns categories using an integer column, modulo a base.
      Category is computed as (column_value - offset)%modulo.
      """
  
      # couldn't find anything in the datashape docs about how to check if a CType is an integer, so
      # just define a big set
      IntegerTypes = {ct.bool_, ct.uint8, ct.uint16, ct.uint32, ct.uint64, ct.int8, ct.int16,
                      ct.int32, ct.int64}
  
      def __init__(self, column, modulo, offset=0):
          super().__init__(column)
          self.offset = offset
          self.modulo = modulo
  
      def _hashable_inputs(self):
          return super()._hashable_inputs() + (self.offset, self.modulo)
  
      def categories(self, in_dshape):
          return list(range(self.modulo))
  
      def validate(self, in_dshape):
          if self.column not in in_dshape.dict:
              raise ValueError("specified column not found")
          if in_dshape.measure[self.column] not in self.IntegerTypes:
              raise ValueError("input must be an integer column")
  
      def apply(self, df, cuda):
          result = (df[self.column] - self.offset) % self.modulo
          if cudf and isinstance(df, cudf.Series):
              if Version(cudf.__version__) >= Version("22.02"):
                  return result.to_cupy()
              return result.to_gpu_array()
          else:
              return result.values
  
  class category_binning(category_modulo):
      """
      A variation on category_codes that assigns categories by binning a continuous-valued column.
      The number of categories returned is always nbins+1.
      The last category (nbin) is for NaNs in the data column, as well as for values under/over the
      binned interval (when include_under or include_over is False).
  
      Parameters
      ----------
      column:   column to use
      lower:    lower bound of first bin
      upper:    upper bound of last bin
      nbins:     number of bins
      include_under: if True, values below bin 0 are assigned to category 0
      include_over:  if True, values above the last bin (nbins-1) are assigned to category nbin-1
      """
  
      def __init__(self, column, lower, upper, nbins, include_under=True, include_over=True):
          super().__init__(column, nbins + 1)  # +1 category for NaNs and clipped values
          self.bin0 = lower
          self.binsize = (upper - lower) / float(nbins)
          self.nbins = nbins
          self.bin_under = 0 if include_under else nbins
          self.bin_over  = nbins-1 if include_over else nbins
  
      def _hashable_inputs(self):
          return super()._hashable_inputs() + (self.bin0, self.binsize, self.bin_under, self.bin_over)
  
      def validate(self, in_dshape):
          if self.column not in in_dshape.dict:
              raise ValueError("specified column not found")
  
      def apply(self, df, cuda):
          if cudf and isinstance(df, cudf.DataFrame):
              if Version(cudf.__version__) >= Version("22.02"):
                  values = df[self.column].to_cupy(na_value=cp.nan)
              else:
                  values = cp.array(df[self.column].to_gpu_array(fillna=True))
              nan_values = cp.isnan(values)
          else:
              values = df[self.column].to_numpy()
              nan_values = np.isnan(values)
  
          index_float = (values - self.bin0) / self.binsize
          # NaN values are corrected below, so set them to zero to avoid warnings when
          # converting from float to int.
          index_float[nan_values] = 0
          index = index_float.astype(int)
          index[index < 0] = self.bin_under
          index[index >= self.nbins] = self.bin_over
          index[nan_values] = self.nbins
          return index
  
  
  class category_values(CategoryPreprocess):
      """Extract a category and a value column from a dataframe as (2,N) numpy array of values."""
      def __init__(self, categorizer, value_column):
          super().__init__(value_column)
          self.categorizer = categorizer
  
      @property
      def inputs(self):
          return (self.categorizer.column, self.column)
  
      @property
      def cat_column(self):
          """Returns name of categorized column"""
          return self.categorizer.column
  
      def categories(self, input_dshape):
          return self.categorizer.categories
  
      def validate(self, in_dshape):
          return self.categorizer.validate(in_dshape)
  
      def apply(self, df, cuda):
          a = self.categorizer.apply(df, cuda)
          if cudf and isinstance(df, cudf.DataFrame):
              import cupy
              if self.column == SpecialColumn.RowIndex:
                  nullval = -1
              elif df[self.column].dtype.kind == 'f':
                  nullval = np.nan
              else:
                  nullval = 0
              a = cupy.asarray(a)
              if self.column == SpecialColumn.RowIndex:
                  b = extract(SpecialColumn.RowIndex).apply(df, cuda)
              elif Version(cudf.__version__) >= Version("22.02"):
                  b = df[self.column].to_cupy(na_value=nullval)
              else:
                  b = cupy.asarray(df[self.column].fillna(nullval))
              return cupy.stack((a, b), axis=-1)
          else:
              if self.column == SpecialColumn.RowIndex:
                  b = extract(SpecialColumn.RowIndex).apply(df, cuda)
              else:
                  b = df[self.column].values
              return np.stack((a, b), axis=-1)
  
  
  class Reduction(Expr):
      """Base class for per-bin reductions."""
      def __init__(self, column: str | SpecialColumn | None=None):
          self.column = column
          self._nan_check_column = None
  
      @property
      def nan_check_column(self):
          if self._nan_check_column is not None:
              return extract(self._nan_check_column)
          else:
              return None
  
      def uses_cuda_mutex(self) -> UsesCudaMutex:
          """Return ``True`` if this Reduction needs to use a CUDA mutex to
          ensure that it is threadsafe across CUDA threads.
  
          If the CUDA append functions are all atomic (i.e. using functions from
          the numba.cuda.atomic module) then this is ``False``, otherwise it is
          ``True``.
          """
          return UsesCudaMutex.No
  
      def uses_row_index(self, cuda, partitioned):
          """Return ``True`` if this Reduction uses a row index virtual column.
  
          For some reductions the order of the rows of supplied data is
          important. These include ``first`` and ``last`` reductions as well as
          ``where`` reductions that return a row index. In some situations the
          order is intrinsic such as ``first`` reductions that are processed
          sequentially (i.e. on a CPU without using Dask) and no extra column is
          required. But in situations of parallel processing (using a GPU or
          Dask) extra information is needed that is provided by a row index
          virtual column.
  
          Returning ``True`` from this function will cause a row index column to
          be created and passed to the ``append`` functions in the usual manner.
          """
          return False
  
      def validate(self, in_dshape):
          if self.column == SpecialColumn.RowIndex:
              return
          if self.column not in in_dshape.dict:
              raise ValueError("specified column not found")
          if not isnumeric(in_dshape.measure[self.column]):
              raise ValueError("input must be numeric")
  
      @property
      def inputs(self):
          return (extract(self.column),)
  
      def is_categorical(self):
          """Return ``True`` if this is or contains a categorical reduction."""
          return False
  
      def is_where(self):
          """Return ``True`` if this is a ``where`` reduction or directly wraps
          a where reduction."""
          return False
  
      def _antialias_requires_2_stages(self):
          # Return True if this Reduction must be processed with 2 stages,
          # False if it doesn't matter.
          # Overridden in derived classes as appropriate.
          return False
  
      def _antialias_stage_2(self, self_intersect, array_module) -> tuple[AntialiasStage2]:
          # Only called if using antialiased lines. Overridden in derived classes.
          # Returns a tuple containing an item for each constituent reduction.
          # Each item is (AntialiasCombination, zero_value)).
          raise NotImplementedError(f"{type(self)}._antialias_stage_2 is not defined")
  
      def _build_bases(self, cuda, partitioned):
          return (self,)
  
      def _build_combine_temps(self, cuda, partitioned):
          # Temporaries (i.e. not returned to user) that are reductions, the
          # aggs of which are passed to the combine() function but not the
          # append() functions, as opposed to _build_temps() which are passed
          # to both append() and combine().
          return ()
  
      def _build_temps(self, cuda=False):
          # Temporaries (i.e. not returned to user) that are reductions, the
          # aggs of which are passed to both append() and combine() functions.
          return ()
  
      def _build_create(self, required_dshape):
          fields = getattr(required_dshape.measure, "fields", None)
          if fields is not None and len(required_dshape.measure.fields) > 0:
              # If more than one field then they all have the same dtype so can just take the first.
              first_field = required_dshape.measure.fields[0]
              required_dshape = dshape(first_field[1])
  
          if isinstance(required_dshape, Option):
              required_dshape = dshape(required_dshape.ty)
  
          if required_dshape == dshape(ct.bool_):
              return self._create_bool
          elif required_dshape == dshape(ct.float32):
              return self._create_float32_nan
          elif required_dshape == dshape(ct.float64):
              return self._create_float64_nan
          elif required_dshape == dshape(ct.int64):
              return self._create_int64
          elif required_dshape == dshape(ct.uint32):
              return self._create_uint32
          else:
              raise NotImplementedError(f"Unexpected dshape {dshape}")
  
      def _build_append(self, dshape, schema, cuda, antialias, self_intersect):
          if cuda:
              if antialias and self.column is None:
                  return self._append_no_field_antialias_cuda
              elif antialias:
                  return self._append_antialias_cuda
              elif self.column is None:
                  return self._append_no_field_cuda
              else:
                  return self._append_cuda
          else:
              if antialias and self.column is None:
                  return self._append_no_field_antialias
              elif antialias:
                  return self._append_antialias
              elif self.column is None:
                  return self._append_no_field
              else:
                  return self._append
  
      def _build_combine(self, dshape, antialias, cuda, partitioned, categorical = False):
          return self._combine
  
      def _build_finalize(self, dshape):
          return self._finalize
  
      @staticmethod
      def _create_bool(shape, array_module):
          return array_module.zeros(shape, dtype='bool')
  
      @staticmethod
      def _create_float32_nan(shape, array_module):
          return array_module.full(shape, array_module.nan, dtype='f4')
  
      @staticmethod
      def _create_float64_nan(shape, array_module):
          return array_module.full(shape, array_module.nan, dtype='f8')
  
      @staticmethod
      def _create_float64_empty(shape, array_module):
          return array_module.empty(shape, dtype='f8')
  
      @staticmethod
      def _create_float64_zero(shape, array_module):
          return array_module.zeros(shape, dtype='f8')
  
      @staticmethod
      def _create_int64(shape, array_module):
          return array_module.full(shape, -1, dtype='i8')
  
      @staticmethod
      def _create_uint32(shape, array_module):
          return array_module.zeros(shape, dtype='u4')
  
      def __repr__(self):
          return f"{type(self).__name__}({self.column!r})"
  
  
  class OptionalFieldReduction(Reduction):
      """Base class for things like ``count`` or ``any`` for which the field is optional"""
      def __init__(self, column=None):
          super().__init__(column)
  
      @property
      def inputs(self):
          return (extract(self.column),) if self.column is not None else ()
  
      def validate(self, in_dshape):
          if self.column is not None:
              super().validate(in_dshape)
  
      @staticmethod
      def _finalize(bases, cuda=False, **kwargs):
          return xr.DataArray(bases[0], **kwargs)
  
  
  class SelfIntersectingOptionalFieldReduction(OptionalFieldReduction):
      """
      Base class for optional field reductions for which self-intersecting
      geometry may or may not be desirable.
      Ignored if not using antialiasing.
      """
      def __init__(self, column=None, self_intersect=True):
          super().__init__(column)
          self.self_intersect = self_intersect
  
      def _antialias_requires_2_stages(self):
          return not self.self_intersect
  
      def _build_append(self, dshape, schema, cuda, antialias, self_intersect):
          if antialias and not self_intersect:
              # append functions specific to antialiased lines without self_intersect
              if cuda:
                  if self.column is None:
                      return self._append_no_field_antialias_cuda_not_self_intersect
                  else:
                      return self._append_antialias_cuda_not_self_intersect
              else:
                  if self.column is None:
                      return self._append_no_field_antialias_not_self_intersect
                  else:
                      return self._append_antialias_not_self_intersect
  
          # Fall back to base class implementation
          return super()._build_append(dshape, schema, cuda, antialias, self_intersect)
  
      def _hashable_inputs(self):
          # Reductions with different self_intersect attributes much have different hashes otherwise
          # toolz.memoize will treat them as the same to give incorrect results.
          return super()._hashable_inputs() + (self.self_intersect,)
  
  
  class count(SelfIntersectingOptionalFieldReduction):
      """Count elements in each bin, returning the result as a uint32, or a
      float32 if using antialiasing.
  
      Parameters
      ----------
      column : str, optional
          If provided, only counts elements in ``column`` that are not ``NaN``.
          Otherwise, counts every element.
      """
      def out_dshape(self, in_dshape, antialias, cuda, partitioned):
          return dshape(ct.float32) if antialias else dshape(ct.uint32)
  
      def _antialias_stage_2(self, self_intersect, array_module) -> tuple[AntialiasStage2]:
          if self_intersect:
              return (AntialiasStage2(AntialiasCombination.SUM_1AGG, array_module.nan),)
          else:
              return (AntialiasStage2(AntialiasCombination.SUM_2AGG, array_module.nan),)
  
      # CPU append functions
      @staticmethod
      @ngjit
      def _append(x, y, agg, field):
          if not isnull(field):
              agg[y, x] += 1
              return 0
          return -1
  
      @staticmethod
      @ngjit
      def _append_antialias(x, y, agg, field, aa_factor, prev_aa_factor):
          if not isnull(field):
              if isnull(agg[y, x]):
                  agg[y, x] = aa_factor - prev_aa_factor
              else:
                  agg[y, x] += aa_factor - prev_aa_factor
              return 0
          return -1
  
      @staticmethod
      @ngjit
      def _append_antialias_not_self_intersect(x, y, agg, field, aa_factor, prev_aa_factor):
          if not isnull(field):
              if isnull(agg[y, x]) or aa_factor > agg[y, x]:
                  agg[y, x] = aa_factor
                  return 0
          return -1
  
      @staticmethod
      @ngjit
      def _append_no_field(x, y, agg):
          agg[y, x] += 1
          return 0
  
      @staticmethod
      @ngjit
      def _append_no_field_antialias(x, y, agg, aa_factor, prev_aa_factor):
          if isnull(agg[y, x]):
              agg[y, x] = aa_factor - prev_aa_factor
          else:
              agg[y, x] += aa_factor - prev_aa_factor
          return 0
  
      @staticmethod
      @ngjit
      def _append_no_field_antialias_not_self_intersect(x, y, agg, aa_factor, prev_aa_factor):
          if isnull(agg[y, x]) or aa_factor > agg[y, x]:
              agg[y, x] = aa_factor
              return 0
          return -1
  
      # GPU append functions
      @staticmethod
      @nb_cuda.jit(device=True)
      def _append_antialias_cuda(x, y, agg, field, aa_factor, prev_aa_factor):
          value = field*aa_factor
          if not isnull(value):
              old = cuda_atomic_nanmax(agg, (y, x), value)
              if isnull(old) or old < value:
                  return 0
          return -1
  
      @staticmethod
      @nb_cuda.jit(device=True)
      def _append_no_field_antialias_cuda_not_self_intersect(x, y, agg, aa_factor, prev_aa_factor):
          if not isnull(aa_factor):
              old = cuda_atomic_nanmax(agg, (y, x), aa_factor)
              if isnull(old) or old < aa_factor:
                  return 0
          return -1
  
      @staticmethod
      @nb_cuda.jit(device=True)
      def _append_cuda(x, y, agg, field):
          if not isnull(field):
              nb_cuda.atomic.add(agg, (y, x), 1)
              return 0
          return -1
  
      @staticmethod
      @nb_cuda.jit(device=True)
      def _append_no_field_antialias_cuda(x, y, agg, aa_factor, prev_aa_factor):
          if not isnull(aa_factor):
              old = cuda_atomic_nanmax(agg, (y, x), aa_factor)
              if isnull(old) or old < aa_factor:
                  return 0
          return -1
  
      @staticmethod
      @nb_cuda.jit(device=True)
      def _append_no_field_cuda(x, y, agg):
          nb_cuda.atomic.add(agg, (y, x), 1)
          return 0
  
      def _build_combine(self, dshape, antialias, cuda, partitioned, categorical = False):
          if antialias:
              return self._combine_antialias
          else:
              return self._combine
  
      @staticmethod
      def _combine(aggs):
          return aggs.sum(axis=0, dtype='u4')
  
      @staticmethod
      def _combine_antialias(aggs):
          ret = aggs[0]
          for i in range(1, len(aggs)):
              nansum_in_place(ret, aggs[i])
          return ret
  
      def __repr__(self):
          return "count()"
  
  
  class _count_ignore_antialiasing(count):
      """Count reduction but ignores antialiasing. Used by mean reduction.
      """
      def out_dshape(self, in_dshape, antialias, cuda, partitioned):
          return dshape(ct.uint32)
  
      def _antialias_stage_2(self, self_intersect, array_module) -> tuple[AntialiasStage2]:
          if self_intersect:
              return (AntialiasStage2(AntialiasCombination.SUM_1AGG, 0),)
          else:
              return (AntialiasStage2(AntialiasCombination.SUM_2AGG, 0),)
  
      @staticmethod
      @ngjit
      def _append_antialias(x, y, agg, field, aa_factor, prev_aa_factor):
          if not isnull(field) and prev_aa_factor == 0.0:
              agg[y, x] += 1
              return 0
          return -1
  
      @staticmethod
      @ngjit
      def _append_antialias_not_self_intersect(x, y, agg, field, aa_factor, prev_aa_factor):
          if not isnull(field) and prev_aa_factor == 0.0:
              agg[y, x] += 1
              return 0
          return -1
  
  
  class by(Reduction):
      """Apply the provided reduction separately per category.
  
      Parameters
      ----------
      cats: str or CategoryPreprocess instance
          Name of column to aggregate over, or a categorizer object that returns categories.
          Resulting aggregate has an outer dimension axis along the categories present.
      reduction : Reduction
          Per-category reduction function.
      """
      def __init__(self, cat_column, reduction=count()):
          super().__init__()
  
          # set basic categorizer
          if isinstance(cat_column, CategoryPreprocess):
              self.categorizer = cat_column
          elif isinstance(cat_column, str):
              self.categorizer = category_codes(cat_column)
          else:
              raise TypeError("first argument must be a column name or a CategoryPreprocess instance")
  
          self.column = self.categorizer.column # for backwards compatibility with count_cat
  
          self.columns = (self.categorizer.column,)
          if (columns := getattr(reduction, 'columns', None)) is not None:
              # Must reverse columns (from where reduction) so that val_column property
              # is the column that is returned to the user.
              self.columns += columns[::-1]
          else:
              self.columns += (getattr(reduction, 'column', None),)
  
          self.reduction = reduction
          # if a value column is supplied, set category_values preprocessor
          if self.val_column is not None:
              self.preprocess = category_values(self.categorizer, self.val_column)
          else:
              self.preprocess = self.categorizer
  
      def __hash__(self):
          return hash((type(self), self._hashable_inputs(), self.categorizer._hashable_inputs(),
                       self.reduction))
  
      def _build_temps(self, cuda=False):
          return tuple(by(self.categorizer, tmp) for tmp in self.reduction._build_temps(cuda))
  
      @property
      def cat_column(self):
          return self.columns[0]
  
      @property
      def val_column(self):
          return self.columns[1]
  
      def validate(self, in_dshape):
          self.preprocess.validate(in_dshape)
          self.reduction.validate(in_dshape)
  
      def out_dshape(self, input_dshape, antialias, cuda, partitioned):
          cats = self.categorizer.categories(input_dshape)
          red_shape = self.reduction.out_dshape(input_dshape, antialias, cuda, partitioned)
          return dshape(Record([(c, red_shape) for c in cats]))
  
      @property
      def inputs(self):
          return (self.preprocess,)
  
      def is_categorical(self):
          return True
  
      def is_where(self):
          return self.reduction.is_where()
  
      @property
      def nan_check_column(self):
          return self.reduction.nan_check_column
  
      def uses_cuda_mutex(self) -> UsesCudaMutex:
          return self.reduction.uses_cuda_mutex()
  
      def uses_row_index(self, cuda, partitioned):
          return self.reduction.uses_row_index(cuda, partitioned)
  
      def _antialias_requires_2_stages(self):
          return self.reduction._antialias_requires_2_stages()
  
      def _antialias_stage_2(self, self_intersect, array_module) -> tuple[AntialiasStage2]:
          ret = self.reduction._antialias_stage_2(self_intersect, array_module)
          return (AntialiasStage2(combination=ret[0].combination,
                                  zero=ret[0].zero,
                                  n_reduction=ret[0].n_reduction,
                                  categorical=True),)
  
      def _build_create(self, required_dshape):
          n_cats = len(required_dshape.measure.fields)
          return lambda shape, array_module: self.reduction._build_create(
              required_dshape)(shape + (n_cats,), array_module)
  
      def _build_bases(self, cuda, partitioned):
          bases = self.reduction._build_bases(cuda, partitioned)
          if len(bases) == 1 and bases[0] is self:
              return bases
          return tuple(by(self.categorizer, base) for base in bases)
  
      def _build_append(self, dshape, schema, cuda, antialias, self_intersect):
          return self.reduction._build_append(dshape, schema, cuda, antialias, self_intersect)
  
      def _build_combine(self, dshape, antialias, cuda, partitioned, categorical = False):
          return self.reduction._build_combine(dshape, antialias, cuda, partitioned, True)
  
      def _build_combine_temps(self, cuda, partitioned):
          return self.reduction._build_combine_temps(cuda, partitioned)
  
      def _build_finalize(self, dshape):
          cats = list(self.categorizer.categories(dshape))
  
          def finalize(bases, cuda=False, **kwargs):
              # Return a modified copy of kwargs. Cannot modify supplied kwargs as it
              # may be used by multiple reductions, e.g. if a summary reduction.
              kwargs = copy.deepcopy(kwargs)
              kwargs['dims'] += [self.cat_column]
              kwargs['coords'][self.cat_column] = cats
              return self.reduction._build_finalize(dshape)(bases, cuda=cuda, **kwargs)
  
          return finalize
  
      def __repr__(self):
          return f"{type(self).__name__}(column={self.column!r}, reduction={self.reduction!r})"
  
  class any(OptionalFieldReduction):
      """Whether any elements in ``column`` map to each bin.
  
      Parameters
      ----------
      column : str, optional
          If provided, any elements in ``column`` that are ``NaN`` are skipped.
      """
      def out_dshape(self, in_dshape, antialias, cuda, partitioned):
          return dshape(ct.float32) if antialias else dshape(ct.bool_)
  
      def _antialias_stage_2(self, self_intersect, array_module) -> tuple[AntialiasStage2]:
          return (AntialiasStage2(AntialiasCombination.MAX, array_module.nan),)
  
      # CPU append functions
      @staticmethod
      @ngjit
      def _append(x, y, agg, field):
          if not isnull(field):
              agg[y, x] = True
              return 0
          return -1
  
      @staticmethod
      @ngjit
      def _append_antialias(x, y, agg, field, aa_factor, prev_aa_factor):
          if not isnull(field):
              if isnull(agg[y, x]) or aa_factor > agg[y, x]:
                  agg[y, x] = aa_factor
                  return 0
          return -1
  
      @staticmethod
      @ngjit
      def _append_no_field(x, y, agg):
          agg[y, x] = True
          return 0
  
      @staticmethod
      @ngjit
      def _append_no_field_antialias(x, y, agg, aa_factor, prev_aa_factor):
          if isnull(agg[y, x]) or aa_factor > agg[y, x]:
              agg[y, x] = aa_factor
              return 0
          return -1
  
      # GPU append functions
      _append_cuda =_append
      _append_no_field_cuda = _append_no_field
  
      def _build_combine(self, dshape, antialias, cuda, partitioned, categorical = False):
          if antialias:
              return self._combine_antialias
          else:
              return self._combine
  
      @staticmethod
      def _combine(aggs):
          return aggs.sum(axis=0, dtype='bool')
  
      @staticmethod
      def _combine_antialias(aggs):
          ret = aggs[0]
          for i in range(1, len(aggs)):
              nanmax_in_place(ret, aggs[i])
          return ret
  
  
  class _upsample(Reduction):
      """"Special internal class used for upsampling"""
      def out_dshape(self, in_dshape, antialias, cuda, partitioned):
          return dshape(Option(ct.float64))
  
      @staticmethod
      def _finalize(bases, cuda=False, **kwargs):
          return xr.DataArray(bases[0], **kwargs)
  
      @property
  