diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2017, Manuel Bärenz + +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 Manuel Bärenz 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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,80 @@ +* README +-------- + +Rhine is a library for synchronous and asynchronous Functional Reactive Programming (FRP). +It separates the aspects of clocking, scheduling and resampling +from each other, and ensures clock-safety on the type level. + +Complex reactive programs often process data at different rates. +For example, games, GUIs and media applications +may output audio and video signals, or receive +user input at unpredictable times. +Coordinating these different rates is a hard problem in general. +If not enough care is taken, buffer underruns and overflows, space and time leaks, +accidental synchronisation of independent sub-systems, +and concurrency issues such as dead-locks may all occur. + +Rhine tackles these problems by annotating +the signal processing components with clocks, +which hold the information when data will be +input, processed and output. +Different components of the signal network +will become active at different times, or work +at different rates. If components running under different clocks need to communicate, it +has to be decided when each component be- +comes active ("scheduling"), and how data is +transferred between the different rates ("resampling"). +Rhine separates all these aspects from each +other, and from the individual signal processing of each subsystem. +It offers a flexible API to all of them and implements several +reusable standard solutions. In the places +where these aspects need to intertwine, typing +constraints on clocks come into effect, enforcing clock safety. + +A typical example, which can be run as `cabal run Demonstration`, +would be: + +``` + -- | Create a simple message containing the time stamp since program start, + -- for each tick of the clock. + -- Since 'createMessage' works for arbitrary clocks (and doesn't need further input data), + -- it is a 'Behaviour'. + -- @td@ is the 'TimeDomain' of any clock used to sample, + -- and it needs to be constrained in order for time differences + -- to have a 'Show' instance. + createMessage + :: (Monad m, Show (Diff td)) + => String + -> Behaviour m td String + createMessage str + = timeInfoOf sinceStart >-> arr show + >-> arr (("Clock " ++ str ++ " has ticked at: ") ++) + + -- | Output a message /every second/ (= every 1000 milliseconds). + -- Let us assume we want to assure that 'printEverySecond' + -- is only called every second, + -- then we constrain its type signature with the clock @Millisecond 1000@. + printEverySecond :: Show a => SyncSF IO (Millisecond 1000) a () + printEverySecond = arrMSync print + + -- | Specialise 'createMessage' to a specific clock. + ms500 :: SyncSF IO (Millisecond 500) () String + ms500 = createMessage "500 MS" + + + ms1200 :: SyncSF IO (Millisecond 1200) () String + ms1200 = createMessage "1200 MS" + + -- | Create messages every 500 ms and every 1200 ms, + -- collecting all of them in a list, + -- which is output every second. + main :: IO () + main = flow $ + ms500 @@ waitClock **@ concurrently @** ms1200 @@ waitClock + >-- collect -@- concurrently --> + printEverySecond @@ waitClock + + -- | Uncomment the following for a type error (the clocks don't match): + + -- typeError = ms500 >>> printEverySecond +``` diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/examples/Demonstration.hs b/examples/Demonstration.hs new file mode 100644 --- /dev/null +++ b/examples/Demonstration.hs @@ -0,0 +1,52 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE TypeFamilies #-} + +import FRP.Rhine +import FRP.Rhine.Clock.Realtime.Millisecond +import FRP.Rhine.Schedule.Concurrently +import FRP.Rhine.ResamplingBuffer.Collect + +-- | Create a simple message containing the time stamp since program start, +-- for each tick of the clock. +-- Since 'createMessage' works for arbitrary clocks (and doesn't need further input data), +-- it is a 'Behaviour'. +-- @td@ is the 'TimeDomain' of any clock used to sample, +-- and it needs to be constrained in order for time differences +-- to have a 'Show' instance. +createMessage + :: (Monad m, Show (Diff td)) + => String + -> Behaviour m td String +createMessage str + = timeInfoOf sinceStart >-> arr show + >-> arr (("Clock " ++ str ++ " has ticked at: ") ++) + +-- | Output a message /every second/ (= every 1000 milliseconds). +-- Let us assume we want to assure that 'printEverySecond' +-- is only called every second, +-- then we constrain its type signature with the clock @Millisecond 1000@. +printEverySecond :: Show a => SyncSF IO (Millisecond 1000) a () +printEverySecond = arrMSync print + +-- | Specialise 'createMessage' to a specific clock. +ms500 :: SyncSF IO (Millisecond 500) () String +ms500 = createMessage "500 MS" + + +ms1200 :: SyncSF IO (Millisecond 1200) () String +ms1200 = createMessage "1200 MS" + +-- | Create messages every 500 ms and every 1200 ms, +-- collecting all of them in a list, +-- which is output every second. +main :: IO () +main = flow $ + ms500 @@ waitClock **@ concurrently @** ms1200 @@ waitClock + >-- collect -@- concurrently --> + printEverySecond @@ waitClock + +-- | Uncomment the following for a type error (the clocks don't match): + +-- typeError = ms500 >>> printEverySecond diff --git a/examples/HelloWorld.hs b/examples/HelloWorld.hs new file mode 100644 --- /dev/null +++ b/examples/HelloWorld.hs @@ -0,0 +1,6 @@ +{-# LANGUAGE DataKinds #-} +import FRP.Rhine +import FRP.Rhine.Clock.Realtime.Millisecond + +main :: IO () +main = flow $ arrMSync_ (putStrLn "Hello World!") @@ (waitClock :: Millisecond 100) diff --git a/examples/test/Test.hs b/examples/test/Test.hs new file mode 100644 --- /dev/null +++ b/examples/test/Test.hs @@ -0,0 +1,38 @@ +{-# LANGUAGE Arrows #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE TypeFamilies #-} + + +-- rhine +import FRP.Rhine +import FRP.Rhine.Clock.Realtime.Millisecond +import FRP.Rhine.Schedule.Concurrently + +-- | Calculates and prints the rounding errors that accumulate +-- when calculating the time since the start of the simulation +-- via an Euler integral. +showRoundingError + :: Diff (TimeDomainOf cl) ~ Double + => String -> SyncSF IO cl () () +showRoundingError clName = proc () -> do + correct <- timeInfoOf sinceStart -< () + simulated <- arr_ 1 >>> integral -< () + liftS putStrLn -< + "Clock " ++ clName + ++ " ticks at time " ++ show correct + ++ " and simulates " ++ show simulated + ++ " => rounding error " ++ show (correct - simulated) + +-- | Show the rounding error for the 1000 milliseconds clock. +showREMS1000 :: SyncSF IO (Millisecond 1000) () () +showREMS1000 = showRoundingError "Millisecond 1000" + +-- | Show the rounding error for the 350 milliseconds clock. +showREMS350 :: SyncSF IO (Millisecond 350) () () +showREMS350 = showRoundingError "Millisecond 350" + +-- | The main program runs both synchronous signal functions in parallel, +-- using a concurrent (GHC threads) schedule. +main :: IO () +main = flow $ showREMS350 @@ waitClock **@ concurrently @** showREMS1000 @@ waitClock diff --git a/rhine.cabal b/rhine.cabal new file mode 100644 --- /dev/null +++ b/rhine.cabal @@ -0,0 +1,120 @@ +name: rhine + +version: 0.1.0.0 + +synopsis: Functional Reactive Programming with type-level clocks + +description: + Rhine is a library for synchronous and asynchronous Functional Reactive Programming (FRP). + It separates the aspects of clocking, scheduling and resampling + from each other, and ensures clock-safety on the type level. + Signal processing units can be annotated by clocks, + which hold the information when data will be + input, processed and output. + Different components of the signal network + will become active at different times, or work + at different rates. + To schedule the components and allow them to communicate, + several standard scheduling and resampling solutions are implemented. + Own schedules and resampling buffers can be implemented in a reusable fashion. + A (synchronous) program outputting "Hello World" every tenth of a second looks like this: + @flow $ arrMSync_ (putStrLn "Hello World!") @@ (waitClock :: Millisecond 100)@ + + +license: BSD3 + +license-file: LICENSE + +author: Manuel Bärenz + +maintainer: maths@manuelbaerenz.de + +category: FRP + +build-type: Simple + +extra-doc-files: README.md + +cabal-version: >=1.18 + +source-repository head + type: git + location: git@github.com:turion/rhine.git + +source-repository this + type: git + location: git@github.com:turion/rhine.git + tag: v0.1.0.0 + + +library + exposed-modules: + Control.Monad.Schedule + FRP.Rhine + FRP.Rhine.Clock + FRP.Rhine.Clock.Count + FRP.Rhine.Clock.FixedRate + FRP.Rhine.Clock.Realtime.Audio + FRP.Rhine.Clock.Realtime.Busy + FRP.Rhine.Clock.Realtime.Millisecond + FRP.Rhine.Clock.Step + FRP.Rhine.Reactimation + FRP.Rhine.Reactimation.Tick + FRP.Rhine.ResamplingBuffer + FRP.Rhine.ResamplingBuffer.Collect + FRP.Rhine.ResamplingBuffer.FIFO + FRP.Rhine.ResamplingBuffer.Interpolation + FRP.Rhine.ResamplingBuffer.KeepLast + FRP.Rhine.ResamplingBuffer.MSF + FRP.Rhine.ResamplingBuffer.Timeless + FRP.Rhine.ResamplingBuffer.Util + FRP.Rhine.Schedule + FRP.Rhine.Schedule.Concurrently + FRP.Rhine.Schedule.Trans + FRP.Rhine.SF + FRP.Rhine.SF.Combinators + FRP.Rhine.SyncSF + FRP.Rhine.TimeDomain + + -- LANGUAGE extensions used by modules in this package. + -- other-extensions: + + -- Other library packages from which modules are imported. + build-depends: base >= 4.7 && < 5 + , dunai >= 0.1.1 && < 0.2 + , transformers >= 0.5 && < 0.6 + , time >= 1.6 && < 1.7 + , free >= 4.12 && < 4.13 + , containers >= 0.5 && < 0.6 + + -- Directories containing source files. + hs-source-dirs: src + + ghc-options: -Wall + + -- Base language which the package is written in. + default-language: Haskell2010 + +executable test + hs-source-dirs: examples/test + main-is: Test.hs + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N + build-depends: base >= 4.8 && <5 + , rhine + default-language: Haskell2010 + +executable HelloWorld + hs-source-dirs: examples + main-is: HelloWorld.hs + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N + build-depends: base >= 4.8 && <5 + , rhine + default-language: Haskell2010 + +executable Demonstration + hs-source-dirs: examples + main-is: Demonstration.hs + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N + build-depends: base >= 4.8 && <5 + , rhine + default-language: Haskell2010 diff --git a/src/Control/Monad/Schedule.hs b/src/Control/Monad/Schedule.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/Schedule.hs @@ -0,0 +1,100 @@ +{-# LANGUAGE DeriveFunctor #-} +module Control.Monad.Schedule where + + +-- base +import Control.Concurrent + +-- transformers +import Control.Monad.IO.Class + +-- free +import Control.Monad.Trans.Free + + +-- TODO Implement Time via StateT + +{- | +A functor implementing a syntactical "waiting" action. + +* 'diff' represents the duration to wait. +* 'a' is the encapsulated value. +-} +data Wait diff a = Wait diff a + deriving Functor + +{- | +Values in @ScheduleT diff m@ are delayed computations with side effects in 'm'. +Delays can occur between any two side effects, with lengths specified by a 'diff' value. +These delays don't have any semantics, it can be given to them with 'runScheduleT'. +-} +type ScheduleT diff = FreeT (Wait diff) + + +-- | The side effect that waits for a specified amount. +wait :: Monad m => diff -> ScheduleT diff m () +wait diff = FreeT $ return $ Free $ Wait diff $ return () + +-- | Supply a semantic meaning to 'Wait'. +-- For every occurrence of @Wait diff@ in the @ScheduleT diff m a@ value, +-- a waiting action is executed, depending on 'diff'. +runScheduleT :: Monad m => (diff -> m ()) -> ScheduleT diff m a -> m a +runScheduleT waitAction = iterT $ \(Wait n ma) -> waitAction n >> ma + +-- | Run a 'ScheduleT' value in a 'MonadIO', +-- interpreting the times as milliseconds. +runScheduleIO + :: (MonadIO m, Integral n) + => ScheduleT n m a -> m a +runScheduleIO = runScheduleT $ liftIO . threadDelay . (* 1000) . fromIntegral + +-- TODO The definition and type signature are both a mouthful. Is there a simpler concept? +-- | Runs two values in 'ScheduleT' concurrently +-- and returns the first one that yields a value +-- (defaulting to the first argument), +-- and a continuation for the other value. +race + :: (Ord diff, Num diff, Monad m) + => ScheduleT diff m a -> ScheduleT diff m b + -> ScheduleT diff m (Either + ( a, ScheduleT diff m b) + (ScheduleT diff m a, b) + ) +race (FreeT ma) (FreeT mb) = FreeT $ do + -- Perform the side effects to find out how long each 'ScheduleT' values need to wait. + aWait <- ma + bWait <- mb + case aWait of + -- 'a' doesn't need to wait. Return immediately and leave the continuation for 'b'. + Pure a -> return $ Pure $ Left (a, FreeT $ return bWait) + -- 'a' needs to wait, so we need to inspect 'b' as well and see which one needs to wait longer. + Free (Wait aDiff aCont) -> case bWait of + -- 'b' doesn't need to wait. Return immediately and leave the continuation for 'a'. + Pure b -> return $ Pure $ Right (wait aDiff >> aCont, b) + -- Both need to wait. Which one needs to wait longer? + Free (Wait bDiff bCont) -> if aDiff <= bDiff + -- 'a' yields first, or both are done simultaneously. + then runFreeT $ do + -- Perform the wait action that we've deconstructed + wait aDiff + -- Recurse, since more wait actions might be hidden in 'a' and 'b'. 'b' doesn't need to wait as long, since we've already waited for 'aDiff'. + race aCont $ wait (bDiff - aDiff) >> bCont + -- 'b' yields first. Analogously. + else runFreeT $ do + wait bDiff + race (wait (aDiff - bDiff) >> aCont) bCont + +-- | Runs both schedules concurrently and returns their results at the end. +async + :: (Ord diff, Num diff, Monad m) + => ScheduleT diff m a -> ScheduleT diff m b + -> ScheduleT diff m (a, b) +async aSched bSched = do + ab <- race aSched bSched + case ab of + Left (a, bCont) -> do + b <- bCont + return (a, b) + Right (aCont, b) -> do + a <- aCont + return (a, b) diff --git a/src/FRP/Rhine.hs b/src/FRP/Rhine.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine.hs @@ -0,0 +1,28 @@ +{- | +This module reexports most common names and combinators you will need to work with Rhine. +It does not export specific clocks, resampling buffers or schedules, +so you will have to import those yourself, e.g. like this: + +@ +{-# LANGUAGE DataKinds #-} +import FRP.Rhine +import FRP.Rhine.Clock.Realtime.Millisecond + +main :: IO () +main = flow $ arrMSync_ (putStrLn "Hello World!") @@ (waitClock :: Millisecond 100) +@ +-} +module FRP.Rhine (module X) where + +-- dunai +import Data.MonadicStreamFunction as X + +-- rhine +import FRP.Rhine.Clock as X +import FRP.Rhine.Reactimation as X +import FRP.Rhine.ResamplingBuffer as X +import FRP.Rhine.Schedule as X +import FRP.Rhine.SF as X +import FRP.Rhine.SF.Combinators as X +import FRP.Rhine.SyncSF as X +import FRP.Rhine.TimeDomain as X diff --git a/src/FRP/Rhine/Clock.hs b/src/FRP/Rhine/Clock.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/Clock.hs @@ -0,0 +1,141 @@ +{-# LANGUAGE Arrows #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE TypeFamilies #-} +module FRP.Rhine.Clock where + +-- dunai +import Data.MonadicStreamFunction + +-- rhine +import FRP.Rhine.TimeDomain + +-- * The 'Clock' type class + +{- | +A clock creates a stream of time stamps, +possibly together with side effects in a monad 'm' +that cause the environment to wait until the specified time is reached. + +Since we want to leverage Haskell's type system to annotate signal functions by their clocks, +each clock must be an own type, 'cl'. +Different values of the same clock type should tick at the same speed, +and only differ in implementation details. +Often, clocks are singletons. +-} +class TimeDomain (TimeDomainOf cl) => Clock m cl where + -- | The time domain, i.e. type of the time stamps the clock creates. + type TimeDomainOf cl + -- | Additional information that the clock may output at each tick, + -- e.g. if a realtime promise was met, if an event occurred, + -- if one of its subclocks (if any) ticked. + type Tag cl + -- | The method that produces to a clock value a running clock, + -- i.e. an effectful stream of tagged time stamps together with an initialisation time. + startClock + :: cl -- ^ The clock value, containing e.g. settings or device parameters + -> m (MSF m () (TimeDomainOf cl, Tag cl), TimeDomainOf cl) -- ^ The stream of time stamps, and the initial time + + +-- * Auxiliary definitions and utilities + +-- | An annotated, rich time stamp. +data TimeInfo cl = TimeInfo + { -- | Time passed since the last tick + sinceTick :: Diff (TimeDomainOf cl) + -- | Time passed since the initialisation of the clock + , sinceStart :: Diff (TimeDomainOf cl) + -- | The absolute time of the current tick + , absolute :: TimeDomainOf cl + -- | The tag annotation of the current tick + , tag :: Tag cl + } + +-- | A utility that changes the tag of a 'TimeInfo'. +retag + :: (TimeDomainOf cl1 ~ TimeDomainOf cl2) + => (Tag cl1 -> Tag cl2) + -> TimeInfo cl1 -> TimeInfo cl2 +retag f TimeInfo {..} = TimeInfo { tag = f tag, .. } + + +-- | Given a clock value and an initial time, +-- generate a stream of time stamps. +genTimeInfo + :: (Monad m, Clock m cl) + => cl -> TimeDomainOf cl + -> MSF m (TimeDomainOf cl, Tag cl) (TimeInfo cl) +genTimeInfo _ initialTime = proc (absolute, tag) -> do + lastTime <- iPre initialTime -< absolute + returnA -< TimeInfo + { sinceTick = absolute `diffTime` lastTime + , sinceStart = absolute `diffTime` initialTime + , .. + } + + +-- * Certain universal building blocks to produce new clocks from given ones + +-- | Applying a morphism of time domains yields a new clock. +data RescaledClock cl td = RescaledClock + { unscaledClock :: cl + , rescale :: TimeDomainOf cl -> td + } + + +instance (Monad m, TimeDomain td, Clock m cl) + => Clock m (RescaledClock cl td) where + type TimeDomainOf (RescaledClock cl td) = td + type Tag (RescaledClock cl td) = Tag cl + startClock (RescaledClock cl f) = do + (runningClock, initTime) <- startClock cl + return + ( runningClock >>> first (arr f) + , f initTime + ) + +-- | Instead of a mere function as morphism of time domains, +-- we can transform one time domain into the other with a monadic stream function. +data RescaledClockS m cl td tag = RescaledClockS + { unscaledClockS :: cl + -- ^ The clock before the rescaling + , rescaleS :: TimeDomainOf cl + -> m (MSF m (TimeDomainOf cl, Tag cl) (td, tag), td) + -- ^ The rescaling stream function, and rescaled initial time, + -- depending on the initial time before rescaling + } + +instance (Monad m, TimeDomain td, Clock m cl) + => Clock m (RescaledClockS m cl td tag) where + type TimeDomainOf (RescaledClockS m cl td tag) = td + type Tag (RescaledClockS m cl td tag) = tag + startClock RescaledClockS {..} = do + (runningClock, initTime) <- startClock unscaledClockS + (rescaling, rescaledInitTime) <- rescaleS initTime + return + ( runningClock >>> rescaling + , rescaledInitTime + ) + + +-- | Applying a monad morphism yields a new clock. +data HoistClock m1 m2 cl = HoistClock + { hoistedClock :: cl + , monadMorphism :: forall a . m1 a -> m2 a + } + +instance (Monad m1, Monad m2, Clock m1 a) + => Clock m2 (HoistClock m1 m2 a) where + type TimeDomainOf (HoistClock m1 m2 cl) = TimeDomainOf cl + type Tag (HoistClock m1 m2 cl) = Tag cl + startClock HoistClock {..} = do + (runningClock, initialTime) <- monadMorphism $ startClock hoistedClock + let hoistMSF = liftMSFPurer + -- TODO Look out for API changes in dunai here + return + ( hoistMSF monadMorphism runningClock + , initialTime + ) diff --git a/src/FRP/Rhine/Clock/Count.hs b/src/FRP/Rhine/Clock/Count.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/Clock/Count.hs @@ -0,0 +1,16 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE TypeFamilies #-} +module FRP.Rhine.Clock.Count where + + +-- rhine +import FRP.Rhine + +-- | A singleton clock that counts the ticks. +data Count = Count -- Sesame street anyone? + +instance Monad m => Clock m Count where + type TimeDomainOf Count = Integer + type Tag Count = () + startClock _ = return (count &&& arr (const ()), 0) diff --git a/src/FRP/Rhine/Clock/FixedRate.hs b/src/FRP/Rhine/Clock/FixedRate.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/Clock/FixedRate.hs @@ -0,0 +1,21 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE TypeFamilies #-} + +module FRP.Rhine.Clock.FixedRate where + + +-- rhine +import FRP.Rhine + + +-- | A side-effect-free clock ticking at a fixed rate. +newtype FixedRate = FixedRate Double + +instance Monad m => Clock m FixedRate where + type TimeDomainOf FixedRate = Double + type Tag FixedRate = () + startClock (FixedRate timeStep) = return + ( arr (const timeStep) >>> sumS &&& arr (const ()) + , 0 + ) diff --git a/src/FRP/Rhine/Clock/Realtime/Audio.hs b/src/FRP/Rhine/Clock/Realtime/Audio.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/Clock/Realtime/Audio.hs @@ -0,0 +1,155 @@ +{-# LANGUAGE Arrows #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE KindSignatures #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE TypeFamilies #-} + +{-# OPTIONS_GHC -Wno-unticked-promoted-constructors #-} + +module FRP.Rhine.Clock.Realtime.Audio + ( AudioClock (..) + , AudioRate (..) + , PureAudioClock (..) + , pureAudioClockF + ) + where + +-- base +import GHC.Float (double2Float) +import GHC.TypeLits (Nat, natVal, KnownNat) +import Data.Time.Clock + +-- transformers? +-- TODO Delete as soon as dunai is updated +import Control.Monad.Trans.Class (lift) + + +-- dunai +import Control.Monad.Trans.MSF.Except + +-- rhine +import FRP.Rhine + +-- | Rates at which audio signals are typically sampled. +data AudioRate + = Hz44100 + | Hz48000 + | Hz96000 + +-- | Converts an 'AudioRate' to its corresponding rate as an 'Integral'. +rateToIntegral :: Integral a => AudioRate -> a +rateToIntegral Hz44100 = 44100 +rateToIntegral Hz48000 = 48000 +rateToIntegral Hz96000 = 96000 + + +-- TODO Test extensively +{- | +A clock for audio analysis and synthesis. +It internally processes samples in buffers of size 'bufferSize', +(the programmer does not have to worry about this), +at a sample rate of 'rate' +(of type 'AudioRate'). +Both these parameters are in the type signature, +so it is not possible to compose signals with different buffer sizes +or sample rates. + +After processing a buffer, the clock will wait the remaining time +until the next buffer must be processed, +using system UTC time. +The tag of the clock specifies whether the attempt to finish the last buffer in real time was successful. +A value of 'Nothing' represents success, +a value of @Just double@ represents a lag of 'double' seconds. +-} +data AudioClock (rate :: AudioRate) (bufferSize :: Nat) = AudioClock + +class AudioClockRate (rate :: AudioRate) where + theRate :: AudioClock rate bufferSize -> AudioRate + theRateIntegral :: Integral a => AudioClock rate bufferSize -> a + theRateIntegral = rateToIntegral . theRate + theRateNum :: Num a => AudioClock rate bufferSize -> a + theRateNum = fromInteger . theRateIntegral + +instance AudioClockRate Hz44100 where + theRate _ = Hz44100 + +instance AudioClockRate Hz48000 where + theRate _ = Hz48000 + +instance AudioClockRate Hz96000 where + theRate _ = Hz96000 + + +theBufferSize + :: (KnownNat bufferSize, Integral a) + => AudioClock rate bufferSize -> a +theBufferSize = fromInteger . natVal + + +instance (KnownNat bufferSize, AudioClockRate rate) => Clock IO (AudioClock rate bufferSize) where + type TimeDomainOf (AudioClock rate bufferSize) = UTCTime + type Tag (AudioClock rate bufferSize) = Maybe Double + + startClock audioClock = do + let + step = picosecondsToDiffTime -- The only sufficiently precise conversion function + $ round (10 ^ (12 :: Integer) / theRateNum audioClock :: Double) + bufferSize = theBufferSize audioClock + once f = try $ arrM (lift . f) >>> throwS -- TODO Delete once dunai is updated + once_ = once . const + + runningClock :: UTCTime -> Maybe Double -> MSF IO () (UTCTime, Maybe Double) + runningClock initialTime maybeWasLate = safely $ do + bufferFullTime <- try $ proc () -> do + n <- count -< () + let nextTime = (realToFrac step * fromIntegral (n :: Int)) `addUTCTime` initialTime + _ <- throwOn' -< (n >= bufferSize, nextTime) + returnA -< (nextTime, if n == 0 then maybeWasLate else Nothing) + currentTime <- once_ getCurrentTime + let + lateDiff = realToFrac $ currentTime `diffUTCTime` bufferFullTime + late = if lateDiff > 0 then Just lateDiff else Nothing + safe $ runningClock bufferFullTime late + initialTime <- getCurrentTime + return + ( runningClock initialTime Nothing + , initialTime + ) + +{- | +A side-effect free clock for audio synthesis and analysis. +The sample rate is given by 'rate' (of type 'AudioRate'). +Since this clock does not wait for the completion of buffers, +the producer or the consumer of the signal has the obligation to +synchronise the signal with the system clock, if realtime is desired. +Otherwise, the clock is also suitable e.g. for batch processing of audio files. +-} +data PureAudioClock (rate :: AudioRate) = PureAudioClock + +class PureAudioClockRate (rate :: AudioRate) where + thePureRate :: PureAudioClock rate -> AudioRate + thePureRateIntegral :: Integral a => PureAudioClock rate -> a + thePureRateIntegral = rateToIntegral . thePureRate + thePureRateNum :: Num a => PureAudioClock rate -> a + thePureRateNum = fromInteger . thePureRateIntegral + + +instance (Monad m, PureAudioClockRate rate) => Clock m (PureAudioClock rate) where + type TimeDomainOf (PureAudioClock rate) = Double + type Tag (PureAudioClock rate) = () + + startClock audioClock = return + ( arr (const (1 / thePureRateNum audioClock)) >>> sumS &&& arr (const ()) + , 0 + ) + + +-- | A rescaled version of 'PureAudioClock' with 'TimeDomain' 'Float'. +type PureAudioClockF (rate :: AudioRate) = RescaledClock (PureAudioClock rate) Float + +pureAudioClockF :: PureAudioClockF rate +pureAudioClockF = RescaledClock + { unscaledClock = PureAudioClock + , rescale = double2Float +} diff --git a/src/FRP/Rhine/Clock/Realtime/Busy.hs b/src/FRP/Rhine/Clock/Realtime/Busy.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/Clock/Realtime/Busy.hs @@ -0,0 +1,28 @@ +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE TypeFamilies #-} +module FRP.Rhine.Clock.Realtime.Busy where + +-- base +import Data.Time.Clock + +-- rhine +import FRP.Rhine + +{- | +A clock that ticks without waiting. +All time passed between ticks amounts to computation time, +side effects, time measurement and framework overhead. +-} +data Busy = Busy + +instance Clock IO Busy where + type TimeDomainOf Busy = UTCTime + type Tag Busy = () + + startClock _ = do + initialTime <- getCurrentTime + return + ( arrM_ getCurrentTime + &&& arr (const ()) + , initialTime + ) diff --git a/src/FRP/Rhine/Clock/Realtime/Millisecond.hs b/src/FRP/Rhine/Clock/Realtime/Millisecond.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/Clock/Realtime/Millisecond.hs @@ -0,0 +1,67 @@ +{-# LANGUAGE Arrows #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE KindSignatures #-} +{-# LANGUAGE TypeFamilies #-} +module FRP.Rhine.Clock.Realtime.Millisecond where + +-- base +import Data.Time.Clock +import Control.Concurrent (threadDelay) +import GHC.TypeLits (Nat, KnownNat) + + +-- rhine +import FRP.Rhine +import FRP.Rhine.Clock.Step + +{- | +A clock ticking every 'n' milliseconds, +in real time. +Since 'n' is in the type signature, +it is ensured that when composing two signals on a 'Millisecond' clock, +they will be driven at the same rate. + +The tag of this clock is 'Bool', +where 'True' represents successful realtime, +and 'False' a lag. +-} +type Millisecond (n :: Nat) = RescaledClockS IO (Step n) UTCTime Bool +-- TODO Consider changing the tag to Maybe Double + +-- | This clock simply sleeps 'n' milliseconds after each tick. +-- The current time is measured, but no adjustment is made. +-- Consequently, the tag is constantly 'False', +-- since the clock will accumulate the computation time as lag. +sleepClock :: KnownNat n => Millisecond n +sleepClock = sleepClock_ Step + where + sleepClock_ :: Step n -> Millisecond n + sleepClock_ cl = RescaledClockS cl $ const $ do + now <- getCurrentTime + return + ( arrM_ (threadDelay (fromInteger $ stepsize cl * 1000) >> getCurrentTime) + *** arr (const False) + , now + ) + + +-- TODO Test whether realtime detection really works here, +-- e.g. with a getLine signal +-- | A more sophisticated implementation that measures the time after each tick, +-- and waits for the remaining time until the next tick. +-- If the next tick should already have occurred, +-- the tag is set to 'False', representing a failed real time attempt. +waitClock :: KnownNat n => Millisecond n +waitClock = RescaledClockS Step $ \_ -> do + initTime <- getCurrentTime + let + runningClock = proc (n, ()) -> do + beforeSleep <- arrM_ getCurrentTime -< () + let + diff :: Double + diff = realToFrac $ beforeSleep `diffUTCTime` initTime + remaining = fromInteger $ n * 1000 - round (diff * 1000000) + _ <- arrM threadDelay -< remaining + now <- arrM_ getCurrentTime -< () -- TODO Test whether this is a performance penalty + returnA -< (now, diff > 0) + return (runningClock, initTime) diff --git a/src/FRP/Rhine/Clock/Step.hs b/src/FRP/Rhine/Clock/Step.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/Clock/Step.hs @@ -0,0 +1,63 @@ +{-# LANGUAGE Arrows #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE TypeOperators #-} +module FRP.Rhine.Clock.Step where + + +-- base +import GHC.TypeLits + + +-- rhine +import FRP.Rhine + + +-- | A pure (side effect free) clock ticking at multiples of 'n'. +-- The tick rate is in the type signature, +-- which prevents composition of signals at different rates. +data Step (n :: Nat) where + Step :: KnownNat n => Step n -- TODO Does the constraint bring any benefit? + +-- | Extract the type-level natural number as an integer. +stepsize :: Step n -> Integer +stepsize step@Step = natVal step + +instance Monad m => Clock m (Step n) where + type TimeDomainOf (Step n) = Integer + type Tag (Step n) = () + startClock cl = return + ( count >>> arr (* stepsize cl) + &&& arr (const ()) + , 0 + ) + + +-- | Two 'Step' clocks can always be scheduled without side effects. +scheduleStep + :: Monad m + => Schedule m (Step n1) (Step n2) +scheduleStep = Schedule f where + f cl1 cl2 = return (msf, 0) + where + n1 = stepsize cl1 + n2 = stepsize cl2 + msf = concatS $ proc _ -> do + k <- arr (+1) <<< count -< () + returnA -< [ (k, Left ()) | k `mod` n1 == 0 ] + ++ [ (k, Right ()) | k `mod` n2 == 0 ] + + +-- * To be ported to dunai + +-- TODO Will be in dunai +concatS :: Monad m => MSF m () [b] -> MSF m () b +concatS msf = MSF $ \_ -> tick msf [] + where + tick msf (b:bs) = return (b, MSF $ \_ -> tick msf bs) + tick msf [] = do + (bs, msf') <- unMSF msf () + tick msf' bs diff --git a/src/FRP/Rhine/Reactimation.hs b/src/FRP/Rhine/Reactimation.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/Reactimation.hs @@ -0,0 +1,79 @@ +{-# LANGUAGE GADTs #-} +{-# LANGUAGE RecordWildCards #-} +module FRP.Rhine.Reactimation where + + +-- dunai +import Data.MonadicStreamFunction + +-- rhine +import FRP.Rhine.Clock +import FRP.Rhine.Reactimation.Tick +import FRP.Rhine.Schedule +import FRP.Rhine.SF + + +{- | +An 'SF' together with a clock of matching type 'cl', +A 'Rhine' is a reactive program, possibly with open inputs and outputs. +If the input and output types 'a' and 'b' are both '()', +that is, the 'Rhine' is "closed", +then it is a standalone reactive program +that can be run with the function 'flow'. +-} +data Rhine m cl a b = Rhine + { sf :: SF m cl a b + , clock :: cl + } + + +-- * Running a Rhine + +{- | +Takes a closed 'Rhine' (with trivial input and output), +and runs it indefinitely. +All input is created, and all output is consumed by means of side effects +in a monad 'm'. + +Basic usage (synchronous case): + +@ +sensor :: SyncSF MyMonad MyClock () a +sensor = arrMSync_ produceData + +processing :: SyncSF MyMonad MyClock a b +processing = ... + +actuator :: SyncSF MyMonad MyClock b () +actuator = arrMSync consumeData + +mainSF :: SyncSF MyMonad MyClock () () +mainSF = sensor >-> processing >-> actuator + +main :: MyMonad () +main = flow $ mainSF @@ clock +@ +-} +-- TODO Can we chuck the constraints into Clock m cl? +flow + :: ( Monad m, Clock m cl + , TimeDomainOf cl ~ TimeDomainOf (Leftmost cl) + , TimeDomainOf cl ~ TimeDomainOf (Rightmost cl) + ) + => Rhine m cl () () -> m () +flow Rhine {..} = do + (runningClock, initTime) <- startClock clock + -- Run the main loop + flow' runningClock $ createTickable + (trivialResamplingBuffer clock) + sf + (trivialResamplingBuffer clock) + initTime + where + flow' runningClock tickable = do + -- Fetch the next time stamp from the stream, wait if necessary + ((now, tag), runningClock') <- unMSF runningClock () + -- Process the part of the signal network that is scheduled to run + tickable' <- tick tickable now tag + -- Loop + flow' runningClock' tickable' diff --git a/src/FRP/Rhine/Reactimation/Tick.hs b/src/FRP/Rhine/Reactimation/Tick.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/Reactimation/Tick.hs @@ -0,0 +1,241 @@ +{-# LANGUAGE GADTs #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE RecordWildCards #-} +module FRP.Rhine.Reactimation.Tick where + +-- transformers +import Control.Monad.Trans.Reader + +-- dunai +import Data.MonadicStreamFunction + +-- rhine +import FRP.Rhine.Clock +import FRP.Rhine.ResamplingBuffer +import FRP.Rhine.Schedule +import FRP.Rhine.SF +import FRP.Rhine.TimeDomain + + +{- | A signal function ('SF') enclosed by matching 'ResamplingBuffer's and further auxiliary data, +such that it can be stepped with each arriving tick from a clock 'cl'. +They play a similar role like 'ReactHandle's in dunai. + +The type parameters: + +* 'm': The monad in which the 'SF' and the 'ResamplingBuffer's produce side effects +* 'cla': The (irrelevant) input clock of the left 'ResamplingBuffer' +* 'clb': The clock at which the left 'ResamplingBuffer' produces output +* 'cl': The clock at which the 'SF' ticks +* 'clc': The clock at which the right 'ResamplingBuffer' accepts input +* 'cld': The (irrelevant) output clock of the right 'ResamplingBuffer' +* 'a': The (irrelevant) input type of the left 'ResamplingBuffer' +* 'b': The input type of the 'SF' +* 'c': The output type of the 'SF' +* 'd': The (irrelevant) output type of the right 'ResamplingBuffer' +-} +data Tickable m cla clb cl clc cld a b c d = Tickable + { -- | The left buffer from which the input is taken. + buffer1 :: ResamplingBuffer m cla clb a b + -- | The signal function that will process the data. + , ticksf :: SF m cl b c + -- | The right buffer in which the output is stored. + , buffer2 :: ResamplingBuffer m clc cld c d + -- | The leftmost clock of the signal function, 'cl', + -- may be a parallel subclock of the buffer clock. + -- 'parClockInL' specifies in which position 'Leftmost cl' + -- is a parallel subclock of 'clb'. + , parClockInL :: ParClockInclusion (Leftmost cl) clb + -- | The same on the output side. + , parClockInR :: ParClockInclusion (Rightmost cl) clc + -- | The last times when the different parts of the signal tree have ticked. + , lastTime :: LastTime cl + -- | The time when the whole clock was initialised. + , initTime :: TimeDomainOf cl + } + + +-- | Initialise the tree of last tick times. +initLastTime :: SF m cl a b -> TimeDomainOf cl -> LastTime cl +initLastTime (Synchronous _) initTime = LeafLastTime initTime +initLastTime (Sequential sf1 _ sf2) initTime = + SequentialLastTime + (initLastTime sf1 initTime) + (initLastTime sf2 initTime) +initLastTime (Parallel sf1 sf2) initTime = + ParallelLastTime + (initLastTime sf1 initTime) + (initLastTime sf2 initTime) + +-- | Initialise a 'Tickable' from a signal function, +-- two matching enclosing resampling buffers and an initial time. +createTickable + :: ResamplingBuffer m cla (Leftmost cl) a b + -> SF m cl b c + -> ResamplingBuffer m (Rightmost cl) cld c d + -> TimeDomainOf cl + -> Tickable m cla (Leftmost cl) cl (Rightmost cl) cld a b c d +createTickable buffer1 ticksf buffer2 initTime = Tickable + { parClockInL = ParClockRefl + , parClockInR = ParClockRefl + , lastTime = initLastTime ticksf initTime + , .. + } + +{- | In this function, one tick, or step of an asynchronous signal function happens. +The 'TimeInfo' holds the information which part of the signal tree will tick. +This information is encoded in the 'Tag' of the 'TimeInfo', +which is of type 'Either tag1 tag2' in case of a 'SequentialClock' or a 'ParallelClock', +encoding either a tick for the left clock or the right clock. +-} +tick :: ( Monad m, Clock m cl + , TimeDomainOf cla ~ TimeDomainOf cl + , TimeDomainOf clb ~ TimeDomainOf cl + , TimeDomainOf clc ~ TimeDomainOf cl + , TimeDomainOf cld ~ TimeDomainOf cl + , TimeDomainOf (Leftmost cl) ~ TimeDomainOf cl + , TimeDomainOf (Rightmost cl) ~ TimeDomainOf cl + ) + => Tickable m cla clb cl clc cld a b c d + -> TimeDomainOf cl -- ^ Timestamp of the present tick + -> Tag cl -- ^ 'Tag' of the overall clock; contains the information which subsystem will become active + -> m (Tickable m cla clb cl clc cld a b c d) +-- Only if we have reached a leaf of the tree, data is actually processed. +tick Tickable + { ticksf = Synchronous syncsf + , lastTime = LeafLastTime lastTime + , .. } now tag = do + let + ti = TimeInfo + { sinceTick = diffTime now lastTime + , sinceStart = diffTime now initTime + , absolute = now + , tag = tag + } + -- Get an input value from the left buffer + (b, buffer1') <- get buffer1 $ retag (parClockTagInclusion parClockInL) ti + -- Run it through the synchronous signal function + (c, syncsf') <- unMSF syncsf b `runReaderT` ti + -- Put the output into the right buffer + buffer2' <- put buffer2 (retag (parClockTagInclusion parClockInR) ti) c + return Tickable + { buffer1 = buffer1' + , ticksf = Synchronous syncsf' + , buffer2 = buffer2' + , lastTime = LeafLastTime now + , .. } +-- The left part of a sequential composition is stepped. +tick tickable@Tickable + { ticksf = Sequential sf1 bufferMiddle sf2 + , lastTime = SequentialLastTime lastTimeL lastTimeR + , initTime + , parClockInL + } now (Left tag) = do + leftTickable <- tick Tickable + { buffer1 = buffer1 tickable + , ticksf = sf1 + , buffer2 = bufferMiddle + , parClockInL = parClockInL + , parClockInR = ParClockRefl + , lastTime = lastTimeL + , initTime = initTime + } now tag + return $ tickable + { buffer1 = buffer1 leftTickable + , ticksf = Sequential (ticksf leftTickable) (buffer2 leftTickable) sf2 + , lastTime = SequentialLastTime (lastTime leftTickable) lastTimeR + } +-- The right part of a sequential composition is stepped. +tick tickable@Tickable + { ticksf = Sequential sf1 bufferMiddle sf2 + , lastTime = SequentialLastTime lastTimeL lastTimeR + , initTime + , parClockInR + } now (Right tag) = do + rightTickable <- tick Tickable + { buffer1 = bufferMiddle + , ticksf = sf2 + , buffer2 = buffer2 tickable + , parClockInL = ParClockRefl + , parClockInR = parClockInR + , lastTime = lastTimeR + , initTime = initTime + } now tag + return $ tickable + { buffer2 = buffer2 rightTickable + , ticksf = Sequential sf1 (buffer1 rightTickable) (ticksf rightTickable) + , lastTime = SequentialLastTime lastTimeL (lastTime rightTickable) + } +-- A parallel composition is stepped. +tick tickable@Tickable + { ticksf = Parallel sfA sfB + , lastTime = ParallelLastTime lastTimeA lastTimeB + , initTime + , parClockInL + , parClockInR + } now tag = case tag of + Left tagL -> do + leftTickable <- tick Tickable + { buffer1 = buffer1 tickable + , ticksf = sfA + , buffer2 = buffer2 tickable + , parClockInL = ParClockInL parClockInL + , parClockInR = ParClockInL parClockInR + , lastTime = lastTimeA + , initTime = initTime + } now tagL + return $ tickable + { buffer1 = buffer1 leftTickable + , ticksf = Parallel (ticksf leftTickable) sfB + , buffer2 = buffer2 leftTickable + , lastTime = ParallelLastTime (lastTime leftTickable) lastTimeB + } + Right tagR -> do + rightTickable <- tick Tickable + { buffer1 = buffer1 tickable + , ticksf = sfB + , buffer2 = buffer2 tickable + , parClockInL = ParClockInR parClockInL + , parClockInR = ParClockInR parClockInR + , lastTime = lastTimeB + , initTime = initTime + } now tagR + return $ tickable + { buffer1 = buffer1 rightTickable + , ticksf = Parallel sfA (ticksf rightTickable) + , buffer2 = buffer2 rightTickable + , lastTime = ParallelLastTime lastTimeA (lastTime rightTickable) + } +tick Tickable + { ticksf = Synchronous _ + , lastTime = SequentialLastTime {} + } _ _ = error "Impossible pattern in tick" +tick Tickable + { ticksf = Synchronous _ + , lastTime = ParallelLastTime {} + } _ _ = error "Impossible pattern in tick" +tick Tickable + { ticksf = Sequential {} + , lastTime = LeafLastTime _ + } _ _ = error "Impossible pattern in tick" +tick Tickable + { ticksf = Parallel {} + , lastTime = LeafLastTime _ + } _ _ = error "Impossible pattern in tick" + +-- TODO It seems wasteful to unwrap and rewrap log(N) Tickables +-- (where N is the size of the clock tree) each tick, +-- but I have no better idea. + +{- | A 'ResamplingBuffer' producing only units. +(Slightly more efficient and direct implementation than the one in 'FRP.Rhine.Timeless' +that additionally unifies the clock types in a way needed for the tick implementation.) +-} +trivialResamplingBuffer + :: Monad m => cl + -> ResamplingBuffer m (Rightmost cl) (Leftmost cl) () () +trivialResamplingBuffer _ = go + where + go = ResamplingBuffer {..} + put _ _ = return go + get _ = return ((), go) diff --git a/src/FRP/Rhine/ResamplingBuffer.hs b/src/FRP/Rhine/ResamplingBuffer.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/ResamplingBuffer.hs @@ -0,0 +1,55 @@ +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE TypeFamilies #-} +module FRP.Rhine.ResamplingBuffer where + +-- rhine +import FRP.Rhine.Clock + + +-- base +import Control.Arrow (second) + +-- A quick note on naming conventions, to whoever cares: +-- . Call a single clock cl. +-- . Call several clocks cl1, cl2 etc. in most situations. +-- . Call it cla, clb etc. when they are Leftmost or Rightmost clocks, +-- i.e. associated to particular boundary types a, b etc., + +{- | A stateful buffer from which one may 'get' a value, +or to which one may 'put' a value, +depending on the clocks. +`ResamplingBuffer`s can be clock-polymorphic, +or specific to certain clocks. + +* 'm': Monad in which the 'ResamplingBuffer' may have side effects +* 'cla': The clock at which data enters the buffer +* 'clb': The clock at which data leaves the buffer +* 'a': The input type +* 'b': The output type +-} +data ResamplingBuffer m cla clb a b = ResamplingBuffer + { put + :: TimeInfo cla + -> a + -> m ( ResamplingBuffer m cla clb a b) + -- ^ Store one input value of type 'a' at a given time stamp, + -- and return a continuation. + , get + :: TimeInfo clb + -> m (b, ResamplingBuffer m cla clb a b) + -- ^ Retrieve one output value of type 'b' at a given time stamp, + -- and a continuation. + } + + +-- | Hoist a 'ResamplingBuffer' along a monad morphism. +hoistResamplingBuffer + :: (Monad m1, Monad m2) + => (forall c. m1 c -> m2 c) + -> ResamplingBuffer m1 cla clb a b + -> ResamplingBuffer m2 cla clb a b +hoistResamplingBuffer hoist ResamplingBuffer {..} = ResamplingBuffer + { put = (((hoistResamplingBuffer hoist <$>) . hoist) .) . put + , get = (second (hoistResamplingBuffer hoist) <$>) . hoist . get + } diff --git a/src/FRP/Rhine/ResamplingBuffer/Collect.hs b/src/FRP/Rhine/ResamplingBuffer/Collect.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/ResamplingBuffer/Collect.hs @@ -0,0 +1,50 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE RecordWildCards #-} +module FRP.Rhine.ResamplingBuffer.Collect where + +-- containers +import Data.Sequence + +-- rhine +import FRP.Rhine.ResamplingBuffer +import FRP.Rhine.ResamplingBuffer.Timeless + +-- | Collects all input in a list, with the newest element at the head, +-- which is returned and emptied upon `get`. +collect :: Monad m => ResamplingBuffer m cl1 cl2 a [a] +collect = timelessResamplingBuffer AsyncMealy {..} [] + where + amPut as a = return $ a : as + amGet as = return (as, []) + + +-- | Reimplementation of 'collect' with sequences, +-- which gives a performance benefit if the sequence needs to be reversed or searched. +collectSequence :: Monad m => ResamplingBuffer m cl1 cl2 a (Seq a) +collectSequence = timelessResamplingBuffer AsyncMealy {..} empty + where + amPut as a = return $ a <| as + amGet as = return (as, empty) + +-- | 'pureBuffer' collects all input values lazily in a list +-- and processes it when output is required. +-- Semantically, @pureBuffer f == collect >>-^ arr f@, +-- but 'pureBuffer' is slightly more efficient. +pureBuffer :: Monad m => ([a] -> b) -> ResamplingBuffer m cl1 cl2 a b +pureBuffer f = timelessResamplingBuffer AsyncMealy {..} [] + where + amPut as a = return (a : as) + amGet as = return (f as, []) + +-- TODO Test whether strictness works here, or consider using deepSeq +-- | A buffer collecting all incoming values with a folding function. +-- It is strict, i.e. the state value 'b' is calculated on every 'put'. +foldBuffer + :: Monad m + => (a -> b -> b) -- ^ The folding function + -> b -- ^ The initial value + -> ResamplingBuffer m cl1 cl2 a b +foldBuffer f = timelessResamplingBuffer AsyncMealy {..} + where + amPut b a = let !b' = f a b in return b' + amGet b = return (b, b) diff --git a/src/FRP/Rhine/ResamplingBuffer/FIFO.hs b/src/FRP/Rhine/ResamplingBuffer/FIFO.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/ResamplingBuffer/FIFO.hs @@ -0,0 +1,34 @@ +{-# LANGUAGE RecordWildCards #-} +module FRP.Rhine.ResamplingBuffer.FIFO where + +-- base +import Prelude hiding (length) + +-- containers +import Data.Sequence + +-- rhine +import FRP.Rhine.ResamplingBuffer +import FRP.Rhine.ResamplingBuffer.Timeless + +-- * FIFO (first-in-first-out) buffers + +-- | An unbounded FIFO buffer. +-- If the buffer is empty, it will return 'Nothing'. +fifo :: Monad m => ResamplingBuffer m cl1 cl2 a (Maybe a) +fifo = timelessResamplingBuffer AsyncMealy {..} empty + where + amPut as a = return $ a <| as + amGet as = case viewr as of + EmptyR -> return (Nothing, empty) + as' :> a -> return (Just a , as' ) + + +-- | An unbounded FIFO buffer that also returns its current size. +fifoWatch :: Monad m => ResamplingBuffer m cl1 cl2 a (Maybe a, Int) +fifoWatch = timelessResamplingBuffer AsyncMealy {..} empty + where + amPut as a = return $ a <| as + amGet as = case viewr as of + EmptyR -> return ((Nothing, 0 ), empty) + as' :> a -> return ((Just a , length as'), as' ) diff --git a/src/FRP/Rhine/ResamplingBuffer/Interpolation.hs b/src/FRP/Rhine/ResamplingBuffer/Interpolation.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/ResamplingBuffer/Interpolation.hs @@ -0,0 +1,30 @@ +{-# LANGUAGE Arrows #-} +{-# LANGUAGE TypeFamilies #-} +module FRP.Rhine.ResamplingBuffer.Interpolation where + + +-- dunai +import Data.VectorSpace + +-- rhine +import FRP.Rhine +import FRP.Rhine.ResamplingBuffer.KeepLast +import FRP.Rhine.ResamplingBuffer.Util + +-- | A simple linear interpolation based on the last calculated position and velocity. +linear + :: ( Monad m, Clock m cl1, Clock m cl2 + , VectorSpace v + , Groundfield v ~ Diff (TimeDomainOf cl1) + , Groundfield v ~ Diff (TimeDomainOf cl2) + ) + => v -- ^ The initial velocity (derivative of the signal) + -> v -- ^ The initial position + -> ResamplingBuffer m cl1 cl2 v v +linear initVelocity initPosition + = (derivativeFrom initPosition &&& syncId) &&& timeInfoOf sinceStart + ^->> keepLast ((initVelocity, initPosition), 0) + >>-^ proc ((velocity, lastPosition), sinceStart1) -> do + sinceStart2 <- timeInfoOf sinceStart -< () + let diff = sinceStart2 - sinceStart1 + returnA -< lastPosition ^+^ velocity ^* diff diff --git a/src/FRP/Rhine/ResamplingBuffer/KeepLast.hs b/src/FRP/Rhine/ResamplingBuffer/KeepLast.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/ResamplingBuffer/KeepLast.hs @@ -0,0 +1,13 @@ +{-# LANGUAGE RecordWildCards #-} +module FRP.Rhine.ResamplingBuffer.KeepLast where + +import FRP.Rhine.ResamplingBuffer +import FRP.Rhine.ResamplingBuffer.Timeless + +-- | Always keeps the last input value, +-- or in case of no input an initialisation value. +keepLast :: Monad m => a -> ResamplingBuffer m cl1 cl2 a a +keepLast = timelessResamplingBuffer AsyncMealy {..} + where + amPut _ a = return a + amGet a = return (a, a) diff --git a/src/FRP/Rhine/ResamplingBuffer/MSF.hs b/src/FRP/Rhine/ResamplingBuffer/MSF.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/ResamplingBuffer/MSF.hs @@ -0,0 +1,32 @@ +{-# LANGUAGE RecordWildCards #-} +module FRP.Rhine.ResamplingBuffer.MSF where + +-- rhine +import FRP.Rhine + +-- | Given a monadic stream function that accepts +-- a varying number of inputs (a list), +-- a `ResamplingBuffer` can be formed +-- that collects all input in a timestamped list +msfBuffer + :: Monad m + => MSF m (TimeInfo cl2, [(TimeInfo cl1, a)]) b + -- ^ The monadic stream function that consumes + -- a single time stamp for the moment when an output value is required, + -- and a list of timestamped inputs, + -- and outputs a single value. + -- The list will contain the /newest/ element in the head. + -> ResamplingBuffer m cl1 cl2 a b +msfBuffer msf = msfBuffer' msf [] + where + msfBuffer' + :: Monad m + => MSF m (TimeInfo cl2, [(TimeInfo cl1, a)]) b + -> [(TimeInfo cl1, a)] + -> ResamplingBuffer m cl1 cl2 a b + msfBuffer' msf as = ResamplingBuffer {..} + where + put ti1 a = return $ msfBuffer' msf $ (ti1, a) : as + get ti2 = do + (b, msf') <- unMSF msf (ti2, as) + return (b, msfBuffer msf') diff --git a/src/FRP/Rhine/ResamplingBuffer/Timeless.hs b/src/FRP/Rhine/ResamplingBuffer/Timeless.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/ResamplingBuffer/Timeless.hs @@ -0,0 +1,41 @@ +{-# LANGUAGE RecordWildCards #-} +module FRP.Rhine.ResamplingBuffer.Timeless where + +import FRP.Rhine + +-- | An asynchronous, effectful Mealy machine description. +-- (Input and output do not happen simultaneously.) +-- It can be used to create 'ResamplingBuffer's. +data AsyncMealy m s a b = AsyncMealy + { amPut :: s -> a -> m s -- ^ Given the previous state and an input value, return the new state. + , amGet :: s -> m (b, s) -- ^ Given the previous state, return an output value and a new state. + } + +-- | A resampling buffer that is unaware of the time information of the clock, +-- and thus clock-polymorphic. +-- It is built from an asynchronous Mealy machine description. +-- Whenever 'get' is called on @timelessResamplingBuffer machine s@, +-- the method 'amGet' is called on @machine@ with state @s@, +-- discarding the time stamp. Analogously for 'put'. +timelessResamplingBuffer + :: Monad m + => AsyncMealy m s a b -- The asynchronous Mealy machine from which the buffer is built + -> s -- ^ The initial state + -> ResamplingBuffer m cl1 cl2 a b +timelessResamplingBuffer AsyncMealy {..} = go + where + go s = + let + put _ a = go <$> amPut s a + get _ = do + (b, s') <- amGet s + return (b, go s') + in ResamplingBuffer {..} + +-- | A resampling buffer that only accepts and emits units. +trivialResamplingBuffer :: Monad m => ResamplingBuffer m cl1 cl2 () () +trivialResamplingBuffer = timelessResamplingBuffer AsyncMealy + { amPut = const (const (return ())) + , amGet = const (return ((), ())) + } + () diff --git a/src/FRP/Rhine/ResamplingBuffer/Util.hs b/src/FRP/Rhine/ResamplingBuffer/Util.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/ResamplingBuffer/Util.hs @@ -0,0 +1,66 @@ +{-# LANGUAGE RankNTypes #-} +module FRP.Rhine.ResamplingBuffer.Util where + +-- transformers +import Control.Monad.Trans.Reader (runReaderT) + +-- rhine +import FRP.Rhine + +-- * Utilities to build 'ResamplingBuffer's from smaller components + +infix 2 >>-^ +-- | Postcompose a 'ResamplingBuffer' with a matching 'SyncSF'. +(>>-^) :: Monad m + => ResamplingBuffer m cl1 cl2 a b + -> SyncSF m cl2 b c + -> ResamplingBuffer m cl1 cl2 a c +resBuf >>-^ syncSF = ResamplingBuffer put_ get_ + where + put_ theTimeInfo a = (>>-^ syncSF) <$> put resBuf theTimeInfo a + get_ theTimeInfo = do + (b, resBuf') <- get resBuf theTimeInfo + (c, syncSF') <- unMSF syncSF b `runReaderT` theTimeInfo + return (c, resBuf' >>-^ syncSF') + + +infix 1 ^->> +-- | Precompose a 'ResamplingBuffer' with a matching 'SyncSF'. +(^->>) :: Monad m + => SyncSF m cl1 a b + -> ResamplingBuffer m cl1 cl2 b c + -> ResamplingBuffer m cl1 cl2 a c +syncSF ^->> resBuf = ResamplingBuffer put_ get_ + where + put_ theTimeInfo a = do + (b, syncSF') <- unMSF syncSF a `runReaderT` theTimeInfo + resBuf' <- put resBuf theTimeInfo b + return $ syncSF' ^->> resBuf' + get_ theTimeInfo = second (syncSF ^->>) <$> get resBuf theTimeInfo + + +infix 4 *-* +-- | Parallely compose two 'ResamplingBuffer's. +(*-*) :: Monad m + => ResamplingBuffer m cl1 cl2 a b + -> ResamplingBuffer m cl1 cl2 c d + -> ResamplingBuffer m cl1 cl2 (a, c) (b, d) +resBuf1 *-* resBuf2 = ResamplingBuffer put_ get_ + where + put_ theTimeInfo (a, c) = do + resBuf1' <- put resBuf1 theTimeInfo a + resBuf2' <- put resBuf2 theTimeInfo c + return $ resBuf1' *-* resBuf2' + get_ theTimeInfo = do + (b, resBuf1') <- get resBuf1 theTimeInfo + (d, resBuf2') <- get resBuf2 theTimeInfo + return ((b, d), resBuf1' *-* resBuf2') + +-- | Given a 'ResamplingBuffer' where the output type depends on the input type polymorphically, +-- we can produce a timestamped version that simply annotates every input value +-- with the 'TimeInfo' when it arrived. +timestamped + :: Monad m + => (forall b. ResamplingBuffer m cl clf b (f b)) + -> ResamplingBuffer m cl clf a (f (a, TimeInfo cl)) +timestamped resBuf = (syncId &&& timeInfo) ^->> resBuf diff --git a/src/FRP/Rhine/SF.hs b/src/FRP/Rhine/SF.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/SF.hs @@ -0,0 +1,55 @@ +{-# LANGUAGE GADTs #-} +{-# LANGUAGE RankNTypes #-} +module FRP.Rhine.SF where + + +-- rhine +import FRP.Rhine.Clock +import FRP.Rhine.ResamplingBuffer +import FRP.Rhine.Schedule +import FRP.Rhine.SyncSF + + +{- | 'SF' is an abbreviation for "signal function". +It represents a side-effectful asynchronous /__s__ignal __f__unction/, or signal network, +where input, data processing (including side effects) and output +need not happen at the same time. + +The type parameters are: + +* 'm': The monad in which side effects take place. +* 'cl': The clock of the whole signal network. + It may be sequentially or parallely composed from other clocks. +* 'a': The input type. Input arrives at the rate @Leftmost cl@. +* 'b': The output type. Output arrives at the rate @Rightmost cl@. +-} +data SF m cl a b where + -- | A synchronous monadic stream function is the basic building block. + -- For such an 'SF', data enters and leaves the system at the same rate as it is processed. + Synchronous + :: ( cl ~ Leftmost cl, cl ~ Rightmost cl) + => SyncSF m cl a b + -> SF m cl a b + -- | Two 'SF's may be sequentially composed if there is a matching 'ResamplingBuffer' between them. + Sequential + :: ( Clock m clab, Clock m clcd + , TimeDomainOf clab ~ TimeDomainOf clcd + , TimeDomainOf clab ~ TimeDomainOf (Rightmost clab) + , TimeDomainOf clcd ~ TimeDomainOf (Leftmost clcd) + ) + => SF m clab a b + -> ResamplingBuffer m (Rightmost clab) (Leftmost clcd) b c + -> SF m clcd c d + -> SF m (SequentialClock m clab clcd) a d + -- | Two 'SF's with the same input and output data may be parallely composed. + Parallel + :: ( Clock m cl1, Clock m cl2 + , TimeDomainOf cl1 ~ TimeDomainOf (Rightmost cl1) + , TimeDomainOf cl2 ~ TimeDomainOf (Rightmost cl2) + , TimeDomainOf cl1 ~ TimeDomainOf cl2 + , TimeDomainOf cl1 ~ TimeDomainOf (Leftmost cl1) + , TimeDomainOf cl2 ~ TimeDomainOf (Leftmost cl2) + ) + => SF m cl1 a b + -> SF m cl2 a b + -> SF m (ParallelClock m cl1 cl2) a b diff --git a/src/FRP/Rhine/SF/Combinators.hs b/src/FRP/Rhine/SF/Combinators.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/SF/Combinators.hs @@ -0,0 +1,141 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE TypeFamilies #-} + +{- | General mnemonic for combinators: + +* @ annotates a data processing unit such as a signal function or a buffer + with temporal information like a clock or a schedule. +* @*@ composes parallely. +* @>@ composes sequentially. +-} +module FRP.Rhine.SF.Combinators where + + +-- rhine +import FRP.Rhine.Clock +import FRP.Rhine.ResamplingBuffer +import FRP.Rhine.Reactimation +import FRP.Rhine.Schedule +import FRP.Rhine.SF +import FRP.Rhine.SyncSF + + +-- * Combinators and syntactic sugar for high-level composition of signal functions. + + +infix 5 @@ +-- | Create a synchronous 'Rhine' by combining a synchronous SF with a matching clock. +-- Synchronicity is ensured by requiring that data enters (@Leftmost cl@) +-- and leaves (@Rightmost cl@) the system at the same as it is processed (@cl@). +(@@) :: ( cl ~ Leftmost cl + , cl ~ Rightmost cl ) + => SyncSF m cl a b -> cl -> Rhine m cl a b +(@@) = Rhine . Synchronous + + +-- | A point at which sequential asynchronous composition +-- ("resampling") of signal functions can happen. +data ResamplingPoint m cla clb a b = ResamplingPoint + (ResamplingBuffer m (Rightmost cla) (Leftmost clb) a b) + (Schedule m cla clb) +-- TODO Make a record out of it? +-- TODO This is aesthetically displeasing. +-- For the buffer, the associativity doesn't matter, but for the Schedule, +-- we sometimes need to specify particular brackets in order for it to work. +-- This is confusing. +-- There would be a workaround if there were pullbacks of schedules... + +-- | Syntactic sugar for 'ResamplingPoint'. +infix 8 -@- +(-@-) :: ResamplingBuffer m (Rightmost cl1) (Leftmost cl2) a b + -> Schedule m cl1 cl2 + -> ResamplingPoint m cl1 cl2 a b +(-@-) = ResamplingPoint + +-- | A purely syntactical convenience construction +-- enabling quadruple syntax for sequential composition, as described below. +infix 2 >-- +data RhineAndResamplingPoint m cl1 cl2 a c = forall b. + RhineAndResamplingPoint (Rhine m cl1 a b) (ResamplingPoint m cl1 cl2 b c) + +-- | Syntactic sugar for 'RhineAndResamplingPoint'. +(>--) :: Rhine m cl1 a b -> ResamplingPoint m cl1 cl2 b c -> RhineAndResamplingPoint m cl1 cl2 a c +(>--) = RhineAndResamplingPoint + +{- | The combinators for sequential composition allow for the following syntax: + +@ +rh1 :: Rhine m cl1 a b +rh1 = ... + +rh2 :: Rhine m cl2 c d +rh2 = ... + +rb :: ResamplingBuffer m (Rightmost cl1) (Leftmost cl2) b c +rb = ... + +sched :: Schedule m cl1 cl2 +sched = ... + +rh :: Rhine m (SequentialClock cl1 cl2) a d +rh = rh1 >-- rb -@- sched --> rh2 +@ +-} +infixr 1 --> +(-->) :: ( Clock m cl1 + , Clock m cl2 + , TimeDomainOf cl1 ~ TimeDomainOf cl2 + , TimeDomainOf (Rightmost cl1) ~ TimeDomainOf cl1 + , TimeDomainOf (Leftmost cl2) ~ TimeDomainOf cl2 + , Clock m (Rightmost cl1) + , Clock m (Leftmost cl2) + ) + => RhineAndResamplingPoint m cl1 cl2 a b + -> Rhine m cl2 b c + -> Rhine m (SequentialClock m cl1 cl2) a c +RhineAndResamplingPoint (Rhine sf1 cl1) (ResamplingPoint rb cc) --> (Rhine sf2 cl2) + = Rhine (Sequential sf1 rb sf2) (SequentialClock cl1 cl2 cc) + +-- | A purely syntactical convenience construction +-- allowing for ternary syntax for parallel composition, described below. +data RhineParallelAndSchedule m cl1 cl2 a b = RhineParallelAndSchedule (Rhine m cl1 a b) (Schedule m cl1 cl2) + +-- | Syntactic sugar for 'RhineParallelAndSchedule'. +infix 4 **@ +(**@) + :: Rhine m cl1 a b + -> Schedule m cl1 cl2 + -> RhineParallelAndSchedule m cl1 cl2 a b +(**@) = RhineParallelAndSchedule + +{- | The combinators for parallel composition allow for the following syntax: + +@ +rh1 :: Rhine m cl1 a b +rh1 = ... + +rh2 :: Rhine m cl2 a b +rh2 = ... + +sched :: Schedule m cl1 cl2 +sched = ... + +rh :: Rhine m (ParallelClock cl1 cl2) a b +rh = rh1 **\@ sched \@** rh2 +@ +-} +infix 3 @** +(@**) :: ( Clock m cl1 + , Clock m cl2 + , TimeDomainOf cl1 ~ TimeDomainOf (Rightmost cl1) + , TimeDomainOf cl2 ~ TimeDomainOf (Rightmost cl2) + , TimeDomainOf cl1 ~ TimeDomainOf (Leftmost cl1) + , TimeDomainOf cl2 ~ TimeDomainOf (Leftmost cl2) + , TimeDomainOf cl1 ~ TimeDomainOf cl2 + ) + => RhineParallelAndSchedule m cl1 cl2 a b + -> Rhine m cl2 a b + -> Rhine m (ParallelClock m cl1 cl2) a b +RhineParallelAndSchedule (Rhine sf1 cl1) schedule @** (Rhine sf2 cl2) + = Rhine (Parallel sf1 sf2) (ParallelClock cl1 cl2 schedule) diff --git a/src/FRP/Rhine/Schedule.hs b/src/FRP/Rhine/Schedule.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/Schedule.hs @@ -0,0 +1,143 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE TypeFamilies #-} + +module FRP.Rhine.Schedule where + +-- dunai +import Data.MonadicStreamFunction + +-- rhine +import FRP.Rhine.Clock + +-- * The schedule type + +-- | A schedule implements a combination of two clocks. +-- It outputs a time stamp and an 'Either' value, +-- which specifies which of the two subclocks has ticked. +data Schedule m cl1 cl2 + = (TimeDomainOf cl1 ~ TimeDomainOf cl2) + => Schedule + { startSchedule + :: cl1 -> cl2 + -> m (MSF m () (TimeDomainOf cl1, Either (Tag cl1) (Tag cl2)), TimeDomainOf cl1) + } +-- The type constraint in the constructor is actually useful when pattern matching on 'Schedule', +-- which is interesting since a constraint like 'Monad m' is useful. +-- When reformulating as a GADT, it might get used, +-- but that would mean that we can't use record syntax. + + +-- * Utilities to create new schedules from existing ones + +-- | Lift a schedule along a monad morphism. +hoistSchedule + :: (Monad m1, Monad m2) + => (forall a . m1 a -> m2 a) + -> Schedule m1 cl1 cl2 + -> Schedule m2 cl1 cl2 +hoistSchedule hoist Schedule {..} = Schedule startSchedule' + where + startSchedule' cl1 cl2 = hoist + $ first (hoistMSF hoist) <$> startSchedule cl1 cl2 + hoistMSF = liftMSFPurer + -- TODO This should be a dunai issue + +-- | Swaps the clocks for a given schedule. +flipSchedule + :: Monad m + => Schedule m cl1 cl2 + -> Schedule m cl2 cl1 +flipSchedule Schedule {..} = Schedule startSchedule_ + where + startSchedule_ cl2 cl1 = first (arr (second swapEither) <<<) <$> startSchedule cl1 cl2 + swapEither :: Either a b -> Either b a -- TODO Why is stuff like this not in base? Maybe send pull request... + swapEither (Left a) = Right a + swapEither (Right b) = Left b + +-- * Composite clocks + + +-- | Two clocks can be combined with a schedule as a clock +-- for an asynchronous sequential composition of signal functions. +data SequentialClock m cl1 cl2 + = TimeDomainOf cl1 ~ TimeDomainOf cl2 + => SequentialClock + { sequentialCl1 :: cl1 + , sequentialCl2 :: cl2 + , sequentialSchedule :: Schedule m cl1 cl2 + } + + +instance (Monad m, Clock m cl1, Clock m cl2) + => Clock m (SequentialClock m cl1 cl2) where + type TimeDomainOf (SequentialClock m cl1 cl2) = TimeDomainOf cl1 + type Tag (SequentialClock m cl1 cl2) = Either (Tag cl1) (Tag cl2) + startClock SequentialClock {..} + = startSchedule sequentialSchedule sequentialCl1 sequentialCl2 + + +-- | Two clocks can be combined with a schedule as a clock +-- for an asynchronous parallel composition of signal functions. +data ParallelClock m cl1 cl2 + = TimeDomainOf cl1 ~ TimeDomainOf cl2 + => ParallelClock + { parallelCl1 :: cl1 + , parallelCl2 :: cl2 + , parallelSchedule :: Schedule m cl1 cl2 + } + +instance (Monad m, Clock m cl1, Clock m cl2) + => Clock m (ParallelClock m cl1 cl2) where + type TimeDomainOf (ParallelClock m cl1 cl2) = TimeDomainOf cl1 + type Tag (ParallelClock m cl1 cl2) = Either (Tag cl1) (Tag cl2) + startClock ParallelClock {..} + = startSchedule parallelSchedule parallelCl1 parallelCl2 + + +-- * Navigating the clock tree + +-- | The clock that represents the rate at which data enters the system. +type family Leftmost cl where + Leftmost (SequentialClock m cl1 cl2) = Leftmost cl1 + Leftmost (ParallelClock m cl1 cl2) = ParallelClock m (Leftmost cl1) (Leftmost cl2) + Leftmost cl = cl + +-- | The clock that represents the rate at which data leaves the system. +type family Rightmost cl where + Rightmost (SequentialClock m cl1 cl2) = Rightmost cl2 + Rightmost (ParallelClock m cl1 cl2) = ParallelClock m (Rightmost cl1) (Rightmost cl2) + Rightmost cl = cl + + +-- | A tree representing possible last times to which +-- the constituents of a clock may have ticked. +data LastTime cl where + SequentialLastTime + :: LastTime cl1 -> LastTime cl2 + -> LastTime (SequentialClock m cl1 cl2) + ParallelLastTime + :: LastTime cl1 -> LastTime cl2 + -> LastTime (ParallelClock m cl1 cl2) + LeafLastTime :: TimeDomainOf cl -> LastTime cl + + +-- | An inclusion of a clock into a tree of parallel compositions of clocks. +data ParClockInclusion clS cl where + ParClockInL + :: ParClockInclusion (ParallelClock m clL clR) cl + -> ParClockInclusion clL cl + ParClockInR + :: ParClockInclusion (ParallelClock m clL clR) cl + -> ParClockInclusion clR cl + ParClockRefl :: ParClockInclusion cl cl + +-- | Generates a tag for the composite clock from a tag of a leaf clock, +-- given a parallel clock inclusion. +parClockTagInclusion :: ParClockInclusion clS cl -> Tag clS -> Tag cl +parClockTagInclusion (ParClockInL parClockInL) tag = parClockTagInclusion parClockInL $ Left tag +parClockTagInclusion (ParClockInR parClockInR) tag = parClockTagInclusion parClockInR $ Right tag +parClockTagInclusion ParClockRefl tag = tag diff --git a/src/FRP/Rhine/Schedule/Concurrently.hs b/src/FRP/Rhine/Schedule/Concurrently.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/Schedule/Concurrently.hs @@ -0,0 +1,33 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE TypeFamilies #-} +module FRP.Rhine.Schedule.Concurrently where + +-- base +import Control.Concurrent + +-- rhine +import FRP.Rhine + + +-- | Runs two clocks in separate GHC threads +-- and collects the results in the foreground thread. +-- Caution: The data processing will still happen in the same thread +-- (since data processing and scheduling are separated concerns). +concurrently :: (Clock IO cl1, Clock IO cl2, TimeDomainOf cl1 ~ TimeDomainOf cl2) => Schedule IO cl1 cl2 +concurrently = Schedule $ \cl1 cl2 -> do + iMVar <- newEmptyMVar + mvar <- newEmptyMVar + _ <- forkIO $ do + (runningClock, initTime) <- startClock cl1 + putMVar iMVar initTime + reactimate $ runningClock >>> second (arr Left) >>> arrM (putMVar mvar) + _ <- forkIO $ do + (runningClock, initTime) <- startClock cl2 + putMVar iMVar initTime + reactimate $ runningClock >>> second (arr Right) >>> arrM (putMVar mvar) + initTime <- takeMVar iMVar -- The first clock to be initialised sets the first time stamp + _ <- takeMVar iMVar -- Initialise the second clock + return (arrM_ $ takeMVar mvar, initTime) + +-- TODO These threads can't be killed from outside easily since we've lost their ids +-- => make a MaybeT or ExceptT variant diff --git a/src/FRP/Rhine/Schedule/Trans.hs b/src/FRP/Rhine/Schedule/Trans.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/Schedule/Trans.hs @@ -0,0 +1,65 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE TypeFamilies #-} +module FRP.Rhine.Schedule.Trans where + +-- rhine +import Control.Monad.Schedule +import FRP.Rhine + + +-- * Universal schedule for the 'ScheduleT' monad transformer + +-- | Two clocks in the 'ScheduleT' monad transformer +-- can always be canonically scheduled. +-- Indeed, this is the purpose for which 'ScheduleT' was defined. +schedule + :: ( Monad m + , Clock (ScheduleT (Diff (TimeDomainOf cl1)) m) cl1 + , Clock (ScheduleT (Diff (TimeDomainOf cl1)) m) cl2 + , TimeDomainOf cl1 ~ TimeDomainOf cl2 + , Ord (Diff (TimeDomainOf cl1)) + , Num (Diff (TimeDomainOf cl1)) + ) + => Schedule (ScheduleT (Diff (TimeDomainOf cl1)) m) cl1 cl2 +schedule = Schedule {..} + where + startSchedule cl1 cl2 = do + (runningClock1, initTime) <- startClock cl1 + (runningClock2, _) <- startClock cl2 + return + ( runningSchedule cl1 cl2 runningClock1 runningClock2 + , initTime + ) + + -- Combines the two individual running clocks to one running clock. + runningSchedule + :: ( Monad m + , Clock (ScheduleT (Diff (TimeDomainOf cl1)) m) cl1 + , Clock (ScheduleT (Diff (TimeDomainOf cl2)) m) cl2 + , TimeDomainOf cl1 ~ TimeDomainOf cl2 + , Ord (Diff (TimeDomainOf cl1)) + , Num (Diff (TimeDomainOf cl1)) + ) + => cl1 -> cl2 + -> MSF (ScheduleT (Diff (TimeDomainOf cl1)) m) () (TimeDomainOf cl1, Tag cl1) + -> MSF (ScheduleT (Diff (TimeDomainOf cl1)) m) () (TimeDomainOf cl2, Tag cl2) + -> MSF (ScheduleT (Diff (TimeDomainOf cl1)) m) () (TimeDomainOf cl1, Either (Tag cl1) (Tag cl2)) + runningSchedule cl1 cl2 rc1 rc2 = MSF $ \_ -> do + -- Race both clocks against each other + raceResult <- race (unMSF rc1 ()) (unMSF rc2 ()) + case raceResult of + -- The first clock ticks first... + Left (((td, tag1), rc1'), cont2) -> return + -- so we can emit its time stamp... + ( (td, Left tag1) + -- and continue. + , runningSchedule cl1 cl2 rc1' (MSF $ const cont2) + ) + -- The second clock ticks first... + Right (cont1, ((td, tag2), rc2')) -> return + -- so we can emit its time stamp... + ( (td, Right tag2) + -- and continue. + , runningSchedule cl1 cl2 (MSF $ const cont1) rc2' + ) diff --git a/src/FRP/Rhine/SyncSF.hs b/src/FRP/Rhine/SyncSF.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/SyncSF.hs @@ -0,0 +1,177 @@ +{-# LANGUAGE Arrows #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE TypeFamilies #-} + +module FRP.Rhine.SyncSF where + + +-- base +import Control.Arrow +import qualified Control.Category (id) + +-- transformers +import Control.Monad.Trans.Reader + ( ReaderT, ask, asks, mapReaderT, withReaderT) + +-- dunai +import Data.MonadicStreamFunction + (MSF, liftMSFPurer, liftMSFTrans, arrM, arrM_, sumFrom, delay, feedback) +import Data.VectorSpace + +-- rhine +import FRP.Rhine.Clock +import FRP.Rhine.TimeDomain + + +-- * Synchronous signal functions and behaviours + +-- | A (synchronous) monadic stream function +-- with the additional side effect of being time-aware, +-- that is, reading the current 'TimeInfo' of the clock 'cl'. +type SyncSF m cl a b = MSF (ReaderT (TimeInfo cl) m) a b + +-- | A (side-effectful) behaviour is a time-aware stream +-- that doesn't depend on a particular clock. +-- 'td' denotes the time domain. +type Behaviour m td a = forall cl. td ~ TimeDomainOf cl => SyncSF m cl () a + +-- | Compatibility to U.S. american spelling. +type Behavior m td a = Behaviour m td a + + +-- * Utilities to create 'SyncSF's from simpler data + +-- TODO Test in which situations it makes sense not to change cl +-- | Hoist a 'SyncSF' along a monad morphism. +hoistSyncSF + :: (Monad m1, Monad m2) + => (forall c. m1 c -> m2 c) + -> SyncSF m1 cl a b + -> SyncSF m2 (HoistClock m1 m2 cl) a b +hoistSyncSF hoist = liftMSFPurer $ withReaderT (retag id) . mapReaderT hoist + +-- | A monadic stream function without dependency on time +-- is a 'SyncSF' for any clock. +timeless :: Monad m => MSF m a b -> SyncSF m cl a b +timeless = liftMSFTrans + +-- | Utility to lift Kleisli arrows directly to 'SyncSF's. +arrMSync :: Monad m => (a -> m b) -> SyncSF m cl a b +arrMSync = timeless . arrM + +-- | Version without input. +arrMSync_ :: Monad m => m b -> SyncSF m cl a b +arrMSync_ = timeless . arrM_ + +-- | Read the environment variable, i.e. the 'TimeInfo'. +timeInfo :: Monad m => SyncSF m cl a (TimeInfo cl) +timeInfo = arrM_ ask + +{- | Utility to apply functions to the current 'TimeInfo', +such as record selectors: +@ +printAbsoluteTime :: SyncSF IO cl () () +printAbsoluteTime = timeInfoOf absolute >>> arrMSync print +@ +-} +timeInfoOf :: Monad m => (TimeInfo cl -> b) -> SyncSF m cl a b +timeInfoOf f = arrM_ $ asks f + +-- * Useful aliases + +{- | Alias for 'Control.Category.>>>' (sequential composition) +with higher operator precedence, designed to work with the other operators, e.g.: + +> syncsf1 >-> syncsf2 @@ clA **@ sched @** syncsf3 >-> syncsf4 @@ clB +-} +infixr 4 >-> +(>->) :: Monad m + => SyncSF m cl a b + -> SyncSF m cl b c + -> SyncSF m cl a c +(>->) = (>>>) + +-- | Alias for 'Control.Arrow.<<<'. +infixl 4 <-< +(<-<) :: Monad m + => SyncSF m cl b c + -> SyncSF m cl a b + -> SyncSF m cl a c +(<-<) = (<<<) + +{- | Output a constant value. +Specialises e.g. to this type signature: + +> arr_ :: Monad m => b -> SyncSF m cl a b +-} +arr_ :: Arrow a => b -> a c b +arr_ = arr . const + + +-- | The identity synchronous stream function. +syncId :: Monad m => SyncSF m cl a a +syncId = Control.Category.id + + +-- * Basic signal processing components + +-- | The output of @integralFrom v0@ is the numerical Euler integral +-- of the input, with initial offset @v0@. +integralFrom + :: ( Monad m, VectorSpace v + , Groundfield v ~ Diff (TimeDomainOf cl)) + => v -> SyncSF m cl v v +integralFrom v0 = proc v -> do + _sinceTick <- timeInfoOf sinceTick -< () + sumFrom v0 -< _sinceTick *^ v + +-- | Euler integration, with zero initial offset. +integral + :: ( Monad m, VectorSpace v + , Groundfield v ~ Diff (TimeDomainOf cl)) + => SyncSF m cl v v +integral = integralFrom zeroVector + + +-- | The output of @derivativeFrom v0@ is the numerical derivative of the input, +-- with a Newton difference quotient. +-- The input is initialised with @v0@. +derivativeFrom + :: ( Monad m, VectorSpace v + , Groundfield v ~ Diff (TimeDomainOf cl)) + => v -> SyncSF m cl v v +derivativeFrom v0 = proc v -> do + vLast <- delay v0 -< v + TimeInfo {..} <- timeInfo -< () + returnA -< (v ^-^ vLast) ^/ sinceTick + +-- | Numerical derivative with input initialised to zero. +derivative + :: ( Monad m, VectorSpace v + , Groundfield v ~ Diff (TimeDomainOf cl)) + => SyncSF m cl v v +derivative = derivativeFrom zeroVector + + +-- | An average, or low pass. It will average out, or filter, +-- all features below a given time scale. +averageFrom + :: ( Monad m, VectorSpace v + , Groundfield v ~ Diff (TimeDomainOf cl)) + => v -- ^ The initial position + -> Diff (TimeDomainOf cl) -- ^ The time scale on which the signal is averaged + -> SyncSF m cl v v +averageFrom v0 t = feedback v0 $ proc (v, vAvg) -> do + TimeInfo {..} <- timeInfo -< () + let vAvg' = (v ^* sinceTick ^+^ vAvg ^* t) ^/ (sinceTick + t) + returnA -< (vAvg', vAvg') + + +-- | An average, or low pass, initialised to zero. +average + :: ( Monad m, VectorSpace v + , Groundfield v ~ Diff (TimeDomainOf cl)) + => Diff (TimeDomainOf cl) -- ^ The time scale on which the signal is averaged + -> SyncSF m cl v v +average = averageFrom zeroVector diff --git a/src/FRP/Rhine/TimeDomain.hs b/src/FRP/Rhine/TimeDomain.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Rhine/TimeDomain.hs @@ -0,0 +1,46 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE TypeFamilies #-} +module FRP.Rhine.TimeDomain where + +-- time +import Data.Time.Clock (UTCTime, diffUTCTime) + +-- dunai +import Data.VectorSpace.Specific () + + +-- | A time domain is an affine space representing a notion of time, +-- such as real time, simulated time, steps, or a completely different notion. +class TimeDomain td where + type Diff td + diffTime :: td -> td -> Diff td + + +instance TimeDomain UTCTime where + type Diff UTCTime = Double + diffTime t1 t2 = realToFrac $ diffUTCTime t1 t2 + +instance TimeDomain Double where + type Diff Double = Double + diffTime = (-) + +instance TimeDomain Float where + type Diff Float = Float + diffTime = (-) + +instance TimeDomain Integer where + type Diff Integer = Integer + diffTime = (-) + +instance TimeDomain () where + type Diff () = () + diffTime _ _ = () + +-- | Any 'Num' can be wrapped to form a 'TimeDomain'. +newtype NumTimeDomain a = NumTimeDomain { fromNumTimeDomain :: a } + deriving Num + +instance Num a => TimeDomain (NumTimeDomain a) where + type Diff (NumTimeDomain a) = NumTimeDomain a + diffTime = (-)