Upload operator.py with huggingface_hub

operator.py

@@ -3,6 +3,7 @@ from dataclasses import field
 from typing import Any, Dict, Generator, List, Optional, Union
 
 from .artifact import Artifact
+from .random_utils import nested_seed
 from .stream import MultiStream, Stream
 
 
@@ -31,11 +32,11 @@ class OperatorError(Exception):
 
 class StreamingOperator(Artifact):
     """
-    Base class for all stream operators in the streaming model. 
+    Base class for all stream operators in the streaming model.
 
     Stream operators are a key component of the streaming model and are responsible for processing continuous data streams.
-    They perform operations such as transformations, aggregations, joins, windowing and more on these streams. 
-    There are several types of stream operators, including source operators, processing operators, etc. 
+    They perform operations such as transformations, aggregations, joins, windowing and more on these streams.
+    There are several types of stream operators, including source operators, processing operators, etc.
 
     As a `StreamingOperator`, this class is responsible for performing operations on a stream, and must be implemented by all other specific types of stream operators in the system.
     When called, a `StreamingOperator` must return a MultiStream.
@@ -59,13 +60,14 @@ class StreamingOperator(Artifact):
 
 class StreamSource(StreamingOperator):
     """
-    A class representing a stream source operator in the streaming system. 
+    A class representing a stream source operator in the streaming system.
 
     A stream source operator is a special type of `StreamingOperator` that generates a data stream without taking any input streams. It serves as the starting point in a stream processing pipeline, providing the initial data that other operators in the pipeline can process.
 
     When called, a `StreamSource` should generate a `MultiStream`. This behavior must be implemented by any classes that inherit from `StreamSource`.
 
     """
+
     @abstractmethod
     def __call__(self) -> MultiStream:
         pass
@@ -73,15 +75,17 @@ class StreamSource(StreamingOperator):
 
 class SourceOperator(StreamSource):
     """
-    A class representing a source operator in the streaming system. 
+    A class representing a source operator in the streaming system.
 
     A source operator is responsible for generating the data stream from some source, such as a database or a file. This is the starting point of a stream processing pipeline. The `SourceOperator` class is a type of `StreamSource`, which is a special type of `StreamingOperator` that generates an output stream but does not take any input streams.
 
     When called, a `SourceOperator` invokes its `process` method, which should be implemented by all subclasses to generate the required `MultiStream`.
 
     """
+
     def __call__(self) -> MultiStream:
-        return self.process()
+        with nested_seed():
+            return self.process()
 
     @abstractmethod
     def process(self) -> MultiStream:
@@ -90,16 +94,17 @@ class SourceOperator(StreamSource):
 
 class StreamInitializerOperator(StreamSource):
     """
-    A class representing a stream initializer operator in the streaming system. 
+    A class representing a stream initializer operator in the streaming system.
 
-    A stream initializer operator is a special type of `StreamSource` that is capable of taking parameters during the stream generation process. This can be useful in situations where the stream generation process needs to be customized or configured based on certain parameters. 
+    A stream initializer operator is a special type of `StreamSource` that is capable of taking parameters during the stream generation process. This can be useful in situations where the stream generation process needs to be customized or configured based on certain parameters.
 
     When called, a `StreamInitializerOperator` invokes its `process` method, passing any supplied arguments and keyword arguments. The `process` method should be implemented by all subclasses to generate the required `MultiStream` based on the given arguments and keyword arguments.
 
     """
-    
+
     def __call__(self, *args, **kwargs) -> MultiStream:
-        return self.process(*args, **kwargs)
+        with nested_seed():
+            return self.process(*args, **kwargs)
 
     @abstractmethod
     def process(self, *args, **kwargs) -> MultiStream:
@@ -112,8 +117,10 @@ class MultiStreamOperator(StreamingOperator):
 
     A multi-stream operator is a type of `StreamingOperator` that operates on an entire MultiStream object at once. It takes a `MultiStream` as input and produces a `MultiStream` as output. The `process` method should be implemented by subclasses to define the specific operations to be performed on the input `MultiStream`.
     """
+
     def __call__(self, multi_stream: Optional[MultiStream] = None) -> MultiStream:
-        return self._process_multi_stream(multi_stream)
+        with nested_seed():
+            return self._process_multi_stream(multi_stream)
 
     def _process_multi_stream(self, multi_stream: Optional[MultiStream] = None) -> MultiStream:
         result = self.process(multi_stream)
@@ -131,6 +138,7 @@ class SingleStreamOperator(MultiStreamOperator):
 
     A single-stream operator is a type of `MultiStreamOperator` that operates on individual `Stream` objects within a `MultiStream`. It iterates through each `Stream` in the `MultiStream` and applies the `process` method. The `process` method should be implemented by subclasses to define the specific operations to be performed on each `Stream`.
     """
+
     def _process_multi_stream(self, multi_stream: MultiStream) -> MultiStream:
         result = {}
         for stream_name, stream in multi_stream.items():
@@ -157,8 +165,9 @@ class PagedStreamOperator(SingleStreamOperator):
 
     A paged-stream operator is a type of `SingleStreamOperator` that operates on a page of instances in a `Stream` at a time, where a page is a subset of instances. The `process` method should be implemented by subclasses to define the specific operations to be performed on each page.
     """
+
     page_size: int = 1000
-    
+
     def _process_stream(self, stream: Stream, stream_name: str = None) -> Generator:
         page = []
         for instance in stream:
@@ -179,6 +188,7 @@ class SingleStreamReducer(StreamingOperator):
 
     A single-stream reducer is a type of `StreamingOperator` that operates on individual `Stream` objects within a `MultiStream` and reduces each `Stream` to a single output value. The `process` method should be implemented by subclasses to define the specific reduction operation to be performed on each `Stream`.
     """
+
     def __call__(self, multi_stream: Optional[MultiStream] = None) -> Dict[str, Any]:
         result = {}
         for stream_name, stream in multi_stream.items():
@@ -198,6 +208,7 @@ class StreamInstanceOperator(SingleStreamOperator):
 
     A stream instance operator is a type of `SingleStreamOperator` that operates on individual instances within a `Stream`. It iterates through each instance in the `Stream` and applies the `process` method. The `process` method should be implemented by subclasses to define the specific operations to be performed on each instance.
     """
+
     def _process_stream(self, stream: Stream, stream_name: str = None) -> Generator:
         for instance in stream:
             yield self._process_instance(instance, stream_name)
@@ -216,6 +227,7 @@ class StreamInstanceOperatorValidator(StreamInstanceOperator):
 
     A stream instance operator validator is a type of `StreamInstanceOperator` that includes a validation step. It operates on individual instances within a `Stream` and validates the result of processing each instance.
     """
+
     @abstractmethod
     def validate(self, instance):
         pass
@@ -235,6 +247,7 @@ class InstanceOperator(Artifact):
 
     An instance operator is a type of `Artifact` that operates on a single instance (represented as a dict) at a time. It takes an instance as input and produces a transformed instance as output.
     """
+
     def __call__(self, data: dict) -> dict:
         return self.process(data)
 
@@ -243,12 +256,13 @@ class InstanceOperator(Artifact):
         pass
 
 
-class FieldOperator(Artifact):
+class BaseFieldOperator(Artifact):
     """
     A class representing a field operator in the streaming system.
 
     A field operator is a type of `Artifact` that operates on a single field within an instance. It takes an instance and a field name as input, processes the field, and updates the field in the instance with the processed value.
     """
+
     def __call__(self, data: Dict[str, Any], field: str) -> dict:
         value = self.process(data[field])
         data[field] = value
@@ -263,10 +277,11 @@ class InstanceOperatorWithGlobalAccess(StreamingOperator):
     """
     A class representing an instance operator with global access in the streaming system.
 
-    An instance operator with global access is a type of `StreamingOperator` that operates on individual instances within a `Stream` and can also access other streams. 
+    An instance operator with global access is a type of `StreamingOperator` that operates on individual instances within a `Stream` and can also access other streams.
     It uses the `accessible_streams` attribute to determine which other streams it has access to.
     In order to make this efficient and to avoid qudratic complexity, it caches the accessible streams by default.
     """
+
     accessible_streams: Union[MultiStream, List[str]] = None
     cache_accessible_streams: bool = True
 
@@ -303,6 +318,7 @@ class SequntialOperator(MultiStreamOperator):
 
     A sequential operator is a type of `MultiStreamOperator` that applies a sequence of other operators to a `MultiStream`. It maintains a list of `StreamingOperator`s and applies them in order to the `MultiStream`.
     """
+
     steps: List[StreamingOperator] = field(default_factory=list)
 
     def process(self, multi_stream: Optional[MultiStream] = None) -> MultiStream:
@@ -317,6 +333,7 @@ class SourceSequntialOperator(SequntialOperator):
 
     A source sequential operator is a type of `SequntialOperator` that starts with a source operator. The first operator in its list of steps is a `StreamSource`, which generates the initial `MultiStream` that the other operators then process.
     """
+
     def __call__(self) -> MultiStream:
         return super().__call__()
 
@@ -333,8 +350,10 @@ class SequntialOperatorInitilizer(SequntialOperator):
 
     A sequential operator initializer is a type of `SequntialOperator` that starts with a stream initializer operator. The first operator in its list of steps is a `StreamInitializerOperator`, which generates the initial `MultiStream` based on the provided arguments and keyword arguments.
     """
+
     def __call__(self, *args, **kwargs) -> MultiStream:
-        return self.process(*args, **kwargs)
+        with nested_seed():
+            return self.process(*args, **kwargs)
 
     def process(self, *args, **kwargs) -> MultiStream:
         assert isinstance(