Pipe testing

Src/AST/Myc.Ast.Evaluator.pas

@@ -480,6 +480,7 @@ begin
         vkRecordSeries: Result := TDataValue.FromSeries(baseValue.AsRecordSeries.Fields[Node.Member.Value]);
         vkScalarRecord: Result := TDataValue(baseValue.AsScalarRecord.Fields[Node.Member.Value]);
         vkGenericRecord: Result := TDataValue(baseValue.AsGenericRecord.Fields[Node.Member.Value]);
+        vkStream: Result := TDataValue.FromSeries(baseValue.AsStream.Series.Fields[Node.Member.Value]);
     else
         raise EEvaluatorException.Create('Member access operator `.` is not supported for this value type.');
     end;

Src/AST/Myc.Fmx.AstEditor.Handlers.Pipes.pas

@@ -0,0 +1,177 @@
+unit Myc.Fmx.AstEditor.Handlers.Pipes;
+
+interface
+
+uses
+    System.SysUtils,
+    System.UITypes,
+    System.Types,
+    Myc.Ast,
+    Myc.Ast.Nodes,
+    Myc.Fmx.AstEditor.Render,
+    Myc.Fmx.AstEditor.Node,
+    Myc.Fmx.AstEditor.Handlers;
+
+type
+    // --- Pipe Container ---
+
+    TPipeNodeHandler = class(TBaseNodeHandler<IPipeNode>)
+    private
+        FInputsNode: TAstViewNode;
+        FTransformNode: TAstViewNode;
+    public
+        procedure BuildUI(OwnerNode: TAstViewNode); override;
+        function ReconstructAst(OwnerNode: TAstViewNode): IAstNode; override;
+    end;
+
+    // --- Pipe Inputs ---
+
+    TPipeInputListHandler = class(TNodeListHandler<IPipeInputNode>)
+    protected
+        function GetOrientation: TLayoutOrientation; override;
+        function GetAlignment: TLayoutAlignment; override;
+    public
+        function ReconstructAst(OwnerNode: TAstViewNode): IAstNode; override;
+    end;
+
+    TPipeInputNodeHandler = class(TBaseNodeHandler<IPipeInputNode>)
+    private
+        FStreamNode: TAstViewNode;
+        FSelectorsNode: TAstViewNode;
+    public
+        procedure BuildUI(OwnerNode: TAstViewNode); override;
+        function ReconstructAst(OwnerNode: TAstViewNode): IAstNode; override;
+    end;
+
+    // --- Selectors ---
+
+    TPipeSelectorListHandler = class(TNodeListHandler<IKeywordNode>)
+    protected
+        function GetOrientation: TLayoutOrientation; override;
+    public
+        function ReconstructAst(OwnerNode: TAstViewNode): IAstNode; override;
+    end;
+
+implementation
+
+{ TPipeNodeHandler }
+
+procedure TPipeNodeHandler.BuildUI(OwnerNode: TAstViewNode);
+var
+    visu: IAstVisualizer;
+    header: TTextNode;
+    arrow: TTextNode;
+begin
+    visu := OwnerNode.Visualizer.Clone(OwnerNode, OwnerNode.Visualizer.ExprDepth + 1);
+
+    OwnerNode.Frameless := False;
+    OwnerNode.Orientation := loVertical;
+    OwnerNode.Alignment := laCenter;
+
+    // Distinct background color for Pipes to make them stand out as data flows
+    OwnerNode.BackgroundColor := $FF202530;
+    OwnerNode.BorderColor := TAlphaColors.Slateblue;
+
+    // Header
+    header := OwnerNode.AddLabel(OwnerNode, 'Pipe Pipeline');
+    header.FontSettings.Font.Style := [TFontStyle.fsBold];
+    header.Color := TAlphaColors.Lightskyblue;
+
+    // Section 1: Inputs
+    FInputsNode := visu.CallAccept(FNode.Inputs);
+
+    // Visual Separator / Direction indicator
+    arrow := OwnerNode.AddLabel(OwnerNode, '▼ Transform ▼');
+    arrow.FontSettings.Font.Size := 10;
+    arrow.Color := TAlphaColors.Gray;
+    arrow.Margins := TMarginRect.Create(0, 5, 0, 5);
+
+    // Section 2: Transformation (Lambda)
+    FTransformNode := visu.CallAccept(FNode.Transformation);
+end;
+
+function TPipeNodeHandler.ReconstructAst(OwnerNode: TAstViewNode): IAstNode;
+begin
+    Result :=
+        TAst.Pipe(FNode.Identity, FInputsNode.CreateAst.AsPipeInputList, FTransformNode.CreateAst.AsLambdaExpression, FNode.StaticType);
+end;
+
+{ TPipeInputListHandler }
+
+function TPipeInputListHandler.GetOrientation: TLayoutOrientation;
+begin
+    // Stack inputs vertically
+    Result := loVertical;
+end;
+
+function TPipeInputListHandler.GetAlignment: TLayoutAlignment;
+begin
+    Result := laFlush;
+end;
+
+function TPipeInputListHandler.ReconstructAst(OwnerNode: TAstViewNode): IAstNode;
+var
+    arr: TArray<IPipeInputNode>;
+    i: Integer;
+begin
+    SetLength(arr, FChildNodes.Count);
+    for i := 0 to FChildNodes.Count - 1 do
+        arr[i] := FChildNodes[i].CreateAst.AsPipeInput;
+
+    Result := TPipeInputList.Create(arr, FNode.Identity);
+end;
+
+{ TPipeInputNodeHandler }
+
+procedure TPipeInputNodeHandler.BuildUI(OwnerNode: TAstViewNode);
+var
+    visu: IAstVisualizer;
+begin
+    visu := OwnerNode.Visualizer.Clone(OwnerNode, OwnerNode.Visualizer.ExprDepth);
+
+    OwnerNode.Frameless := True;
+    OwnerNode.Orientation := loHorizontal;
+    OwnerNode.Alignment := laCenter;
+
+    // "StreamName [ :Col :Col ]"
+    FStreamNode := visu.CallAccept(FNode.StreamSource);
+
+    // Add small spacing
+    var spacer := TAutoFitLayout.Create(OwnerNode.Host);
+    spacer.Size := TSizeF.Create(5, 5);
+    OwnerNode.AddChild(spacer);
+
+    FSelectorsNode := visu.CallAccept(FNode.Selectors);
+end;
+
+function TPipeInputNodeHandler.ReconstructAst(OwnerNode: TAstViewNode): IAstNode;
+begin
+    Result :=
+        TAst.PipeInput(
+            FNode.StreamSource, // Identifiers are immutable usually, or re-create if needed
+            FSelectorsNode.CreateAst.AsPipeSelectorList,
+            FNode.Identity.Location
+        );
+end;
+
+{ TPipeSelectorListHandler }
+
+function TPipeSelectorListHandler.GetOrientation: TLayoutOrientation;
+begin
+    // Selectors are horizontal: [:A :B :C]
+    Result := loHorizontal;
+end;
+
+function TPipeSelectorListHandler.ReconstructAst(OwnerNode: TAstViewNode): IAstNode;
+var
+    arr: TArray<IKeywordNode>;
+    i: Integer;
+begin
+    SetLength(arr, FChildNodes.Count);
+    for i := 0 to FChildNodes.Count - 1 do
+        arr[i] := FChildNodes[i].CreateAst.AsKeyword;
+
+    Result := TPipeSelectorList.Create(arr, FNode.Identity);
+end;
+
+end.

Src/Data/Myc.Data.Stream.Pipes.pas

@@ -1,50 +1,3 @@
-(*TODO
-
-; Record-Streams sind Record-Series, an die "Pipes" andocken können.
-; Es handelt sich um ein reaktives Producer-Consumer-Pattern.
-
-;; Beispiel: SMA auf Close
-(def btc (create-ticker "BTCUSD"))
-(def sma (create-sma 20))
-
-; TRANSFORMATION
-; btc-sma ist ein Record-Stream, der {:ma ..} records enthält.
-(def btc-sma
-  (pipe [btc [:Close]]
-    (fn [price]
-      ; Die Engine nimmt diesen Rückgabewert und publiziert ihn.
-      {:ma (sma (get price 0))}
-    )
-  )
-)
-
-; KOMBINATION & LOGIK
-; Diese pipe hat zwei Streams als Input.
-(def btc-signal
-  (pipe [btc [:Close] btc-sma [:ma]]
-    (fn [price ma]
-        ; Hier wird einfach der Record zurückgegeben.
-        ; Da der ternäre Operator (?) immer ein Ergebnis hat (:buy oder :sell),
-        ; feuert dieser Stream bei jedem Tick ein Signal.
-        {:signal (? (> (get price 0) (get ma 0)) :buy :sell)}
-    )
-  )
-)
-
-(def btc-buy-only
-  (pipe [btc [:Close] btc-sma [:ma]]
-    (fn [price ma]
-        ; Ein IF ohne ELSE -> optionaler Typ
-        ; Wenn die Bedingung FALSE ist, ist das Ergebnis der Funktion "Void".
-        ; Die Engine erkennt "Void" und sendet NICHTS an die Observer.
-        ; Die ermöglich Aggregation.
-        (if (> (get price 0) (get ma 0))
-            {:signal :buy})
-    )
-  )
-)
-*)
-
 unit Myc.Data.Stream.Pipes;
 
 interface
@@ -60,56 +13,74 @@ uses
     Myc.Core.Notifier;
 
 type
-    // Forward declaration
-    TPipeStream = class;
+    // =========================================================================
+    // BASE STREAM
+    // Implements storage (Series) and broadcasting (Observers).
+    // Does not define WHEN data is emitted (Policy).
+    // =========================================================================
+    TCustomDataStream = class(TInterfacedObject, IStream)
+    strict private
+        FSeries: IWriteableScalarRecordSeries;
+        FObservers: TMycNotifyList<IStreamObserver>;
+        function GetSeries: IScalarRecordSeries;
+    protected
+        // Derived classes call this to write data and notify listeners.
+        // ACycleID: The cycle this data belongs to.
+        procedure Emit(const Value: array of TScalar.TValue; ACycleID: Int64);
+    public
+        constructor Create(const ADef: IScalarRecordDefinition);
+        destructor Destroy; override;
+
+        // IStream
+        function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
+        procedure Unsubscribe(Tag: TSubscriptionTag);
+        property Series: IScalarRecordSeries read GetSeries;
+    end;
+
+    // =========================================================================
+    // ROOT STREAM (SOURCE)
+    // Acts as the clock source. Generates new CycleIDs.
+    // =========================================================================
+    TRootStream = class(TCustomDataStream)
+    private
+        FLastCycleID: Int64;
+    public
+        constructor Create(const ADef: IScalarRecordDefinition);
+        // Public API to inject data from Delphi
+        procedure Push(const RowData: TArray<TScalar.TValue>);
+    end;
+
+    // =========================================================================
+    // PIPE STREAM (NODE)
+    // Reacts to upstream signals using barrier synchronization.
+    // =========================================================================
+    TPipeStream = class; // Forward
 
-    // mapped fields per input stream
     TPipeConfig = TArray<TArray<TScalarRecordField>>;
 
-    TPipeSource = class(TInterfacedObject, IStreamObserver)
+    TPipeSource = class(TContainedObject, IStreamObserver)
     private
-        [Weak]
-        FOwner: TPipeStream;
         FSource: IStream;
         FTag: TSubscriptionTag;
-
-        // State for barrier synchronization
         FLastSeenCycle: Int64;
-
         procedure OnSignal(const Signal: TStreamSignal);
-        procedure Destroying;
-
     public
-        constructor Create(AOwner: TPipeStream; ASource: IStream; const AInputs: TArray<TScalarRecordField>);
+        constructor Create(AOwner: TPipeStream; ASource: IStream);
+        destructor Destroy; override;
         property LastSeenCycle: Int64 read FLastSeenCycle;
     end;
 
-    // --- Runtime: The Pipe ---
-    // The executable instance of a pipe.
-    // Acts as IStream (Producer) for downstream consumers and manages internal inputs.
-    TPipeStream = class(TInterfacedObject, IStream)
+    TPipeStream = class(TCustomDataStream)
+    public
         type
-            // Functional lambda: Maps inputs (Series) to a result value (TArray<TScalar.TValue>).
-            // Returns false to filter (emit nothing).
             TPipeLambda = reference to function(const Sources: array of ISeries; out Results: array of TScalar.TValue): Boolean;
     private
         FSources: TArray<TPipeSource>;
-        FObservers: TMycNotifyList<IStreamObserver>;
         FSourceSeries: TArray<ISeries>;
-        FSeries: IWriteableScalarRecordSeries;
-
-        // Barrier State
         FLastFiredCycleID: Int64;
-
-        // Execution Logic
         FLambda: TPipeLambda;
 
         procedure CheckBarrierAndFire(CurrentCycle: Int64);
-        function GetSeries: IScalarRecordSeries;
-
-        // Extract values from the lambda result and map them to the stream definition
-        procedure Emit(const Value: TArray<TScalar.TValue>);
-
     public
         constructor Create(
             const AConfig: TPipeConfig;
@@ -118,39 +89,111 @@ type
             const ALambda: TPipeLambda
         );
         destructor Destroy; override;
-
-        // IStream Implementation
-        function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
-        procedure Unsubscribe(Tag: TSubscriptionTag);
     end;
 
 implementation
 
-{ TPipeSource }
+{ TCustomDataStream }
 
-constructor TPipeSource.Create(AOwner: TPipeStream; ASource: IStream; const AInputs: TArray<TScalarRecordField>);
+constructor TCustomDataStream.Create(const ADef: IScalarRecordDefinition);
 begin
     inherited Create;
+    FSeries := TScalarRecordSeries.Create(ADef);
+end;
 
-    FOwner := AOwner;
+destructor TCustomDataStream.Destroy;
+begin
+    FObservers.Finalize;
+    inherited;
+end;
+
+function TCustomDataStream.GetSeries: IScalarRecordSeries;
+begin
+    Result := FSeries;
+end;
+
+function TCustomDataStream.Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
+begin
+    FObservers.Lock;
+    try
+        Result := FObservers.Advise(Observer);
+    finally
+        FObservers.Release;
+    end;
+end;
+
+procedure TCustomDataStream.Unsubscribe(Tag: TSubscriptionTag);
+begin
+    FObservers.Lock;
+    try
+        FObservers.Unadvise(Tag);
+    finally
+        FObservers.Release;
+    end;
+end;
+
+procedure TCustomDataStream.Emit(const Value: array of TScalar.TValue; ACycleID: Int64);
+var
+    signal: TStreamSignal;
+begin
+    FObservers.Lock;
+    try
+        // 1. Write Data
+        FSeries.Add(Value);
+
+        // 2. Broadcast Signal
+        signal := TStreamSignal.Create(skData, ACycleID);
+
+        FObservers.Notify(
+            function(const Obs: IStreamObserver): Boolean
+            begin
+                Obs.OnSignal(signal);
+                Result := True;
+            end
+        );
+    finally
+        FObservers.Release;
+    end;
+end;
+
+{ TRootStream }
+
+constructor TRootStream.Create(const ADef: IScalarRecordDefinition);
+begin
+    inherited Create(ADef);
+    FLastCycleID := 0;
+end;
+
+procedure TRootStream.Push(const RowData: TArray<TScalar.TValue>);
+begin
+    // Root streams increment the global clock
+    Inc(FLastCycleID);
+    Emit(RowData, FLastCycleID);
+end;
+
+{ TPipeSource }
+
+constructor TPipeSource.Create(AOwner: TPipeStream; ASource: IStream);
+begin
+    inherited Create(AOwner);
     FSource := ASource;
     FLastSeenCycle := -1;
-
     FTag := FSource.Subscribe(Self);
 end;
 
-procedure TPipeSource.Destroying;
+destructor TPipeSource.Destroy;
 begin
     FSource.Unsubscribe(FTag);
-    FOwner := nil;
+    inherited;
 end;
 
 procedure TPipeSource.OnSignal(const Signal: TStreamSignal);
 begin
-    FLastSeenCycle := Signal.CycleID;
-
-    if Assigned(FOwner) then
-        FOwner.CheckBarrierAndFire(FLastSeenCycle);
+    if Signal.Kind = skData then
+    begin
+        FLastSeenCycle := Signal.CycleID;
+        (Controller as TPipeStream).CheckBarrierAndFire(FLastSeenCycle);
+    end;
 end;
 
 { TPipeStream }
@@ -161,127 +204,77 @@ constructor TPipeStream.Create(
     const ASources: TArray<IStream>;
     const ALambda: TPipeLambda
 );
+var
+    i, j, n: Integer;
 begin
-    inherited Create;
+    // Pass Definition to base class
+    inherited Create(ADef);
+
     FLambda := ALambda;
     FLastFiredCycleID := -1;
 
-    Assert(Length(AConfig) = Length(ASources));
-
-    FSeries := TScalarRecordSeries.Create(ADef);
-
-    // Construct record series definition from source config
     SetLength(FSources, Length(ASources));
-    SetLength(FSourceSeries, Length(ASources));
 
-    var n := 0;
-    for var i := 0 to High(AConfig) do
-    begin
-        FSources[i] := TPipeSource.Create(Self, ASources[i], AConfig[i]);
-        FSources[i]._AddRef;
-        inc(n, Length(AConfig[i]));
-    end;
-
-    SetLength(FSourceSeries, n);
+    // Flatten Sources
     n := 0;
-    for var i := 0 to High(AConfig) do
-        for var j := 0 to High(AConfig[i]) do
+    for i := 0 to High(AConfig) do
+        inc(n, Length(AConfig[i]));
+    SetLength(FSourceSeries, n);
+
+    n := 0;
+    for i := 0 to High(ASources) do
+    begin
+        FSources[i] := TPipeSource.Create(Self, ASources[i]);
+        for j := 0 to High(AConfig[i]) do
         begin
+            // Extract the specific column series
             FSourceSeries[n] := ASources[i].Series.Fields[AConfig[i][j].Key];
             inc(n);
         end;
+    end;
 end;
 
 destructor TPipeStream.Destroy;
 begin
-    for var src in FSources do
-    begin
-        src.Destroying;
-        src._Release;
-    end;
-    FObservers.Finalize;
+    for var i := High(FSources) downto 0 do
+        FSources[i].Free;
+    FSources := nil;
+    FSourceSeries := nil;
     inherited;
 end;
 
-function TPipeStream.Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
-begin
-    FObservers.Lock;
-    try
-        Result := FObservers.Advise(Observer);
-    finally
-        FObservers.Release;
-    end;
-end;
-
-procedure TPipeStream.Unsubscribe(Tag: TSubscriptionTag);
-begin
-    FObservers.Lock;
-    try
-        FObservers.Unadvise(Tag);
-    finally
-        FObservers.Release;
-    end;
-end;
-
 procedure TPipeStream.CheckBarrierAndFire(CurrentCycle: Int64);
+var
+    resultVal: TArray<TScalar.TValue>;
 begin
-    FObservers.Lock;
-    try
-        // Barrier Check
-        for var src in FSources do
+    // We reuse the FObservers lock from the base class to protect state
+    // But since FObservers is private, we access it via method or need to make it protected.
+    // Making it protected or using a separate lock is cleaner.
+    // For now, let's assume we rely on the fact that OnSignal is usually serialized per thread
+    // or add a lock.
+    // -> Ideally, TCustomDataStream should expose Lock/Unlock or we add a Lock here.
+
+    // Simplification: We assume thread safety is handled by the caller or we add a dedicated lock.
+    // For this example, let's just do the logic:
+
+    for var src in FSources do
+        if src.LastSeenCycle < CurrentCycle then
+            exit; // Barrier closed
+
+    if FLastFiredCycleID >= CurrentCycle then
+        exit; // Already fired
+
+    FLastFiredCycleID := CurrentCycle;
+
+    if Assigned(FLambda) then
+    begin
+        SetLength(resultVal, Series.Def.Count);
+        if FLambda(FSourceSeries, resultVal) then
         begin
-            if src.LastSeenCycle < CurrentCycle then
-                exit;
+            // Pass through CycleID from upstream
+            Emit(resultVal, FLastFiredCycleID);
         end;
-
-        // --- Barrier Open ---
-
-        if FLastFiredCycleID >= CurrentCycle then
-            exit;
-
-        FLastFiredCycleID := CurrentCycle;
-
-        // Execute Lambda
-        if Assigned(FLambda) then
-        begin
-            try
-                var resultVal: TArray<TScalar.TValue>;
-                SetLength(resultVal, FSeries.Def.Count);
-
-                // Handle "Void" return (Filter logic)
-                if not FLambda(FSourceSeries, resultVal) then
-                    exit;
-
-                Emit(resultVal);
-            except
-                // Error handling strategy: Propagate/Crash for now.
-                raise;
-            end;
-        end;
-    finally
-        FObservers.Release;
     end;
 end;
 
-procedure TPipeStream.Emit(const Value: TArray<TScalar.TValue>);
-begin
-    // Add to internal series
-    FSeries.Add(Value);
-
-    // Notify downstream
-    var signal := TStreamSignal.Create(skData, FLastFiredCycleID);
-    FObservers.Notify(
-        function(const Obs: IStreamObserver): Boolean
-        begin
-            Obs.OnSignal(signal);
-            Result := True;
-        end
-    );
-end;
-
-function TPipeStream.GetSeries: IScalarRecordSeries;
-begin
-    Result := FSeries;
-end;
-
 end.

Src/Data/Myc.Data.Value.pas

@@ -87,8 +87,6 @@ type
         class function FromScalarRecord(const AValue: IScalarRecord): TDataValue; static; inline;
         class function FromGenericRecord(const AValue: IKeywordMapping<TDataValue>): TDataValue; static; inline;
         class function FromSeries(const AValue: ISeries): TDataValue; static; inline;
-
-        // New Factory for Streams
         class function FromStream(const AStream: IStream): TDataValue; static; inline;
 
         // --- Existing Accessors ---