Example ported Streaming Report Model in Fsharp

Program.fs

open System
open FSharp.Data
open FSharpx.Collections

// Example file filtered_subread_summary.csv 
// passedFilter = 1 True

// MovieName,HoleNumber,Start,End,Length,PassedFilter
// m120201_042231_42129_c100275262550000001523007907041260_s1_p0,9,0,1756,1756,1
// m120201_042231_42129_c100275262550000001523007907041260_s1_p0,11,1513,4949,3436,1
// m120201_042231_42129_c100275262550000001523007907041260_s1_p0,13,0,68,68,1
// m120201_042231_42129_c100275262550000001523007907041260_s1_p0,13,117,519,402,1


type Subread = {movieName: string; holeNumber: int; readStart: int; readEnd: int; readLength: int; passedFilter: int}


[<AbstractClass>]
type Accumulator() = 
    abstract member ToAttribute : unit -> float
    abstract member apply : Subread -> unit


type MeanAccumulator(total: int, nrecords: int) = 
    inherit Accumulator()
    // ReadLength mean
    let mutable total = total
    let mutable nrecords = nrecords

    override this.apply(subread: Subread) = 
        total <- total + subread.readLength
        nrecords <- nrecords + 1

    override this.ToAttribute() = float(total / nrecords)


type TotalAccumulator(total: int) = 
    inherit Accumulator()
    // Total Number of Records
    let mutable total = total

    override this.apply(subread: Subread) = total <- total + 1
    override this.ToAttribute() = float(total)


type TotalBasesAccumulator(total: int) =
    inherit Accumulator()

    let mutable total = total

    override this.apply(subread: Subread) = total <- total + subread.readLength
    override this.ToAttribute() = float(total)


[<AbstractClass>]
type BaseFilter() = 
    abstract member apply : Subread -> bool

type FilterByReadLength(minReadLength: int) = 
    inherit BaseFilter()
    let minReadLength = minReadLength

    override this.apply(subread: Subread) =
        subread.readLength > minReadLength

type IsPassedFilter() =
    inherit BaseFilter()
    override this.apply(subread: Subread) =
        subread.passedFilter = 1            


type StatModel(accumulators: list<Accumulator>, filters: list<BaseFilter>) =
    member this.accumulators = accumulators
    member this.filters = filters

    member this.apply(subread: Subread) =
        for a in accumulators do
            let passedFilter = this.filters |> List.forall( fun f -> (f.apply(subread) = true))
            if passedFilter then a.apply(subread)


let runAnalysis(subreads: list<Subread>, models: list<StatModel>) = 
    for s in subreads do
        for m in models do
            m.apply(s)


let printResults(models: list<StatModel>) =
    for m in models do
        printfn "Model summary %O" m
        for a in m.accumulators do
            printfn "%f" (a.ToAttribute())

                    
let generateMockSubreads (n: int) = 
    let r = System.Random()
    [ for i in 1 .. 10 -> {movieName = "m_blah_stuff"; holeNumber = i; readStart = 1; readEnd = 10; readLength = r.Next(1, 1000); passedFilter = r.Next(0, 2)}]


[<EntryPoint>]
let main (args : string[]) = 
    
    // mock subreads
    let subreads = generateMockSubreads(5)

    //let subread = {movieName = "m_blah_stuff"; holeNumber = 100; readStart = 1; readEnd = 10; readLength = 9; passedFilter = 1}
    //printfn subread.ToString()

    // ******* PreFilter Accumulators
    let preFilters : BaseFilter list = []
    let preAccumulators : Accumulator list = [new MeanAccumulator(0, 0); new TotalAccumulator(0); new TotalBasesAccumulator(0)]
    let preStatModels = new StatModel(preAccumulators, preFilters)

    // ******* PostFilter Accumulators
    let postFilters : BaseFilter list = [new IsPassedFilter(); new FilterByReadLength(100)]
    let postAccumulators : Accumulator list = [new MeanAccumulator(0, 0); new TotalAccumulator(0); new TotalBasesAccumulator(0)]
    let postStatModels = new StatModel(postAccumulators, postFilters)

    let statModels : StatModel list = [preStatModels; postStatModels]

    printfn "Running Analysis"
    runAnalysis(subreads, statModels)

    printResults(statModels)

    printfn "Completed running program"
    0