()
| 1370 | |
| 1371 | |
| 1372 | def test_conditional(): |
| 1373 | @experimental_pipeline_def(enable_conditionals=True) |
| 1374 | def conditional_pipeline(): |
| 1375 | true = types.Constant(np.array(True), device="cpu") |
| 1376 | false = types.Constant(np.array(False), device="cpu") |
| 1377 | if true and true or not false: |
| 1378 | output = types.Constant(np.array([42]), device="cpu") |
| 1379 | else: |
| 1380 | output = types.Constant(np.array([0]), device="cpu") |
| 1381 | return output |
| 1382 | |
| 1383 | cond_pipe = conditional_pipeline(batch_size=5, num_threads=1, device_id=None) |
| 1384 | cond_pipe.run() |
| 1385 | |
| 1386 | @pipeline_def |
| 1387 | def explicit_conditional_ops_pipeline(): |
| 1388 | value = types.Constant(np.array([42]), device="cpu") |
| 1389 | pred = fn.random.coin_flip(dtype=types.DALIDataType.BOOL) |
| 1390 | pred_validated = fn._conditional.validate_logical( |
| 1391 | pred, expression_name="or", expression_side="right" |
| 1392 | ) |
| 1393 | true, false = fn._conditional.split(value, predicate=pred) |
| 1394 | true = true + 10 |
| 1395 | merged = fn._conditional.merge(true, false, predicate=pred) |
| 1396 | negated = fn._conditional.not_(pred) |
| 1397 | return merged, negated, pred_validated |
| 1398 | |
| 1399 | pipe = explicit_conditional_ops_pipeline(batch_size=5, num_threads=1, device_id=None) |
| 1400 | pipe.run() |
| 1401 | |
| 1402 | |
| 1403 | def get_shape_data(): |
nothing calls this directly
no test coverage detected