하스켈의 파이프 및 콜백

Question

portaudio를 사용하여 일부 오디오를 처리 중입니다. haskell FFI 바인딩은 처리 할 오디오 데이터가있을 때마다 사용자 정의 콜백을 호출합니다. 이 콜백은 I/O없이 매우 신속하게 이상적으로 처리되어야합니다. 응용 프로그램이 실시간으로 오디오에 반응 할 필요가 없기 때문에 오디오 입력을 저장하고 신속하게 반환하고 싶었습니다. (지금은 오디오 데이터를 파일로 저장하고 나중에 간단한 음성 인식 시스템을 구성 할 것입니다. .

나는 pipes이라는 생각을 좋아하고 그 라이브러리를 사용할 수 있다고 생각했습니다. 문제는 콜백을 통해 들어오는 데이터를 반환하는 Producer을 만드는 방법을 모르겠다는 것입니다.

유스 케이스를 어떻게 처리합니까? 여기

---

내가 도움이 경우에, 지금 함께 일하고 있어요 무엇을 (... 내가 것 데이텀 MVAR는 지금 당장 작동하지 않습니다하지만 난 서열에 모든 데이터를 저장하기 싫어 그 대신) 끝으로 등장하면서 오히려 처리 :

{-# 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" 

Answer 1

가장 간단한 해결책은 콜백과 Producer 사이의 통신에 MVar의를 사용하는 것입니다.

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 

스트림 콜백이 MVar-Just input를 작성하고 귀하의 마무리 콜백이 Producer을 종료 Nothing을 쓸 것입니다 방법은 다음과 같습니다.

이

>>> 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) 
>>> 

편집 :

여기에 어떻게 작동하는지 보여주는 ghci 예입니다 The pipes-concurrency library 지금이 기능을 제공하며, 심지어 section in the tutorial 콜백에서 데이터를 가져 오기 위해 그것을 사용하는 방법을 구체적으로 설명하고있다.