Merge pull request #144 from SainsburyWellcomeCentre/aeon-db

Add database package core operators

src/Aeon.Database/Aeon.Database.csproj

@@ -0,0 +1,24 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <Title>Project Aeon - Database</Title>
+    <Description>Provides querying and schema functionality for Project Aeon databases.</Description>
+    <PackageTags>Bonsai Rx Project Aeon Database</PackageTags>
+    <TargetFramework>net472</TargetFramework>
+    <VersionPrefix>0.1.0</VersionPrefix>
+    <VersionSuffix>build230831</VersionSuffix>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <EmbeddedResource Include="**\*.bonsai" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="MySqlConnector" Version="2.2.7" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Aeon.Acquisition\Aeon.Acquisition.csproj" />
+  </ItemGroup>
+
+</Project>

src/Aeon.Database/CreateConnection.cs

@@ -0,0 +1,27 @@
+using System;
+using System.ComponentModel;
+using System.Reactive.Linq;
+using Bonsai;
+using MySqlConnector;
+
+namespace Aeon.Database
+{
+    [DefaultProperty(nameof(ConnectionString))]
+    [Description("Creates a connection to the MySQL server using the specified connection string.")]
+    public class CreateConnection : Source<MySqlConnection>
+    {
+        [Editor("Bonsai.Design.RichTextEditor, Bonsai.Design", DesignTypes.UITypeEditor)]
+        [Description("Specifies the parameters used to establish a connection to the MySQL server.")]
+        public string ConnectionString { get; set; }
+
+        public override IObservable<MySqlConnection> Generate()
+        {
+            return Observable.Defer(async () =>
+            {
+                var connection = new MySqlConnection(ConnectionString);
+                await connection.OpenAsync();
+                return Observable.Return(connection);
+            });
+        }
+    }
+}

src/Aeon.Database/EnumerateColony.cs

@@ -0,0 +1,45 @@
+using System;
+using System.ComponentModel;
+using System.Linq;
+using System.Reactive.Linq;
+using Bonsai;
+using MySqlConnector;
+
+namespace Aeon.Database
+{
+    [Description("Enumerates all records in the colony table for each MySQL connection in the sequence.")]
+    public class EnumerateColony : Combinator<MySqlConnection, ColonyRecord>
+    {
+        public override IObservable<ColonyRecord> Process(IObservable<MySqlConnection> source)
+        {
+            return source.SelectMany(connection => ObservableDatabase.Query<ColonyRecord>(
+                "SELECT * FROM `#colony`;",
+                connection));
+        }
+    }
+
+    public class ColonyRecord
+    {
+        public string Subject { get; set; }
+
+        public float? ReferenceWeight { get; set; }
+
+        public SubjectSex Sex { get; set; }
+
+        public DateTime? SubjectBirthDate { get; set; }
+
+        public string Note { get; set; }
+
+        public override string ToString()
+        {
+            return $"({Subject}, wt: {ReferenceWeight}, sex: {Sex}, dob: {SubjectBirthDate}, {Note})";
+        }
+    }
+
+    public enum SubjectSex
+    {
+        Male = 'M',
+        Female = 'F',
+        Unspecified = 'U'
+    }
+}

src/Aeon.Database/ExecuteQuery.cs

@@ -0,0 +1,23 @@
+using System;
+using System.ComponentModel;
+using System.Linq;
+using System.Reactive.Linq;
+using Bonsai;
+using MySqlConnector;
+
+namespace Aeon.Database
+{
+    [DefaultProperty(nameof(QueryString))]
+    [Description("Runs the specified SQL query against each MySQL connection in the sequence.")]
+    public class ExecuteQuery : Combinator<MySqlConnection, MySqlDataReader>
+    {
+        [Editor("Bonsai.Design.RichTextEditor, Bonsai.Design", DesignTypes.UITypeEditor)]
+        [Description("Specifies the full SQL query to run against the MySQL connection.")]
+        public string QueryString { get; set; }
+
+        public override IObservable<MySqlDataReader> Process(IObservable<MySqlConnection> source)
+        {
+            return source.SelectMany(connection => ObservableDatabase.Query(QueryString, connection));
+        }
+    }
+}

src/Aeon.Database/ObservableDatabase.cs

@@ -0,0 +1,126 @@
+using System;
+using System.Reactive.Linq;
+using MySqlConnector;
+
+namespace Aeon.Database
+{
+    static class ObservableDatabase
+    {
+        public static IObservable<MySqlDataReader> Query(string queryString, MySqlConnection connection)
+        {
+            return Query(queryString, connection, reader => reader);
+        }
+
+        public static IObservable<TResult> Query<TResult>(string queryString, MySqlConnection connection)
+        {
+            return Query(
+                queryString,
+                connection,
+                RecordReader<TResult>.Instance.Validate,
+                RecordReader<TResult>.Instance.Select);
+        }
+
+        public static IObservable<TResult> Query<TResult>(
+            string queryString,
+            MySqlConnection connection,
+            Func<MySqlDataReader, TResult> selector)
+        {
+            return Query(queryString, connection, reader => { }, selector);
+        }
+
+        public static IObservable<TResult> Query<TResult>(
+            string queryString,
+            MySqlConnection connection,
+            Action<MySqlDataReader> validator,
+            Func<MySqlDataReader, TResult> selector)
+        {
+            return Observable.Create<TResult>(async (observer, cancellationToken) =>
+            {
+                using var command = new MySqlCommand(queryString, connection);
+                using var reader = await command.ExecuteReaderAsync(cancellationToken);
+
+                validator(reader);
+                while (await reader.ReadAsync(cancellationToken))
+                {
+                    var result = selector(reader);
+                    observer.OnNext(result);
+                }
+            });
+        }
+
+        public static bool? BooleanField(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetBoolean(ordinal);
+        }
+
+        public static char? CharField(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetChar(ordinal);
+        }
+
+        public static sbyte? SByteField(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetSByte(ordinal);
+        }
+
+        public static byte? ByteField(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetByte(ordinal);
+        }
+
+        public static short? Int16Field(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetInt16(ordinal);
+        }
+
+        public static ushort? UInt16Field(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetUInt16(ordinal);
+        }
+
+        public static int? Int32Field(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetInt32(ordinal);
+        }
+
+        public static uint? UInt32Field(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetUInt32(ordinal);
+        }
+
+        public static long? Int64Field(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetInt64(ordinal);
+        }
+
+        public static ulong? UInt64Field(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetUInt64(ordinal);
+        }
+
+        public static float? FloatField(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetFloat(ordinal);
+        }
+
+        public static double? DoubleField(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetDouble(ordinal);
+        }
+
+        public static decimal? DecimalField(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetDecimal(ordinal);
+        }
+
+        public static DateTime? DateTimeField(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetDateTime(ordinal);
+        }
+
+        public static string StringField(this MySqlDataReader reader, int ordinal)
+        {
+            return reader.IsDBNull(ordinal) ? null : reader.GetString(ordinal);
+        }
+    }
+}

src/Aeon.Database/Properties/AssemblyInfo.cs

@@ -0,0 +1,6 @@
+using Bonsai;
+
+// General Information about an assembly is controlled through the following 
+// set of attributes. Change these attribute values to modify the information
+// associated with an assembly.
+[assembly: XmlNamespacePrefix("clr-namespace:Aeon.Database", "aeondb")]

src/Aeon.Database/Properties/launchSettings.json

@@ -0,0 +1,10 @@
+{
+  "profiles": {
+    "Bonsai": {
+      "commandName": "Executable",
+      "executablePath": "$(registry:HKEY_CURRENT_USER\\Software\\Bonsai Foundation\\Bonsai@InstallDir)Bonsai.exe",
+      "commandLineArgs": "--lib:$(TargetDir).",
+      "nativeDebugging": true
+    }
+  }
+}