diff --git a/apps/Dump.hs b/apps/Dump.hs --- a/apps/Dump.hs +++ b/apps/Dump.hs @@ -1,15 +1,19 @@ -{-# LANGUAGE NamedFieldPuns, TemplateHaskell #-} +{-# LANGUAGE NamedFieldPuns #-} {-# OPTIONS_GHC -fno-warn-orphans #-} module Main where +import Control.Concurrent (threadDelay) import Control.Monad import Data.Aeson (ToJSON (..), encode) -import qualified Data.ByteString.Char8 as BS8 +import qualified Data.ByteString as BS import qualified Data.ByteString.Lazy.Char8 as BSL8 import qualified Data.ByteString.Base64 as Base64 +import Data.Function (fix) +import Data.IORef import Data.List import Data.List.Split (splitOn) +import Data.Time.Clock import Options.Applicative hiding (action) import System.IO import Text.Read @@ -18,6 +22,7 @@ import Hemokit import Hemokit.Start +import Hemokit.Internal.Utils (withJustM) import WebsocketUtils (makeJsonOrShowWSServer, JsonShowable (..)) @@ -25,13 +30,15 @@ data DumpArgs = DumpArgs { emotivArgs :: EmotivArgs , mode :: DumpMode -- ^ What to dump. + , realtime :: Bool -- ^ In case fromFile is used, throttle to 128 Hz. , listDevices :: Bool -- ^ Do not do anything, print available devices. , json :: Bool -- ^ Whether to format the output as JSON. , serve :: Maybe (String, Int) -- ^ Serve via websockets on host:port. } --- | Whether to dump raw data, hardware-sent packages or cumulative states. -data DumpMode = Raw | Packets | State deriving (Eq, Show) +-- | Whether to dump raw data, hardware-sent packages, cumulative states, +-- or measurements of device-computer latency. +data DumpMode = Raw | Packets | State | Measure deriving (Eq, Show) -- | Parser for `DumpArgs`. @@ -41,8 +48,11 @@ <*> nullOption ( long "mode" <> reader parseDumpMode <> value State - <> help "What to dump. Can be 'raw', 'packets', or 'state'" ) + <> help "What to dump. Can be 'raw', 'packets', 'state' or 'measure'" ) <*> switch + ( long "realtime" + <> help "In case --from-file is used, throttle data to 128 Hz like on real device" ) + <*> switch ( long "list" <> help "Show all available Emotiv devices and exit" ) <*> switch @@ -64,6 +74,7 @@ "raw" -> return Raw "packets" -> return Packets "state" -> return State + "measure" -> return Measure _ -> fail "Mode is not valid. Must be 'raw', 'packets', or 'state'." @@ -84,11 +95,11 @@ main = do DumpArgs{ emotivArgs , mode + , realtime , listDevices , json , serve - } <- execParser $ info (helper <*> dumpArgsParser) - (progDesc "Dumps Emotiv data") + } <- parseArgs "Dumps Emotiv data" dumpArgsParser if listDevices -- Only list devices then getEmotivDevices >>= putStrLn . ("Available devices:\n" ++) . ppShow @@ -104,25 +115,55 @@ -- Print to stdout or serve via websockets? Show the datatype or format via JSON? -- `output` accepts anything that's JSON-formattable and showable (wrapped in JsonShowable). output <- case serve of - Nothing -> return (putStrLn . if json then BSL8.unpack . encode else show) + -- TODO use Data.ByteString.Lazy.UTF8.fromString instead of BSL8 to prevent unicode errors + Nothing -> return (BSL8.putStrLn . if json then encode else BSL8.pack . show) Just (host, port) -> makeJsonOrShowWSServer host port json + -- For --mode measure: See how long a 0-128 cycle takes + timeRef <- newIORef =<< getCurrentTime + countRef <- newIORef (0 :: Int) + -- Packet loop - forever $ do + fix $ \loop -> do + timeBefore <- getCurrentTime + -- Output accumulative state, device-sent packet, or raw data? - case mode of - Packets -> do (_, packet) <- readEmotiv device - output (JsonShowable packet) - State -> do (state, _) <- readEmotiv device - output (JsonShowable state) + moreInput <- case mode of + Packets -> readEmotiv device `withJustM` \(_, packet) -> + output (JsonShowable packet) - Raw -> do rawBytes <- readEmotivRaw device - if json then output (JsonShowable rawBytes) - else BS8.putStr (emotivRawDataBytes rawBytes) + State -> readEmotiv device `withJustM` \(state, _) -> + output (JsonShowable state) - hFlush stdout + Raw -> readEmotivRaw device `withJustM` \rawBytes -> do + if json then output (JsonShowable rawBytes) + else BS.putStr (emotivRawDataBytes rawBytes) + hFlush stdout -- flush so that consuming apps immediately get it + + Measure -> readEmotivRaw device `withJustM` \_ -> do + -- When a full cycle is done, print how long it took. + count <- readIORef countRef + modifyIORef' countRef (+1) + when (count == 128) $ do + cycleTime <- diffUTCTime <$> getCurrentTime <*> readIORef timeRef + output . JsonShowable $ toDoule cycleTime + writeIORef countRef 0 + writeIORef timeRef =<< getCurrentTime + where + toDoule x = fromRational (toRational x) :: Double + + + -- When realtime is on, throttle the reading to 1/129 (a real + -- device's frequency). But take into the account the time that + -- we have spent reading from the device. + when realtime $ do + timeTaken <- (`diffUTCTime` timeBefore) <$> getCurrentTime + let delayUs = 1000000 `div` 129 - round (timeTaken * 1000000) + when (delayUs > 0) $ threadDelay delayUs + + when moreInput loop -- * JSON instances diff --git a/apps/DumpConduit.hs b/apps/DumpConduit.hs new file mode 100644 --- /dev/null +++ b/apps/DumpConduit.hs @@ -0,0 +1,189 @@ +{-# LANGUAGE NamedFieldPuns, ExistentialQuantification #-} +{-# OPTIONS_GHC -fno-warn-orphans #-} + +module Main where + +import Control.Concurrent (threadDelay) +import Control.Monad +import Control.Monad.IO.Class +import Data.Aeson (ToJSON (..), encode) +import qualified Data.ByteString as BS +import qualified Data.ByteString.Lazy.Char8 as BSL8 +import qualified Data.ByteString.Base64 as Base64 +import Data.Conduit +import qualified Data.Conduit.List as CL +import Data.Function (fix) +import Data.IORef +import Data.List +import Data.List.Split (splitOn) +import Data.Time.Clock +import Options.Applicative hiding (action) +import System.IO +import Text.Read +import Text.Show.Pretty + +import Hemokit +import Hemokit.Conduit +import Hemokit.Start + + +-- | Arguments for the EEG dump application. +data DumpArgs = DumpArgs + { emotivArgs :: EmotivArgs + , mode :: DumpMode -- ^ What to dump. + , realtime :: Bool -- ^ In case fromFile is used, throttle to 128 Hz. + , listDevices :: Bool -- ^ Do not do anything, print available devices. + , json :: Bool -- ^ Whether to format the output as JSON. + , serve :: Maybe (String, Int) -- ^ Serve via websockets on host:port. + } + +-- | Whether to dump raw data, hardware-sent packages, cumulative states, +-- or measurements of device-computer latency. +data DumpMode = Raw | Packets | State | Measure deriving (Eq, Show) + + +-- | Parser for `DumpArgs`. +dumpArgsParser :: Parser DumpArgs +dumpArgsParser = DumpArgs + <$> emotivArgsParser + <*> nullOption + ( long "mode" + <> reader parseDumpMode <> value State + <> help "What to dump. Can be 'raw', 'packets', 'state' or 'measure'" ) + <*> switch + ( long "realtime" + <> help "In case --from-file is used, throttle data to 128 Hz like on real device" ) + <*> switch + ( long "list" + <> help "Show all available Emotiv devices and exit" ) + <*> switch + ( long "json" + <> help "Format output as JSON" ) + <*> (optional . nullOption) + ( long "serve" <> metavar "HOST:PORT" + <> eitherReader parseHostPort + <> help ("Serve output via websockets, e.g. 127.0.0.1:1234 " ++ + "(port 1234, only localhost) or 0.0.0.0:1234 (all interfaces)") ) + where + -- TODO https://github.com/pcapriotti/optparse-applicative/issues/48 + eitherReader str2either = reader (either fail return . str2either) + + +-- | `DumpMode` command line parser. +parseDumpMode :: Monad m => String -> m DumpMode +parseDumpMode s = case s of + "raw" -> return Raw + "packets" -> return Packets + "state" -> return State + "measure" -> return Measure + _ -> fail "Mode is not valid. Must be 'raw', 'packets', or 'state'." + + +-- | Parses host and port from a string like "0.0.0.0:1234". +parseHostPort :: String -> Either String (String, Int) +parseHostPort hostPort = case readMaybe portStr of + Nothing -> Left $ show portStr ++ " is not a valid port number" + Just p -> Right (host, p) + where + (host, portStr) = splitLast ":" hostPort + + splitLast :: String -> String -> (String, String) + splitLast sep s = let sp = splitOn sep s -- splitOn never returns [] + in (intercalate sep (init sp), last sp) + + +main :: IO () +main = do + DumpArgs{ emotivArgs + , mode + , realtime + , listDevices + , json + , serve + } <- parseArgs "Dumps Emotiv data" dumpArgsParser + + if listDevices -- Only list devices + then getEmotivDevices >>= putStrLn . ("Available devices:\n" ++) . ppShow + else do + + e'device <- getEmotivDeviceFromArgs emotivArgs + + -- Do we have a device? + case e'device of + Left err -> error err + Right device -> do + + -- Print to stdout or serve via websockets? Show the datatype or format via JSON? + let outputSink :: (ToJSON i, Show i) => Sink i IO () + outputSink = case serve of + Nothing | json -> asJson $ CL.mapM_ BSL8.putStrLn + | otherwise -> CL.mapM_ print + Just (host, port) | json -> asJson $ websocketSink host port + | otherwise -> websocketSink host port + where + asJson = mapInput encode (const Nothing) + + throttled = if realtime then ($= throttle) else id + + -- Output accumulative state, device-sent packet, or raw data? + case mode of + Packets -> throttled (emotivPackets device) $$ outputSink + + State -> throttled (emotivStates device) $$ outputSink + + Raw -> throttled (rawSource device) $$ if json then outputSink + else CL.mapM_ (putStrBsFlush . emotivRawDataBytes) + Measure -> throttled (rawSource device) $= measureConduit $$ outputSink + + where + putStrBsFlush bs = BS.putStr bs >> hFlush stdout + + measureConduit = do + -- For --mode measure: See how long a 0-128 cycle takes + timeRef <- liftIO $ newIORef =<< getCurrentTime + countRef <- liftIO $ newIORef (0 :: Int) + + let yieldCyleTimes = do + -- When a full cycle is done, print how long it took. + count <- liftIO $ readIORef countRef + <* modifyIORef' countRef (+1) + when (count == 128) $ do + cycleTime <- liftIO $ diffUTCTime <$> getCurrentTime <*> readIORef timeRef + yield $ toDoule cycleTime + liftIO $ do writeIORef countRef 0 + writeIORef timeRef =<< getCurrentTime + where + toDoule x = fromRational (toRational x) :: Double + + awaitForever (const yieldCyleTimes) + + +-- When realtime is on, throttle the reading to 1/129 (a real +-- device's frequency). But take into the account the time that +-- we have spent reading from the device. +throttle :: (MonadIO m) => Conduit i m i +throttle = fix $ \loop -> do + + timeBefore <- liftIO getCurrentTime + m'x <- await + + case m'x of + Nothing -> return () + Just x -> do + timeTaken <- liftIO $ (`diffUTCTime` timeBefore) <$> getCurrentTime + let delayUs = 1000000 `div` 129 - round (timeTaken * 1000000) + when (delayUs > 0) $ liftIO $ threadDelay delayUs + yield x + loop + + +-- * JSON instances + +instance ToJSON EmotivPacket +instance ToJSON EmotivState + +instance ToJSON EmotivRawData where + toJSON = toJSON . Base64.encode . emotivRawDataBytes + +instance ToJSON Sensor where + toJSON = toJSON . show diff --git a/apps/FFT.hs b/apps/FFT.hs --- a/apps/FFT.hs +++ b/apps/FFT.hs @@ -5,59 +5,114 @@ import Data.Complex import Data.Conduit import qualified Data.Conduit.List as CL +import Data.List +import Data.Vector (Vector) import qualified Data.Vector as V import Numeric.FFT.Vector.Unnormalized -import System.Environment +import System.IO import Text.Printf import Hemokit +import Hemokit.Start +import Hemokit.Internal.Utils (untilNothing) + +rollingFFTConduit :: (Monad m) => Int -> ConduitM (Vector Double) [Vector Double] m () +rollingFFTConduit size = mapOutput (map (V.map magnitude . execute fft . ground) . transposeV 14) (rollingBuffer size) + where + fft = plan dftR2C size + + packets :: EmotivDevice -> Source IO EmotivState -packets d = forever (liftIO (readEmotiv d) >>= yield . fst) +packets d = void $ untilNothing (liftIO (readEmotiv d)) (yield . fst) buffer :: Monad m => Int -> Conduit a m [ a ] buffer n = forever (CL.take n >>= yield) + +-- | Rolls a buffer of size n over the input, always taking one element in, +-- throwing an old one out. +-- Only starts returning buffers once the buffer is filled. +-- +-- Implemented using a Difference List. +-- This allows fast skipping of buffers, e.g. for using only every 5th one. +rollingBuffer :: (Monad m) => Int -> Conduit a m [ a ] +rollingBuffer 0 = return () +rollingBuffer n | n < 0 = error "rollingBuffer: negative buffer size" + | otherwise = fillup 0 id + where + -- Consume until buffer is filled with n elements. + fillup have front + | have < n = await >>= maybe (return ()) (\x -> fillup (have+1) (front . (x:))) + | otherwise = roll front + -- Then keep kicking one element out, taking a new element in, yielding the buffer each time. + roll front = do yield (front []) + await >>= maybe (return ()) (\x -> roll (tail . front . (x:))) + + printAll :: Sink [V.Vector Double] IO () -printAll = awaitForever $ \tds -> liftIO $ putStrLn (unlines (map showFFT tds)) +-- printAll = awaitForever $ \tds -> liftIO $ putStrLn (unlines (map showFFT tds)) +-- printAll = awaitForever $ \tds -> liftIO $ putStrLn (unlines (map graphFFT tds)) +printAll = do + liftIO $ hSetBuffering stdout (BlockBuffering (Just 8000)) + awaitForever $ \tds -> liftIO $ putStrLn (unlines (map graphFFT [last tds])) >> hFlush stdout -- >> threadDelay 1000000 --- Convert a length M list of length N vectors into a length N list of length M vectors. +-- | Converts a length M list of length N vectors into a length N list of length M vectors. +-- Example: [ v1a v1b v1c ] [ v1a, v2a ] +-- [ v2a v2b v2c ] -> [ v1b, v2b ] +-- [ v1c, v2c ] transposeV :: Int -> [ V.Vector a ] -> [ V.Vector a ] transposeV n vs = [ V.fromList (map (V.! i) vs) | i <- [ 0 .. n - 1 ] ] showFFT :: V.Vector Double -> String --- showFFT ms = map (toChar . (/(V.maximum ms))) $ V.toList ms -showFFT ms = unwords . map (formatNumber . (/ V.maximum ms)) $ V.toList ms +showFFT ms = unwords . V.toList . V.map (formatNumber . maxed) $ ms where - formatNumber n = printf "%2.0f" (n*100) - toChar m - | m < 0.25 = ' ' - | m < 0.5 = '.' - | m < 0.75 = 'o' - | otherwise = '#' + formatNumber n = printf "%.3f" n + -- formatNumber n = printf "%2.0f" n -norm :: V.Vector Double -> V.Vector Double -norm v = V.map (subtract avg) v + -- simple = id + -- distributed = (/ V.sum ms) + maxed = (/ V.maximum ms) + + +graphFFT :: V.Vector Double -> String +-- graphFFT ms = (unlines . transpose . V.toList . V.map (formatNumber . maxed) $ ms) ++ showFFT ms +graphFFT ms = (unlines . transpose . V.toList . V.map (formatNumber . maxed) $ ms) + where + maxed = (/ V.maximum ms) + formatNumber n = replicate space ' ' ++ replicate filled '|' + where + chars = 40 + filled = floor (n * fromIntegral chars) + space = chars - filled + + +toChar :: Double -> Char +toChar m + | m < 0.25 = ' ' + | m < 0.5 = '.' + | m < 0.75 = 'o' + | otherwise = '#' + +-- | Reduces a data series by its average. +-- This is useful to bring a signal moving around at some level "to the ground". +-- Example: 4 5 4 3 -> 0 1 0 -1 +ground :: V.Vector Double -> V.Vector Double +ground v = V.map (subtract avg) v where avg = V.sum v / fromIntegral (V.length v) main :: IO () main = do - args <- getArgs - let model = if "--developer" `elem` args then Developer else Consumer - - devices <- getEmotivDevices - device <- case devices of - [] -> error "No devices found." - _ -> openEmotivDevice model (last devices) + m'device <- getEmotivDeviceFromArgs =<< parseArgs "FFT on Emotiv data" emotivArgsParser - let size = 256 - let fft = plan dftR2C size + case m'device of + Left err -> error err + Right device -> do - let sensorData = mapOutput (V.map fromIntegral . sensors) (packets device) - let fftConduit = mapOutput (map (V.map magnitude . execute fft . norm) . transposeV 14) (buffer size) + let sensorData = mapOutput (V.map fromIntegral . sensors) (packets device) - sensorData $= fftConduit $$ printAll + sensorData $= rollingFFTConduit 256 $$ printAll diff --git a/apps/Headmap.hs b/apps/Headmap.hs --- a/apps/Headmap.hs +++ b/apps/Headmap.hs @@ -29,7 +29,7 @@ setBlack >> drawStr name emotivStateMvar <- newEmptyMVar - forkIO $ do + _ <- forkIO $ do putMVar emotivStateMvar undefined putStrLn "Waiting for EEG data..." withDataFromLastEEG Consumer (void . swapMVar emotivStateMvar . fst) @@ -94,7 +94,7 @@ win <- windowNew vbox <- vBoxNew False 0 closeButton <- buttonNewWithLabel "Close" - onClicked closeButton (widgetDestroy win) + _ <- onClicked closeButton (widgetDestroy win) canvas <- drawingAreaNew _ <- canvas `onSizeRequest` return (Requisition width height) _ <- canvas `on` exposeEvent $ tryEvent $ updateCanvas canvas (renderAct win) @@ -105,7 +105,7 @@ widgetShow canvas widgetShowAll win - onDestroy win mainQuit + _ <- onDestroy win mainQuit -- flush mainGUI -- Flush all commands that are waiting to be sent to the graphics server. diff --git a/apps/Mouse.hs b/apps/Mouse.hs --- a/apps/Mouse.hs +++ b/apps/Mouse.hs @@ -11,7 +11,9 @@ import Hemokit +import Hemokit.Internal.Utils (untilNothing) + main :: IO () main = do args <- getArgs @@ -36,13 +38,11 @@ runRobotWithConnection (moveBy ((-x) `quot` 10) (y `quot` 10)) xc threadDelay 10000 - forever $ do - (state, _) <- readEmotiv device + void $ untilNothing (readEmotiv device) $ \(state, _) -> do + -- print (qualities state) print state -- putStrLn $ show (gyroX state) ++ " " ++ show (gyroY state) hFlush stdout modifyIORef' xy $ \(x,y) -> (x + gyroX state, y + gyroY state) - - return (battery state, qualities state) diff --git a/bench/ConduitRollingBufferBench.hs b/bench/ConduitRollingBufferBench.hs new file mode 100644 --- /dev/null +++ b/bench/ConduitRollingBufferBench.hs @@ -0,0 +1,55 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Main (main) where + +import Criterion.Main +import Control.Monad.Identity +import Data.Conduit +import qualified Data.Conduit.List as CL + + + +main :: IO () +main = defaultMain + [ bench "rollingBuffer" $ nf (benchBuf rollingBuffer 1000) [1..10000::Int] + , bench "rollingBuffer DL" $ nf (benchBuf rollingBufferDL 1000) [1..10000::Int] + , bench "last rollingBuffer" $ nf (last . benchBuf rollingBuffer 1000) [1..10000::Int] + , bench "last rollingBuffer DL" $ nf (last . benchBuf rollingBufferDL 1000) [1..10000::Int] + , bench "5th rollingBuffer" $ nf (every5th . benchBuf rollingBuffer 1000) [1..10000::Int] + , bench "5th rollingBuffer DL" $ nf (every5th . benchBuf rollingBufferDL 1000) [1..10000::Int] + ] + where + benchBuf bufFn bufSize inputList = runIdentity (CL.sourceList inputList $= bufFn bufSize $$ CL.consume) + + +every5th :: [a] -> [a] +every5th (x:_:_:_:_:xs) = x : every5th xs +every5th _ = [] + + +-- Implemented with lists + reverse. +rollingBuffer :: (Monad m) => Int -> Conduit a m [ a ] +rollingBuffer 0 = return () +rollingBuffer n = fillup 0 [] + where + -- Consume until buffer is filled with n elements. + fillup have buf + | have < n = await >>= maybe (return ()) (\x -> fillup (have+1) (x:buf)) + | otherwise = roll buf + -- Then keep kicking one element out, taking a new element in, yielding the buffer each time. + roll buf = do yield (reverse buf) + await >>= maybe (return ()) (\x -> roll (x : init buf)) + + +-- Implemented with a Difference List. +rollingBufferDL :: (Monad m) => Int -> Conduit a m [ a ] +rollingBufferDL 0 = return () +rollingBufferDL n = fillup 0 id + where + -- Consume until buffer is filled with n elements. + fillup have front + | have < n = await >>= maybe (return ()) (\x -> fillup (have+1) (front . (x:))) + | otherwise = roll front + -- Then keep kicking one element out, taking a new element in, yielding the buffer each time. + roll front = do yield (front []) + await >>= maybe (return ()) (\x -> roll (tail . front . (x:))) diff --git a/hemokit.cabal b/hemokit.cabal --- a/hemokit.cabal +++ b/hemokit.cabal @@ -1,5 +1,5 @@ name: hemokit -version: 0.4.2 +version: 0.5 license: MIT copyright: 2013 Niklas Hambüchen , Patrick Chilton author: Niklas Hambüchen , Patrick Chilton @@ -44,16 +44,22 @@ library exposed-modules: Hemokit + Hemokit.Conduit + Hemokit.Internal.Utils Hemokit.Start build-depends: base >= 4 && < 5 + , aeson >= 0.6.1.0 , bytestring >= 0.9.2.1 , cipher-aes >= 0.1.7 + , conduit >= 1 , deepseq >= 1.2 , deepseq-generics >= 0.1 , hidapi >= 0.1.1 + , mtl >= 2.1.2 , optparse-applicative >= 0.5.2.1 , vector >= 0.9 + , websockets >= 0.7.3.0 hs-source-dirs: src default-language: Haskell2010 @@ -92,11 +98,33 @@ , optparse-applicative >= 0.5.2.1 , pretty-show >= 1.0 , split >= 0.2.2 + , time >= 1.4.0.1 , transformers >= 0.3.0.0 , websockets >= 0.7.3.0 ghc-options: -Wall +executable hemokit-dump-conduit + default-language: Haskell2010 + hs-source-dirs: + apps + main-is: + DumpConduit.hs + build-depends: + base + , hemokit + , aeson >= 0.6.1.0 + , base64-bytestring >= 1.0.0.1 + , bytestring >= 0.9.2.1 + , conduit >= 1 + , optparse-applicative >= 0.5.2.1 + , pretty-show >= 1.0 + , split >= 0.2.2 + , time >= 1.4.0.1 + , transformers >= 0.3.0.0 + ghc-options: -Wall + + executable hemokit-fft default-language: Haskell2010 hs-source-dirs: @@ -157,6 +185,22 @@ Bench.hs build-depends: base + , hemokit + , criterion >= 0.8.0.0 + ghc-options: -Wall + + +benchmark bench-rollingbuffer + default-language: Haskell2010 + type: exitcode-stdio-1.0 + hs-source-dirs: + bench + main-is: + ConduitRollingBufferBench.hs + build-depends: + base + , conduit + , mtl , hemokit , criterion >= 0.8.0.0 ghc-options: -Wall diff --git a/src/Hemokit.hs b/src/Hemokit.hs --- a/src/Hemokit.hs +++ b/src/Hemokit.hs @@ -19,6 +19,7 @@ , getEmotivDevices , openEmotivDevice , openEmotivDeviceFile + , openEmotivDeviceHandle , readEmotiv , EmotivException (..) , SerialNumber () @@ -60,7 +61,6 @@ import Control.Applicative import Control.DeepSeq.Generics import Control.Exception -import Control.Monad import Crypto.Cipher.AES import Data.Bits ((.|.), (.&.), shiftL, shiftR) import Data.Char @@ -406,6 +406,12 @@ openEmotivDeviceFile :: EmotivModel -> SerialNumber -> String -> IO EmotivDevice openEmotivDeviceFile model sn path = do h <- openFile path ReadMode + openEmotivDeviceHandle model sn h + + +-- | Creates an `EmotivDevice` device from an open file handle. +openEmotivDeviceHandle :: EmotivModel -> SerialNumber -> Handle -> IO EmotivDevice +openEmotivDeviceHandle model sn h = do stateRef <- newIORef Nothing return $ EmotivDevice { rawDevice = HandleDevice h @@ -416,13 +422,21 @@ -- | Reads one 32 byte packet from the device and decrypts it to raw data. -readEmotivRaw :: EmotivDevice -> IO EmotivRawData +-- +-- Returns Nothing on end of input (or if there are < 32 bytes before it). +-- +-- Note that if the EEG is (turned) off, this function block until +-- it is turned on again. +readEmotivRaw :: EmotivDevice -> IO (Maybe EmotivRawData) readEmotivRaw EmotivDevice{ rawDevice, serial, emotivModel } = do d32 <- case rawDevice of HidapiDevice d -> HID.read d 32 HandleDevice d -> BS.hGet d 32 - return $ decrypt serial emotivModel d32 + -- If we get less than the requested 32 bytes, we're at EOF. + return $ if BS.length d32 < 32 + then Nothing + else Just $ decrypt serial emotivModel d32 -- | Given a device and a Emotiv raw data, parses the raw data into an @@ -473,12 +487,20 @@ -- for that device. -- -- Returns both the packet read from the device and the updated state. -readEmotiv :: EmotivDevice -> IO (EmotivState, EmotivPacket) -readEmotiv device = updateEmotivState device =<< readEmotivRaw device +-- +-- Returns Nothing on end of input (or if there are < 32 bytes before it). +-- +-- Note that if the EEG is (turned) off, this function block until +-- it is turned on again. +readEmotiv :: EmotivDevice -> IO (Maybe (EmotivState, EmotivPacket)) +readEmotiv device = do m'raw <- readEmotivRaw device + case m'raw of + Nothing -> return Nothing + Just raw -> Just <$> updateEmotivState device raw -- | Opens and reads from the last available device, giving all data from it --- to the given function. +-- to the given function. Stops if end of input is reached. -- -- Intended for use with ghci. -- @@ -487,13 +509,16 @@ -- >withDataFromLastEEG Consumer print -- >withDataFromLastEEG Consumer (print . packetQuality . snd) -- >withDataFromLastEEG Consumer (putStrLn . unwords . map show . V.toList . qualities . fst) -withDataFromLastEEG :: EmotivModel -> ((EmotivState, EmotivPacket) -> IO a) -> IO a +withDataFromLastEEG :: EmotivModel -> ((EmotivState, EmotivPacket) -> IO ()) -> IO () withDataFromLastEEG model f = do devices <- getEmotivDevices device <- case devices of [] -> error "No devices found." _ -> openEmotivDevice model (last devices) - forever $ readEmotiv device >>= f + let run = do m'd <- readEmotiv device + case m'd of Nothing -> return () -- terminate + Just d -> f d >> run + run -- * NFData instances diff --git a/src/Hemokit/Conduit.hs b/src/Hemokit/Conduit.hs new file mode 100644 --- /dev/null +++ b/src/Hemokit/Conduit.hs @@ -0,0 +1,59 @@ +module Hemokit.Conduit where + +import Control.Concurrent (forkIO) +import Control.Concurrent.Chan +import Control.Monad +import Control.Monad.Trans +import Data.Aeson (ToJSON (..), encode) +import Data.ByteString.Lazy (ByteString) +import Data.Conduit +import qualified Network.WebSockets as WS + +import Hemokit + +import Hemokit.Internal.Utils (untilNothing) + + +rawSource :: (MonadIO m) => EmotivDevice -> Source m EmotivRawData +rawSource dev = void $ untilNothing (liftIO (readEmotivRaw dev)) yield + +parsePackets :: (MonadIO m) => EmotivDevice -> Conduit EmotivRawData m (EmotivState, EmotivPacket) +parsePackets dev = awaitForever (\raw -> liftIO (updateEmotivState dev raw) >>= yield) + + +-- * Convenience + +emotivStates :: (MonadIO m) => EmotivDevice -> Source m EmotivState +emotivStates dev = rawSource dev $= mapOutput fst (parsePackets dev) + +emotivPackets :: (MonadIO m) => EmotivDevice -> Source m EmotivPacket +emotivPackets dev = rawSource dev $= mapOutput snd (parsePackets dev) + + +-- * JSON + +-- TODO check if we really need this since it doesn't do any monadic thing +jsonConduit :: (Monad m, ToJSON i) => Conduit i m ByteString +jsonConduit = awaitForever (yield . encode) + + +-- * Websockets + +websocketSink :: (MonadIO m, ToJSON a) => String -> Int -> Sink a m () +websocketSink host port = do + chan <- liftIO $ newChan + + -- Server loop: Send what comes in via the chan; Nothing shuts down + let jsonWSServerFromChan :: WS.Request -> WS.WebSockets WS.Hybi10 () + jsonWSServerFromChan = \req -> do + WS.acceptRequest req + void $ untilNothing (liftIO (readChan chan)) (WS.sendTextData . encode) + + -- Fork off Websocket server + _ <- liftIO $ forkIO $ WS.runServer host port jsonWSServerFromChan + + -- Send messages to server via the chan + void $ awaitForever (liftIO . writeChan chan . Just) + + -- Tell server to shut down + liftIO $ writeChan chan Nothing diff --git a/src/Hemokit/Internal/Utils.hs b/src/Hemokit/Internal/Utils.hs new file mode 100644 --- /dev/null +++ b/src/Hemokit/Internal/Utils.hs @@ -0,0 +1,18 @@ +module Hemokit.Internal.Utils + ( withJustM + , untilNothing + ) where + + +-- | If the monad retuns a Just, runs the function on its contents. +-- Returns True if the action was executed. +withJustM :: (Monad m) => m (Maybe a) -> (a -> m ()) -> m Bool +withJustM act f = act >>= maybe (return False) (\x -> f x >> return True) + + +-- | Runs the monadic action as long as the producer returns Justs. +-- Returns True if the action was ever executed. +untilNothing :: (Monad m) => m (Maybe a) -> (a -> m ()) -> m Bool +untilNothing act f = act `withJustM` (\x -> f x >> again) + where + again = untilNothing act f >> return () -- void would be nicer diff --git a/src/Hemokit/Start.hs b/src/Hemokit/Start.hs --- a/src/Hemokit/Start.hs +++ b/src/Hemokit/Start.hs @@ -8,11 +8,13 @@ ( EmotivArgs (..) , emotivArgsParser , parseModel + , parseArgs , getEmotivDeviceFromArgs ) where import Data.List import Options.Applicative +import System.IO (stdin) import Hemokit hiding (serial) @@ -54,13 +56,21 @@ maybeReader mbFn msg = reader $ maybe (fail msg) pure . mbFn +-- | Runs a command line parser. The given program description is used for the +-- --help message. +parseArgs :: String -> Parser a -> IO a +parseArgs programDescription parser = execParser $ info (helper <*> parser) + (progDesc programDescription) + + -- | Depending on some common EEG-choice-related user input, list devices or -- try to open the correct device. getEmotivDeviceFromArgs :: EmotivArgs -> IO (Either String EmotivDevice) getEmotivDeviceFromArgs EmotivArgs{ model, serial, fromFile } = case fromFile of -- File given, use device file / file handle - Just f | Just s <- serial -> Right <$> openEmotivDeviceFile model s f + Just f | Just s <- serial -> Right <$> if f == "-" then openEmotivDeviceHandle model s stdin + else openEmotivDeviceFile model s f | otherwise -> fail "A serial number must be provided when using --from-file" -- No file given, use HIDAPI to select the device