Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Synchronizer monitor mean timestamps only used when devices report heartbeats #224

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Synchronizer monitor mean timestamps only used when devices report heartbeats #224

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f4ed4e4
Add create group operator
glopesdev Oct 21, 2023
e101212
Ensure all created group observables are hot
glopesdev Oct 16, 2024
184ae27
Merge pull request #165 from SainsburyWellcomeCentre/aeon-acquisition
glopesdev Oct 16, 2024
8cd7b22
Mean timestamps only used if not NaN
aspaNeuro Oct 24, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
108 changes: 108 additions & 0 deletions src/Aeon.Acquisition/CreateGroup.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Linq;
using System.Linq.Expressions;
using System.Reactive.Linq;
using Bonsai;
using Bonsai.Expressions;

namespace Aeon.Acquisition
{
[Description("Creates a sequence of grouped observables for each key in the input sequence.")]
public class CreateGroup : WorkflowExpressionBuilder
{
static readonly Range<int> argumentRange = Range.Create(lowerBound: 1, upperBound: 1);

public CreateGroup()
: this(new ExpressionBuilderGraph())
{
}

public CreateGroup(ExpressionBuilderGraph workflow)
: base(workflow)
{
}

public override Range<int> ArgumentRange => argumentRange;

public override Expression Build(IEnumerable<Expression> arguments)
{
var source = arguments.FirstOrDefault();
if (source == null)
{
throw new InvalidOperationException("There must be at least one input to the create group operator.");
}

Type keyType, elementType;
var sourceType = source.Type.GetGenericArguments()[0];
var selectorParameter = Expression.Parameter(source.Type);
if (sourceType.IsGenericType && sourceType.GetGenericTypeDefinition() == typeof(IGroupedObservable<,>))
{
var sourceTypeArguments = sourceType.GetGenericArguments();
keyType = sourceTypeArguments[0];
elementType = sourceTypeArguments[1];
}
else
{
keyType = sourceType;
elementType = null;
}

return BuildWorkflow(new[] { selectorParameter }, null, selectorBody =>
{
var selector = Expression.Lambda(selectorBody, selectorParameter);
var resultType = selectorBody.Type.GetGenericArguments()[0];

return Expression.Call(
typeof(CreateGroup),
nameof(Process),
elementType != null
? new[] { keyType, elementType, resultType }
: new[] { keyType, resultType },
source, selector);
});
}

static IObservable<IGroupedObservable<TKey, TResult>> Process<TKey, TResult>(
IObservable<TKey> source,
Func<IObservable<TKey>, IObservable<TResult>> selector)
{
return source.Select(key => new GroupedObservable<TKey, TResult>(
key,
selector(Observable.Return(key))));
}

static IObservable<IGroupedObservable<TKey, TResult>> Process<TKey, TSource, TResult>(
IObservable<IGroupedObservable<TKey, TSource>> source,
Func<IObservable<IGroupedObservable<TKey, TSource>>, IObservable<TResult>> selector)
{
return source.Select(group => new GroupedObservable<TKey, TResult>(
group.Key,
selector(Observable.Return(group))));
}

class GroupedObservable<TKey, TElement> : IGroupedObservable<TKey, TElement>
{
public GroupedObservable(TKey key, IObservable<TElement> source)
{
if (source is null)
{
throw new ArgumentNullException(nameof(source));
}

Key = key;
Source = source.Publish().RefCount();
}

public TKey Key { get; }

private IObservable<TElement> Source { get; }

public IDisposable Subscribe(IObserver<TElement> observer)
{
return Source.Subscribe(observer);
}
}
}
}
29 changes: 19 additions & 10 deletions src/Aeon.Acquisition/SynchronizerMonitor.bonsai
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
<?xml version="1.0" encoding="utf-8"?>
<WorkflowBuilder Version="2.8.1"
<WorkflowBuilder Version="2.8.2"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:rx="clr-namespace:Bonsai.Reactive;assembly=Bonsai.Core"
xmlns:harp="clr-namespace:Bonsai.Harp;assembly=Bonsai.Harp"
Expand Down Expand Up @@ -91,13 +91,20 @@
<Expression xsi:type="MemberSelector">
<Selector>Item2</Selector>
</Expression>
<Expression xsi:type="harp:Parse">
<harp:Register xsi:type="harp:ParseMessagePayload">
<harp:PayloadType>Timestamp</harp:PayloadType>
<harp:IsArray>false</harp:IsArray>
</harp:Register>
</Expression>
<Expression xsi:type="Combinator">
<Combinator xsi:type="rx:Zip" />
aspaNeuro marked this conversation as resolved.
Show resolved Hide resolved
</Expression>
<Expression xsi:type="scr:ExpressionTransform">
<scr:Expression>new(
Item1 as Stats,
Item2 as Names)</scr:Expression>
Item2 as Names,
Item3 as SyncTimestamp)</scr:Expression>
</Expression>
<Expression xsi:type="SubscribeSubject">
<Name>HeartbeatSources</Name>
Expand All @@ -111,7 +118,7 @@ Item2 as Names)</scr:Expression>
<Expression xsi:type="scr:ExpressionTransform">
<scr:Expression>new(
Item1.Stats.Mean as MeanTimestamp,
DateTime(1904, 1, 1) + TimeSpan.FromSeconds(Item1.Stats.Mean) as MeanUtcTimestamp,
DateTime(1904, 1, 1) + TimeSpan.FromSeconds(Double.IsNan(Item1.Stats.Mean) ? Item1.SyncTimestamp : Item1.Stats.Mean) as MeanUtcTimestamp,
Item2 as ExpectedDeviceCount,
Item1.Stats.Count as DeviceCount,
Item1.Stats.Maximum - Item1.Stats.Minimum as MaxDifference,
Expand All @@ -128,18 +135,20 @@ string.Join("\t", Item1.Names) as Elements)</scr:Expression>
<Edge From="5" To="8" Label="Source1" />
<Edge From="6" To="7" Label="Source1" />
<Edge From="7" To="8" Label="Source2" />
<Edge From="7" To="13" Label="Source1" />
<Edge From="8" To="9" Label="Source1" />
<Edge From="9" To="10" Label="Source1" />
<Edge From="9" To="12" Label="Source1" />
<Edge From="10" To="11" Label="Source1" />
<Edge From="11" To="13" Label="Source1" />
<Edge From="12" To="13" Label="Source2" />
<Edge From="13" To="14" Label="Source1" />
<Edge From="14" To="17" Label="Source1" />
<Edge From="15" To="16" Label="Source1" />
<Edge From="16" To="17" Label="Source2" />
<Edge From="17" To="18" Label="Source1" />
<Edge From="11" To="14" Label="Source1" />
<Edge From="12" To="14" Label="Source2" />
<Edge From="13" To="14" Label="Source3" />
<Edge From="14" To="15" Label="Source1" />
<Edge From="15" To="18" Label="Source1" />
<Edge From="16" To="17" Label="Source1" />
<Edge From="17" To="18" Label="Source2" />
<Edge From="18" To="19" Label="Source1" />
<Edge From="19" To="20" Label="Source1" />
</Edges>
</Workflow>
</WorkflowBuilder>