add Arrow.jl conversion methods so that Samples/SamplesInfo can be (de)serialized as Arrow values (#68)

Author: jrevels

Project.toml

@@ -1,7 +1,7 @@
 name = "Onda"
 uuid = "e853f5be-6863-11e9-128d-476edb89bfb5"
 authors = ["Beacon Biosignals, Inc."]
-version = "0.13.1"
+version = "0.13.2"
 
 [deps]
 Arrow = "69666777-d1a9-59fb-9406-91d4454c9d45"
@@ -17,9 +17,10 @@ TranscodingStreams = "3bb67fe8-82b1-5028-8e26-92a6c54297fa"
 UUIDs = "cf7118a7-6976-5b1a-9a39-7adc72f591a4"
 
 [compat]
-Arrow = "~1.2.4"
+Arrow = "1.3"
 CodecZstd = "0.6, 0.7"
 ConstructionBase = "1.0"
+DataFrames = "0.22.7"
 MsgPack = "1.1"
 Tables = "1.2"
 TimeSpans = "0.2.2"

examples/tour.jl

@@ -8,7 +8,7 @@
 # before and/or alongside the completion of this tour; it explains the
 # purpose/structure of the format.
 
-using Onda, TimeSpans, DataFrames, Dates, UUIDs, Test, ConstructionBase
+using Onda, TimeSpans, DataFrames, Dates, UUIDs, Test, ConstructionBase, Arrow
 
 #####
 ##### generate some mock data
@@ -204,6 +204,17 @@ rows = ["c3", 4, "f3"]
 f_channels = ["fp1", "f3","f7", "fz", "fp2", "f4", "f8"]
 @test eeg[r"f", span].data == view(eeg, channel.(Ref(eeg), f_channels), span_range).data
 
+# Onda overloads the necessary Arrow.jl machinery to enable individual sample data
+# segments (specifically, `Samples` and `SamplesInfo` values) to be (de)serialized
+# to/from Arrow for storage or IPC purposes; see below for an example. Note that if
+# you wanted to use Arrow as a storage format for whole sample data files w/ Onda,
+# it'd make more sense to create an `AbstractLPCMFormat` subtype for your Arrow <-> LPCM
+# mapping (an example of this can be seen in `examples/flac.jl`).
+x = (a=[eeg], b=[eeg.info])
+y = Arrow.Table(Arrow.tobuffer(x))
+@test x.a == y.a
+@test x.b == y.b
+
 # Note that `Samples` is not an `AbstractArray` subtype; the special indexing
 # behavior above is only defined for convenient data manipulation. It is fine
 # to access the sample data matrix directly via the `data` field if you need

src/Onda.jl

@@ -5,11 +5,6 @@ using TimeSpans, ConstructionBase
 using Arrow, Tables
 using MsgPack, TranscodingStreams, CodecZstd
 
-function __init__()
-    Arrow.ArrowTypes.registertype!(TimeSpan, TimeSpan)
-    return nothing
-end
-
 #####
 ##### includes/exports
 #####

src/samples.jl

@@ -495,3 +495,31 @@ function Base.show(io::IO, samples::Samples)
         show(io, "text/plain", samples.data)
     end
 end
+
+#####
+##### Arrow conversion
+#####
+
+const SamplesArrowType{T,S} = NamedTuple{(:data, :info, :encoded),Tuple{Vector{T},S,Bool}} where {S<:SamplesInfoArrowType}
+
+const SAMPLES_ARROW_NAME = Symbol("JuliaLang.Samples")
+
+Arrow.ArrowTypes.arrowname(::Type{<:Samples}) = SAMPLES_ARROW_NAME
+
+Arrow.ArrowTypes.ArrowType(::Type{<:Samples{D,S}}) where {D,S} = SamplesArrowType{eltype(D),Arrow.ArrowTypes.ArrowType(S)}
+
+function Arrow.ArrowTypes.toarrow(samples::Samples)
+    return (data=vec(samples.data),
+            info=Arrow.ArrowTypes.toarrow(samples.info),
+            encoded=samples.encoded)
+end
+
+function Arrow.ArrowTypes.JuliaType(::Val{SAMPLES_ARROW_NAME}, ::Type{SamplesArrowType{T,S}}) where {T,S}
+    return Samples{Matrix{T},Arrow.ArrowTypes.JuliaType(Val(SAMPLES_INFO_ARROW_NAME), S)}
+end
+
+function Arrow.ArrowTypes.fromarrow(::Type{<:Samples}, arrow_data, arrow_info, arrow_encoded)
+    info = SamplesInfo(arrow_info; validate=false)
+    data = reshape(arrow_data, (channel_count(info), :))
+    return Samples(data, info, arrow_encoded)
+end

src/signals.jl

@@ -277,6 +277,35 @@ function Signal(info::SamplesInfo; recording, file_path, file_format, span, cust
                   info.sample_offset_in_unit, info.sample_type, info.sample_rate, custom...)
 end
 
+#####
+##### Arrow conversion
+#####
+
+const SamplesInfoArrowType{R,O,SR} = NamedTuple{(:kind, :channels, :sample_unit, :sample_resolution_in_unit, :sample_offset_in_unit, :sample_type, :sample_rate),
+                                                Tuple{String,Vector{String},String,R,O,String,SR}}
+
+const SAMPLES_INFO_ARROW_NAME = Symbol("JuliaLang.SamplesInfo")
+
+Arrow.ArrowTypes.arrowname(::Type{<:SamplesInfo}) = SAMPLES_INFO_ARROW_NAME
+
+Arrow.ArrowTypes.ArrowType(::Type{<:SamplesInfo{<:Any,<:Any,<:Any,R,O,<:Any,SR}}) where {R,O,SR} = SamplesInfoArrowType{R,O,SR}
+
+function Arrow.ArrowTypes.toarrow(info::SamplesInfo)
+    return (kind=convert(String, info.kind),
+            channels=convert(Vector{String}, info.channels),
+            sample_unit=convert(String, info.sample_unit),
+            sample_resolution_in_unit=info.sample_resolution_in_unit,
+            sample_offset_in_unit=info.sample_offset_in_unit,
+            sample_type=onda_sample_type_from_julia_type(info.sample_type),
+            sample_rate=info.sample_rate)
+end
+
+function Arrow.ArrowTypes.JuliaType(::Val{SAMPLES_INFO_ARROW_NAME}, ::Type{SamplesInfoArrowType{R,O,SR}}) where {R,O,SR}
+    return SamplesInfo{String,Vector{String},String,R,O,<:LPCM_SAMPLE_TYPE_UNION,SR}
+end
+
+Arrow.ArrowTypes.fromarrow(::Type{<:SamplesInfo}, fields...) = SamplesInfo(fields...; validate=false)
+
 #####
 ##### duck-typed utilities
 #####

src/utilities.jl

@@ -54,6 +54,11 @@ TimeSpans.istimespan(::NamedTupleTimeSpan) = true
 TimeSpans.start(x::NamedTupleTimeSpan) = x.start
 TimeSpans.stop(x::NamedTupleTimeSpan) = x.stop
 
+const TIME_SPAN_ARROW_NAME = Symbol("JuliaLang.TimeSpan")
+
+Arrow.ArrowTypes.arrowname(::Type{TimeSpan}) = TIME_SPAN_ARROW_NAME
+ArrowTypes.JuliaType(::Val{TIME_SPAN_ARROW_NAME}) = TimeSpan
+
 #####
 ##### zstd_compress/zstd_decompress
 #####