1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105
|
# BSD 3-Clause License; see https://github.com/scikit-hep/awkward/blob/main/LICENSE
from __future__ import annotations
import numpy as np
import pytest
import awkward as ak
to_list = ak.operations.to_list
def test_one_level():
base = ak.operations.zip(
{"a": ak.operations.zip({"x": [1, 2, 3]}), "b": [1, 2, 3]},
depth_limit=1,
)
what = ak.Array([1.1, 2.2, 3.3], check_valid=True)
assert to_list(ak.operations.with_field(base, what, where=["a", "y"])) == [
{"b": 1, "a": {"x": 1, "y": 1.1}},
{"b": 2, "a": {"x": 2, "y": 2.2}},
{"b": 3, "a": {"x": 3, "y": 3.3}},
]
assert ak.operations.fields(base) == ["a", "b"]
base["a", "y"] = what
assert to_list(base) == [
{"b": 1, "a": {"x": 1, "y": 1.1}},
{"b": 2, "a": {"x": 2, "y": 2.2}},
{"b": 3, "a": {"x": 3, "y": 3.3}},
]
def test_two_level():
base = ak.operations.zip(
{
"A": ak.operations.zip(
{"a": ak.operations.zip({"x": [1, 2, 3]}), "b": [1, 2, 3]}
),
"B": [1, 2, 3],
},
depth_limit=1,
)
what = ak.Array([1.1, 2.2, 3.3], check_valid=True)
assert to_list(ak.operations.with_field(base, what, where=["A", "a", "y"])) == [
{"B": 1, "A": {"b": 1, "a": {"x": 1, "y": 1.1}}},
{"B": 2, "A": {"b": 2, "a": {"x": 2, "y": 2.2}}},
{"B": 3, "A": {"b": 3, "a": {"x": 3, "y": 3.3}}},
]
assert ak.operations.fields(base) == ["A", "B"]
base["A", "a", "y"] = what
assert to_list(base) == [
{"B": 1, "A": {"b": 1, "a": {"x": 1, "y": 1.1}}},
{"B": 2, "A": {"b": 2, "a": {"x": 2, "y": 2.2}}},
{"B": 3, "A": {"b": 3, "a": {"x": 3, "y": 3.3}}},
]
def test_replace_the_only_field():
base = ak.operations.zip({"a": ak.operations.zip({"x": [1, 2, 3]})}, depth_limit=1)
what = ak.Array([1.1, 2.2, 3.3], check_valid=True)
assert to_list(ak.operations.with_field(base, what, where=["a", "y"])) == [
{"a": {"x": 1, "y": 1.1}},
{"a": {"x": 2, "y": 2.2}},
{"a": {"x": 3, "y": 3.3}},
]
assert ak.operations.fields(base) == ["a"]
base["a", "y"] = what
assert to_list(base) == [
{"a": {"x": 1, "y": 1.1}},
{"a": {"x": 2, "y": 2.2}},
{"a": {"x": 3, "y": 3.3}},
]
def test_check_no_field():
base = ak.operations.zip({"a": ak.operations.zip({"x": [1, 2, 3]})}, depth_limit=1)
what = ak.Array([1.1, 2.2, 3.3], check_valid=True)
assert to_list(ak.operations.with_field(base, what)) == [
{"a": {"x": 1}, "1": 1.1},
{"a": {"x": 2}, "1": 2.2},
{"a": {"x": 3}, "1": 3.3},
]
with pytest.raises(ValueError):
ak.operations.with_field(what, what)
content1 = ak.contents.NumpyArray(np.array([1, 2, 3]))
recordarray = ak.contents.RecordArray([content1], None)
what = ak.Array([1.1, 2.2, 3.3], check_valid=True)
assert to_list(ak.operations.with_field(recordarray, what)) == [
(1, 1.1),
(2, 2.2),
(3, 3.3),
]
assert to_list(ak.operations.with_field(recordarray, what, "a")) == [
{"0": 1, "a": 1.1},
{"0": 2, "a": 2.2},
{"0": 3, "a": 3.3},
]
assert ak.operations.fields(recordarray) == ["0"]
|