packages feed

pipes-rt (empty) → 0.3.0

raw patch · 6 files changed

+356/−0 lines, 6 filesdep +basedep +mwc-randomdep +pipessetup-changed

Dependencies added: base, mwc-random, pipes, pipes-rt, time

Files

+ LICENSE view
@@ -0,0 +1,27 @@+Copyright (c) 2013, Greg Hale+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 the {organization} nor the names of its+  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 HOLDER 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.
+ README.md view
@@ -0,0 +1,72 @@+pipes-rt+========++[![Build Status](https://secure.travis-ci.org/ImAlsoGreg/pipes-rt.png)](http://travis-ci.org/ImAlsoGreg/pipes-rt)++A few pipes to yield values at a steady rate, as a poisson process, or according to the values themselves++For example:++```+λ: import Pipes+λ: import qualified Pipes.Prelude as PP+λ: import Pipes.RealTime+λ: import Data.Time+λ: import Data.Time.Clock ++λ: -- Pass values at 2 Hz+λ: runEffect $ each [1..4] >-> steadyCat 2 >-> PP.print+  [1/2 second pause...]+1 [1/2 second pause...]+2 [1/2 second pause...]+3 [1/2 second pause...]+4 [1/2 second pause...]++λ: -- Pass Values at 100 Hz, printing arrival times+λ: runEffect $ for (each [1..10] >-> steadyCat 100) (const $ lift (getCurrentTime >>= print))+2013-10-10 19:55:53.944484 UTC  +2013-10-10 19:55:53.954939 UTC      [          ]+2013-10-10 19:55:53.964623 UTC    [              ]+2013-10-10 19:55:53.975125 UTC  [   .. pauses ..    ]+2013-10-10 19:55:53.984759 UTC    [              ]+2013-10-10 19:55:53.994345 UTC      [          ]+2013-10-10 19:55:54.004886 UTC+2013-10-10 19:55:54.01449 UTC+2013-10-10 19:55:54.025124 UTC+2013-10-10 19:55:54.034661 UTC++λ: -- Pass values with Poisson timing, average 4 Hz, print data and arrival time+λ: runEffect $ for (each "Testing" >-> poissonCat 4) (\c -> lift (getCurrentTime >>= \t -> print (c,t)))+('T',2013-10-10 19:57:29.707621 UTC)+('e',2013-10-10 19:57:29.710815 UTC)+('s',2013-10-10 19:57:29.71766 UTC)+('t',2013-10-10 19:57:29.726371 UTC)+('i',2013-10-10 19:57:29.74401 UTC)+('n',2013-10-10 19:57:29.744338 UTC)+('g',2013-10-10 19:57:29.759882 UTC)++λ: -- Get timestamps from the data being piped+λ: import Data.Char+λ: let timeOfChar = (/ 10) . fromIntegral . (\c -> ord c - ord 'a')+λ: runEffect $ for (each "abcdwxyz" >-> relativeTimeCat timeOfChar) (\v -> lift (getCurrentTime >>= \t -> print (v,t)))+('a',2013-11-07 15:54:05.645025 UTC)   [ .. short pause .. ]+('b',2013-11-07 15:54:05.745847 UTC)   [ .. short pause .. ]+('c',2013-11-07 15:54:05.845771 UTC)   [ .. short pause .. ]+('d',2013-11-07 15:54:05.945533 UTC)   [ .. long  pause .. ]+('w',2013-11-07 15:54:07.847302 UTC)   [ .. short pause .. ]+('x',2013-11-07 15:54:07.946071 UTC)   [ .. short pause .. ]+('y',2013-11-07 15:54:08.045846 UTC)   [ .. short pause .. ]+('z',2013-11-07 15:54:08.145573 UTC)   [ .. short pause .. ]++λ: -- "Delay" the output by -2 seconds, which means+λ: -- skip ahead by 2 seconds+λ: let myPrint = (\v -> (getCurrentTime >>= \t -> print (v,t))) :: Char -> IO ()+λ: runEffect $ for (each "abcdwxyz" >-> relativeTimeCatDelayedBy timeOfChar (-0.2)) (lift . myPrint)+('c',2013-11-08 02:55:37.131626 UTC)+('d',2013-11-08 02:55:37.232347 UTC)   [ .. we discarded the early data .. ]+('w',2013-11-08 02:55:39.134008 UTC)   [ .. and jumped in immediately   .. ]  +('x',2013-11-08 02:55:39.232772 UTC)   [ .. at 'c'                      .. ]+('y',2013-11-08 02:55:39.332545 UTC)+('z',2013-11-08 02:55:39.432303 UTC)++```
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ examples/Examples.hs view
@@ -0,0 +1,78 @@+module Main where++import Pipes+import Pipes.RealTime+import Control.Monad+import Data.Time+import Control.Concurrent+import System.IO++data TestTypeA = TestValueA Double String+               deriving (Show)++relTimeOfA :: TestTypeA -> Double+relTimeOfA (TestValueA t _) = t++data TestTypeB = TestValueB UTCTime String+               deriving (Show)++timeOfB :: TestTypeB -> UTCTime+timeOfB (TestValueB t _) = t++-- Some test data that can associated relative timestamps+testDataA :: [TestTypeA]+testDataA = map (\t -> TestValueA t ("Data" ++ show t)) [0.5, 0.75..6]++-- Build some test data with associated absolute timestamps (using the t0 parameter,+-- so that the timestamps fall near the time that we run the example).+makeTestDataB :: UTCTime -> [TestTypeB]+makeTestDataB t0 = map (\(TestValueA t _) -> TestValueB (addUTCTime (doubleToDiff t) t0) "Payload") testDataA+  where doubleToDiff t =+          let tUTC =  UTCTime (ModifiedJulianDay 0) (picosecondsToDiffTime $ floor (t/1e-12))+              tZero = UTCTime (ModifiedJulianDay 0) (picosecondsToDiffTime 0)+          in diffUTCTime tUTC tZero++main :: IO ()+main = do++  hSetBuffering stdout NoBuffering++  putStrLn "\nGenerate some values at half-second intervals." >> drumRoll+  runEffect $ each [1..10 :: Int] >-> steadyCat 2 >-> printWithTime++  putStrLn "\nGenerate some values with poisson timing, 8 Hz" >> drumRoll+  runEffect $ each [1..20 :: Int] >-> poissonCat 8 >-> printWithTime+  threadDelay (truncate (1e6 :: Double))++  putStrLn "\nGenerate some values at their preferred times relative to now." >> drumRoll+  runEffect $ each testDataA >-> relativeTimeCat relTimeOfA >-> printWithTime++  putStrLn "\nSame data, delay the generator by 2 seconds." >> drumRoll+  runEffect $ each testDataA >-> relativeTimeCatDelayedBy relTimeOfA 2 >-> printWithTime++  putStrLn "\nSame data, advance the generator by 2 seconds, dropping too-early values" >> drumRoll+  runEffect $ each testDataA >-> relativeTimeCatDelayedBy relTimeOfA (-2) >-> printWithTime++  putStrLn "\nGenerate some values at their preferred absolute times." >> drumRoll+  do+    now <- getCurrentTime+    runEffect $ each (makeTestDataB now) >-> timeCat timeOfB >-> printWithTime++  putStrLn "\nSame UTC timestamped data, advance the generator by 2 seconds dropping too-early values" >> drumRoll+  do+    now <- getCurrentTime+    runEffect $ each (makeTestDataB now) >-> timeCatDelayedBy timeOfB (-2) >-> printWithTime+++printWithTime :: (Show a) => Consumer a IO r+printWithTime = forever $ do+  now <- lift getCurrentTime+  v <- await+  lift . putStrLn . unwords $ ["At time", show now, "got value", show v]++drumRoll :: IO ()+drumRoll = replicateM_ 3 (putStr "." >> threadDelay 500000) >> putStr "\n"+++  +
+ lib/Pipes/RealTime.hs view
@@ -0,0 +1,138 @@+module Pipes.RealTime (++  -- *Pipes throttled by their own timestamps+  timeCat,+  timeCatDelayedBy,+  relativeTimeCat,+  relativeTimeCatDelayedBy,++  -- *Pipes throttled by you+  steadyCat,+  poissonCat,+  poissonCatConst,+  genPoissonCat,+  catAtTimes,+  catAtRelativeTimes,++  ) where++import Prelude hiding (dropWhile)+import Control.Monad+import Pipes+import Pipes.Prelude (chain, dropWhile)+import Control.Concurrent (threadDelay)+import Data.Time.Clock+import Data.Time.Calendar++import System.Random.MWC+import qualified System.Random.MWC.Distributions as MWCDists++{-| Yield values some time after the effect is run,+    according to their relative timestamps.  Assumes that+    values arrive in ascending time order. Values with+    negative relative timestamps are discarded -}+relativeTimeCat :: (a -> Double) -> Pipe a a IO r+relativeTimeCat toRelTime = do+  t0 <- lift getCurrentTime+  dropWhile (( < 0 ) . toRelTime) >->+    chain (\v -> pauseUntil (doubleToNomDiffTime (toRelTime v) `addUTCTime` t0))++{-| Yield values at their timestamps, but delay+    by some time (given in seconds).  Passing+    a negative delay advances the generator,+    discarding events happening before the effect -}+relativeTimeCatDelayedBy :: (a -> Double) -> Double -> Pipe a a IO r+relativeTimeCatDelayedBy toTime delay = relativeTimeCat toTime'+     where toTime' = ((+ delay) . toTime)++{-| Yield values at the absolute times given by their timestamps.+    Assumes that they arrive in ascending time order. Values with timestamps+    earlier than the starting time of the effect are discarded -}+timeCat :: (a -> UTCTime) -> Pipe a a IO r+timeCat toTime = do+  t0 <- lift getCurrentTime+  dropWhile (( < t0 ) . toTime) >->+    chain (pauseUntil . toTime)++{-| Yield values at their absolute timesteps, but delay+    or advance their production by some time (given in+    seconds).  Values with timestamps less than zero+    after adjustment are discarded -}+timeCatDelayedBy :: (a -> UTCTime) -> Double -> Pipe a a IO r+timeCatDelayedBy toTime delay = do+  timeCat $ toTime'+  where toTime' = (doubleToNomDiffTime delay `addUTCTime`) . toTime+  +{-| Yield values at steady rate (Hz) -}+steadyCat :: Double -> Pipe a a IO r+steadyCat rate = do+  t0 <- lift getCurrentTime+  loop t0+  where+    dtUTC = doubleToNomDiffTime (1/rate)+    loop t =+      let t' = dtUTC `addUTCTime` t in do+        lift $ pauseUntil t'+        v <- await+        yield v+        loop t'++{-| Constant-rate Poisson process yielding values, randomized by IO -}+poissonCat :: Double -> Pipe a a IO r+poissonCat rate = lift createSystemRandom >>= \gen ->+  genPoissonCat gen rate++{-| Constant-rate Poisson process with a fixed seed -+    the same random every time -}+poissonCatConst :: Double -> Pipe a a IO r+poissonCatConst rate = lift create >>=  \gen ->+  genPoissonCat gen rate++{-| Constant-rate Poisson process yielding values, seeded by you -}+genPoissonCat :: GenIO -> Double -> Pipe a a IO r+genPoissonCat gen rate = do+  t0 <- lift getCurrentTime+  loop t0+  where+    loop t = do+      v <- await+      dt <- lift $ MWCDists.exponential rate gen+      let t' = addUTCTime (doubleToNomDiffTime dt) t+      lift $ pauseUntil t'+      yield v+      loop t'++{-|Yield values at a set of absolute times.+   Yield remaining values immediately if the+   time list becomes empty -}+catAtTimes :: [UTCTime] -> Pipe a a IO r+catAtTimes []     = cat+catAtTimes (t:ts) = do+  lift $ pauseUntil t+  v <- await+  yield v+  catAtTimes ts++{-|Yield values at a set of times relative to the first received value.+   Yield remaining values immediately if the time list becomes empty -}+catAtRelativeTimes :: [Double] -> Pipe a a IO r+catAtRelativeTimes []       = cat+catAtRelativeTimes ts@(_:_) = lift absTimes >>= catAtTimes +  where absTimes = +          getCurrentTime >>= \t0 ->+          return $ map (\d -> doubleToNomDiffTime d `addUTCTime` t0) ts+++pauseUntil :: UTCTime -> IO ()+pauseUntil t = do+  now <- getCurrentTime+  case compare now t of+    LT -> threadDelay (truncate (diffUTCTime t now * 1000000))+    _  -> return ()++doubleToNomDiffTime :: Double -> NominalDiffTime+doubleToNomDiffTime x =+  let d0 = ModifiedJulianDay 0+      t0 = UTCTime d0 (picosecondsToDiffTime 0)+      t1 = UTCTime d0 (picosecondsToDiffTime $ floor (x/1e-12))+  in  diffUTCTime t1 t0
+ pipes-rt.cabal view
@@ -0,0 +1,39 @@+Name:               pipes-rt+Version:            0.3.0+License:            BSD3+License-File:       LICENSE+Copyright:          (c) 2013 Greg Hale+Author:             Greg Hale+Maintainer:         Greg Hale <imalsogreg@gmail.com>+Stability:          Experimental+Homepage:           http://github.com/ImAlsoGreg/pipes-rt+Bug-Reports:        http://github.com/ImAlsoGreg/pipes-rt/issues+Category:           Control, Pipes+Build-Type:         Simple+Synopsis:           A few pipes to control the timing of yields+Cabal-Version:      >=1.8+Extra-Source-Files: README.md+Description:+  Use this library to yield values downstream according to different timing rules.  For example, use the relTimeCat pipe, and your data will be yielded at each datum's specified time since the running of the effect.  Or use poissonCat to yield values with poisson timing.++Source-Repository head+  Type:     git+  Location: git://github.com/ImAlsoGreg/pipes-rt++Library+  Hs-Source-Dirs: lib+  Build-Depends:+    base        (>= 4.2 && < 4.8)+   ,pipes       (>= 4   && < 4.1)+   ,time        (>= 1.4 && < 1.5)+   ,mwc-random  (>= 0.13 && < 0.14)+  Exposed-Modules:+    Pipes.RealTime++Executable PipesRealTimeExample+  Hs-Source-Dirs: examples+  Build-Depends:  base+                , pipes+                , pipes-rt+                , time+  Main-Is:        Examples.hs