Data Transformers

import { useMessageDataSource } from '@electricui/core-timeseries'
import { useDataTransformer } from '@electricui/timeseries-react'
import { map, filter } from '@electricui/dataflow'
 
const ChartPage = () => {
  const sensorDS = useMessageDataSource('tempK')
 
  const transformedDS = useDataTransformer(() => {
    const celsiusEvents = map(sensorDS, (data, time) => {
      return data + 273.15
    })
  
    const validEvents = filter(celsiusEvents, (data, time) => {
      if (data >= -40 && data <= 500 ) {
        return true
      }
      return false
    })
 
    return validEvents
  })
 
  return (
    <ChartContainer>
      <LineChart dataSource={transformedDS} />
      <RealTimeDomain window={10000} />
    </ChartContainer>
  )
}

import { ChartContainer, LineChart } from '@electricui/components-desktop-charts'
import { useMessageDataSource, Query, Queryable } from '@electricui/core-timeseries'
import { useDataTransformer } from '@electricui/timeseries-react'
import { forEach, map, interleave } from '@electricui/dataflow'
 
type XYZEvent = {
  x: number
  y: number
  z: number
}
type ColorEvent = {
  color: string
}
type MixedEvent = XYZEvent & ColorEvent
 
function colorMixer(
  colorQueryable: Queryable<ColorEvent>,
  xyzQueryable: Queryable<XYZEvent>,
) {
  let currentColorState = 'blue'
 
  const colorSetter = forEach(colorQueryable, (data, time) => {
    currentColorState = data.color
  })
 
  const colorXYZ = map(xyzQueryable, (data, time) => ({
    x: data.x,
    y: data.y,
    z: data.z,
    color: currentColorState,
  }))
 
  return interleave([colorXYZ, colorSetter])
}
 
export const ColorMixingChart = () => {
  const stateColDataSource = useMessageDataSource<ColorEvent>('state_col')
  const posDataSource = useMessageDataSource<XYZEvent>('pos')
 
  const dataTransformer = useDataTransformer(() => {
    return colorMixer(stateColDataSource, posDataSource)
  })
 
  return (
    <ChartContainer>
      <LineChart
        dataSource={dataTransformer}
        accessor={data => data.z}
        lineWidth={4}
        colorAccessor={(data, time) => data.color}
      />
    </ChartContainer>
  )
}

   ┌─────────┐
   │         │   ┌─────┐                        ┌───┐
   │  Color  ├───►Green├────────────────────────►Red│
   │         │   └──┬──┘                        └─┬─┘
   └─────────┘      │                             │
                    │                             │
                    │                             │
                 ┌──▼──┐     ┌─────┐     ┌─────┐  │  ┌─────┐
                 │Green│     │Green│     │Green│  │  │ Red │
                 │  1  ├─────►  2  ├─────►  3  ├──┴──►  4  │
                 └──▲──┘     └──▲──┘     └──▲──┘     └──▲──┘
                    │           │           │           │
                    │           │           │           │
   ┌─────────┐      │           │           │           │
   │         │     ┌┴┐         ┌┴┐         ┌┴┐         ┌┴┐
   │   XYZ   ├─────►1├─────────►2├─────────►3├─────────►4│
   │         │     └─┘         └─┘         └─┘         └─┘
   └─────────┘

// @filename: rollingAverage.ts
import {
  prepare,
  map,
  interleave,
} from '@electricui/dataflow'
import {
  FixedQueue,
  Queryable,
  Query,
  ORDERING,
  PERSISTENCE,
  Event
} from '@electricui/core-timeseries'
 
export function rollingAverage(
  dataSource: Queryable<number>,
  bufferSize: number,
) {
  // Allocate our circular buffer
  const buffer = new FixedQueue<number>(bufferSize)
 
  // Prefetch up to `bufferSize` before the start of the query
  const prefetch = prepare(dataSource, query => {
    return query.clone()
      .limit(bufferSize) // Fetch up to `bufferSize` events
      .end(query.getStart()) // before the start of the current query
      .order(ORDERING.DESC) // ordered latest to earliest, so the _last_ events are fetched
      .persist(PERSISTENCE.IMMEDIATE) // fetching only events that are immediately available
  })
 
  // The resultant batch will be ordered back-to-front, but the interleave operator
  // will automatically re-order the incoming events intelligently.
  // The prefetched events and the 'current' events will be combined into the same stream.
  const allEvents = interleave([prefetch, dataSource])
 
  // The actual averaging occurs here:
  const averageEvents = map(allEvents, (data, time) => {
    // Add the event to the queue
    buffer.push(data)
 
    // Calculate the average over the queue
    let total = 0
    buffer.forEach(val => {
      total += val
    })
    const average = total / buffer.length
 
    // Return the average event
    return average
  })
 
  // Return the averaged event stream
  return averageEvents
}

const OverviewPage = () => {
  const sensorDataSource = useMessageDataSource('adc')
 
  const averagedSensorDataSource = useDataTransformer(() => {
    // Average the last 10 events
    return rollingAverage(sensorDataSource, 10)
  })
 
  return (
    <ChartContainer>
      <LineChart dataSource={averagedSensorDataSource} />
      <RealTimeDomain window={10000} />
    </ChartContainer>
  )
}