| 236 | |
| 237 | |
| 238 | class FromMap(PartitionsFiltered, BlockwiseIO): |
| 239 | _parameters = [ |
| 240 | "func", |
| 241 | "iterables", |
| 242 | "args", |
| 243 | "kwargs", |
| 244 | "user_meta", |
| 245 | "enforce_metadata", |
| 246 | "user_divisions", |
| 247 | "label", |
| 248 | "_partitions", |
| 249 | ] |
| 250 | _defaults = { |
| 251 | "user_meta": no_default, |
| 252 | "enforce_metadata": False, |
| 253 | "user_divisions": None, |
| 254 | "label": None, |
| 255 | "_partitions": None, |
| 256 | } |
| 257 | _absorb_projections = False |
| 258 | |
| 259 | @functools.cached_property |
| 260 | def _name(self): |
| 261 | if self.label is None: |
| 262 | return funcname(self.func).lower() + "-" + self.deterministic_token |
| 263 | else: |
| 264 | return self.label + "-" + self.deterministic_token |
| 265 | |
| 266 | @functools.cached_property |
| 267 | def _meta(self): |
| 268 | if self.operand("user_meta") is not no_default: |
| 269 | meta = self.operand("user_meta") |
| 270 | return make_meta(meta) |
| 271 | else: |
| 272 | vals = [v[0] for v in self.iterables] |
| 273 | meta = delayed(self.func)(*vals, *self.args, **self.kwargs) |
| 274 | return delayed(make_meta)(meta).compute() |
| 275 | |
| 276 | def _divisions(self): |
| 277 | if self.operand("user_divisions"): |
| 278 | return self.operand("user_divisions") |
| 279 | else: |
| 280 | npartitions = len(self.iterables[0]) |
| 281 | return (None,) * (npartitions + 1) |
| 282 | |
| 283 | @property |
| 284 | def apply_func(self): |
| 285 | if self.enforce_metadata: |
| 286 | return apply_and_enforce |
| 287 | return self.func |
| 288 | |
| 289 | @functools.cached_property |
| 290 | def apply_kwargs(self): |
| 291 | kwargs = self.kwargs |
| 292 | if self.enforce_metadata: |
| 293 | kwargs = kwargs.copy() |
| 294 | kwargs.update( |
| 295 | { |