mediabus (empty) → 0.2.0.0
raw patch · 36 files changed
+2788/−0 lines, 36 filesdep +QuickCheckdep +arraydep +asyncsetup-changed
Dependencies added: QuickCheck, array, async, base, binary, bytestring, cereal, conduit, conduit-combinators, conduit-extra, conduit-network-rtp, containers, data-default, deepseq, hspec, lens, lifted-async, monad-control, mtl, network, parallel, primitive, process, random, resourcet, singletons, spool, stm, streaming-commons, tagged, template-haskell, text, time, transformers, type-spec, vector
Files
- LICENSE +30/−0
- Setup.hs +2/−0
- mediabus.cabal +208/−0
- specs/Data/MediaBus/Audio/ResampleSpec.hs +58/−0
- specs/Data/MediaBus/ReorderSpec.hs +172/−0
- specs/Data/MediaBus/SampleSpec.hs +45/−0
- specs/Data/MediaBus/SequenceSpec.hs +31/−0
- specs/Data/MediaBus/StreamSegmentSpec.hs +98/−0
- specs/Data/MediaBus/StreamSpec.hs +60/−0
- specs/Data/MediaBus/TicksSpec.hs +38/−0
- specs/Spec.hs +1/−0
- src/Data/MediaBus.hs +26/−0
- src/Data/MediaBus/AsyncConduit.hs +119/−0
- src/Data/MediaBus/Audio/Alaw.hs +131/−0
- src/Data/MediaBus/Audio/Channels.hs +69/−0
- src/Data/MediaBus/Audio/Raw.hs +99/−0
- src/Data/MediaBus/Audio/Resample.hs +41/−0
- src/Data/MediaBus/BlankMedia.hs +24/−0
- src/Data/MediaBus/Clock.hs +92/−0
- src/Data/MediaBus/Conduit.hs +59/−0
- src/Data/MediaBus/DebugSink.hs +49/−0
- src/Data/MediaBus/Discontinous.hs +48/−0
- src/Data/MediaBus/Monotone.hs +36/−0
- src/Data/MediaBus/OrderedBy.hs +13/−0
- src/Data/MediaBus/Payload.hs +9/−0
- src/Data/MediaBus/Reorder.hs +102/−0
- src/Data/MediaBus/Sample.hs +158/−0
- src/Data/MediaBus/Segment.hs +49/−0
- src/Data/MediaBus/Sequence.hs +94/−0
- src/Data/MediaBus/Series.hs +135/−0
- src/Data/MediaBus/SourceId.hs +37/−0
- src/Data/MediaBus/Stream.hs +302/−0
- src/Data/MediaBus/StreamSegment.hs +59/−0
- src/Data/MediaBus/Ticks.hs +231/−0
- src/Data/MediaBus/Transcoder.hs +22/−0
- src/Data/MediaBus/Transport/Udp.hs +41/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Sven Heyll, Lindenbaum GmbH (c) 2016++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Sven Heyll nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ mediabus.cabal view
@@ -0,0 +1,208 @@+name: mediabus+version: 0.2.0.0+synopsis: Multimedia streaming on top of Conduit+description: Please see README.md+homepage: https://github.com/lindenbaum/mediabus+license: BSD3+license-file: LICENSE+author: Sven Heyll+maintainer: sven.heyll@lindenbaum.eu+copyright: 2016,2017 Sven Heyll, Lindenbaum GmbH+category: Media+build-type: Simple+-- extra-source-files:+cabal-version: >=1.10++library+ hs-source-dirs: src+ exposed-modules:+ Data.MediaBus+ , Data.MediaBus.AsyncConduit+ , Data.MediaBus.Audio.Alaw+ , Data.MediaBus.Audio.Channels+ , Data.MediaBus.Audio.Raw+ , Data.MediaBus.Audio.Resample+ , Data.MediaBus.BlankMedia+ , Data.MediaBus.Clock+ , Data.MediaBus.Conduit+ , Data.MediaBus.DebugSink+ , Data.MediaBus.Discontinous+ , Data.MediaBus.Monotone+ , Data.MediaBus.OrderedBy+ , Data.MediaBus.Segment+ , Data.MediaBus.StreamSegment+ , Data.MediaBus.Payload+ , Data.MediaBus.Reorder+ , Data.MediaBus.Sample+ , Data.MediaBus.Sequence+ , Data.MediaBus.Series+ , Data.MediaBus.SourceId+ , Data.MediaBus.Stream+ , Data.MediaBus.Ticks+ , Data.MediaBus.Transcoder+ , Data.MediaBus.Transport.Udp+ default-language: Haskell2010+ ghc-options: -O2 -Wall -funbox-strict-fields -fno-warn-unused-do-bind -fprint-explicit-kinds+ default-extensions: ApplicativeDo+ , BangPatterns+ , ConstraintKinds+ , CPP+ , DataKinds+ , DefaultSignatures+ , DeriveDataTypeable+ , DeriveFoldable+ , DeriveFunctor+ , DeriveGeneric+ , DeriveLift+ , DeriveTraversable+ , DuplicateRecordFields+ , EmptyDataDecls+ , EmptyCase+ , FlexibleInstances+ , FlexibleContexts+ , FunctionalDependencies+ , GADTs+ , GeneralizedNewtypeDeriving+ , KindSignatures+ , LambdaCase+ , MultiParamTypeClasses+ , MultiWayIf+ , NamedFieldPuns+ , OverloadedStrings+ , QuasiQuotes+ , RecordWildCards+ , RankNTypes+ , ScopedTypeVariables+ , StandaloneDeriving+ , TemplateHaskell+ , TupleSections+ , TypeApplications+ , TypeFamilies+ , TypeInType+ , TypeOperators+ , TypeSynonymInstances+ , UnicodeSyntax+ build-depends: QuickCheck+ , array+ , async+ , base >= 4.9 && < 5+ , bytestring+ , cereal+ , conduit+ , conduit-combinators+ , conduit-extra+ , containers+ , data-default+ , deepseq+ , lens+ , lifted-async+ , monad-control+ , mtl+ , network+ , parallel+ , primitive+ , process+ , random+ , resourcet+ , spool == 0.1+ , stm+ , streaming-commons+ , tagged+ , text+ , time+ , transformers+ , vector++test-suite tests+ type: exitcode-stdio-1.0+ hs-source-dirs: specs+ main-is: Spec.hs+ other-modules: Data.MediaBus.Audio.ResampleSpec+ , Data.MediaBus.StreamSegmentSpec+ , Data.MediaBus.ReorderSpec+ , Data.MediaBus.SampleSpec+ , Data.MediaBus.SequenceSpec+ , Data.MediaBus.StreamSpec+ , Data.MediaBus.TicksSpec+ default-language: Haskell2010+ build-depends: QuickCheck+ , array+ , async+ , base >= 4.9 && < 5+ , binary+ , bytestring+ , conduit+ , conduit-combinators+ , conduit-extra+ , conduit-network-rtp+ , containers+ , data-default+ , deepseq+ , hspec+ , lens+ , monad-control+ , mtl+ , singletons+ , spool == 0.1+ , stm+ , tagged+ , template-haskell+ , text+ , time+ , transformers+ , type-spec+ , vector+ default-language: Haskell2010+ default-extensions: ApplicativeDo+ , Arrows+ , BangPatterns+ , ConstraintKinds+ , CPP+ , DataKinds+ , DefaultSignatures+ , DeriveDataTypeable+ , DeriveFoldable+ , DeriveFunctor+ , DeriveGeneric+ , DeriveLift+ , DeriveTraversable+ , DuplicateRecordFields+ , EmptyDataDecls+ , EmptyCase+ , FlexibleInstances+ , FlexibleContexts+ , FunctionalDependencies+ , GADTs+ , GeneralizedNewtypeDeriving+ , KindSignatures+ , LambdaCase+ , MultiParamTypeClasses+ , MultiWayIf+ , NamedFieldPuns+ , OverloadedStrings+ , QuasiQuotes+ , RankNTypes+ , ScopedTypeVariables+ , StandaloneDeriving+ , TemplateHaskell+ , TupleSections+ , TypeApplications+ , TypeFamilies+ , TypeInType+ , TypeOperators+ , TypeSynonymInstances+ , UnicodeSyntax+ ghc-options: -threaded+ -rtsopts -with-rtsopts=-N+ -j +RTS -A256m -n2m -RTS+ -Wall+ -O0+ -fno-warn-unused-binds+ -fno-warn-orphans+ -fno-warn-unused-do-bind+ -fno-warn-missing-signatures+ -fno-warn-type-defaults++source-repository head+ type: git+ location: https://github.com/lindenbaum/mediabus
+ specs/Data/MediaBus/Audio/ResampleSpec.hs view
@@ -0,0 +1,58 @@+module Data.MediaBus.Audio.ResampleSpec ( spec ) where++import Data.MediaBus+import Conduit+import Test.QuickCheck+import Test.Hspec+import Control.Monad.State.Strict+import Data.Default++spec :: Spec+spec = describe "Resampling of S16 samples from 8 to 16 kHz" $ do+ it "interpolates between samples" $+ let lastVal = 0+ in+ property $+ \samples -> sampleBufferToList (resampleAndConsume (singleFrameFromList samples)+ lastVal)+ `shouldBe` expectedResamplingResult samples lastVal+ it "interpolates also between frames" $+ let lastVal = 0+ in+ property $+ \samplesLists -> sampleBufferToList (resampleAndConsume (framesFromLists samplesLists)+ lastVal)+ `shouldBe` expectedResamplingResult (join samplesLists)+ lastVal++expectedResamplingResult :: [S16 8000] -> S16 8000 -> [S16 16000]+expectedResamplingResult xs lastVal =+ concatMap (\(x, y) -> [ setAudioSampleRateTo16kHz (avgSamples x y)+ , setAudioSampleRateTo16kHz y+ ])+ (zip (lastVal : xs) xs)++resampleAndConsume :: Source Identity (Stream SrcId32 SeqNum32 Ticks32At48000 () (SampleBuffer (S16 8000)))+ -> S16 8000+ -> SampleBuffer (S16 16000)+resampleAndConsume vvv lastVal =+ runConduitPure (vvv .|+ resample8to16kHz' lastVal .|+ concatStreamContents)++singleFrameFromList :: Monad m+ => [S16 8000]+ -> Source m (Stream SrcId32 SeqNum32 Ticks32At48000 () (SampleBuffer (S16 8000)))+singleFrameFromList x = mapOutput (MkStream . Next)+ (mapOutput (MkFrame () def)+ (yield (sampleBufferFromList x)) .|+ deriveFrameTimestamp 0)++framesFromLists :: Monad m+ => [[S16 8000]]+ -> Source m (Stream SrcId32 SeqNum32 Ticks32At48000 () (SampleBuffer (S16 8000)))+framesFromLists xs = mapOutput (MkStream . Next)+ (mapOutput (MkFrame () def)+ (mapM_ (yield . sampleBufferFromList)+ xs) .|+ deriveFrameTimestamp 0)
+ specs/Data/MediaBus/ReorderSpec.hs view
@@ -0,0 +1,172 @@+module Data.MediaBus.ReorderSpec ( spec ) where++import Conduit+import Data.Conduit.List+import Data.MediaBus+import Test.Hspec+import Data.Proxy+import Data.Word+import Test.QuickCheck++spec :: Spec+spec = describe "reorderFramesBySeqNumC" $ do+ let runC inputs = runConduitPure (sourceList inputs .|+ annotateTypeC (Proxy :: Proxy (Stream () Word8 () () ()))+ (reorderFramesBySeqNumC 3) .|+ consume)++ it "yields at least one 'Start' for each incoming 'Start'" $+ property $+ \inputs -> let countStarts = foldr (\x n -> case x of+ MkStream (Start _) -> n + 1+ MkStream (Next _) -> n)+ 0+ in+ countStarts (runC inputs) `shouldSatisfy`+ (>= countStarts inputs)+ it "yields exactly the given input if the input is ordered and gap-free" $+ let inputs = [ startFrame 0 ] +++ [ nextFrame x+ | x <- [0 .. 10] ]+ in+ runC inputs `shouldBe`+ inputs+ it "reorders out of order packets (1)" $+ let inputs = [ startFrame 0+ , nextFrame 0+ , nextFrame 2+ , nextFrame 3+ , nextFrame 1+ ]+ expected = [ startFrame 0+ , nextFrame 0+ , nextFrame 1+ , nextFrame 2+ , nextFrame 3+ ]+ in+ runC inputs `shouldBe`+ expected+ it "reorders out of order packets (2)" $+ let inputs = [ startFrame 0, nextFrame 2, nextFrame 1, nextFrame 0 ]+ expected = [ startFrame 0, nextFrame 0, nextFrame 1, nextFrame 2 ]+ in+ runC inputs `shouldBe`+ expected+ it "skips over missing frames when the queue is full" $+ let inputs = [ startFrame 0+ , nextFrame 0+ , nextFrame 1+ , nextFrame 3+ , nextFrame 4+ , nextFrame 5+ ]+ expected = [ startFrame 0+ , nextFrame 0+ , nextFrame 1+ , nextFrame 3+ , nextFrame 4+ , nextFrame 5+ ]+ in+ runC inputs `shouldBe`+ expected+ it "restarts at the incoming sequence numbering after too many frame drops" $+ let inputs = [ startFrame 10+ , nextFrame 10+ , nextFrame 11+ , nextFrame 12+ , nextFrame 0+ , nextFrame 1+ , nextFrame 2+ , nextFrame 3+ , nextFrame 4+ ]+ expected = [ startFrame 10+ , nextFrame 10+ , nextFrame 11+ , nextFrame 12+ , startFrame 2+ , nextFrame 2+ , nextFrame 3+ , nextFrame 4+ ]+ in+ runC inputs `shouldBe`+ expected+ it "restarts at the incoming sequence numbering after too many frame drops, AND flushes the queued elements first" $+ let inputs = [ startFrame 10+ , nextFrame 13+ , nextFrame 14+ , nextFrame 0+ , nextFrame 1+ , nextFrame 2+ , nextFrame 3+ , nextFrame 4+ ]+ expected = [ startFrame 10+ , nextFrame 13+ , nextFrame 14+ , startFrame 2+ , nextFrame 2+ , nextFrame 3+ , nextFrame 4+ ]+ in+ runC inputs `shouldBe`+ expected+ it "flushes and resets its internal state after every 'Start'" $+ let inputs = [ startFrame 10+ , nextFrame 13+ , nextFrame 14+ , startFrame 40+ , nextFrame 40+ , nextFrame 41+ , startFrame 50+ , nextFrame 0+ , nextFrame 1+ , startFrame 60+ , nextFrame 50+ , nextFrame 51+ , nextFrame 52+ ]+ expected = [ startFrame 10+ , nextFrame 13+ , nextFrame 14+ , startFrame 40+ , nextFrame 40+ , nextFrame 41+ , startFrame 50+ , startFrame 60+ , startFrame 52+ , nextFrame 52+ ]+ in+ runC inputs `shouldBe`+ expected+ it "yields monotone increasing frames higher than the start-frame" $+ property $+ \inputs -> runC inputs `shouldSatisfy`+ isMonoIncreasingAndHigherThanStartSeqNum++isMonoIncreasingAndHigherThanStartSeqNum :: [Stream () Word8 () () ()] -> Bool+isMonoIncreasingAndHigherThanStartSeqNum [] =+ True+isMonoIncreasingAndHigherThanStartSeqNum (MkStream (Start (MkFrameCtx () () sn ())) : rest) =+ isMonoIncreasingAndHigherThanStartSeqNumN sn rest+isMonoIncreasingAndHigherThanStartSeqNum rest@(MkStream (Next (MkFrame () sn ())) : _) =+ isMonoIncreasingAndHigherThanStartSeqNumN sn rest++isMonoIncreasingAndHigherThanStartSeqNumN _ [] =+ True+isMonoIncreasingAndHigherThanStartSeqNumN _ (MkStream (Start (MkFrameCtx () () sn ())) : rest) =+ isMonoIncreasingAndHigherThanStartSeqNumN sn rest+isMonoIncreasingAndHigherThanStartSeqNumN sn (MkStream (Next (MkFrame () snFrame ())) : rest)+ | snFrame >= sn = isMonoIncreasingAndHigherThanStartSeqNumN snFrame rest+ | otherwise = False++startFrame :: Word8 -> Stream () Word8 () () ()+startFrame sn = MkStream (Start (MkFrameCtx () () sn ()))++nextFrame :: Word8 -> Stream () Word8 () () ()+nextFrame sn = MkStream (Next (MkFrame () sn ()))
+ specs/Data/MediaBus/SampleSpec.hs view
@@ -0,0 +1,45 @@+module Data.MediaBus.SampleSpec ( spec ) where++import Data.MediaBus+import Control.Monad+import Test.Hspec+import Control.Lens+import Data.Char+import qualified Data.Vector.Generic.Mutable as V++spec :: Spec+spec = describe "SampleBuffer" $ do+ it "can be mapped over with eachSample" $+ (sampleBufferFromList "Hello" & eachSample %~ toUpper) `shouldBe`+ sampleBufferFromList "HELLO"+ it "can be mapped over with eachSample changing the type" $+ (sampleBufferFromList "Hello" & sampleBuffer . sampleVector .+ each %~+ const True) `shouldBe`+ sampleBufferFromList (Prelude.replicate 5 True)+ describe "mutateSamples" $+ it "modifies in-place" $+ let f v =+ -- imperative safe destructive updates+ let n = V.length v+ in+ forM_ [0 .. (n - 1) `div` 2] (\i -> V.swap v i (n - 1 - i))+ in+ mutateSamples f (sampleBufferFromList [1 .. 4 :: Int]) `shouldBe`+ sampleBufferFromList [4,3 .. 1]+ describe "unsafeMutateSamples" $+ it "modifies in-place and can return values" $+ let f v =+ -- imperative safe destructive updates+ let n = V.length v+ in+ forM [0 .. (n - 1) `div` 2]+ (\i -> do+ V.swap v i (n - 1 - i)+ return i)+ in+ unsafeMutateSamples f (sampleBufferFromList [1 .. 4 :: Int]) `shouldBe`+ ([ 0, 1 ], sampleBufferFromList [4,3 .. 1])++data TestFormat = MkTestFormat+ deriving Show
+ specs/Data/MediaBus/SequenceSpec.hs view
@@ -0,0 +1,31 @@+module Data.MediaBus.SequenceSpec ( spec ) where++import Conduit as C+import Data.Conduit.List ( consume, sourceList )+import Data.Word+import Test.Hspec+import Test.QuickCheck+import Data.MediaBus++-- -----------------------------------------------------------------------------+-- * Tests/Specs+-- -----------------------------------------------------------------------------+spec :: Spec+spec =+ describe "synchronizeToSeqNum" $+ it "produces dense, strictly monotonic output" $+ property synchronizeToSeqNumIsMonotone++synchronizeToSeqNumIsMonotone :: NonEmptyList (SeqNum ())+ -> Word64+ -> Expectation+synchronizeToSeqNumIsMonotone (NonEmpty xs) startVal = do+ let inEvents = sourceList xs :: Producer Identity (SeqNum ())+ (e0 : rest) = runConduitPure (inEvents .|+ synchronizeToSeqNum startVal .|+ consume)+ e0 `shouldBe` MkSeqNum startVal++ (rest `zip` drop 1 rest) `shouldSatisfy`+ all (not .+ uncurry succeeds)
+ specs/Data/MediaBus/StreamSegmentSpec.hs view
@@ -0,0 +1,98 @@+module Data.MediaBus.StreamSegmentSpec ( spec ) where++import Test.Hspec+import Data.Conduit.List+import Conduit+import Test.QuickCheck+import Data.MediaBus+import Data.Word+import qualified Data.Vector.Storable as V+import Control.Lens+import Data.Proxy++spec :: Spec+spec = describe "segmentC" $ do+ it "only outputs segments with a valid length" $+ property $+ \ls -> withTestData ls+ PT5+ (\_ outputs -> (Prelude.drop 1 outputs) `shouldSatisfy`+ all ((== getDuration PT5) . getDuration))+ it "only drops less than the static duration" $+ property $+ \ls -> withTestData ls+ PT5+ (\inputs outputs -> (sum (getDuration <$> inputs) -+ sum (getDuration <$> outputs)) `shouldSatisfy`+ (< getDuration PT5))+ it "maintains strictly monotonic sequence numbers" $+ property $+ \ls -> withTestData ls+ PT5+ (const (`shouldSatisfy` seqNumStrictlyMonotoneIncreasing))+ it "maintains strictly monotonic timestamps" $+ property $+ \ls -> withTestData ls+ PT5+ (const (`shouldSatisfy` ticksStrictlyMonotoneIncreasing (nominalDiffTime # getStaticDuration PT5)))++withTestData :: HasStaticDuration d+ => [TestLen]+ -> PT d+ -> ([Stream () Word8 (Ticks 8000 Word32) () (SampleBuffer (S16 8000))]+ -> [Stream () Word8 (Ticks 8000 Word32) () (Segment d (SampleBuffer (S16 8000)))]+ -> res)+ -> res+withTestData ls pt f = let inputs = mkTestInputs ls+ outputs = runSegmentC inputs pt+ in+ f inputs outputs++seqNumStrictlyMonotoneIncreasing outs =+ let res = view seqNum <$> outs+ in+ all (== 1) $ zipWith (-) (Prelude.drop 2 res) (Prelude.drop 1 res)++ticksStrictlyMonotoneIncreasing dur outs =+ let res = view timestamp' <$> outs+ in+ all (== dur) $ zipWith (-) (Prelude.drop 2 res) (Prelude.drop 1 res)++runSegmentC :: (HasStaticDuration d, HasDuration c, CanSegment c, Monoid c)+ => [Stream () Word8 (Ticks 8000 Word32) () c]+ -> PT d+ -> [Stream () Word8 (Ticks 8000 Word32) () (Segment d c)]+runSegmentC inputs _p = runConduitPure (sourceList inputs .|+ segmentC .|+ consume)++mkTestInputs :: [TestLen]+ -> [Stream () Word8 (Ticks 8000 Word32) () (SampleBuffer (S16 8000))]+mkTestInputs = reverse .+ snd .+ foldl (\((ts0, sn0), acc0) (MkTestLen len) ->+ ( (ts0 + fromIntegral len, sn0 + 1)+ , mkTestPacket sn0 ts0 len : acc0+ ))+ ((0, 0), [ mkTestStartPacket ])+ where+ mkTestPacket sn ts len =+ MkStream (Next (MkFrame ts sn (MkSampleBuffer (V.replicate len 0))))++ mkTestStartPacket = MkStream (Start (MkFrameCtx () 0 0 ()))++data PT d where+ PT0 :: PT (0 :@ 1)+ PT5 :: PT (5 :@ 1000)+ PT10 :: PT (80 :@ 8000)+ PT20 :: PT (320 :@ 16000)++instance HasStaticDuration d =>+ HasDuration (PT d) where+ getDuration _ = getStaticDuration (Proxy :: Proxy d)++newtype TestLen = MkTestLen Int+ deriving (Read, Show, Ord, Eq, Num)++instance Arbitrary TestLen where+ arbitrary = MkTestLen <$> choose (0, 1000)
+ specs/Data/MediaBus/StreamSpec.hs view
@@ -0,0 +1,60 @@+module Data.MediaBus.StreamSpec ( spec ) where++import Test.Hspec+import Test.QuickCheck+import Conduit+import Data.Conduit.List+import Data.MediaBus+import Control.Monad+import Data.Word+import Control.Lens++spec :: Spec+spec = describe "Stream conduits" $+ describe "Functor instance: fmap" $+ it "should have the same result as 'overFrameContentC (...) f'" pending++-- fmap f = overFrameContentC undefined (mapInput _framePayload (const Nothing) (const f))+_helloWorld :: IO ()+_helloWorld = void $+ runConduit (yieldMany ("Hello world" :: String) .|+ traceShowC 1 "YO" .|+ consume)++_sampleSomeStream :: IO ALittleOutOfOrder+_sampleSomeStream = MkALittleOutOfOrder <$> sample' arbitrary++_yieldStream :: Monad m+ => ALittleOutOfOrder+ -> Source m (Stream () (SeqNum Word16) () () ())+_yieldStream (MkALittleOutOfOrder frames) =+ yieldMany frames .|+ traceShowC 1 "ORIGINAL"++_reorderSomeFrames = void $+ _sampleSomeStream >>=+ (\fs -> runConduit (_yieldStream fs .|+ reorderFramesBySeqNumC 2 .|+ traceShowC 1 " ORDERED" .|+ consume))++newtype ALittleOutOfOrder =+ MkALittleOutOfOrder { fromALittleOutOfOrder :: [Stream () (SeqNum Word16) () () ()]+ }++instance Arbitrary ALittleOutOfOrder where+ arbitrary = do+ start <- (MkStream . Start) <$> arbitrary+ (len :: Int) <- arbitrary+ MkALittleOutOfOrder <$> loop (start ^. seqNum) (len + 10) [ start ]+ where+ loop (MkSeqNum lastSeq) n acc+ | n == 0 = return acc+ | otherwise = do+ nextSeq' <- choose (lastSeq + 1, lastSeq + 2)+ ts <- arbitrary+ c <- arbitrary+ let nextSeq = MkSeqNum nextSeq'+ loop nextSeq+ (n - 1)+ (MkStream (Next (MkFrame ts nextSeq c)) : acc)
+ specs/Data/MediaBus/TicksSpec.hs view
@@ -0,0 +1,38 @@+module Data.MediaBus.TicksSpec ( spec ) where++import Data.MediaBus+import Test.QuickCheck+import Test.Hspec+import Control.Lens+import Data.Word++spec :: Spec+spec = do+ describe "nominalDiffTime At8kHzU32" $+ it "is isomorphic" $+ property $+ \tix -> (view (from nominalDiffTime)+ (view nominalDiffTime (mkTicks at8kHzU32 tix)))+ `shouldBe`+ mkTicks at8kHzU32 tix+ describe "nominalDiffTime At48kHzU64" $+ it "is isomorphic" $+ property $+ \tix -> (view (from nominalDiffTime)+ (view nominalDiffTime (mkTicks at48kHzU64 tix)))+ `shouldBe`+ mkTicks at48kHzU64 tix+ describe "convertTicks between At16kHzU32 and At16kHzU32" $+ it "is idempotent" $+ property $+ \tix -> let t = mkTicks at16kHzU32 tix+ in+ convertTicks t `shouldBe` t+ describe "convertTicks between At8kHzU32 and At16kHzU64" $+ it "preserves nominalDiffTime when converting back and forth" $+ property $+ \tix -> let t = mkTicks at8kHzU32 tix+ in+ view nominalDiffTime+ (convertTicks t :: Ticks 16000 Word64) `shouldBe`+ view nominalDiffTime t
+ specs/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}
+ src/Data/MediaBus.hs view
@@ -0,0 +1,26 @@+module Data.MediaBus ( module X ) where++import Data.MediaBus.AsyncConduit as X+import Data.MediaBus.Audio.Alaw as X+import Data.MediaBus.Audio.Channels as X+import Data.MediaBus.Audio.Raw as X+import Data.MediaBus.Audio.Resample as X+import Data.MediaBus.BlankMedia as X+import Data.MediaBus.Clock as X+import Data.MediaBus.Conduit as X+import Data.MediaBus.DebugSink as X+import Data.MediaBus.Discontinous as X+import Data.MediaBus.Monotone as X+import Data.MediaBus.OrderedBy as X+import Data.MediaBus.Segment as X+import Data.MediaBus.Payload as X+import Data.MediaBus.Reorder as X+import Data.MediaBus.Sample as X+import Data.MediaBus.Sequence as X+import Data.MediaBus.Series as X+import Data.MediaBus.SourceId as X+import Data.MediaBus.Stream as X+import Data.MediaBus.StreamSegment as X+import Data.MediaBus.Ticks as X+import Data.MediaBus.Transcoder as X+import Data.MediaBus.Transport.Udp as X
+ src/Data/MediaBus/AsyncConduit.hs view
@@ -0,0 +1,119 @@+module Data.MediaBus.AsyncConduit+ ( withAsyncPolledSource+ , PayloadQ()+ , mkPayloadQ+ , payloadQSink+ , payloadQSource+ ) where++import Control.Exception ( evaluate )+import Control.Monad.State+import Data.Time.Clock+import Control.Concurrent.Async.Lifted+import Control.Concurrent.STM+import Control.Concurrent ( threadDelay )+import Control.Parallel.Strategies ( NFData, rdeepseq+ , withStrategy )+import GHC.TypeLits+import System.Random+import Conduit+import Data.MediaBus.Clock+import Data.MediaBus.Stream+import Data.MediaBus.Discontinous+import Data.MediaBus.Ticks+import Control.Lens+import Data.Default+import Debug.Trace+import Data.Proxy+import Text.Printf++data PollPayloadSourceSt s t =+ MkPollPayloadSourceSt { _ppSeqNum :: !s+ , _ppTicks :: !t+ }++makeLenses ''PollPayloadSourceSt++withAsyncPolledSource :: (MonadResource m, MonadBaseControl IO m, KnownNat r, Integral t, Integral s, Default c, Default p, HasStaticDuration c, HasDuration c, NFData c, NFData p, NFData s, NFData t, Random i, Random t, Random s, Show c)+ => Int+ -> Source m (Stream i s (Ticks r t) p c)+ -> (( Async ()+ , Source m (Stream i s (Ticks r t) p (Discontinous c))+ )+ -> m o)+ -> m o+withAsyncPolledSource !frameQueueLen !src !f = do+ !pq <- mkPayloadQ frameQueueLen+ withAsync (runConduit (src .| payloadQSink pq))+ (\a -> f (void a, payloadQSource pq))++data PayloadQ a = MkPayloadQ { _payloadQSegmentDuration :: !NominalDiffTime+ , _payloadQPollIntervall :: !NominalDiffTime+ , _payloadQRing :: !(TBQueue a)+ }++mkPayloadQ :: forall m a.+ (HasStaticDuration a, MonadBaseControl IO m)+ => Int+ -> m (PayloadQ a)+mkPayloadQ qlen = MkPayloadQ segmentDuration+ (fromIntegral qlen * 0.5 * segmentDuration) <$> liftBase (newTBQueueIO qlen)+ where+ segmentDuration = getStaticDuration (Proxy :: Proxy a)++payloadQSink :: (NFData a, MonadBaseControl IO m, Show a)+ => PayloadQ a+ -> Sink (Stream i s t p a) m ()+payloadQSink (MkPayloadQ _ _ !ringRef) =+ awaitForever go+ where+ go !x = do+ maybe (return ()) pushInRing (x ^? payload)+ return ()+ where+ pushInRing !buf' = liftBase $ do+ !buf <- evaluate $ withStrategy rdeepseq buf'+ atomically $ do+ isFull <- isFullTBQueue ringRef+ when isFull (void $ readTBQueue ringRef)+ writeTBQueue ringRef buf++payloadQSource :: (Random i, NFData c, NFData p, Default p, HasStaticDuration c, HasDuration c, MonadBaseControl IO m, KnownNat r, Integral t, Integral s, NFData t, NFData s)+ => PayloadQ c+ -> Source m (Stream i s (Ticks r t) p (Discontinous c))+payloadQSource (MkPayloadQ pTime pollIntervall ringRef) =+ evalStateC (MkPollPayloadSourceSt 0 0) $ do+ yieldStart+ go False+ where+ go wasMissing = do+ res <- liftBase $ race (atomically $ readTBQueue ringRef) sleep+ case res of+ Left buf -> yieldNextBuffer (Got buf) >> go False+ Right dt -> yieldMissing dt wasMissing >> go True++ sleep = liftBase (do+ !(t0 :: ClockTime UtcClock) <- now+ threadDelay (_ticks pollIntervallMicros)+ !t1 <- now+ return (_utcClockTimeDiff (diffTime t1 t0)))+ yieldMissing !dt !wasMissing = do+ unless wasMissing+ (traceM (printf "*** UNDERFLOW: Missing %s" (show dt)))+ replicateM_ (floor (dt / pTime)) (yieldNextBuffer Missing)+ yieldStart = (MkFrameCtx <$> liftBase randomIO+ <*> use ppTicks+ <*> use ppSeqNum+ <*> pure def) >>=+ yieldStartFrameCtx++ pollIntervallMicros :: Ticks 1000000 Int+ pollIntervallMicros = nominalDiffTime # pollIntervall+++ yieldNextBuffer !buf = do+ let !bufferDuration = nominalDiffTime # getDuration buf+ !ts <- ppTicks <<+= bufferDuration+ !sn <- ppSeqNum <<+= 1+ frm <- liftBase (evaluate (withStrategy rdeepseq $ MkFrame ts sn buf))+ yieldNextFrame frm
+ src/Data/MediaBus/Audio/Alaw.hs view
@@ -0,0 +1,131 @@+module Data.MediaBus.Audio.Alaw+ ( ALaw(..)+ , alawSample+ ) where++import Foreign.Storable+import Data.MediaBus.Stream+import Data.MediaBus.Audio.Raw+import Data.MediaBus.Audio.Channels+import Data.MediaBus.BlankMedia+import Data.MediaBus.Ticks+import Data.MediaBus.Sample+import Data.MediaBus.Transcoder+import Data.Bits+import Data.Word+import Data.Int+import Control.Lens+import Data.Proxy+import Data.Function ( on )+import Test.QuickCheck ( Arbitrary(..) )+import GHC.Generics ( Generic )+import Control.Parallel.Strategies ( NFData, rdeepseq, withStrategy )++newtype ALaw = MkALaw { _alawSample :: Word8 }+ deriving (Show, Storable, Num, Eq, Bits, Arbitrary, Generic)++instance NFData ALaw++makeLenses ''ALaw++instance Ord ALaw where+ compare = compare `on` (decodeALawSample . _alawSample)++instance HasDuration (Proxy ALaw) where+ getDuration _ = 1 / 8000+ getDurationTicks _ = convertTicks (MkTicks 1 :: Ticks 8000 Int)++instance HasChannelLayout ALaw where+ channelLayout _ = SingleChannel++instance Transcoder (SampleBuffer ALaw) (SampleBuffer (S16 8000)) where+ transcode = over (framePayload . eachSample)+ (withStrategy rdeepseq .+ MkS16 . decodeALawSample . _alawSample)++instance Transcoder (SampleBuffer (S16 8000)) (SampleBuffer ALaw) where+ transcode = over (framePayload . eachSample)+ (withStrategy rdeepseq .+ MkALaw . encodeALawSample . _s16Sample)++instance IsAudioSample ALaw where+ type GetAudioSampleRate ALaw = 8000+ type SetAudioSampleRate ALaw x = ALaw+ avgSamples !x !y = MkALaw .+ encodeALawSample .+ _s16Sample $+ (avgSamples `on` (mkS16 . decodeALawSample . _alawSample)) x y+ where+ mkS16 :: Int16 -> S16 8000+ mkS16 = MkS16+ setAudioSampleRate _ = id++instance CanBeBlank ALaw where+ blank = MkALaw 0xD5++decodeALawSample :: Word8 -> Int16+decodeALawSample !a' = let !a = a' `xor` 85+ !quant_mask = 15+ !quant_shift = 4+ !seg_mask = 112+ !seg_shift = 4+ tBase, tAbs, seg :: Int16+ !seg = (fromIntegral a .&. seg_mask) `shiftR`+ seg_shift+ !tBase = (fromIntegral a .&. quant_mask) `shiftL`+ quant_shift+ !tAbs = case seg of+ 0 -> tBase + 8+ 1 -> tBase + 264+ _ -> (tBase + 264) `shiftL`+ fromIntegral (seg - 1)+ !isPos = testBit a 7+ in+ if isPos then tAbs else tAbs * (-1)++-- | See http://opensource.apple.com//source/tcl/tcl-20/tcl_ext/snack/snack/generic/g711.c+--+-- > Linear Input Code Compressed Code+-- > ----------------- ---------------+-- > 0000000wxyza 000wxyz+-- > 0000001wxyza 001wxyz+-- > 000001wxyzab 010wxyz+-- > 00001wxyzabc 011wxyz+-- > 0001wxyzabcd 100wxyz+-- > 001wxyzabcde 101wxyz+-- > 01wxyzabcdef 110wxyz+-- > 1wxyzabcdefg 111wxyz+--+-- For further information see John C. Bellamy's Digital Telephony, 1982, John+-- Wiley & Sons, pps 98-111 and 472-476.+encodeALawSample :: Int16 -> Word8+encodeALawSample !pcmVal' =+ let !pcmVal = pcmVal' `shiftR` 3 -- to 13 bit+ (!mask, !pcmValAbs) = if pcmVal >= 0+ then ( 0xD5 -- sign (7th) bit = 1+ , pcmVal+ )+ else ( 0x55 -- sign bit = 0+ , (-1) * pcmVal - 1+ )+ -- !segments = [0x1F,0x3F,0x7F,0xFF,0x1FF,0x3FF,0x7FF,0xFFF] :: [+ !segment+ | pcmValAbs <= 31 = 0+ | pcmValAbs <= 63 = 1+ | pcmValAbs <= 127 = 2+ | pcmValAbs <= 255 = 3+ | pcmValAbs <= 511 = 4+ | pcmValAbs <= 1023 = 5+ | pcmValAbs <= 2047 = 6+ | pcmValAbs <= 4095 = 7+ | otherwise = 8+ !res = if segment == 8+ then 0x7F+ else let !segShift = if segment < 2+ then 1+ else fromIntegral segment+ in+ shiftL segment 4 .|.+ (shiftR pcmValAbs segShift .&. 0xF)+ in+ fromIntegral res `xor` mask
+ src/Data/MediaBus/Audio/Channels.hs view
@@ -0,0 +1,69 @@+module Data.MediaBus.Audio.Channels+ ( ChannelLayout(..)+ , ChannelPair(..)+ , leftSample+ , rightSample+ , HasChannelLayout(..)+ ) where++import Control.Lens+import Foreign.Storable+import Data.MediaBus.Ticks+import Data.MediaBus.BlankMedia+import Test.QuickCheck+import Data.Proxy+import GHC.Generics ( Generic )+import Control.DeepSeq++data ChannelLayout = SingleChannel | ChannelPair+ deriving (Show, Eq, Ord, Enum, Generic)++instance (CanBeBlank r) =>+ CanBeBlank (ChannelPair r) where+ blank = MkChannelPair blank blank++instance NFData ChannelLayout++class HasChannelLayout c where+ channelLayout :: c -> ChannelLayout++data ChannelPair a = MkChannelPair { _leftSample :: a+ , _rightSample :: a+ }+ deriving (Show, Eq, Ord, Generic)++instance NFData a =>+ NFData (ChannelPair a)++instance Arbitrary a =>+ Arbitrary (ChannelPair a) where+ arbitrary = MkChannelPair <$> arbitrary <*> arbitrary++makeLenses ''ChannelPair++instance HasChannelLayout a =>+ HasChannelLayout (ChannelPair a) where+ channelLayout MkChannelPair{_leftSample,_rightSample} =+ case (channelLayout _leftSample, channelLayout _rightSample) of+ (SingleChannel, SingleChannel) ->+ ChannelPair+ other -> error ("Sorry this channel layout is not supported: " +++ show other)++instance (HasDuration (Proxy a)) =>+ HasDuration (Proxy (ChannelPair a)) where+ getDuration _ = getDuration (Proxy :: Proxy a)++instance Storable s =>+ Storable (ChannelPair s) where+ sizeOf s = 2 * sizeOf s+ alignment = alignment+ peekByteOff ptr off = do+ l <- peekByteOff ptr off+ let rOffset = sizeOf l+ r <- peekByteOff ptr (rOffset + off)+ return (MkChannelPair l r)+ pokeByteOff ptr off (MkChannelPair l r) = do+ pokeByteOff ptr off l+ let rOffset = sizeOf l+ pokeByteOff ptr (off + rOffset) r
+ src/Data/MediaBus/Audio/Raw.hs view
@@ -0,0 +1,99 @@+{-# LANGUAGE TemplateHaskell #-}++module Data.MediaBus.Audio.Raw+ ( S16(..)+ , s16Sample+ , IsAudioSample(..)+ , setAudioSampleRateTo8kHz+ , setAudioSampleRateTo16kHz+ , setAudioSampleRateTo48kHz+ , doubleAudioSampleRate+ ) where++import Foreign.Storable+import Data.MediaBus.Audio.Channels+import Data.MediaBus.BlankMedia+import Data.Int+import Control.Lens+import Test.QuickCheck+import Data.Bits+import Data.Typeable+import Data.MediaBus.Ticks+import GHC.TypeLits+import Data.Function ( on )+import GHC.Generics ( Generic )+import Control.DeepSeq++newtype S16 (rate :: Nat) = MkS16 { _s16Sample :: Int16 }+ deriving (Typeable, Storable, Num, Eq, Ord, Arbitrary, Generic)++instance NFData (S16 rate)++instance KnownNat r =>+ HasDuration (Proxy (S16 r)) where+ getDuration _ = 1 / fromInteger (natVal (Proxy :: Proxy r))+ getDurationTicks _ = convertTicks (MkTicks 1 :: Ticks r Int)++instance Show (S16 r) where+ show (MkS16 x) = show x++makeLenses ''S16++class (KnownNat (GetAudioSampleRate a), SetAudioSampleRate a (GetAudioSampleRate a) ~ a, Show a, Storable a, Eq a, Ord a, Arbitrary a) =>+ IsAudioSample a where+ type GetAudioSampleRate a :: Nat+ type SetAudioSampleRate a (b :: Nat)+ avgSamples :: a -> a -> a+ setAudioSampleRate :: KnownNat r => proxy r -> a -> SetAudioSampleRate a r++instance KnownNat r =>+ IsAudioSample (S16 r) where+ type SetAudioSampleRate (S16 r) x = S16 x+ type GetAudioSampleRate (S16 r) = r+ avgSamples (MkS16 !x) (MkS16 !y) =+ MkS16 $+ if abs x < 16382 &&+ abs y < 16382+ then (x + y) `unsafeShiftR` 1+ else (x `unsafeShiftR` 1) + (y `unsafeShiftR` 1)+ setAudioSampleRate _ (MkS16 !x) =+ MkS16 x++instance CanBeBlank (S16 r) where+ blank = MkS16 0++instance (IsAudioSample a) =>+ IsAudioSample (ChannelPair a) where+ type SetAudioSampleRate (ChannelPair a) x = ChannelPair (SetAudioSampleRate a x)+ type GetAudioSampleRate (ChannelPair a) = GetAudioSampleRate a+ avgSamples !x !y = MkChannelPair ((avgSamples `on` _leftSample) x y)+ ((avgSamples `on` _rightSample) x y)+ setAudioSampleRate !p (MkChannelPair !l !r) =+ MkChannelPair (setAudioSampleRate p l) (setAudioSampleRate p r)++setAudioSampleRateTo8kHz :: (IsAudioSample x, y ~ SetAudioSampleRate x 8000)+ => x+ -> y+setAudioSampleRateTo8kHz =+ setAudioSampleRate (Proxy :: Proxy 8000)++setAudioSampleRateTo16kHz :: (IsAudioSample x, y ~ SetAudioSampleRate x 16000)+ => x+ -> y+setAudioSampleRateTo16kHz =+ setAudioSampleRate (Proxy :: Proxy 16000)++setAudioSampleRateTo48kHz :: (IsAudioSample x, y ~ SetAudioSampleRate x 48000)+ => x+ -> y+setAudioSampleRateTo48kHz =+ setAudioSampleRate (Proxy :: Proxy 48000)++doubleAudioSampleRate :: forall a b.+ (IsAudioSample a, b ~ SetAudioSampleRate a (GetAudioSampleRate a + GetAudioSampleRate a), KnownNat (GetAudioSampleRate a + GetAudioSampleRate a))+ => a+ -> b+doubleAudioSampleRate = setAudioSampleRate (Proxy :: Proxy (GetAudioSampleRate a + GetAudioSampleRate a))++instance HasChannelLayout (S16 r) where+ channelLayout _ = SingleChannel
+ src/Data/MediaBus/Audio/Resample.hs view
@@ -0,0 +1,41 @@+module Data.MediaBus.Audio.Resample ( resample8to16kHz' ) where++import Data.MediaBus.Stream+import Data.MediaBus.Sample+import Data.MediaBus.Audio.Raw+import Conduit+import qualified Data.Vector.Storable as V+import qualified Data.Vector.Storable.Mutable as M+import Control.Monad.State.Strict+import Control.Lens+import Control.Parallel.Strategies ( NFData, rdeepseq, using )++resample8to16kHz' :: (NFData s, NFData t, NFData p, NFData i, IsAudioSample sa, GetAudioSampleRate sa ~ 8000, Monad m, IsAudioSample (SetAudioSampleRate sa 16000), NFData (SetAudioSampleRate sa 16000))+ => sa+ -> Conduit (Stream i s t p (SampleBuffer sa)) m (Stream i s t p (SampleBuffer (SetAudioSampleRate sa 16000)))+resample8to16kHz' !sa = evalStateC sa (mapPayloadMC' resample)+ where+ resample !sb+ | sampleCount sb == 0 = return (MkSampleBuffer mempty)+ | otherwise = do+ !lastVal <- get+ let !lastVal' = V.last (sb ^. sampleVector)+ put lastVal'+ let !sb' = createSampleBufferFrom (interpolate lastVal) sb+ return sb'+ where+ interpolate !lastVal !vIn = do+ let !lenOut = 2 * lenIn+ !vOut <- M.new lenOut+ void $ lerpSamples vOut (doubleAudioSampleRate lastVal) 0+ return (vOut `using` rdeepseq)+ where+ !lenIn = V.length vIn+ lerpSamples !vOut !prev !i =+ when (i < lenIn) $ do+ !nextHalfRate <- V.indexM vIn i+ let !next = doubleAudioSampleRate nextHalfRate+ let !smpl = avgSamples prev next+ M.unsafeWrite vOut (2 * i) smpl+ M.unsafeWrite vOut (2 * i + 1) next+ lerpSamples vOut next (i + 1)
+ src/Data/MediaBus/BlankMedia.hs view
@@ -0,0 +1,24 @@+module Data.MediaBus.BlankMedia+ ( CanGenerateBlankMedia(..)+ , CanBeBlank(..)+ ) where++import Data.MediaBus.Ticks+import Data.MediaBus.Segment+import Data.Time.Clock+import GHC.TypeLits+import Control.Lens+import Data.Proxy++class CanGenerateBlankMedia a where+ blankFor :: NominalDiffTime -> a+ blankFor dt = blankForTicks (nominalDiffTime # dt :: Ticks 1000000000000 Integer)+ blankForTicks :: (Integral i, KnownNat r) => Ticks r i -> a+ blankForTicks ticks = blankFor (from nominalDiffTime # ticks)++class CanBeBlank a where+ blank :: a++instance (HasStaticDuration d, CanGenerateBlankMedia a) =>+ CanBeBlank (Segment d a) where+ blank = MkSegment (blankFor (getStaticDuration (Proxy :: Proxy d)))
+ src/Data/MediaBus/Clock.hs view
@@ -0,0 +1,92 @@+module Data.MediaBus.Clock+ ( IsClock(..)+ , timeSince+ , UtcClock(..)+ , useUtcClock+ , _utcClockTimeDiff+ , _utcClockTime+ , utcClockTimeDiff+ ) where++import Conduit+import Control.Lens+import Data.Default+import Data.Function ( on )+import Data.Kind+import Data.MediaBus.Monotone+import Data.Proxy+import Data.Time.Calendar+import Data.Time.Clock+import Data.Word+import Test.QuickCheck+import GHC.Generics ( Generic )+import Control.DeepSeq++-- | Clocks can generate reference times, and they can convert these to tickss. Tickss are mere integrals+class (Default (ClockTimeDiff c), Ord (ClockTimeDiff c), Eq (ClockTimeDiff c), Num (ClockTimeDiff c), Show (ClockTime c), Eq (ClockTime c), Show (ClockTimeDiff c), LocalOrd (ClockTimeDiff c)) =>+ IsClock c where+ data ClockTime c+ data ClockTimeDiff c+ type MonadClock c (m :: Type -> Type) :: Constraint+ now :: MonadClock c m => m (ClockTime c)+ timeAsTimeDiff :: ClockTime c -> ClockTimeDiff c+ diffTime :: ClockTime c -> ClockTime c -> ClockTimeDiff c+ timeAddTimeDiff :: ClockTime c -> ClockTimeDiff c -> ClockTime c++timeSince :: (IsClock c, MonadClock c m, Monad m) => ClockTime c -> m (ClockTimeDiff c)+timeSince t0 = do+ t1 <- now+ return (diffTime t1 t0)++data UtcClock = MkUtcClock+ deriving Generic++instance NFData UtcClock++useUtcClock :: Proxy UtcClock+useUtcClock = Proxy++instance IsClock UtcClock where+ newtype ClockTime UtcClock = MkUtcClockTime{_utcClockTime :: UTCTime}+ deriving (Eq, Generic)+ newtype ClockTimeDiff UtcClock = MkUtcClockTimeDiff{_utcClockTimeDiff ::+ NominalDiffTime}+ deriving (Ord, Eq, Num, Generic)+ type MonadClock UtcClock m = MonadIO m+ now = MkUtcClockTime <$> liftIO getCurrentTime+ timeAsTimeDiff (MkUtcClockTime ref) =+ MkUtcClockTimeDiff $ diffUTCTime ref $ UTCTime (toEnum 0) 0+ timeAddTimeDiff (MkUtcClockTime t) (MkUtcClockTimeDiff dt) =+ MkUtcClockTime (addUTCTime dt t)+ diffTime (MkUtcClockTime later) (MkUtcClockTime sooner) =+ MkUtcClockTimeDiff $ diffUTCTime later sooner++instance NFData (ClockTime UtcClock)++instance Show (ClockTime UtcClock) where+ show (MkUtcClockTime t) = show t++instance Show (ClockTimeDiff UtcClock) where+ show (MkUtcClockTimeDiff t) =+ "dt:" ++ show t++instance NFData (ClockTimeDiff UtcClock)++instance Default (ClockTimeDiff UtcClock) where+ def = MkUtcClockTimeDiff $ fromInteger def++instance Arbitrary (ClockTime UtcClock) where+ arbitrary = MkUtcClockTime <$> (UTCTime <$> (ModifiedJulianDay <$> arbitrary)+ <*> (fromInteger <$> arbitrary))++instance Arbitrary (ClockTimeDiff UtcClock) where+ arbitrary = MkUtcClockTimeDiff . fromInteger <$> arbitrary++utcClockTimeDiff :: Lens' (ClockTimeDiff UtcClock) NominalDiffTime+utcClockTimeDiff = lens _utcClockTimeDiff (const MkUtcClockTimeDiff)++instance LocalOrd (ClockTimeDiff UtcClock) where+ succeeds = succeeds `on` roundToSeconds+ where+ roundToSeconds = round . (/ 1000000000000) . _utcClockTimeDiff+ roundToSeconds :: ClockTimeDiff UtcClock -> Word64
+ src/Data/MediaBus/Conduit.hs view
@@ -0,0 +1,59 @@+module Data.MediaBus.Conduit+ ( annotateTypeC+ , annotateTypeCIn+ , annotateTypeCOut+ , annotateTypeSink+ , annotateTypeSource+ , exitAfterC+ , traceShowC+ , traceShowSink+ ) where++import Conduit+import Data.Conduit.List+import Control.Monad.State.Strict as State+import Debug.Trace+import System.Random++annotateTypeC :: proxy a -> Conduit a m a -> Conduit a m a+annotateTypeC _ = id++annotateTypeCIn :: proxy a -> Conduit a m b -> Conduit a m b+annotateTypeCIn _ = id++annotateTypeCOut :: proxy b -> Conduit a m b -> Conduit a m b+annotateTypeCOut _ = id++annotateTypeSource :: proxy a -> Source m a -> Source m a+annotateTypeSource _ = id++annotateTypeSink :: proxy a -> Sink a m r -> Sink a m r+annotateTypeSink _ = id++exitAfterC :: Monad m => Int -> Conduit a m a+exitAfterC 0 = return ()+exitAfterC n = await >>= maybe (return ()) (yield >=> const (exitAfterC (n - 1)))++traceShowC :: (Show a, Monad m) => Double -> String -> Conduit a m a+traceShowC probability msg =+ evalStateC (mkStdGen 100, 0 :: Integer) $+ awaitForever $+ \x -> do+ (g, omitted) <- State.get+ let (p, g') = randomR (0, 1) g+ if p < probability+ then do+ let omittedmsg = if omitted == 0+ then ""+ else " *** " +++ show omitted +++ " messages omitted"+ traceM ((if null msg then "" else msg ++ ": ") ++ show x +++ omittedmsg)+ State.put (g', 0)+ else State.put (g', omitted + 1)+ yield x++traceShowSink :: (Show a, Monad m) => Double -> String -> Consumer a m [a]+traceShowSink probability msg =+ traceShowC probability msg .| consume
+ src/Data/MediaBus/DebugSink.hs view
@@ -0,0 +1,49 @@+module Data.MediaBus.DebugSink+ ( HasDebugPlaybackSink(..)+ , streamDebugPlaybackSink+ ) where++import Data.MediaBus.Audio.Raw+import Data.MediaBus.Sample+import Data.MediaBus.Segment+import Data.MediaBus.Stream+import System.IO ( Handle, hClose )+import System.Process ( shell )+import Data.Streaming.Process ( Inherited(..), streamingProcess+ , waitForStreamingProcess )+import qualified Data.ByteString as B+import Data.Conduit+import Control.Monad.IO.Class+import Data.Proxy+import Text.Printf+import GHC.TypeLits+import Data.Default+import Control.Lens++class HasDebugPlaybackSink s t c where+ debugPlaybackSink :: MonadIO m => Sink (Frame s t c) m ()++instance HasDebugPlaybackSink s t c =>+ HasDebugPlaybackSink s t (Segment d c) where+ debugPlaybackSink = mapInput (over payload _segmentContent)+ (Just . over payload MkSegment)+ debugPlaybackSink++instance KnownNat r =>+ HasDebugPlaybackSink s t (SampleBuffer (S16 r)) where+ debugPlaybackSink = do+ let cp = shell (printf "play -r %d -b 16 -c1 -e signed-integer -t raw -"+ (natVal (Proxy :: Proxy r)))+ (!(sinH :: Handle), Inherited, Inherited, cph) <- streamingProcess cp+ awaitForever (pcmToByteString sinH)+ liftIO (hClose sinH)+ _ <- waitForStreamingProcess cph+ return ()+ where+ pcmToByteString !h (MkFrame _ _ !d) =+ liftIO (B.hPut h (byteStringFromSampleBuffer d))++streamDebugPlaybackSink :: (Default i, HasDebugPlaybackSink s t c, MonadIO m)+ => Sink (Stream i s t p c) m ()+streamDebugPlaybackSink =+ toFramesC .| debugPlaybackSink
+ src/Data/MediaBus/Discontinous.hs view
@@ -0,0 +1,48 @@+module Data.MediaBus.Discontinous+ ( Discontinous(..)+ , _Missing+ , _Got+ , concealMissing+ ) where++import Conduit+import Control.Lens+import Control.Parallel.Strategies ( NFData )+import Data.Default+import Data.MediaBus.Ticks+import Data.MediaBus.Payload+import Data.MediaBus.Stream+import GHC.Generics ( Generic )++-- TODO create a gap detection mechanism, a simple stateful conduit that knows the next timestamp+data Discontinous a = Missing+ | Got !a+ deriving (Show, Generic)++instance NFData a =>+ NFData (Discontinous a)++instance Default (Discontinous a) where+ def = Missing++makePrisms ''Discontinous++instance (HasDuration a) =>+ HasDuration (Discontinous a) where+ getDuration Missing = 0+ getDuration (Got !x) = getDuration x++instance HasPayload a =>+ HasPayload (Discontinous a) where+ type GetPayload (Discontinous a) = GetPayload a+ type SetPayload (Discontinous a) b = Discontinous (SetPayload a b)+ payload = _Got . payload++concealMissing :: (NFData c, Monad m)+ => c+ -> Conduit (Stream i s t p (Discontinous c)) m (Stream i s t p c)+concealMissing conceal =+ mapPayloadC' go+ where+ go (Got !b) = b+ go Missing = conceal-- TODO delete ??
+ src/Data/MediaBus/Monotone.hs view
@@ -0,0 +1,36 @@+module Data.MediaBus.Monotone+ ( LocalOrd(..)+ ) where++import Data.Word+import Data.Int++-- | Class of numbers that are monotone increasing (or decreasing) and have a+-- relative order, that is not necessarily transitive.+--+-- For example, for a series of 'Word8' values: @0 64 128 192 0 64 128 ...@+-- could be interpreted as a monotone series of consecutive increasing values,+-- that wrap around after 255. But note that the 'Ord' instance is not+-- sufficient to express that @0@ is __after__ @192@, since @0 < 192@.+class LocalOrd a where+ succeeds :: a -> a -> Bool+ default succeeds :: (Bounded a, Integral a) => a -> a -> Bool+ x `succeeds` y = (x - y) < ((maxBound - minBound) `div` 2)++instance LocalOrd Word8++instance LocalOrd Word16++instance LocalOrd Word32++instance LocalOrd Word64++instance LocalOrd Int8++instance LocalOrd Int16++instance LocalOrd Int32++instance LocalOrd Int64++instance LocalOrd Int
+ src/Data/MediaBus/OrderedBy.hs view
@@ -0,0 +1,13 @@+module Data.MediaBus.OrderedBy ( OrderedBy(..) ) where++import Data.Function (on)++data OrderedBy a = MkOrderedBy { orderedByCompare :: a -> a -> Ordering+ , orderedByValue :: a+ }++instance Eq (OrderedBy a) where+ l == r = EQ == ((orderedByCompare l `on` orderedByValue) l r)++instance Ord (OrderedBy a) where+ l `compare` r = (orderedByCompare l `on` orderedByValue) l r
+ src/Data/MediaBus/Payload.hs view
@@ -0,0 +1,9 @@+module Data.MediaBus.Payload ( HasPayload(..) ) where++import Control.Lens++class (SetPayload a (GetPayload a) ~ a) =>+ HasPayload a where+ type GetPayload a+ type SetPayload a b+ payload :: Traversal a (SetPayload a b) (GetPayload a) b
+ src/Data/MediaBus/Reorder.hs view
@@ -0,0 +1,102 @@+{-# OPTIONS -Wno-unused-top-binds #-}+module Data.MediaBus.Reorder+ ( reorderFramesBySeqNumC+ , reorderFramesByC+ ) where++import Data.MediaBus.Stream+import Data.MediaBus.Sequence+import Data.MediaBus.OrderedBy+import Data.MediaBus.Series+import qualified Data.Set as Set+import Conduit+import Control.Lens+import Control.Monad.State.Strict+import Data.Function ( on )+import Data.Default++data ReorderSt a b c = MkReorderSt { _expectedRank :: !a+ , _frameQueue :: !(Set.Set (OrderedBy b))+ , _frameDrops :: !Int+ , _lastFrameCtx :: !c+ }++makeLenses ''ReorderSt++reorderFramesBySeqNumC :: (Default s, Default i, Default t, Default p, Num s, Ord s, Monad m)+ => Int+ -> Conduit (Stream i s t p c) m (Stream i s t p c)+reorderFramesBySeqNumC =+ reorderFramesByC seqNum (+ 1)++reorderFramesByC :: (Monad m, Ord rank, Default i, Default t, Default s, Default p, Default rank)+ => Lens' (Stream i s t p c) rank+ -> (rank -> rank)+ -> Int+ -> Conduit (Stream i s t p c) m (Stream i s t p c)+reorderFramesByC !frameRank !getNextRank !maxQueueLen =+ evalStateC (MkReorderSt def Set.empty 0 def) go+ where+ maxDrops = maxQueueLen+ go = do+ awaitForever handleNext+ flushQueue+ where+ handleNext s@(MkStream (Start !ctx)) = do+ flushQueue+ yield s+ put (MkReorderSt (s ^. frameRank) Set.empty 0 ctx)++ handleNext !frm = do+ !expRank <- use expectedRank+ let !currRank = frm ^. frameRank+ case compare currRank expRank of+ EQ -> do+ yieldNext frm+ maybeYieldNextFromQueue+ LT -> do+ -- drop the frame, it lacks behind+ framesDropped <- frameDrops <+= 1+ when (framesDropped == maxDrops) $ do+ flushQueue+ -- yield a new Start frame+ ctx <- use lastFrameCtx+ let start = MkStream (Start ctx) & frameRank .~+ (frm ^. frameRank)+ MkStream (Start ctx') = start+ lastFrameCtx .= ctx'+ yield start+ yieldNext frm++ GT -> do+ frameQueue %= Set.insert (MkOrderedBy rankCmp frm)+ maybeYieldNextFromQueue++ yieldNext !frm = do+ expectedRank .= frm ^. frameRank+ updateExpectedRank+ frameDrops .= 0+ yield frm++ flushQueue = do+ !q <- frameQueue <<.= Set.empty+ mapM_ (yieldNext . orderedByValue) (Set.toAscList q)++ maybeYieldNextFromQueue = do+ !q <- use frameQueue+ !expRank <- use expectedRank+ case Set.minView q of+ Nothing -> return ()+ Just (MkOrderedBy _ !candidate, !q') ->+ let !currRank = candidate ^. frameRank+ !isQueueFull = Set.size q == maxQueueLen+ !isNextInQueue = currRank <= expRank+ in+ when (isQueueFull || isNextInQueue) $ do+ frameQueue .= q'+ yieldNext candidate+ maybeYieldNextFromQueue++ updateExpectedRank = expectedRank %= getNextRank++ rankCmp = compare `on` view frameRank
+ src/Data/MediaBus/Sample.hs view
@@ -0,0 +1,158 @@+{-# LANGUAGE UndecidableInstances #-}++module Data.MediaBus.Sample+ ( SampleBuffer(..)+ , sampleBufferFromByteString+ , byteStringFromSampleBuffer+ , sampleBufferToList+ , sampleBufferFromList+ , sampleVector+ , createSampleBufferFrom+ , HasSampleBuffer(..)+ , type GetSampleBuffer+ , mutateSamples+ , unsafeMutateSamples+ ) where++import Control.Lens+import qualified Data.Vector.Storable as SV+import Data.Vector.Storable.Mutable as M+import Control.Monad.ST ( ST, runST )+import GHC.Exts ( IsList(..) )+import Data.Typeable+import Data.MediaBus.BlankMedia+import Data.MediaBus.Segment+import Data.MediaBus.Ticks+import qualified Data.ByteString as B+import qualified Data.Vector.Storable.ByteString as Spool+import Data.Default+import GHC.Generics ( Generic )+import Control.DeepSeq++-- | A sample is a discrete value of a continuous signal, periodically sampled+-- at the sampling frequency. This is a full buffer of those things.+newtype SampleBuffer sampleType =+ MkSampleBuffer { _sampleVector :: SV.Vector sampleType }+ deriving (Eq, Monoid, Generic)++instance NFData sampleType =>+ NFData (SampleBuffer sampleType)++instance (SV.Storable sampleType, Typeable sampleType, Show sampleType) =>+ Show (SampleBuffer sampleType) where+ show (MkSampleBuffer sampleVec) =+ let l = SV.length sampleVec+ sampleTypeRep = typeRep (Proxy :: Proxy sampleType)+ samples = SV.toList sampleVec+ in+ "SampleBuffer: " +++ show l +++ " × " +++ show sampleTypeRep +++ if l > 10 then "" else " " ++ show samples++makeLenses ''SampleBuffer++instance (CanBeBlank sa, SV.Storable sa, HasDuration (Proxy sa)) =>+ CanGenerateBlankMedia (SampleBuffer sa) where+ blankFor !dur = let !sampleDuration = getDuration (Proxy :: Proxy sa)+ !samples = ceiling (dur / sampleDuration)+ !blankSample = blank+ in+ MkSampleBuffer (SV.replicate samples blankSample)++instance SV.Storable sampleType =>+ Default (SampleBuffer sampleType) where+ def = mempty++instance (HasDuration (Proxy sampleType), SV.Storable sampleType) =>+ HasDuration (SampleBuffer sampleType) where+ getDuration sb = let sampleDur = getDuration (Proxy :: Proxy sampleType)+ in+ sampleDur * fromIntegral (sampleCount sb)+ getDurationTicks sb = getDurationTicks (Proxy :: Proxy sampleType) *+ fromIntegral (sampleCount sb)++instance (SV.Storable a, HasDuration (Proxy a)) =>+ CanSegment (SampleBuffer a) where+ splitAfterDuration proxy buf@(MkSampleBuffer !bufV)+ | getDuration buf >= tPacket =+ let (!nextPacket, !rest) = SV.splitAt n bufV++ in+ Just ( MkSegment (MkSampleBuffer (SV.force nextPacket))+ , MkSampleBuffer rest+ )+ | otherwise = Nothing+ where+ !tPacket = getStaticDuration proxy+ !n = ceiling (tPacket / tSample)+ !tSample = getDuration (Proxy :: Proxy a)++instance SV.Storable s =>+ IsList (SampleBuffer s) where+ type Item (SampleBuffer s) = s+ fromList = sampleBufferFromList+ toList = sampleBufferToList++sampleBufferFromByteString :: SV.Storable a => B.ByteString -> SampleBuffer a+sampleBufferFromByteString =+ MkSampleBuffer . Spool.byteStringToVector++byteStringFromSampleBuffer :: SV.Storable a => SampleBuffer a -> B.ByteString+byteStringFromSampleBuffer =+ Spool.vectorToByteString . _sampleVector++sampleBufferToList :: SV.Storable s => SampleBuffer s -> [s]+sampleBufferToList = SV.toList . _sampleVector++sampleBufferFromList :: SV.Storable s => [s] -> SampleBuffer s+sampleBufferFromList = MkSampleBuffer . SV.fromList++createSampleBufferFrom :: (SV.Storable sample')+ => (forall s.+ SV.Vector sample -> ST s (MVector s sample'))+ -> SampleBuffer sample+ -> SampleBuffer sample'+createSampleBufferFrom f =+ over sampleVector (\ !v -> SV.create (f v))++-- | A type class for media formats, like encodings, sample rate, etc...+class (SV.Storable (GetSampleType s), SetSampleType s (GetSampleType s) ~ s) =>+ HasSampleBuffer s where+ type SetSampleType s t+ type GetSampleType s+ sampleCount :: s -> Int+ eachSample :: SV.Storable t+ => Traversal s (SetSampleType s t) (GetSampleType s) t+ sampleBuffer :: SV.Storable t+ => Lens s (SetSampleType s t) (SampleBuffer (GetSampleType s)) (SampleBuffer t)++type GetSampleBuffer s = SampleBuffer (GetSampleType s)++instance SV.Storable a =>+ HasSampleBuffer (SampleBuffer a) where+ type GetSampleType (SampleBuffer a) = a+ type SetSampleType (SampleBuffer a) t = SampleBuffer t+ sampleCount = SV.length . _sampleVector+ eachSample = sampleBuffer . sampleVector . each+ sampleBuffer = lens id (flip const)++mutateSamples :: SV.Storable a+ => (forall s. M.MVector s a -> ST s ())+ -> SampleBuffer a+ -> SampleBuffer a+mutateSamples f (MkSampleBuffer v) =+ MkSampleBuffer (SV.modify f v)++-- | Unsafe because results can be returned, which might contain the /thawn/ vector.+unsafeMutateSamples :: SV.Storable a+ => (forall s. M.MVector s a -> ST s r)+ -> SampleBuffer a+ -> (r, SampleBuffer a)+unsafeMutateSamples f (MkSampleBuffer !v) =+ runST $ do+ !mv <- SV.unsafeThaw v+ !r <- f mv+ !v' <- SV.unsafeFreeze mv+ return (r, MkSampleBuffer v')
+ src/Data/MediaBus/Segment.hs view
@@ -0,0 +1,49 @@+{-# LANGUAGE UndecidableInstances #-}++module Data.MediaBus.Segment+ ( CanSegment(..)+ , Segment(..)+ , segmentContent+ ) where++import Data.MediaBus.Ticks+import Control.Lens+import Data.Default+import Data.Proxy+import Test.QuickCheck+import Control.DeepSeq+import Text.Printf++-- | A segment is some content with a fixed (type level) duration.+newtype Segment (duration :: StaticTicks) c = MkSegment { _segmentContent :: c }+ deriving (NFData, Default, Arbitrary, Functor, Eq)++instance (HasStaticDuration d, Show c) =>+ Show (Segment d c) where+ show (MkSegment c) = printf "[| %s |%10s]"+ (show c)+ (show (getStaticDuration (Proxy :: Proxy d)))++makeLenses ''Segment++instance KnownStaticTicks d =>+ HasStaticDuration (Segment d x) where+ type SetStaticDuration (Segment d x) pt = Segment pt x+ type GetStaticDuration (Segment d x) = d++instance HasStaticDuration d =>+ HasDuration (Segment d x) where+ getDuration _ = getStaticDuration (Proxy :: Proxy d)++-- | Class of types that support splitting of from the front a packet containing+-- roughly a certain duration.+class CanSegment a where+ -- | Try to split the packet into the a part which has at most the given+ -- duration and a rest. If not possible, e.g. because the input data is+ -- already shorter than the given duration, return `Nothing`.+ splitAfterDuration :: (HasStaticDuration d)+ => proxy d+ -> a+ -> Maybe (Segment d a, a)-- TODO make the repacketization create ONLY valid sized packets, even if that means dropping content+ -- TODO allow repacketization to combine the packets+ -- TODO use 'Segment' to automatically derive the ptime
+ src/Data/MediaBus/Sequence.hs view
@@ -0,0 +1,94 @@+module Data.MediaBus.Sequence+ ( SeqNum(..)+ , type SeqNum8+ , type SeqNum16+ , type SeqNum32+ , type SeqNum64+ , HasSeqNumT(..)+ , HasSeqNum(..)+ , fromSeqNum+ , synchronizeToSeqNum+ ) where++import Test.QuickCheck ( Arbitrary(..) )+import Conduit+import Data.MediaBus.Monotone+import Data.MediaBus.Series+import Control.Lens+import Control.Monad.State.Strict+import Data.Default+import Text.Printf+import GHC.Generics ( Generic )+import Control.DeepSeq+import System.Random+import Data.Word++class SetSeqNum t (GetSeqNum t) ~ t =>+ HasSeqNumT t where+ type GetSeqNum t+ type SetSeqNum t s++class HasSeqNumT t =>+ HasSeqNum t where+ seqNum :: Lens t (SetSeqNum t s) (GetSeqNum t) s++instance (HasSeqNumT a, HasSeqNumT b, GetSeqNum a ~ GetSeqNum b) =>+ HasSeqNumT (Series a b) where+ type GetSeqNum (Series a b) = GetSeqNum a+ type SetSeqNum (Series a b) t = Series (SetSeqNum a t) (SetSeqNum b t)++instance (HasSeqNum a, HasSeqNum b, GetSeqNum a ~ GetSeqNum b) =>+ HasSeqNum (Series a b) where+ seqNum f (Start !a) = Start <$> seqNum f a+ seqNum f (Next !b) = Next <$> seqNum f b++newtype SeqNum s = MkSeqNum { _fromSeqNum :: s }+ deriving (Num, Eq, Bounded, Enum, LocalOrd, Arbitrary, Default, Generic, Random)++type SeqNum8 = SeqNum Word8++type SeqNum16 = SeqNum Word16++type SeqNum32 = SeqNum Word32++type SeqNum64 = SeqNum Word64++instance NFData s =>+ NFData (SeqNum s)++makeLenses ''SeqNum++instance HasSeqNumT (SeqNum s) where+ type GetSeqNum (SeqNum s) = s+ type SetSeqNum (SeqNum s) s' = SeqNum s'++instance HasSeqNum (SeqNum s) where+ seqNum = fromSeqNum++instance Show s =>+ Show (SeqNum s) where+ show (MkSeqNum s) = printf "SEQNUM: %10s" (show s)++instance (Eq a, LocalOrd a) =>+ Ord (SeqNum a) where+ compare !x !y+ | x == y = EQ+ | x `succeeds` y = GT+ | otherwise = LT++deriving instance (Real a, Num a, Eq a, LocalOrd a) => Real+ (SeqNum a)++deriving instance (Integral a, Enum a, Real a, Eq a, LocalOrd a) =>+ Integral (SeqNum a)++synchronizeToSeqNum :: (HasSeqNum a, Monad m, Integral i)+ => i+ -> Conduit a m (SetSeqNum a i)+synchronizeToSeqNum startSeq =+ evalStateC startSeq (awaitForever yieldSeq)+ where+ yieldSeq !a = do+ !nextSeq <- get+ modify (+ 1)+ yield (a & seqNum .~ nextSeq)
+ src/Data/MediaBus/Series.hs view
@@ -0,0 +1,135 @@+module Data.MediaBus.Series+ ( Series(..)+ , type Series'+ , _Next+ , _Start+ , AsSeries(..)+ , AsSeriesStart(..)+ , AsSeriesNext(..)+ , StartingFrom(..)+ , startingFromValue+ , toNextsC'+ , toStartsC'+ , monotoneSeriesC+ ) where++import Control.Lens+import Conduit+import Control.Monad.Reader+import Test.QuickCheck+import Data.Bifunctor+import GHC.Generics ( Generic )+import Control.DeepSeq++class (SetSeriesStart s (GetSeriesStart s) ~ s) =>+ AsSeriesStart s where+ type GetSeriesStart s+ type SetSeriesStart s t+ seriesStart :: Prism s (SetSeriesStart s n) (GetSeriesStart s) n++class (SetSeriesNext s (GetSeriesNext s) ~ s) =>+ AsSeriesNext s where+ type GetSeriesNext s+ type SetSeriesNext s t+ seriesNext :: Prism s (SetSeriesNext s n) (GetSeriesNext s) n++class AsSeries s a b | s -> a, s -> b where+ seriesStart' :: Prism' s a+ seriesNext' :: Prism' s b++instance AsSeries (Either a b) a b where+ seriesStart' = _Left+ seriesNext' = _Right++instance AsSeriesStart (Either a b) where+ type GetSeriesStart (Either a b) = a+ type SetSeriesStart (Either a b) n = (Either n b)+ seriesStart = _Left++instance AsSeriesNext (Either a b) where+ type GetSeriesNext (Either a b) = b+ type SetSeriesNext (Either a b) n = (Either a n)+ seriesNext = _Right++data Series a b = Next { _seriesValue :: !b }+ | Start { _seriesStartValue :: !a }+ deriving (Eq, Generic)++instance (NFData a, NFData b) =>+ NFData (Series a b)++instance (Show a, Show b) =>+ Show (Series a b) where+ show (Start !x) = "(START: " ++ show x ++ ")"+ show (Next !x) = show x++instance (Ord a, Ord b) =>+ Ord (Series a b) where+ compare (Next !l) (Next !r) =+ compare l r+ compare _ _ = EQ++type Series' a = Series a a++instance (Arbitrary a, Arbitrary b) =>+ Arbitrary (Series a b) where+ arbitrary = do+ isNext <- choose (0.0, 1.0)+ if isNext < (0.95 :: Double)+ then Next <$> arbitrary+ else Start <$> arbitrary++makePrisms ''Series++instance AsSeries (Series a b) a b where+ seriesNext' = _Next+ seriesStart' = _Start++instance AsSeriesNext (Series a b) where+ type GetSeriesNext (Series a b) = b+ type SetSeriesNext (Series a b) n = (Series a n)+ seriesNext = _Next++instance AsSeriesStart (Series a b) where+ type GetSeriesStart (Series a b) = a+ type SetSeriesStart (Series a b) n = (Series n b)+ seriesStart = _Start++instance Functor (Series a) where+ fmap = over _Next++instance Bifunctor Series where+ first = over _Start+ second = over _Next++newtype StartingFrom a = MkStartingFrom { _startingFromValue :: a }+ deriving (Eq, Ord, Arbitrary)++makeLenses ''StartingFrom++instance Show a =>+ Show (StartingFrom a) where+ show (MkStartingFrom !x) =+ "(STARTING-FROM: " ++ show x ++ ")"++toNextsC' :: Monad m => Conduit (Series a b) m b+toNextsC' = awaitForever go+ where+ go (Start !_a) = return ()+ go (Next !b) = yield b++toStartsC' :: Monad m => Conduit (Series a b) m a+toStartsC' = awaitForever go+ where+ go (Start !a) = yield a+ go (Next !_b) = return ()++monotoneSeriesC :: Monad m => m a -> (i -> m b) -> Conduit i m (Series a b)+monotoneSeriesC !initSeries !continueSeries = do+ !rStart <- lift initSeries+ yield (Start rStart)+ !mi <- await+ mapM_ (lift . continueSeries >=>+ yield . Next >=>+ const (awaitForever (lift . continueSeries >=> yield . Next)))+ mi
+ src/Data/MediaBus/SourceId.hs view
@@ -0,0 +1,37 @@+module Data.MediaBus.SourceId+ ( HasSourceIdT(..)+ , SourceId(..)+ , type SrcId32+ , type SrcId64+ , sourceId+ ) where++import Control.Lens+import Test.QuickCheck+import Data.Default+import GHC.Generics ( Generic )+import Control.DeepSeq+import Data.Word++class SetSourceId a (GetSourceId a) ~ a =>+ HasSourceIdT a where+ type GetSourceId a+ type SetSourceId a b++-- | Things that can be uniquely identified by a looking at a (much simpler)+-- representation, the 'identity'.+newtype SourceId i = MkSourceId { _sourceId :: i }+ deriving (Eq, Arbitrary, Default, Ord, Generic)++type SrcId32 = SourceId Word32++type SrcId64 = SourceId Word64++instance (NFData i) =>+ NFData (SourceId i)++makeLenses ''SourceId++instance Show i =>+ Show (SourceId i) where+ show (MkSourceId x) = "SOURCE-ID: " ++ show x
+ src/Data/MediaBus/Stream.hs view
@@ -0,0 +1,302 @@+module Data.MediaBus.Stream+ ( HasPayload(..)+ , FrameCtx(..)+ , frameCtxSourceId+ , frameCtxSeqNumRef+ , frameCtxTimestampRef+ , frameCtxInit+ , Frame(..)+ , frameSeqNum+ , frameTimestamp+ , framePayload+ , Stream(..)+ , stream+ , yieldStreamish+ , yieldStreamish'+ , yieldNextFrame+ , yieldNextFrame'+ , yieldStartFrameCtx+ , yieldStartFrameCtx'+ , toFramesC+ , overStreamC+ , mapFramesC+ , mapFramesC'+ , mapSeqNumC+ , mapTicksC+ , mapTicksC'+ , mapPayloadMC+ , mapPayloadMC'+ , mapPayloadC'+ , convertTicksC'+ , foldStream+ , foldStreamM+ , concatStreamContents+ ) where++import Conduit+import Control.Monad+import Control.Lens+import Data.MediaBus.Sequence+import Data.MediaBus.Payload+import Data.MediaBus.Ticks+import Data.MediaBus.Series+import Control.Monad.Writer.Strict ( tell )+import Data.Maybe+import Test.QuickCheck+import Data.Default+import Text.Printf+import GHC.TypeLits+import GHC.Generics ( Generic )+import Control.Parallel.Strategies ( NFData, rdeepseq, withStrategy )++-- TODO remove Ticks' or create Ticks32/64/...+data FrameCtx i s t p = MkFrameCtx { _frameCtxSourceId :: !i+ , _frameCtxTimestampRef :: !t+ , _frameCtxSeqNumRef :: !s+ , _frameCtxInit :: !p+ }+ deriving (Eq, Ord, Generic)++instance (NFData i, NFData s, NFData t, NFData p) =>+ NFData (FrameCtx i s t p)++makeLenses ''FrameCtx++instance HasTimestampT (FrameCtx i s t p) where+ type GetTimestamp (FrameCtx i s t p) = t+ type SetTimestamp (FrameCtx i s t p) t' = (FrameCtx i s t' p)++instance HasTimestamp (FrameCtx i s t p) where+ timestamp = frameCtxTimestampRef++instance HasSeqNumT (FrameCtx i s t p) where+ type GetSeqNum (FrameCtx i s t p) = s+ type SetSeqNum (FrameCtx i s t p) x = FrameCtx i x t p++instance HasDuration (FrameCtx i s t p) where+ getDuration _ = 0++instance HasSeqNum (FrameCtx i s t p) where+ seqNum = frameCtxSeqNumRef++instance (Arbitrary i, Arbitrary s, Arbitrary t, Arbitrary p) =>+ Arbitrary (FrameCtx i s t p) where+ arbitrary = MkFrameCtx <$> arbitrary+ <*> arbitrary+ <*> arbitrary+ <*> arbitrary++instance (Default i, Default s, Default t, Default p) =>+ Default (FrameCtx i s t p) where+ def = MkFrameCtx def def def def++instance (Show i, Show s, Show t, Show p) =>+ Show (FrameCtx i s t p) where+ show (MkFrameCtx sid tsr snr spr) =+ printf "FRAME-CTX: %15s | %15s | %15s | %s"+ (show sid)+ (show snr)+ (show tsr)+ (show spr)++-- | A 'Frame' can be anything that has a start time and is exactly one time+-- unit long, it can respresent anything ranging from an audio buffer with 20ms+-- of audio to a single pulse coded audio sample, of course it could also be a+-- video frame or a chat message.+data Frame s t c = MkFrame { _frameTimestamp :: !t+ , _frameSeqNum :: !s+ , _framePayload :: !c+ }+ deriving (Eq, Ord, Generic)++instance (NFData c, NFData s, NFData t) =>+ NFData (Frame s t c)++deriving instance Functor (Frame s t)++makeLenses ''Frame++instance HasPayload (Frame s t c) where+ type GetPayload (Frame s t c) = c+ type SetPayload (Frame s t c) d = Frame s t d+ payload = framePayload++instance HasTimestampT (Frame s t c) where+ type GetTimestamp (Frame s t c) = t+ type SetTimestamp (Frame s t c) t' = Frame s t' c++instance HasTimestamp (Frame s t c) where+ timestamp = frameTimestamp++instance HasSeqNumT (Frame s t c) where+ type GetSeqNum (Frame s t c) = s+ type SetSeqNum (Frame s t c) x = Frame x t c++instance HasSeqNum (Frame s t c) where+ seqNum = frameSeqNum++instance HasDuration c =>+ HasDuration (Frame s t c) where+ getDuration = getDuration . _framePayload++instance (Arbitrary c, Arbitrary s, Arbitrary t) =>+ Arbitrary (Frame s t c) where+ arbitrary = MkFrame <$> arbitrary <*> arbitrary <*> arbitrary++instance (Default s, Default t, Default c) =>+ Default (Frame s t c) where+ def = MkFrame def def def++instance (Show s, Show t, Show v) =>+ Show (Frame s t v) where+ show (MkFrame ts sn v) =+ printf "FRAME: %15s | %15s | %s" (show sn) (show ts) (show v)++newtype Stream i s t p c = MkStream { _stream :: Streamish i s t p c }+ deriving (Ord, Eq, Arbitrary, Generic)++instance (NFData i, NFData s, NFData t, NFData c, NFData p) =>+ NFData (Stream i s t p c)++type Streamish i s t p c = Series (FrameCtx i s t p) (Frame s t c)++makeLenses ''Stream++instance HasPayload (Stream i s t p c) where+ type GetPayload (Stream i s t p c) = c+ type SetPayload (Stream i s t p c) d = Stream i s t p d+ payload = stream . _Next . payload++instance HasDuration c =>+ HasDuration (Stream i s t p c) where+ getDuration = maybe 0 getDuration . preview (stream . _Next)++instance HasSeqNumT (Stream i s t p c) where+ type GetSeqNum (Stream i s t p c) = s+ type SetSeqNum (Stream i s t p c) x = Stream i x t p c++instance HasSeqNum (Stream i s t p c) where+ seqNum = stream . seqNum++instance HasTimestampT (Stream i s t p c) where+ type GetTimestamp (Stream i s t p c) = t+ type SetTimestamp (Stream i s t p c) t' = Stream i s t' p c++instance HasTimestamp (Stream i s t p c) where+ timestamp = stream . timestamp++instance (Default c, Default s, Default t) =>+ Default (Stream i s t p c) where+ def = MkStream (Next (MkFrame def def def))++instance (Show i, Show s, Show t, Show c, Show p) =>+ Show (Stream i s t p c) where+ show (MkStream s) = show s++yieldStreamish :: Monad m+ => Streamish i s t p c+ -> Conduit a m (Stream i s t p c)+yieldStreamish = yield . MkStream++yieldStreamish' :: (NFData i, NFData s, NFData t, NFData c, NFData p, Monad m)+ => Streamish i s t p c+ -> Conduit a m (Stream i s t p c)+yieldStreamish' = yield . withStrategy rdeepseq . MkStream++yieldNextFrame :: Monad m => Frame s t c -> Conduit a m (Stream i s t p c)+yieldNextFrame = yieldStreamish . Next++yieldNextFrame' :: (NFData i, NFData s, NFData t, NFData c, NFData p, Monad m)+ => Frame s t c+ -> Conduit a m (Stream i s t p c)+yieldNextFrame' = yieldStreamish' . Next++yieldStartFrameCtx :: Monad m+ => FrameCtx i s t p+ -> Conduit a m (Stream i s t p c)+yieldStartFrameCtx = yieldStreamish . Start++yieldStartFrameCtx' :: (NFData i, NFData s, NFData t, NFData c, NFData p, NFData (FrameCtx i s t p), Monad m)+ => FrameCtx i s t p+ -> Conduit a m (Stream i s t p c)+yieldStartFrameCtx' = yieldStreamish' . Start++overStreamC :: Monad m+ => Conduit (Series (FrameCtx i s t p) (Frame s t c)) m (Series (FrameCtx i' s' t' p') (Frame s' t' c'))+ -> Conduit (Stream i s t p c) m (Stream i' s' t' p' c')+overStreamC = mapInput _stream (Just . MkStream) . mapOutput MkStream++toFramesC :: Monad m => Conduit (Stream i s t p c) m (Frame s t c)+toFramesC = awaitForever go+ where+ go (MkStream (Start _)) =+ return ()+ go (MkStream (Next !frm)) =+ yield frm++mapFramesC' :: (NFData i, NFData s, NFData t, NFData c', Monad m)+ => (Frame s t c -> Frame s t c')+ -> Conduit (Stream i s t p c) m (Stream i s t p c')+mapFramesC' !f = mapC (over (stream . _Next) (withStrategy rdeepseq f))++mapFramesC :: Monad m+ => (Frame s t c -> m (Frame s t c'))+ -> Conduit (Stream i s t p c) m (Stream i s t p c')+mapFramesC !f = mapMC (mapMOf (stream . _Next) f)++mapSeqNumC :: Monad m+ => (s -> s')+ -> Conduit (Stream i s t p c) m (Stream i s' t p c)+mapSeqNumC = mapC . over seqNum++mapTicksC :: Monad m+ => (t -> t')+ -> Conduit (Stream i s t p c) m (Stream i s t' p c)+mapTicksC = mapC . over timestamp++mapTicksC' :: (NFData t, Monad m)+ => (t -> t')+ -> Conduit (Stream i s t p c) m (Stream i s t' p c)+mapTicksC' = mapC . withStrategy rdeepseq . over timestamp++mapPayloadMC :: Monad m+ => (c -> m c')+ -> Conduit (Stream i s t p c) m (Stream i s t p c')+mapPayloadMC = mapMC . mapMOf payload++mapPayloadMC' :: (NFData (Stream i s t p c'), Monad m)+ => (c -> m c')+ -> Conduit (Stream i s t p c) m (Stream i s t p c')+mapPayloadMC' !f = mapMC (mapMOf payload f >=> return . withStrategy rdeepseq)++mapPayloadC' :: (NFData c', Monad m)+ => (c -> c')+ -> Conduit (Stream i s t p c) m (Stream i s t p c')+mapPayloadC' !f = mapC (over payload (withStrategy rdeepseq . f))++convertTicksC' :: forall proxy0 proxy1 m r t r' t' i s c p.+ (NFData t, NFData t', KnownNat r, KnownNat r', Integral t, Integral t', Monad m, NFData t')+ => proxy0 '(r, t)+ -> proxy1 '(r', t')+ -> Conduit (Stream i s (Ticks r t) p c) m (Stream i s (Ticks r' t') p c)+convertTicksC' _ _ = mapTicksC' convertTicks++foldStream :: (Monoid o, Monad m)+ => (Stream i s t p c -> o)+ -> Sink (Stream i s t p c) m o+foldStream !f = execWriterC $+ awaitForever $+ tell .+ f++foldStreamM :: (Monoid o, Monad m)+ => (Stream i s t p c -> m o)+ -> Sink (Stream i s t p c) m o+foldStreamM !f = execWriterC $+ awaitForever (lift . lift . f >=> tell)++concatStreamContents :: (Monoid c, Monad m) => Sink (Stream i s t p c) m c+concatStreamContents = foldStream (fromMaybe mempty .+ (^? stream .+ _Next .+ framePayload))
+ src/Data/MediaBus/StreamSegment.hs view
@@ -0,0 +1,59 @@+module Data.MediaBus.StreamSegment+ ( segmentC+ , segmentC'+ ) where++import Conduit+import Data.Monoid+import Data.MediaBus.Stream+import Data.MediaBus.Ticks+import Data.MediaBus.Segment+import Control.Lens+import Data.Default+import Data.MediaBus.Series+import Data.Proxy+import GHC.TypeLits++-- | The packetizer recombines incoming packets into 'Segment's of the given+-- size. The sequence numbers will be offsetted by the number extra frames+-- generated.+segmentC :: (Num s, Monad m, HasDuration c, CanSegment c, Monoid c, Default i, KnownNat r, Integral t, HasStaticDuration d)+ => Conduit (Stream i s (Ticks r t) p c) m (Stream i s (Ticks r t) p (Segment d c))+segmentC = segmentC' Proxy++segmentC' :: (Num s, Monad m, HasDuration c, CanSegment c, Monoid c, Default i, KnownNat r, Integral t, HasStaticDuration d)+ => proxy d+ -> Conduit (Stream i s (Ticks r t) p c) m (Stream i s (Ticks r t) p (Segment d c))+segmentC' dpx = evalStateC (0, Nothing) $ awaitForever go+ where+ segmentDurationInTicks =+ nominalDiffTime # segmentDuration+ segmentDuration = getStaticDuration dpx+ go (MkStream (Next (MkFrame !t !s !cIn))) = do+ !cRest <- _2 <<.= Nothing+ let tsOffset = negate (getDurationTicks cRest)+ !cRest' <- yieldLoop (maybe cIn (<> cIn) cRest) tsOffset+ _2 .= cRest'+ where+ yieldLoop !c !timeOffset =+ if getDuration c == segmentDuration+ then do+ yieldWithAdaptedSeqNumAndTimestamp (MkSegment c)+ return Nothing+ else case splitAfterDuration dpx c of+ Just (!packet, !rest) -> do+ yieldWithAdaptedSeqNumAndTimestamp packet+ _1 += 1+ yieldLoop rest (timeOffset + segmentDurationInTicks)+ Nothing -> do+ -- we just swallowed an incoming packet, therefore we need+ -- to decrease the seqnums+ _1 -=+ 1+ return (Just c)+ where+ yieldWithAdaptedSeqNumAndTimestamp !p = do+ !seqNumOffset <- use _1+ yieldNextFrame (MkFrame (t + timeOffset) (s + seqNumOffset) p)+ go (MkStream (Start !frmCtx)) =+ yieldStartFrameCtx frmCtx
+ src/Data/MediaBus/Ticks.hs view
@@ -0,0 +1,231 @@+module Data.MediaBus.Ticks+ ( HasDuration(..)+ , HasTimestampT(..)+ , HasTimestamp(..)+ , Ticks(..)+ , type Ticks32+ , type Ticks64+ , type Ticks32At8000+ , type Ticks32At16000+ , type Ticks32At48000+ , type Ticks64At8000+ , type Ticks64At16000+ , type Ticks64At48000+ , mkTicks+ , at8kHzU32+ , at16kHzU32+ , at48kHzU32+ , at16kHzU64+ , at48kHzU64+ , nominalDiffTime+ , convertTicks+ , deriveFrameTimestamp+ , StaticTicks(..)+ , KnownStaticTicks(..)+ , Rate(..)+ , type (:@)+ , type StaticTicksRate+ , type StaticTicksTicks+ , HasStaticDuration(..)+ , toStaticDurationProxy+ , getStaticDuration+ , getStaticTicks+ , getStaticRate+ , ticksFromStaticDuration+ , rateFromStaticDuration+ ) where++import Conduit+import Control.Lens+import Control.Monad.State.Strict+import Data.Default+import Data.MediaBus.Monotone+import Data.MediaBus.Series+import Data.Proxy+import Data.Time.Clock+import Data.Word+import GHC.TypeLits+import Test.QuickCheck+import GHC.Generics ( Generic )+import Control.DeepSeq+import System.Random+import Text.Printf++newtype Ticks (rate :: Nat) w = MkTicks { _ticks :: w }+ deriving (Eq, Real, Integral, Enum, LocalOrd, Num, Arbitrary, Default, Generic, Random)++type Ticks32 r = Ticks r Word32+type Ticks64 r = Ticks r Word64++type Ticks32At8000 = Ticks32 8000+type Ticks32At16000 = Ticks32 16000+type Ticks32At48000 = Ticks32 48000++type Ticks64At8000 = Ticks64 8000+type Ticks64At16000 = Ticks64 16000+type Ticks64At48000 = Ticks64 48000++instance NFData w =>+ NFData (Ticks rate w)++mkTicks :: forall proxy rate baseType.+ proxy '(rate, baseType)+ -> baseType+ -> Ticks rate baseType+mkTicks _ = MkTicks++at8kHzU32 :: Proxy '(8000, Word32)+at8kHzU32 = Proxy++at16kHzU32 :: Proxy '(16000, Word32)+at16kHzU32 = Proxy++at48kHzU32 :: Proxy '(48000, Word32)+at48kHzU32 = Proxy++at16kHzU64 :: Proxy '(16000, Word64)+at16kHzU64 = Proxy++at48kHzU64 :: Proxy '(48000, Word64)+at48kHzU64 = Proxy++convertTicks :: (Integral w, Integral w', KnownNat r, KnownNat r')+ => Ticks r w+ -> Ticks r' w'+convertTicks = view (from nominalDiffTime) . view nominalDiffTime++nominalDiffTime :: forall r w.+ (Integral w, KnownNat r)+ => Iso' (Ticks r w) NominalDiffTime+nominalDiffTime = iso (toNDT . _ticks) (MkTicks . fromNDT)+ where+ toNDT = (/ rate) . fromIntegral+ fromNDT = round . (* rate)+ rate = fromInteger $ natVal (Proxy :: Proxy r)++instance (KnownNat r, Integral w, Show w) =>+ Show (Ticks r w) where+ show tix@(MkTicks x) =+ printf "%10s (%10d @ %10d Hz)"+ (show (view nominalDiffTime tix))+ (toInteger x)+ (natVal (Proxy :: Proxy r))++instance (Eq w, LocalOrd w) =>+ Ord (Ticks rate w) where+ (<=) = flip succeeds++-- | Types with a duration (e.g. audio samples).+class HasDuration a where+ getDuration :: a -> NominalDiffTime+ getDuration !x = from nominalDiffTime #+ (getDurationTicks x :: Ticks 1000000000000 Integer)+ getDurationTicks :: (Integral i, KnownNat r) => a -> Ticks r i+ getDurationTicks !x = nominalDiffTime # getDuration x++instance HasDuration a => HasDuration (Maybe a) where+ getDuration Nothing = 0+ getDuration (Just !a) = getDuration a+ getDurationTicks Nothing = 0+ getDurationTicks (Just !a) = getDurationTicks a++-- TODO rename *Timestamp to *Tick+class SetTimestamp t (GetTimestamp t) ~ t =>+ HasTimestampT t where+ type GetTimestamp t+ type SetTimestamp t s++class HasTimestampT t -- TODO inline HasTimestampT again+ =>+ HasTimestamp t where+ timestamp :: Lens t (SetTimestamp t s) (GetTimestamp t) s+ timestamp' :: Lens' t (GetTimestamp t)+ timestamp' = timestamp++instance (HasTimestampT a, HasTimestampT b, GetTimestamp a ~ GetTimestamp b) =>+ HasTimestampT (Series a b) where+ type GetTimestamp (Series a b) = GetTimestamp a+ type SetTimestamp (Series a b) t = Series (SetTimestamp a t) (SetTimestamp b t)++instance (HasTimestamp a, HasTimestamp b, GetTimestamp a ~ GetTimestamp b) =>+ HasTimestamp (Series a b) where+ timestamp f (Start a) = Start <$> timestamp f a+ timestamp f (Next b) = Next <$> timestamp f b++-- * Media Data Synchronization+deriveFrameTimestamp :: (Monad m, KnownNat r, Integral t, HasDuration a, HasTimestamp a)+ => Ticks r t+ -> Conduit a m (SetTimestamp a (Ticks r t))+deriveFrameTimestamp t0 =+ evalStateC t0 (awaitForever yieldSync)+ where+ yieldSync sb = do+ t <- get+ modify (+ (nominalDiffTime # getDuration sb))+ yield (sb & timestamp .~ t)++data StaticTicks where+ MkStaticTicks :: Nat -> Rate -> StaticTicks++data Rate = MkRate Nat++type ticks :@ rate = 'MkStaticTicks ticks ('MkRate rate)++instance (KnownNat r, KnownNat t) =>+ HasStaticDuration (t :@ r) where+ type SetStaticDuration (t :@ r) (t' :@ r') = t' :@ r'+ type GetStaticDuration (t :@ r) = t :@ r++type family StaticTicksRate (s :: StaticTicks) :: Nat where+ StaticTicksRate (t :@ r) = r++type family StaticTicksTicks (s :: StaticTicks) :: Nat where+ StaticTicksTicks (t :@ r) = t++class KnownStaticTicks (s :: StaticTicks) where+ staticTicksVal :: KnownNat r => proxy s -> Ticks r Integer++instance (KnownNat d, KnownNat r) => KnownStaticTicks ('MkStaticTicks d ('MkRate r)) where+ staticTicksVal _ =+ convertTicks (MkTicks (natVal (Proxy :: Proxy d)) :: Ticks r Integer)++class (KnownStaticTicks (GetStaticDuration s), SetStaticDuration s (GetStaticDuration s) ~ s) =>+ HasStaticDuration (s :: k) where+ type SetStaticDuration s (pt :: StaticTicks) :: k'+ type GetStaticDuration s :: StaticTicks++toStaticDurationProxy :: (HasStaticDuration s)+ => proxy s+ -> Proxy (GetStaticDuration s)+toStaticDurationProxy _ = Proxy++getStaticDuration :: forall proxy s.+ HasStaticDuration s+ => proxy s+ -> NominalDiffTime+getStaticDuration px = from nominalDiffTime #+ (staticTicksVal (toStaticDurationProxy px) :: Ticks 1000000000000 Integer)++getStaticTicks :: forall proxy s r t i.+ (KnownNat r, KnownNat t, HasStaticDuration s, GetStaticDuration s ~ (t :@ r), Integral i)+ => proxy s+ -> Ticks r i+getStaticTicks px = ticksFromStaticDuration (toStaticDurationProxy px)++getStaticRate :: forall proxy s r t.+ (KnownNat r, KnownNat t, HasStaticDuration s, GetStaticDuration s ~ (t :@ r))+ => proxy s+ -> Integer+getStaticRate px = rateFromStaticDuration (toStaticDurationProxy px)++ticksFromStaticDuration :: forall proxy rate ticks i.+ (KnownNat rate, KnownNat ticks, Integral i)+ => proxy (ticks :@ rate)+ -> Ticks rate i+ticksFromStaticDuration _ = MkTicks (fromIntegral (natVal (Proxy :: Proxy ticks)))++rateFromStaticDuration :: forall proxy rate ticks.+ (KnownNat rate, KnownNat ticks)+ => proxy (ticks :@ rate)+ -> Integer+rateFromStaticDuration _ = fromIntegral (natVal (Proxy :: Proxy rate))
+ src/Data/MediaBus/Transcoder.hs view
@@ -0,0 +1,22 @@+module Data.MediaBus.Transcoder+ ( Transcoder(..)+ , transcodeStreamC'+ ) where++import Conduit+import Data.MediaBus.Stream+import Data.Kind+import Control.Parallel.Strategies ( NFData )++class Transcoder from to where+ type TranscodingSeqNum from to s :: Constraint+ type TranscodingSeqNum from to s = ()+ type TranscodingTicks from to t :: Constraint+ type TranscodingTicks from to t = ()+ transcode :: (TranscodingSeqNum from to s, TranscodingTicks from to t)+ => Frame s t from+ -> Frame s t to++transcodeStreamC' :: (NFData to, NFData p, NFData i, NFData s, NFData t, NFData p, Monad m, Transcoder from to, TranscodingSeqNum from to s, TranscodingTicks from to t)+ => Conduit (Stream i s t p from) m (Stream i s t p to)+transcodeStreamC' = mapFramesC' transcode
+ src/Data/MediaBus/Transport/Udp.hs view
@@ -0,0 +1,41 @@+module Data.MediaBus.Transport.Udp ( udpDatagramSource ) where++import Control.Lens+import Control.Monad.State.Strict+import Control.Monad.Trans.Resource+import Conduit+import Data.Conduit.Network.UDP+import Data.MediaBus.Clock+import Data.MediaBus.Stream+import Data.MediaBus.SourceId+import Data.MediaBus.Sequence+import Data.Streaming.Network+import Network.Socket ( SockAddr, close )+import qualified Data.ByteString as B+import Data.Default++-- | A UDP source that uses 'MonandResource' to make sure the socket is closed.+udpDatagramSource :: (IsClock c, MonadClock c m, MonadResource m, Num s, Default p)+ => proxy c+ -> Int+ -> HostPreference+ -> Source m (Stream (SourceId (Maybe SockAddr)) (SeqNum s) (ClockTimeDiff c) p B.ByteString)+udpDatagramSource _clk port host = do+ !t0 <- lift now+ bracketP (bindPortUDP port host) close (`sourceSocket` 1024) .|+ evalStateC (Nothing, 0, t0) (awaitForever createFrame)+ where+ createFrame m = do+ let currentSender = msgSender m+ lastSender <- _1 <<.= Just currentSender+ tNow <- lift (lift now)+ when (Just currentSender /= lastSender) $ do+ _2 .= 0+ _3 .= tNow+ yieldStartFrameCtx (MkFrameCtx (MkSourceId (Just currentSender))+ (timeAsTimeDiff tNow)+ 0+ def)+ sn <- _2 <<+= 1+ tStart <- use _3+ yieldNextFrame (MkFrame (diffTime tNow tStart) sn (msgData m))