File size: 50,650 Bytes
341b917
c9c2d08
8fecbbd
 
78663de
8fecbbd
b7c39fe
8fecbbd
c9c2d08
 
 
 
 
 
 
 
 
 
 
b7c39fe
778ad61
78663de
8fecbbd
778ad61
b7c39fe
778ad61
8fecbbd
778ad61
 
8fecbbd
b7c39fe
 
f60252a
778ad61
78663de
 
8fecbbd
d292ceb
778ad61
 
 
78663de
d292ceb
 
 
 
8fecbbd
778ad61
 
 
 
f60252a
 
 
 
 
 
 
8fecbbd
 
 
 
 
 
 
 
 
 
 
 
 
78663de
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d292ceb
8fecbbd
778ad61
 
78663de
 
778ad61
 
 
 
78663de
 
 
 
 
 
 
 
 
 
 
8fecbbd
 
 
78663de
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8fecbbd
 
78663de
8fecbbd
778ad61
 
 
78663de
d292ceb
 
 
 
 
8fecbbd
d292ceb
 
8fecbbd
78663de
 
 
d292ceb
778ad61
 
 
78663de
d292ceb
 
 
78663de
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d292ceb
8fecbbd
778ad61
8fecbbd
 
 
78663de
 
 
8fecbbd
 
 
 
 
 
 
 
 
 
 
 
3d43021
78663de
3d43021
 
78663de
3d43021
 
 
 
78663de
 
 
 
 
3d43021
 
 
8fecbbd
78663de
 
8fecbbd
 
 
 
 
78663de
8fecbbd
 
 
 
 
 
 
b868ef2
 
8fecbbd
 
 
 
78663de
 
 
8fecbbd
 
 
 
 
 
78663de
 
 
8fecbbd
 
 
 
 
 
 
 
 
 
 
 
78663de
8fecbbd
 
 
78663de
 
 
8fecbbd
c9c2d08
b868ef2
 
 
 
 
 
 
78663de
 
 
 
 
 
 
 
 
 
 
 
 
8fecbbd
 
78663de
 
 
 
 
 
 
8fecbbd
 
 
c9c2d08
78663de
c9c2d08
 
 
 
78663de
 
 
c9c2d08
 
 
 
 
 
 
 
 
 
 
78663de
 
 
 
c9c2d08
78663de
 
 
 
 
 
 
 
 
 
c9c2d08
78663de
 
 
c9c2d08
 
78663de
 
 
 
 
 
 
 
 
 
 
 
c9c2d08
78663de
c9c2d08
78663de
 
 
 
c9c2d08
78663de
 
 
 
c9c2d08
78663de
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
c9c2d08
78663de
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
c9c2d08
 
78663de
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
c9c2d08
 
78663de
c9c2d08
 
 
8fecbbd
78663de
 
8fecbbd
 
 
 
 
 
 
 
 
 
c9c2d08
78663de
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
c9c2d08
 
 
 
 
2ec6f71
 
 
 
 
 
 
 
 
78663de
c9c2d08
 
 
 
78663de
 
 
 
 
 
 
c9c2d08
78663de
 
 
 
c9c2d08
 
 
 
 
2ec6f71
c9c2d08
78663de
 
 
c9c2d08
 
 
 
 
 
 
 
 
 
78663de
c9c2d08
78663de
c9c2d08
 
 
78663de
 
 
c9c2d08
78663de
 
c9c2d08
 
 
 
8fecbbd
78663de
8fecbbd
 
 
 
 
778ad61
78663de
 
 
8fecbbd
78663de
 
8fecbbd
 
 
 
 
d292ceb
 
 
8fecbbd
78663de
8fecbbd
 
 
 
 
d292ceb
78663de
 
 
8fecbbd
 
 
d292ceb
778ad61
 
8fecbbd
78663de
8fecbbd
 
 
 
 
 
 
 
 
 
78663de
 
 
8fecbbd
 
 
 
 
 
 
78663de
8fecbbd
 
 
 
 
 
 
 
 
 
 
 
 
78663de
 
 
8fecbbd
 
 
 
 
78663de
8fecbbd
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
78663de
8fecbbd
 
 
78663de
8fecbbd
 
 
 
 
78663de
 
 
 
 
8fecbbd
78663de
8fecbbd
78663de
 
 
 
8fecbbd
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
78663de
 
 
8fecbbd
 
 
778ad61
78663de
d292ceb
 
 
 
8fecbbd
778ad61
 
 
 
 
 
 
 
 
 
b868ef2
78663de
d292ceb
 
 
 
 
 
8fecbbd
b868ef2
f60252a
778ad61
 
f60252a
778ad61
78663de
 
 
778ad61
 
 
 
 
 
 
 
 
 
 
 
78663de
 
 
 
b868ef2
78663de
778ad61
 
 
 
 
78663de
 
 
 
d292ceb
78663de
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d292ceb
 
78663de
d292ceb
8fecbbd
78663de
 
 
 
 
 
 
 
 
778ad61
78663de
778ad61
78663de
 
 
 
 
 
 
 
 
778ad61
 
 
78663de
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d292ceb
78663de
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d292ceb
 
 
 
8fecbbd
778ad61
 
 
 
 
78663de
 
778ad61
 
 
 
 
 
 
 
 
 
 
 
 
 
 
78663de
d292ceb
 
 
 
8fecbbd
778ad61
 
 
 
 
 
 
 
 
 
78663de
 
 
 
 
 
 
778ad61
 
 
 
 
 
78663de
d292ceb
 
 
 
 
8fecbbd
778ad61
 
 
78663de
778ad61
 
 
 
 
 
 
 
 
 
 
78663de
 
 
8fecbbd
 
778ad61
 
 
 
78663de
 
 
 
 
 
d292ceb
78663de
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d292ceb
 
 
 
8fecbbd
778ad61
 
 
 
 
78663de
 
 
 
 
 
 
778ad61
 
 
78663de
d292ceb
 
 
 
 
 
8fecbbd
78663de
778ad61
 
 
 
 
 
78663de
 
 
 
 
778ad61
 
78663de
 
 
 
 
 
 
d292ceb
8fecbbd
d292ceb
78663de
d292ceb
 
 
 
8fecbbd
78663de
 
8fecbbd
 
 
 
78663de
8fecbbd
 
 
 
 
 
 
 
 
 
 
78663de
 
 
 
 
8fecbbd
 
 
 
 
 
78663de
 
 
8fecbbd
 
341b917
 
 
 
 
78663de
341b917
 
 
 
 
 
 
78663de
341b917
 
 
 
 
 
 
 
 
 
 
 
 
78663de
 
 
341b917
78663de
341b917
 
 
 
 
78663de
 
 
341b917
 
 
 
78663de
 
 
341b917
 
 
 
 
 
 
 
 
 
 
 
 
 
 
78663de
 
341b917
 
 
 
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
import collections
import importlib
import uuid
from abc import abstractmethod
from collections import Counter
from copy import deepcopy
from dataclasses import field
from itertools import zip_longest
from typing import (
    Any,
    Callable,
    Dict,
    Generator,
    Iterable,
    List,
    Optional,
    Tuple,
    Union,
)

from .artifact import Artifact, fetch_artifact
from .dataclass import NonPositionalField
from .dict_utils import dict_delete, dict_get, dict_set, is_subpath
from .operator import (
    MultiStream,
    MultiStreamOperator,
    PagedStreamOperator,
    SingleStreamOperator,
    SingleStreamReducer,
    StreamingOperator,
    StreamInitializerOperator,
    StreamInstanceOperator,
    StreamSource,
)
from .random_utils import get_random, nested_seed
from .stream import Stream
from .text_utils import nested_tuple_to_string
from .utils import flatten_dict


class FromIterables(StreamInitializerOperator):
    """Creates a MultiStream from iterables.

    Args:
        iterables (Dict[str, Iterable]): A dictionary where each key-value pair represents a stream name and its corresponding iterable.
    """

    def process(self, iterables: Dict[str, Iterable]) -> MultiStream:
        return MultiStream.from_iterables(iterables)


class IterableSource(StreamSource):
    iterables: Dict[str, Iterable]

    def __call__(self) -> MultiStream:
        return MultiStream.from_iterables(self.iterables)


class MapInstanceValues(StreamInstanceOperator):
    """A class used to map instance values into a stream.

    This class is a type of StreamInstanceOperator,
    it maps values of instances in a stream using predefined mappers.

    Attributes:
        mappers (Dict[str, Dict[str, str]]): The mappers to use for mapping instance values.
            Keys are the names of the fields to be mapped, and values are dictionaries
            that define the mapping from old values to new values.
        strict (bool): If True, the mapping is applied strictly. That means if a value
            does not exist in the mapper, it will raise a KeyError. If False, values
            that are not present in the mapper are kept as they are.
        process_every_value (bool): If True, all fields to be mapped should be lists, and the mapping
            is to be applied to their individual elements. If False, mapping is only applied to a field
            containing a single value.

    Examples:
        MapInstanceValues(mappers={"a": {"1": "hi", "2": "bye"}})
        replaces '1' with 'hi' and '2' with 'bye' in field 'a' in all instances of all streams:
        instance {"a":"1", "b": 2} becomes {"a":"hi", "b": 2}.

        MapInstanceValues(mappers={"a": {"1": "hi", "2": "bye"}}, process_every_element=True)
        Assuming field 'a' is a list of values, potentially including "1"-s and "2"-s, this replaces
        each such "1" with "hi" and "2" -- with "bye" in all instances of all streams:
        instance {"a": ["1", "2"], "b": 2} becomes {"a": ["hi", "bye"], "b": 2}.

        MapInstanceValues(mappers={"a": {"1": "hi", "2": "bye"}}, strict=True)
        To ensure that all values of field 'a' are mapped in every instance, use strict=True.
        Input instance {"a":"3", "b": 2} will raise an exception per the above call,
        because "3" is not a key in the mapper of "a".
    """

    mappers: Dict[str, Dict[str, str]]
    strict: bool = True
    use_query: bool = False
    process_every_value: bool = False

    def verify(self):
        # make sure the mappers are valid
        for key, mapper in self.mappers.items():
            assert isinstance(
                mapper, dict
            ), f"Mapper for given field {key} should be a dict, got {type(mapper)}"
            for k in mapper.keys():
                assert isinstance(
                    k, str
                ), f'Key "{k}" in mapper for field "{key}" should be a string, got {type(k)}'

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        for key, mapper in self.mappers.items():
            value = dict_get(instance, key, use_dpath=self.use_query)
            if value is not None:
                if (self.process_every_value is True) and (not isinstance(value, list)):
                    raise ValueError(
                        f"'process_every_field' == True is allowed only when all fields which have mappers, i.e., {list(self.mappers.keys())} are lists. Instace = {instance}"
                    )
                if isinstance(value, list):
                    if self.process_every_value:
                        for i, val in enumerate(value):
                            val = str(val)  # make sure the value is a string
                            if self.strict and (val not in mapper):
                                raise KeyError(
                                    f"value '{val}' in instance '{instance}' is not found in mapper '{mapper}', associated with field '{key}'."
                                )
                            if val in mapper:
                                # replace just that member of value (value is a list)
                                value[i] = mapper[val]
                                dict_set(instance, key, value, use_dpath=self.use_query)
                    else:  # field is a list, and process_every_value == False
                        if self.strict:  # whole lists can not be mapped by a string-to-something mapper
                            raise KeyError(
                                f"A whole list ({value}) in the instance can not be mapped by a field mapper."
                            )
                else:  # value is not a list, implying process_every_value == False
                    value = str(value)  # make sure the value is a string
                    if self.strict and (value not in mapper):
                        raise KeyError(
                            f"value '{value}' in instance '{instance}' is not found in mapper '{mapper}', associated with field '{key}'."
                        )
                    if value in mapper:
                        dict_set(instance, key, mapper[value], use_dpath=self.use_query)

        return instance


class FlattenInstances(StreamInstanceOperator):
    """Flattens each instance in a stream, making nested dictionary entries into top-level entries.

    Args:
        parent_key (str): A prefix to use for the flattened keys. Defaults to an empty string.
        sep (str): The separator to use when concatenating nested keys. Defaults to "_".
    """

    parent_key: str = ""
    sep: str = "_"

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        return flatten_dict(instance, parent_key=self.parent_key, sep=self.sep)


class AddFields(StreamInstanceOperator):
    """Adds specified fields to each instance in a given stream or all streams (default) If fields exist, updates them.

    Args:
        fields (Dict[str, object]): The fields to add to each instance.
        use_query (bool) : Use '/' to access inner fields
        use_deepcopy (bool) : Deep copy the input value to avoid later modifications

    Examples:
        # Add a 'classes' field with a value of a list "positive" and "negative" to all streams
        AddFields(fields={"classes": ["positive","negatives"]})

        # Add a 'start' field under the 'span' field with a value of 0 to all streams
        AddFields(fields={"span/start": 0}

        # Add a 'classes' field with a value of a list "positive" and "negative" to 'train' stream
        AddFields(fields={"classes": ["positive","negatives"], apply_to_stream=["train"]})

        # Add a 'classes' field on a given list, prevent modification of original list
        # from changing the instance.
        AddFields(fields={"classes": alist}), use_deepcopy=True)
    """

    fields: Dict[str, object]
    use_query: bool = False
    use_deepcopy: bool = False

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        if self.use_query:
            for key, value in self.fields.items():
                if self.use_deepcopy:
                    value = deepcopy(value)
                dict_set(instance, key, value, use_dpath=self.use_query)
        else:
            if self.use_deepcopy:
                self.fields = deepcopy(self.fields)
            instance.update(self.fields)
        return instance


class RemoveFields(StreamInstanceOperator):
    """Remove specified fields to each instance in a stream.

    Args:
        fields (List[str]): The fields to remove from each instance.
    """

    fields: List[str]

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        for field_name in self.fields:
            del instance[field_name]
        return instance


class FieldOperator(StreamInstanceOperator):
    """A general stream that processes the values of a field (or multiple ones.

    Args:
        field (Optional[str]): The field to process, if only a single one is passed Defaults to None
        to_field (Optional[str]): Field name to save, if only one field is to be saved, if None is passed the operator would happen in-place and replace "field" Defaults to None
        field_to_field (Optional[Union[List[Tuple[str, str]], Dict[str, str]]]): Mapping from fields to process to their names after this process, duplicates are allowed. Defaults to None
        process_every_value (bool): Processes the values in a list instead of the list as a value, similar to *var. Defaults to False
        use_query (bool): Whether to use dpath style queries. Defaults to False.
    """

    field: Optional[str] = None
    to_field: Optional[str] = None
    field_to_field: Optional[Union[List[Tuple[str, str]], Dict[str, str]]] = None
    process_every_value: bool = False
    use_query: bool = False
    get_default: Any = None
    not_exist_ok: bool = False

    def verify(self):
        super().verify()

        assert (
            self.field is not None or self.field_to_field is not None
        ), "Must supply a field to work on"
        assert (
            self.to_field is None or self.field_to_field is None
        ), f"Can not apply operator to create both on {self.to_field} and on the mapping from fields to fields {self.field_to_field}"
        assert (
            self.field is None or self.field_to_field is None
        ), f"Can not apply operator both on {self.field} and on the mapping from fields to fields {self.field_to_field}"
        assert (
            self._field_to_field
        ), f"the from and to fields must be defined got: {self._field_to_field}"

    @abstractmethod
    def process_value(self, value: Any) -> Any:
        pass

    def prepare(self):
        if self.to_field is None:
            self.to_field = self.field
        if self.field_to_field is None:
            self._field_to_field = [(self.field, self.to_field)]
        else:
            try:
                self._field_to_field = list(self.field_to_field.items())
            except AttributeError:
                self._field_to_field = self.field_to_field

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        for from_field, to_field in self._field_to_field:
            try:
                old_value = dict_get(
                    instance,
                    from_field,
                    use_dpath=self.use_query,
                    default=self.get_default,
                    not_exist_ok=self.not_exist_ok,
                )
            except Exception as e:
                raise ValueError(
                    f"Failed to get '{from_field}' from {instance} due to : {e}"
                ) from e
            try:
                if self.process_every_value:
                    new_value = [self.process_value(value) for value in old_value]
                else:
                    new_value = self.process_value(old_value)
            except Exception as e:
                raise ValueError(
                    f"Failed to process '{from_field}' from {instance} due to : {e}"
                ) from e
            if self.use_query and is_subpath(from_field, to_field):
                dict_delete(instance, from_field)
            dict_set(
                instance,
                to_field,
                new_value,
                use_dpath=self.use_query,
                not_exist_ok=True,
            )
        return instance


class RenameFields(FieldOperator):
    """Renames fields."""

    def process_value(self, value: Any) -> Any:
        return value

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        res = super().process(instance=instance, stream_name=stream_name)
        vals = [x[1] for x in self._field_to_field]
        for key, _ in self._field_to_field:
            if self.use_query and "/" in key:
                continue
            if key not in vals:
                res.pop(key)
        return res


class AddConstant(FieldOperator):
    """Adds a value, similar to  add + field.

    Args:
        add: sum to add.
    """

    add: Any

    def process_value(self, value: Any) -> Any:
        return self.add + value


class Augmentor(StreamInstanceOperator):
    """A stream that augments the values of either the task input fields before rendering with the template,  or the  input passed to the model after rendering of the template.

    Args:
        augment_model_input: Whether to augment the input to the model.
        augment_task_input:  Whether to augment the task input fields.  The specific fields are defined in the FormTask operator.

    """

    augment_task_input: bool = False
    augment_model_input: bool = False

    def verify(self):
        assert not (
            self.augment_task_input and self.augment_model_input
        ), "Augmentor must set either 'augment_task_input' and 'augment_model_input' but not both"
        assert (
            self.augment_task_input or self.augment_model_input
        ), "Augmentor must set either 'augment_task_input' or 'augment_model_input'"

        super().verify()

    @abstractmethod
    def process_value(self, value: Any) -> Any:
        pass

    def prepare(self):
        pass

    def set_task_input_fields(self, task_input_fields: List[str]):
        self._task_input_fields = [
            "inputs/" + task_input_field for task_input_field in task_input_fields
        ]

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        if self.augment_task_input:
            assert (
                len(self._task_input_fields) > 0
            ), "No augmentable input fields were defined in FormTask, and augmentation was requested. Specify the fields to augment in 'argumentable_inputs' attribute of the FormTask."
            fields = self._task_input_fields
            assert not self.augment_model_input

        if self.augment_model_input:
            fields = ["source"]
            assert not self.augment_task_input

        for field_name in fields:
            try:
                old_value = dict_get(
                    instance,
                    field_name,
                    use_dpath=True,
                    default="",
                    not_exist_ok=False,
                )
            except TypeError as e:
                raise TypeError(f"Failed to get {field_name} from {instance}") from e

            # We are setting a nested seed based on the value processed, to ensure that
            # the augmentation randomizations do not effect other randomization choices and
            # to make the augmentation randomization choices different for each text.
            with nested_seed(str(hash(old_value))):
                try:
                    new_value = self.process_value(old_value)
                except Exception as e:
                    raise RuntimeError(
                        f"Error augmenting value '{old_value}' from '{field_name}' in instance: {instance}"
                    ) from e
            dict_set(instance, field_name, new_value, use_dpath=True, not_exist_ok=True)
        return instance


class NullAugmentor(Augmentor):
    def verify(self):
        pass

    def process_value(self, value: Any) -> Any:
        return value


class AugmentWhitespace(Augmentor):
    """Augments the inputs by replace existing whitespace with other whitespace.

    Currently each whitespace is replaced by a random choice of 1-3 whitespace charaters (spcae, tab, newline).
    """

    def process_value(self, value: Any) -> Any:
        import re

        words = re.split(r"(\s+)", value)
        new_value = ""

        for word in words:
            if word.isspace():
                new_value += get_random().choice(
                    ["\n", "\t", " "]
                ) * get_random().randint(1, 3)
            else:
                new_value += word
        return new_value


class AugmentSuffix(Augmentor):
    r"""Augments the input by appending to it a randomly selected (typically, whitespace) pattern.

    Args:
     suffixes : the potential (typically, whitespace) patterns to select from.
        The dictionary version allows to specify relative weights of the different patterns.
     remove_existing_trailing_whitespaces : allows to first clean existing trailing whitespaces.
        The selected pattern is then appended to the potentially trimmed at its end input.


    Examples:
        to append a '\n' or a '\t' to the end of the input, employ
        AugmentSuffix(augment_model_input=True, suffixes=['\n','\t'])
        If '\n' is preferred over '\t', at 2:1 ratio, employ
        AugmentSuffix(augment_model_input=True, suffixes={'\n':2,'\t':1})
        which will append '\n' twice as often as '\t'.

    """

    suffixes: Optional[Union[List[str], Dict[str, int]]] = [" ", "\n", "\t"]
    remove_existing_trailing_whitespaces: Optional[bool] = False

    def verify(self):
        assert (
            isinstance(self.suffixes, list) or isinstance(self.suffixes, dict)
        ), f"Argument 'suffixes' should be either a list or a dictionary, whereas it is of type {type(self.suffixes)}"

        if isinstance(self.suffixes, dict):
            for k, v in self.suffixes.items():
                assert isinstance(
                    k, str
                ), f"suffixes should map strings, whereas key {k!s} is of type {type(k)}"
                assert isinstance(
                    v, int
                ), f"suffixes should map to ints, whereas value {v!s} is of type {type(v)}"
        else:
            for k in self.suffixes:
                assert isinstance(
                    k, str
                ), f"suffixes should be a list of strings, whereas member {k!s} is of type {type(k)}"

        self.pats = (
            self.suffixes
            if isinstance(self.suffixes, list)
            else [k for k, v in self.suffixes.items()]
        )
        total_weight = (
            len(self.pats)
            if isinstance(self.suffixes, list)
            else sum([v for k, v in self.suffixes.items()])
        )
        self.weights = (
            [1.0 / total_weight] * len(self.pats)
            if isinstance(self.suffixes, list)
            else [float(self.suffixes[p]) / total_weight for p in self.pats]
        )
        super().verify()

    def process_value(self, value: Any) -> Any:
        assert value is not None, "input value should not be None"
        new_value = str(value)
        if self.remove_existing_trailing_whitespaces:
            new_value = new_value.rstrip()
        new_value += get_random().choices(self.pats, self.weights, k=1)[0]

        return new_value


class ShuffleFieldValues(FieldOperator):
    """Shuffles an iterable value."""

    def process_value(self, value: Any) -> Any:
        res = list(value)
        get_random().shuffle(res)
        return res


class JoinStr(FieldOperator):
    """Joins a list of strings (contents of a field), similar to str.join().

    Args:
        separator (str): text to put between values
    """

    separator: str = ","

    def process_value(self, value: Any) -> Any:
        return self.separator.join(str(x) for x in value)


class Apply(StreamInstanceOperator):
    """A class used to apply a python function and store the result in a field.

    Args:
        function (str): name of function.
        to_field (str): the field to store the result
        additional arguments are field names passed to the function

    Examples:
    Store in field  "b" the uppercase string of the value in field "a"
    Apply("a", function=str.upper, to_field="b")

    Dump the json representation of field "t" and store back in the same field.
    Apply("t", function=json.dumps, to_field="t")

    Set the time in a field 'b'.
    Apply(function=time.time, to_field="b")

    """

    __allow_unexpected_arguments__ = True
    function: Callable = NonPositionalField(required=True)
    to_field: str = NonPositionalField(required=True)

    def function_to_str(self, function: Callable) -> str:
        parts = []

        if hasattr(function, "__module__"):
            parts.append(function.__module__)
        if hasattr(function, "__qualname__"):
            parts.append(function.__qualname__)
        else:
            parts.append(function.__name__)

        return ".".join(parts)

    def str_to_function(self, function_str: str) -> Callable:
        splitted = function_str.split(".", 1)
        if len(splitted) == 1:
            return __builtins__[splitted[0]]

        module_name, function_name = splitted
        if module_name in __builtins__:
            obj = __builtins__[module_name]
        elif module_name in globals():
            obj = globals()[module_name]
        else:
            obj = importlib.import_module(module_name)
        for part in function_name.split("."):
            obj = getattr(obj, part)
        return obj

    def prepare(self):
        super().prepare()
        if isinstance(self.function, str):
            self.function = self.str_to_function(self.function)
        self._init_dict["function"] = self.function_to_str(self.function)

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        argv = [instance[arg] for arg in self._argv]
        kwargs = {key: instance[val] for key, val in self._kwargs}

        result = self.function(*argv, **kwargs)

        instance[self.to_field] = result
        return instance


class ListFieldValues(StreamInstanceOperator):
    """Concatenates values of multiple fields into a list, and assigns it to a new field."""

    fields: List[str]
    to_field: str
    use_query: bool = False

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        values = []
        for field_name in self.fields:
            values.append(dict_get(instance, field_name, use_dpath=self.use_query))
        instance[self.to_field] = values
        return instance


class ZipFieldValues(StreamInstanceOperator):
    """Zips values of multiple fields similar to list(zip(*fields))."""

    fields: str
    to_field: str
    longest: bool = False
    use_query: bool = False

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        values = []
        for field_name in self.fields:
            values.append(dict_get(instance, field_name, use_dpath=self.use_query))
        if self.longest:
            zipped = zip_longest(*values)
        else:
            zipped = zip(*values)
        instance[self.to_field] = list(zipped)
        return instance


class IndexOf(StreamInstanceOperator):
    """Finds the location of one value in another (iterable) value similar to to_field=search_in.index(index_of)."""

    search_in: str
    index_of: str
    to_field: str
    use_query: bool = False

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        lst = dict_get(instance, self.search_in, use_dpath=self.use_query)
        item = dict_get(instance, self.index_of, use_dpath=self.use_query)
        instance[self.to_field] = lst.index(item)
        return instance


class TakeByField(StreamInstanceOperator):
    """Takes value from one field based on another field similar to field[index]."""

    field: str
    index: str
    to_field: str = None
    use_query: bool = False

    def prepare(self):
        if self.to_field is None:
            self.to_field = self.field

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        value = dict_get(instance, self.field, use_dpath=self.use_query)
        index_value = dict_get(instance, self.index, use_dpath=self.use_query)
        instance[self.to_field] = value[index_value]
        return instance


class CopyFields(FieldOperator):
    """Copies specified fields from one field to another.

    Args:
        field_to_field (Union[List[List], Dict[str, str]]): A list of lists, where each sublist contains the source field and the destination field, or a dictionary mapping source fields to destination fields.
        use_dpath (bool): Whether to use dpath for accessing fields. Defaults to False.
    """

    def process_value(self, value: Any) -> Any:
        return value


class AddID(StreamInstanceOperator):
    id_field_name: str = "id"

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        instance[self.id_field_name] = str(uuid.uuid4()).replace("-", "")
        return instance


class CastFields(StreamInstanceOperator):
    """Casts specified fields to specified types.

    Args:
        types (Dict[str, str]): A dictionary mapping fields to their new types.
        nested (bool): Whether to cast nested fields. Defaults to False.
        fields (Dict[str, str]): A dictionary mapping fields to their new types.
        defaults (Dict[str, object]): A dictionary mapping types to their default values for cases of casting failure.
    """

    types = {
        "int": int,
        "float": float,
        "str": str,
        "bool": bool,
    }
    fields: Dict[str, str] = field(default_factory=dict)
    failure_defaults: Dict[str, object] = field(default_factory=dict)
    use_nested_query: bool = False
    cast_multiple: bool = False

    def _cast_single(self, value, type, field):
        try:
            return self.types[type](value)
        except Exception as e:
            if field not in self.failure_defaults:
                raise ValueError(
                    f'Failed to cast field "{field}" with value {value} to type "{type}", and no default value is provided.'
                ) from e
            return self.failure_defaults[field]

    def _cast_multiple(self, values, type, field):
        values = [self._cast_single(value, type, field) for value in values]

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        for field_name, type in self.fields.items():
            value = dict_get(instance, field_name, use_dpath=self.use_nested_query)
            if self.cast_multiple:
                casted_value = self._cast_multiple(value, type, field_name)
            else:
                casted_value = self._cast_single(value, type, field_name)
            dict_set(
                instance, field_name, casted_value, use_dpath=self.use_nested_query
            )
        return instance


def recursive_divide(instance, divisor, strict=False):
    if isinstance(instance, dict):
        for key, value in instance.items():
            instance[key] = recursive_divide(value, divisor, strict=strict)
    elif isinstance(instance, list):
        for i, value in enumerate(instance):
            instance[i] = recursive_divide(value, divisor, strict=strict)
    elif isinstance(instance, float):
        instance /= divisor
    elif strict:
        raise ValueError(f"Cannot divide instance of type {type(instance)}")
    return instance


class DivideAllFieldsBy(StreamInstanceOperator):
    divisor: float = 1.0
    strict: bool = False
    recursive: bool = True

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        return recursive_divide(instance, self.divisor, strict=self.strict)


class ArtifactFetcherMixin:
    """Provides a way to fetch and cache artifacts in the system.

    Args:
        cache (Dict[str, Artifact]): A cache for storing fetched artifacts.
    """

    cache: Dict[str, Artifact] = {}

    @classmethod
    def get_artifact(cls, artifact_identifier: str) -> Artifact:
        if artifact_identifier not in cls.cache:
            artifact, artifactory = fetch_artifact(artifact_identifier)
            cls.cache[artifact_identifier] = artifact
        return cls.cache[artifact_identifier]


class ApplyOperatorsField(StreamInstanceOperator, ArtifactFetcherMixin):
    """Applies value operators to each instance in a stream based on specified fields.

    Args:
        value_field (str): The field containing the value to be operated on.
        operators_field (str): The field containing the operators to be applied.
        default_operators (List[str]): A list of default operators to be used if no operators are found in the instance.
    """

    inputs_fields: str

    operators_field: str
    default_operators: List[str] = None
    fields_to_treat_as_list: List[str] = NonPositionalField(default_factory=list)

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        operator_names = instance.get(self.operators_field)
        if operator_names is None:
            assert (
                self.default_operators is not None
            ), f"No operators found in {self.field} field and no default operators provided"
            operator_names = self.default_operators

        if isinstance(operator_names, str):
            operator_names = [operator_names]

        for name in operator_names:
            operator = self.get_artifact(name)
            for field_name in self.inputs_fields:
                value = instance[field_name]
                if field_name in self.fields_to_treat_as_list:
                    instance[field_name] = [operator.process(v) for v in value]
                else:
                    instance[field_name] = operator.process(instance[field_name])

        return instance


class FilterByValues(SingleStreamOperator):
    """Filters a stream, yielding only instances that match specified values in the provided fields.

    Args:
        values (Dict[str, Any]): For each field, the values that instances should match to be included in the output.
    """

    required_values: Dict[str, Any]

    def process(self, stream: Stream, stream_name: Optional[str] = None) -> Generator:
        for instance in stream:
            filter = False
            for key, value in self.required_values.items():
                if key not in instance:
                    raise ValueError(
                        f"Required filter field ('{key}') in FilterByValues is not found in {instance}"
                    )
                if instance[key] != value:
                    filter = True
            if not filter:
                yield instance


class ExtractFieldValues(MultiStreamOperator):
    field: str
    stream_name: str
    overall_top_frequency_percent: Optional[int] = 100
    min_frequency_percent: Optional[int] = 0
    to_field: str
    process_every_value: Optional[bool] = False

    """
    Extract the unique values of a field ('field') of a given stream ('stream_name') and store (the most frequent of) them
    as a list in a new field ('to_field') in all streams.

    More specifically, sort all the unique values encountered in field 'field' by decreasing order of frequency.
    When 'overall_top_frequency_percent' is smaller than 100, trim the list from bottom, so that the total frequency of
    the remaining values makes 'overall_top_frequency_percent' of the total number of instances in the stream.
    When 'min_frequency_percent' is larger than 0, remove from the list any value whose relative frequency makes
    less than 'min_frequency_percent' of the total number of instances in the stream.
    At most one of 'overall_top_frequency_percent' and 'min_frequency_percent' is allowed to move from their default values.

    Examples:

    ExtractFieldValues(stream_name="train", field="label", to_field="classes") - extracts all the unique values of
    field 'label', sorts them by decreasing frequency, and stores the resulting list in field 'classes' of each and
    every instance in all streams.

    ExtractFieldValues(stream_name="train", field="labels", to_field="classes", process_every_value=True) -
    in case that field 'labels' contains a list of values (and not a single value) - track the occurrences of all the possible
    value members in these lists, and report the most frequent values.
    if process_every_value=False, track the most frequent whole lists, and report those (as a list of lists) in field
    'to_field' of each instance of all streams.

    ExtractFieldValues(stream_name="train", field="label", to_field="classes",overall_top_frequency_percent=80) -
    extracts the most frequent possible values of field 'label' that together cover at least 80% of the instances of stream_name,
    and stores them in field 'classes' of each instance of all streams.

    ExtractFieldValues(stream_name="train", field="label", to_field="classes",min_frequency_percent=5) -
    extracts all possible values of field 'label' that cover, each, at least 5% of the instances.
    Stores these values, sorted by decreasing order of frequency, in field 'classes' of each instance in all streams.
    """

    def verify(self):
        assert (
            self.overall_top_frequency_percent <= 100
            and self.overall_top_frequency_percent >= 0
        ), "'overall_top_frequency_percent' must be between 0 and 100"
        assert (
            self.min_frequency_percent <= 100 and self.min_frequency_percent >= 0
        ), "'min_frequency_percent' must be between 0 and 100"
        assert not (
            self.overall_top_frequency_percent < 100 and self.min_frequency_percent > 0
        ), "At most one of 'overall_top_frequency_percent' and 'min_frequency_percent' is allowed to move from their default value"
        super().verify()

    def process(self, multi_stream: MultiStream) -> MultiStream:
        stream = multi_stream[self.stream_name]
        all_values = []
        for instance in stream:
            if (not isinstance(instance[self.field], list)) and (
                self.process_every_value is True
            ):
                raise ValueError(
                    "'process_every_field' is allowed to change to 'True' only for fields whose contents are lists"
                )
            if (not isinstance(instance[self.field], list)) or (
                self.process_every_value is False
            ):
                # either not a list, or is a list but process_every_value == False : view contetns of 'field' as one entity whose occurrences are counted.
                all_values.append(
                    (*instance[self.field],)
                    if isinstance(instance[self.field], list)
                    else instance[self.field]
                )  # convert to a tuple if list, to enable the use of Counter which would not accept
                # a list as an entity to count its occurrences
            else:
                # content of 'field' is a list and process_every_value == True: add one occurrence on behalf of each individual value
                all_values.extend(instance[self.field])
        counter = Counter(
            all_values
        )  # here all_values is a list of individual values, or tupples. Hence, Counter is feasible
        values_and_counts = counter.most_common()
        if self.overall_top_frequency_percent < 100:
            top_frequency = len(all_values) * self.overall_top_frequency_percent / 100.0
            sum_counts = 0
            for _i, p in enumerate(values_and_counts):
                sum_counts += p[1]
                if sum_counts >= top_frequency:
                    break
            values_and_counts = counter.most_common(_i + 1)
        if self.min_frequency_percent > 0:
            min_frequency = self.min_frequency_percent * len(all_values) / 100.0
            while values_and_counts[-1][1] < min_frequency:
                values_and_counts.pop()
        values_to_keep = [
            [*ele[0]] if isinstance(ele[0], tuple) else ele[0]
            for ele in values_and_counts
        ]
        for name in multi_stream:
            for instance in multi_stream[name]:
                instance[self.to_field] = values_to_keep
        return multi_stream


class FilterByListsOfValues(SingleStreamOperator):
    """Filters a stream, yielding only instances that  whose field values are included in the specified value lists.

    Args:
        required_values (Dict[str, List]): For each field, the list of values that instances should match to be included in the output.
    """

    required_values: Dict[str, List]

    def verify(self):
        super().verify()
        for key, value in self.required_values.items():
            if not isinstance(value, list):
                raise ValueError(
                    f"The filter for key ('{key}') in FilterByListsOfValues is not a list but '{value}'"
                )

    def process(self, stream: Stream, stream_name: Optional[str] = None) -> Generator:
        for instance in stream:
            filter = False
            for key, value in self.required_values.items():
                if key not in instance:
                    raise ValueError(
                        f"Required filter field ('{key}') in FilterByListsOfValues is not found in {instance}"
                    )
                if instance[key] not in value:
                    filter = True
            if not filter:
                yield instance


class Intersect(FieldOperator):
    """Intersects the value of a field, which must be a list, with a given list.

    Args:
        allowed_values (list) - list to intersect.
    """

    allowed_values: List[Any]

    def verify(self):
        super().verify()
        if self.process_every_value:
            raise ValueError(
                "'process_every_value=True' is not supported in Intersect operator"
            )

        if not isinstance(self.allowed_values, list):
            raise ValueError(
                f"The allowed_values is not a list but '{self.allowed_values}'"
            )

    def process_value(self, value: Any) -> Any:
        if not isinstance(value, list):
            raise ValueError(f"The value in field is not a list but '{value}'")
        return [e for e in value if e in self.allowed_values]


class RemoveValues(FieldOperator):
    """Removes elements in a field, which must be a list, using a given list of unallowed.

    Args:
        unallowed_values (list) - removed_values.
    """

    unallowed_values: List[Any]

    def verify(self):
        super().verify()
        if self.process_every_value:
            raise ValueError(
                "'process_every_value=True' is not supported in RemoveValues operator"
            )

        if not isinstance(self.unallowed_values, list):
            raise ValueError(
                f"The unallowed_values is not a list but '{self.unallowed_values}'"
            )

    def process_value(self, value: Any) -> Any:
        if not isinstance(value, list):
            raise ValueError(f"The value in field is not a list but '{value}'")
        return [e for e in value if e not in self.unallowed_values]


class Unique(SingleStreamReducer):
    """Reduces a stream to unique instances based on specified fields.

    Args:
        fields (List[str]): The fields that should be unique in each instance.
    """

    fields: List[str] = field(default_factory=list)

    @staticmethod
    def to_tuple(instance: dict, fields: List[str]) -> tuple:
        result = []
        for field_name in fields:
            value = instance[field_name]
            if isinstance(value, list):
                value = tuple(value)
            result.append(value)
        return tuple(result)

    def process(self, stream: Stream) -> Stream:
        seen = set()
        for instance in stream:
            values = self.to_tuple(instance, self.fields)
            if values not in seen:
                seen.add(values)
        return list(seen)


class SplitByValue(MultiStreamOperator):
    """Splits a MultiStream into multiple streams based on unique values in specified fields.

    Args:
        fields (List[str]): The fields to use when splitting the MultiStream.
    """

    fields: List[str] = field(default_factory=list)

    def process(self, multi_stream: MultiStream) -> MultiStream:
        uniques = Unique(fields=self.fields)(multi_stream)

        result = {}

        for stream_name, stream in multi_stream.items():
            stream_unique_values = uniques[stream_name]
            for unique_values in stream_unique_values:
                filtering_values = dict(zip(self.fields, unique_values))
                filtered_streams = FilterByValues(
                    required_values=filtering_values
                )._process_single_stream(stream)
                filtered_stream_name = (
                    stream_name + "_" + nested_tuple_to_string(unique_values)
                )
                result[filtered_stream_name] = filtered_streams

        return MultiStream(result)


class ApplyStreamOperatorsField(SingleStreamOperator, ArtifactFetcherMixin):
    """Applies stream operators to a stream based on specified fields in each instance.

    Args:
        field (str): The field containing the operators to be applied.
        reversed (bool): Whether to apply the operators in reverse order.
    """

    field: str
    reversed: bool = False

    def process(self, stream: Stream, stream_name: Optional[str] = None) -> Generator:
        first_instance = stream.peak()

        operators = first_instance.get(self.field, [])
        if isinstance(operators, str):
            operators = [operators]

        if self.reversed:
            operators = list(reversed(operators))

        for operator_name in operators:
            operator = self.get_artifact(operator_name)
            assert isinstance(
                operator, StreamingOperator
            ), f"Operator {operator_name} must be a SingleStreamOperator"

            stream = operator(MultiStream({"tmp": stream}))["tmp"]

        yield from stream


class ApplyMetric(SingleStreamOperator, ArtifactFetcherMixin):
    """Applies metric operators to a stream based on a metric field specified in each instance.

    Args:
        metric_field (str): The field containing the metrics to be applied.
        calc_confidence_intervals (bool): Whether the applied metric should calculate confidence intervals or not.
    """

    metric_field: str
    calc_confidence_intervals: bool

    def process(self, stream: Stream, stream_name: Optional[str] = None) -> Generator:
        from .metrics import Metric, MetricPipeline, MetricWithConfidenceInterval

        first_instance = stream.peak()

        metric_names = first_instance.get(self.metric_field, [])
        if not metric_names:
            raise RuntimeError(
                f"Missing metric names in field '{self.metric_field}' and instance '{first_instance}'."
            )

        if isinstance(metric_names, str):
            metric_names = [metric_names]

        # Each metric operator computes its score and then sets the main score, overwriting
        # the previous main score value (if any). So, we need to reverse the order of the listed metrics.
        # This will cause the first listed metric to run last, and the main score will be set
        # by the first listed metric (as desired).
        metric_names = list(reversed(metric_names))

        for metric_name in metric_names:
            metric = self.get_artifact(metric_name)
            assert isinstance(
                metric, Metric
            ), f"Operator {metric_name} must be a Metric"

            if not self.calc_confidence_intervals:
                if isinstance(metric, MetricWithConfidenceInterval):
                    metric.disable_confidence_interval_calculation()
                elif isinstance(metric, MetricPipeline) and isinstance(
                    metric.metric, MetricWithConfidenceInterval
                ):
                    metric.metric.disable_confidence_interval_calculation()

            stream = metric(MultiStream({"tmp": stream}))["tmp"]

        yield from stream


class AddFieldNamePrefix(StreamInstanceOperator):
    """Adds a prefix to each field name in each instance of a stream.

    Args:
        prefix_dict (Dict[str, str]): A dictionary mapping stream names to prefixes.
    """

    prefix_dict: Dict[str, str]

    def prepare(self):
        return super().prepare()

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        return {
            self.prefix_dict[stream_name] + key: value
            for key, value in instance.items()
        }


class MergeStreams(MultiStreamOperator):
    """Merges multiple streams into a single stream.

    Args:
        new_stream_name (str): The name of the new stream resulting from the merge.
        add_origin_stream_name (bool): Whether to add the origin stream name to each instance.
        origin_stream_name_field_name (str): The field name for the origin stream name.
    """

    streams_to_merge: List[str] = None
    new_stream_name: str = "all"
    add_origin_stream_name: bool = True
    origin_stream_name_field_name: str = "origin"

    def merge(self, multi_stream):
        for stream_name, stream in multi_stream.items():
            if self.streams_to_merge is None or stream_name in self.streams_to_merge:
                for instance in stream:
                    if self.add_origin_stream_name:
                        instance[self.origin_stream_name_field_name] = stream_name
                    yield instance

    def process(self, multi_stream: MultiStream) -> MultiStream:
        return MultiStream(
            {
                self.new_stream_name: Stream(
                    self.merge, gen_kwargs={"multi_stream": multi_stream}
                )
            }
        )


class Shuffle(PagedStreamOperator):
    """Shuffles the order of instances in each page of a stream.

    Args:
        page_size (int): The size of each page in the stream. Defaults to 1000.
    """

    def process(self, page: List[Dict], stream_name: Optional[str] = None) -> Generator:
        get_random().shuffle(page)
        yield from page


class EncodeLabels(StreamInstanceOperator):
    """Encode labels of specified fields together a into integers.

    Args:
        fields (List[str]): The fields to encode together.
    """

    fields: List[str]

    def _process_multi_stream(self, multi_stream: MultiStream) -> MultiStream:
        self.encoder = {}
        return super()._process_multi_stream(multi_stream)

    def process(
        self, instance: Dict[str, Any], stream_name: Optional[str] = None
    ) -> Dict[str, Any]:
        for field_name in self.fields:
            values = dict_get(instance, field_name, use_dpath=True)
            if not isinstance(values, list):
                values = [values]
            for value in values:
                if value not in self.encoder:
                    self.encoder[value] = len(self.encoder)
            new_values = [self.encoder[value] for value in values]
            dict_set(
                instance, field_name, new_values, use_dpath=True, set_multiple=True
            )

        return instance


class StreamRefiner(SingleStreamOperator):
    max_instances: int = None

    def process(self, stream: Stream, stream_name: Optional[str] = None) -> Generator:
        if self.max_instances is not None:
            yield from stream.take(self.max_instances)
        else:
            yield from stream


class DeterministicBalancer(StreamRefiner):
    """A class used to balance streams deterministically.

    Attributes:
        fields (List[str]): A list of field names to be used in determining the signature of an instance.
        streams (List[str]): A list of stream names to be processed by the balancer.

    Usage:
        balancer = DeterministicBalancer(fields=["field1", "field2"], streams=["stream1", "stream2"])
        balanced_stream = balancer.process(stream)
    """

    fields: List[str]

    def signature(self, instance):
        return str(
            tuple(dict_get(instance, field, use_dpath=True) for field in self.fields)
        )

    def process(self, stream: Stream, stream_name: Optional[str] = None) -> Generator:
        counter = collections.Counter()

        for instance in stream:
            counter[self.signature(instance)] += 1

        if len(counter) == 0:
            return

        lowest_count = counter.most_common()[-1][-1]

        max_total_instances_per_sign = lowest_count
        if self.max_instances is not None:
            max_total_instances_per_sign = min(
                lowest_count, self.max_instances // len(counter)
            )

        counter = collections.Counter()

        for instance in stream:
            sign = self.signature(instance)
            if counter[sign] < max_total_instances_per_sign:
                counter[sign] += 1
                yield instance


class LengthBalancer(DeterministicBalancer):
    segments_boundaries: List[int]

    def signature(self, instance):
        total_len = 0
        for field_name in self.fields:
            total_len += len(dict_get(instance, field_name, use_dpath=True))
        for i, val in enumerate(self.segments_boundaries):
            if total_len < val:
                return i
        return i + 1