11

我正在使用 portaudio 处理一些音频。只要有音频数据需要处理，haskell FFI 绑定就会调用用户定义的回调。这个回调应该很快被处理，理想情况下没有 I/O。我想保存音频输入并快速返回，因为我的应用程序不需要实时响应音频（现在我只是将音频数据保存到文件中；稍后我将构建一个简单的语音识别系统） .

我喜欢这个想法，pipes并认为我可以使用那个库。问题是我不知道如何创建一个Producer返回通过回调传入的数据的方法。

如何处理我的用例？

这是我现在正在使用的内容，以防有帮助（基准 mvar 现在不起作用，但我不喜欢将所有数据存储在 seq 中......我宁愿处理它来代替就在最后）：

{-# LANGUAGE FlexibleInstances, MultiParamTypeClasses #-}

module Main where

import Codec.Wav

import Sound.PortAudio
import Sound.PortAudio.Base
import Sound.PortAudio.Buffer

import Foreign.Ptr
import Foreign.ForeignPtr
import Foreign.C.Types
import Foreign.Storable

import qualified Data.StorableVector as SV
import qualified Data.StorableVector.Base as SVB

import Control.Exception.Base (evaluate)

import Data.Int
import Data.Sequence as Seq

import Control.Concurrent

instance Buffer SV.Vector a where
  fromForeignPtr fp = return . SVB.fromForeignPtr fp
  toForeignPtr = return . (\(a, b, c) -> (a, c)) . SVB.toForeignPtr

-- | Wrap a buffer callback into the generic stream callback type.
buffCBtoRawCB' :: (StreamFormat input, StreamFormat output, Buffer a input, Buffer b output) =>
    BuffStreamCallback input output a b -> StreamCallback input output    
buffCBtoRawCB' func = \a b c d e -> do
    fpA <- newForeignPtr_ d -- We will not free, as callback system will do that for us   
    fpB <- newForeignPtr_ e -- We will not free, as callback system will do that for us
    storeInp <- fromForeignPtr fpA (fromIntegral $ 1 * c)
    storeOut <- fromForeignPtr fpB (fromIntegral $ 0 * c)
    func a b c storeInp storeOut

callback :: MVar (Seq.Seq [Int32]) -> PaStreamCallbackTimeInfo -> [StreamCallbackFlag] -> CULong 
            -> SV.Vector Int32 -> SV.Vector Int32 -> IO StreamResult
callback seqmvar = \timeinfo flags numsamples input output -> do
  putStrLn $ "timeinfo: " ++ show timeinfo ++ "; flags are " ++ show flags ++ " in callback with " ++ show numsamples ++ " samples."  
  print input
  -- write data to output
  --mapM_ (uncurry $ pokeElemOff output) $ zip (map fromIntegral [0..(numsamples-1)]) datum
  --print "wrote data"

  input' <- evaluate $ SV.unpack input  
  modifyMVar_ seqmvar (\s -> return $ s Seq.|> input')

  case flags of
    [] -> return $ if unPaTime (outputBufferDacTime timeinfo) > 0.2 then Complete else Continue
    _ -> return Complete

done doneMVar = do
  putStrLn "total done dood!"
  putMVar doneMVar True
  return ()

main = do

  let samplerate = 16000

  Nothing <- initialize

  print "initialized"

  m <- newEmptyMVar
  datum <- newMVar Seq.empty

  Right s <- openDefaultStream 1 0 samplerate Nothing (Just $ buffCBtoRawCB' (callback datum)) (Just $ done m)
  startStream s

  _ <- takeMVar m -- wait until our callbacks decide they are done!
  Nothing <- terminate

  print "let's see what we've recorded..."

  stuff <- takeMVar datum
  print stuff

  -- write out wav file

  -- let datum = 
  --       audio = Audio { sampleRate = samplerate
  --                   , channelNumber = 1
  --                   , sampleData = datum
  --                   }
  -- exportFile "foo.wav" audio

  print "main done"
4

1 回答 1

13

最简单的解决方案是使用MVars 在回调和Producer. 就是这样：

import Control.Proxy
import Control.Concurrent.MVar

fromMVar :: (Proxy p) => MVar (Maybe a) -> () -> Producer p a IO ()
fromMVar mvar () = runIdentityP loop where
    loop = do
        ma <- lift $ takeMVar mvar
        case ma of
            Nothing -> return ()
            Just a  -> do
                respond a
                loop

您的流回调将写入Just input并且MVar您的终结回调将写入Nothing以终止Producer.

这是一个ghci演示它如何工作的示例：

>>> mvar <- newEmptyMVar :: IO (MVar (Maybe Int))
>>> forkIO $ runProxy $ fromMVar mvar >-> printD
>>> putMVar mvar (Just 1)
1
>>> putMVar mvar (Just 2)
2
>>> putMVar mvar Nothing
>>> putMVar mvar (Just 3)
>>>

编辑：该pipes-concurrency库现在提供此功能，甚至在教程中有一节专门解释如何使用它从回调中获取数据。

于 2013-02-02T07:59:20.360 回答