Fix memory memory leak

- Tuple to strict pair
- Self-recursive loops with more knot tying <- this was the thing

.gitignore

@@ -3,3 +3,5 @@ dist-newstyle
 .direnv
 
 .worktrees/
+
+docs/superpowers

default.nix

@@ -1,6 +1,6 @@
 { mkDerivation, aeson, annotated-exception, async, base, bytestring
-, hedgehog, hspec, hspec-hedgehog, katip, lens, lens-aeson, lib
-, network, stm, text, time, uuid, websockets
+, containers, hedgehog, hspec, hspec-hedgehog, katip, lens
+, lens-aeson, lib, network, stm, text, time, uuid, websockets
 }:
 mkDerivation {
   pname = "home-assistant-controller";
@@ -9,13 +9,13 @@ mkDerivation {
   isLibrary = true;
   isExecutable = true;
   libraryHaskellDepends = [
-    aeson annotated-exception async base bytestring katip lens
-    lens-aeson network stm text time uuid websockets
+    aeson annotated-exception async base bytestring containers katip
+    lens lens-aeson network stm text time uuid websockets
   ];
   executableHaskellDepends = [ base ];
   testHaskellDepends = [
-    aeson annotated-exception async base hedgehog hspec hspec-hedgehog
-    stm text time
+    aeson annotated-exception async base containers hedgehog hspec
+    hspec-hedgehog katip stm text time uuid
   ];
   license = lib.meta.getLicenseFromSpdxId "BSD-3-Clause";
   mainProgram = "home-assistant-controller";

home-assistant-controller.cabal

@@ -62,6 +62,8 @@ library
     exposed-modules:  AFRP
                     , HomeAssistant.Controller
                     , HomeAssistant.Controller.Bedroom
+                    , HomeAssistant.Controller.Children
+                    , HomeAssistant.Controller.Ruuvi
                     , HomeAssistant.Runtime
                     , HomeAssistant.Runtime.Bus
                     , HomeAssistant.Runtime.Connection
@@ -88,6 +90,7 @@ library
                     , annotated-exception
                     , uuid
                     , katip
+                    , containers
 
     -- Directories containing source files.
     hs-source-dirs:   src
@@ -128,11 +131,14 @@ test-suite home-assistant-controller-test
     default-language: GHC2024
 
     -- Modules included in this executable, other than Main.
-    other-modules:    BusSpec
+    other-modules:    AFRPSpec
+                    , BackoffProp
+                    , BedroomSpec
+                    , BusSpec
                     , ConnectionSpec
                     , RuntimeSpec
                     , SupervisorSpec
-                    , BackoffProp
+                    , Support
 
     -- LANGUAGE extensions used by modules in this package.
     -- other-extensions:
@@ -160,4 +166,5 @@ test-suite home-assistant-controller-test
         annotated-exception,
         time,
         uuid,
-        katip
+        katip,
+        containers

secrets.yaml

@@ -1,8 +1,8 @@
 HA_TOKEN: ENC[AES256_GCM,data:zs9nTT2EV/OhBiH9alUuupUvYURXj40lxe7IKKWXdExW+BW98yEDHcQr/KNsQ0Iu9+U2GrsA8WpODeM351kI3Owj3rl082tObBJ2jwF1iyYcXhoH3xvbHb+HN802SBOTIO32m+ZwGvIVPjF8W+67eGKTrIUcqnKj7Ab/TchG+mYPLdil0mYjITY+DCUi0mW1nElCaTTexovj4PHlUdA1xIsCYF+XnTypQv8o+Zy8MsMZtdzNQKJG,iv:Lmxmp5y15sqhQvEO7liVk6x3Y26qtoL39TT2MSqZdok=,tag:1Xvs6YUilP/XH0UxYngTIg==,type:str]
-HA_HOST: ENC[AES256_GCM,data:6ivOMsDGVWsc+xV+Qjk5Y/bbu7cy/g==,iv:nFkLQCNk6DlW5vFq8So44YoICL95GYlhrvj4Rqes9kM=,tag:Nkqnflf4lazGdYnoDkFVVw==,type:str]
+HA_HOST: ENC[AES256_GCM,data:MfA+RReJHnKP31bD4YAoZrc=,iv:Z9xDtzPsUER4jQ12BbM2EtnLNggZhOWzCb7VAL2I02g=,tag:/Eogbd2kT7phdhaR/vGNDA==,type:str]
 sops:
-    lastmodified: "2026-08-20T10:49:23Z"
-    mac: ENC[AES256_GCM,data:Vfks6xgIizhaYD90vEtWFwjFG1qox56/rlSOMzLZ30xKx1x+085hrlSsq4XOgqBXTDAXeRbMXTpZe/U259gNbfHeBYfpQOgwjHg5MrXl8K3G6eIEWkv/OBQfPpR1DVrhfJsN34ehOUkt3h7K+uHad/XAUavwNNaOQ8fInvRCMqo=,iv:XAvgJJONoI8G5bJ09nR7wxnoH/HJVsPkz2ahXP5fOvI=,tag:hprULBSFR/JpLTot9+B/ng==,type:str]
+    lastmodified: "2026-08-21T10:17:59Z"
+    mac: ENC[AES256_GCM,data:GuNyo/3flhR30e6kds1iXcDGmq0Xq2hqvBmAvYtakKwzUzMocP9vImfxs+MfVNiMGirlkwbihrX5mFSbfORjES/FDTs6p0100H1tkBqPTecW88PC6/66Js9umVevGYbgToCNBjV9mUBu7o7SyU+NYLnZ/8X4xdx0y6/MKnuvjUA=,iv:/nL5AIL80iV3sutjMTmHyF62SRezmX9XyJJ6+o9TdkQ=,tag:SNMu17rgMIhdzZVsGmXyfg==,type:str]
     pgp:
         - created_at: "2026-08-20T05:39:51Z"
           enc: |-

src/AFRP.hs

@@ -1,8 +1,10 @@
 {-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE Arrows #-}
 
 module AFRP
   ( Mealy(..)
   , eff
+  , withEntities
   , Event(..)
   , hold
   , events
@@ -18,117 +20,228 @@ module AFRP
   , (>>|)
   , toEvent
   , lMerge
+  , Pair(..)
   , Request(..)
+  , edge
+  , dropFirst
+  , duration
+  , tag
+  , isEvent
+  , delayEvent
+  , sample
+  , rollup
+  , sliding
+  , fixed
+  , debounce
+  , currentTime
+  , onEvent
   ) where
 
 import Control.Category (Category(..), (>>>))
 import Prelude hiding ((.), id)
 import Control.Arrow (Arrow(..), ArrowChoice(..), ArrowLoop(..))
-import Data.Time (UTCTime)
+import Data.Time (UTCTime, NominalDiffTime, diffUTCTime, addUTCTime, TimeZone, LocalTime, utcToLocalTime)
 import Control.Monad.Fix (MonadFix (mfix))
 import Data.Either (fromLeft)
 import Data.Bool (bool)
+import Data.Monoid (Endo(..))
 import Data.UUID (UUID)
+import qualified Data.Set as S
+import qualified Data.Text as T
+
+data Pair a b = Pair !a !b
 
 data Request = Request
   { requestTime :: !UTCTime
+  , requestTimeZone :: !TimeZone
   , requestTraceId :: !UUID
-  }
-  deriving Show
+  } deriving (Show, Eq)
 
-newtype Mealy eff a b = Mealy
-  { runMealy :: forall m. MonadFix m => (forall x. eff x -> m x) -> Request -> a -> m (b, Mealy eff a b) }
+-- | The set of entity ids an arrow subscribes to. Static: it does not
+-- change as the machine steps, so the runtime can read it once to build
+-- trigger subscriptions.
+data Mealy eff a b = Mealy
+  { entities :: S.Set T.Text
+  , runMealy :: forall m. MonadFix m => (forall x. eff x -> m x) -> Request -> a -> m (Pair b (Mealy eff a b))
+  }
+
+
+instance Semigroup b => Semigroup (Mealy eff a b) where
+  Mealy ast af <> Mealy bst bf = Mealy (ast <> bst) $ \nt r a -> do
+    Pair x af' <- af nt r a
+    Pair x' bf' <- bf nt r a
+    pure (Pair (x <> x') (af' <> bf'))
+
+
+instance Monoid b => Monoid (Mealy eff a b) where
+  mempty = m
+    where
+      m = Mealy mempty $ \_ _ _ -> pure (Pair mempty m)
 
 eff :: (Request -> a -> eff b) -> Mealy eff a b
-eff f = Mealy $ \nt req x ->
-  nt (f req x) >>= \b -> pure (b, eff f)
+eff f = m
+  where
+    m = Mealy mempty $ \nt req x ->
+      nt (f req x) >>= \b -> pure (Pair b m)
+
+-- | Override the static entity set of an arrow. Use when a combinator
+-- (e.g. 'switch') hides continuation entities from the runtime's
+-- startup subscription scan.
+withEntities :: S.Set T.Text -> Mealy eff a b -> Mealy eff a b
+withEntities es (Mealy _ f) = Mealy es f
 
 instance Category (Mealy eff) where
-  id = Mealy (\_ _ x -> pure (x, id))
-  (Mealy f) . (Mealy g) = Mealy $ \nt t a -> do
-    (b, g') <- g nt t a
-    (c, f') <- f nt t b
-    pure (c, f' . g')
+  id = Mealy mempty (\_ _ x -> pure (Pair x id))
+  (Mealy ast f) . (Mealy bst g) = Mealy (ast <> bst) $ \nt t a -> do
+    Pair b g' <- g nt t a
+    Pair c f' <- f nt t b
+    pure (Pair c (f' . g'))
 
 instance Arrow (Mealy eff) where
-  arr f = Mealy $ \_ _ b -> pure (f b, arr f)
-  first (Mealy f) = Mealy $ \nt t (b,d) -> do
-    (c, f') <- f nt t b
-    pure ((c, d), first f')
+  arr f = mealy
+    where
+      mealy = Mealy mempty $ \_ _ b -> pure (Pair (f b) mealy)
+  first (Mealy st f) = Mealy st $ \nt t (b,d) -> do
+    Pair c f' <- f nt t b
+    pure (Pair (c, d) (first f'))
 
 instance ArrowChoice (Mealy eff) where
-  left (Mealy f) = Mealy $ \nt t -> \case
+  left (Mealy st f) = lm
+    where
+      lm = Mealy st $ \nt t -> \case
         Left b -> do
-      (c, f') <- f nt t b
-      pure (Left c, left f')
-    Right d -> pure (Right d, left (Mealy f))
+          Pair c f' <- f nt t b
+          pure (Pair (Left c) (left f'))
+        Right d -> pure (Pair (Right d) lm)
 
-instance ArrowLoop (Mealy eff) where
-  loop (Mealy f) = Mealy $ \nt t b -> do
-    ((c,_), f') <- mfix $ \((_,d), _) -> f nt t (b,d)
-    pure (c, loop f')
+-- ArrowLoop is incompatible with strict Pair (strict fields prevent
+-- the lazy knot-tying that mfix requires with loop).
+-- instance ArrowLoop (Mealy eff) where
+--   loop (Mealy st f) = Mealy st $ \nt t b -> do
+--     Pair (c,_) f' <- mfix $ \(Pair (_,d) _) -> f nt t (b,d)
+--     pure (Pair c (loop f'))
 
 instance Functor (Mealy eff a) where
-  fmap f (Mealy g) = Mealy $ \nt t a -> do
-    (b, g') <- g nt t a
-    pure (f b, fmap f g')
+  fmap f (Mealy st g) = Mealy st $ \nt t a -> do
+    Pair b g' <- g nt t a
+    pure (Pair (f b) (fmap f g'))
 
 instance Applicative (Mealy eff a) where
-  pure b = Mealy $ \_ _ _ -> pure (b, pure b)
-  Mealy f <*> Mealy x = Mealy $ \nt t a -> do
-    (f', fNext) <- f nt t a
-    (x', xNext) <- x nt t a
-    pure (f' x', fNext <*> xNext)
+  pure b = Mealy mempty $ \_ _ _ -> pure (Pair b (pure b))
+  Mealy ast f <*> Mealy bst x = Mealy (ast <> bst) $ \nt t a -> do
+    b' <- x nt t a
+    let Pair x' xNext = b'
+    Pair f' fNext <- f nt t a
+    pure (Pair (f' x') (fNext <*> xNext))
 
 data Event a
   = Tick
   | Event a
-  deriving (Show, Functor, Foldable, Traversable)
+  deriving (Show, Eq, Functor, Foldable, Traversable)
+
+instance Semigroup (Event a) where
+  (<>) = lMerge
+
+instance Monoid (Event a) where
+  mempty = Tick
 
 hold :: a -> Mealy eff (Event a) a
-hold a = Mealy $ \_ _ -> \case
-  Tick -> pure (a, hold a)
-  Event a' -> pure (a', hold a')
+hold a = Mealy mempty $ \_ _ -> \case
+  Tick -> pure (Pair a (hold a))
+  Event a' -> pure (Pair a' (hold a'))
 
 events :: Mealy eff (Event a) (Either () a)
 events = arr $ \case
   Tick -> Left ()
   Event a -> Right a
 
+isEvent :: Event a -> Bool
+isEvent Tick = False
+isEvent _ = True
+
+tag :: b -> Event a -> Event b
+tag b ev = b <$ ev
+
 switch :: Mealy eff a (b, Event c) -> (c -> Mealy eff a b) -> Mealy eff a b
-switch (Mealy f) s = Mealy $ \nt t a -> do
-  ((b, ev), f') <- f nt t a
+switch (Mealy st f) s = Mealy st $ \nt t a -> do
+  Pair (b, ev) f' <- f nt t a
   case ev of
-          Tick -> pure (b, switch f' s)
+          Tick -> pure (Pair b (switch f' s))
           Event x -> runMealy (s x) nt t a
 
+sample :: Mealy eff (a, Event b) (Event a)
+sample = arr (uncurry tag)
+
 preMapAccum :: (x -> a -> x) -> x -> (x -> b) -> Mealy eff a b
 preMapAccum f x extract = go x
   where
-    go b = Mealy $ \_ _ a ->
+    go b = Mealy mempty $ \_ _ a ->
       let next = f b a
-      in pure (extract b, go next)
+      in pure (Pair (extract b) (go next))
 
 preMapAccumRequest :: (Request -> x -> a -> x) -> x -> (x -> b) -> Mealy eff a b
 preMapAccumRequest f x extract = go x
   where
-    go b = Mealy $ \_ t a ->
+    go b = Mealy mempty $ \_ t a ->
       let next = f t b a
-      in pure (extract b, go next)
+      in pure (Pair (extract b) (go next))
 
 mapAccum :: (x -> a -> x) -> x -> (x -> b) -> Mealy eff a b
 mapAccum f x extract = go x
   where
-    go b = Mealy $ \_ _ a ->
+    go b = Mealy mempty $ \_ _ a ->
       let next = f b a
-      in pure (extract next, go next)
+      in pure (Pair (extract next) (go next))
 
 mapAccumRequest :: (Request -> x -> a -> x) -> x -> (x -> b) -> Mealy eff a b
 mapAccumRequest f x extract = go x
   where
-    go b = Mealy $ \_ t a ->
+    go b = Mealy mempty $ \_ t a ->
       let next = f t b a
-      in pure (extract next, go next)
+      in pure (Pair (extract next) (go next))
+
+data DelayState x a = DelayState
+  { pending :: x
+  , output  :: !(Event a)
+  }
+
+
+delayEvent :: NominalDiffTime -> Mealy eff (Event a) (Event a)
+delayEvent delay =
+  mapAccumRequest step initial output
+  where
+    initial = DelayState [] Tick
+
+    step req st input =
+      let now = requestTime req
+
+          queued =
+            case input of
+              Tick    -> pending st
+              Event x -> pending st ++ [(delay `addUTCTime` now, x)]
+
+      in case queued of
+           (due, x) : rest
+             | due <= now ->
+                 DelayState rest (Event x)
+
+           _ ->
+                 DelayState queued Tick
+
+debounce :: NominalDiffTime -> Mealy eff (Event a) (Event a)
+debounce delay =
+  mapAccumRequest step initial output
+  where
+    initial = DelayState Nothing Tick
+    step req st input =
+      let now = requestTime req
+          held = case input of
+                      Tick -> pending st
+                      Event x -> Just (delay `addUTCTime` now, x)
+      in case held of
+              Just (due, x)
+                | due <= now -> DelayState Nothing (Event x)
+              _ -> DelayState held Tick
 
 changes :: Eq a => Mealy eff a (Event a)
 changes = mapAccum go Nothing (maybe Tick snd)
@@ -161,3 +274,88 @@ lMerge :: Event a -> Event a -> Event a
 lMerge Tick Tick = Tick
 lMerge (Event a) _ = Event a
 lMerge Tick (Event a) = Event a
+
+edge :: Mealy eff Bool (Event ())
+edge = go False
+  where
+    go True = Mealy mempty $ \_ _ -> \case
+      True -> pure (Pair Tick (go True))
+      False -> pure (Pair Tick (go False))
+    go False = Mealy mempty $ \_ _ -> \case
+      True -> pure (Pair (Event ()) (go True))
+      False -> pure (Pair Tick (go False))
+
+
+-- | Drop the first 'Event' and pass through everything after. Useful for
+-- ignoring a self-triggered event (e.g. a service call that changes the
+-- very entity the arrow listens to).
+dropFirst :: Mealy eff (Event a) (Event a)
+dropFirst = go False
+  where
+    go seen = Mealy mempty $ \_ _ input ->
+      case input of
+        Event _ | not seen -> pure (Pair Tick (go True))
+        _ -> pure (Pair input (go seen))
+
+
+duration :: forall eff a. Mealy eff a NominalDiffTime
+duration = mapAccumRequest go (Nothing @(UTCTime, NominalDiffTime)) (maybe 0 snd)
+  where
+    go :: Request -> Maybe (UTCTime, NominalDiffTime) -> a -> Maybe (UTCTime, NominalDiffTime)
+    go req Nothing _ = Just (requestTime req, requestTime req `diffUTCTime` requestTime req)
+    go req (Just (startTime, _)) _ = Just (startTime, requestTime req `diffUTCTime` startTime)
+
+
+
+-- | Rollup, hold back bursty messages
+--
+-- Consider a case where you have a  bursty set of data. You care to get an immediate response,
+-- but don't want to spam the output
+rollup
+  :: Int -- ^ How many items to pass through before burst protection
+  -> Int -- ` How many seconds to collect the bursty data
+  -> Mealy eff (Event a) (Event [a])
+rollup limit seconds = mapAccumRequest go (Left Tick) (either id (\(_, _, _, ev) -> ev))
+  where
+    e a = Endo ([a] ++)
+    go :: Request -> Either (Event [a]) (UTCTime, Int, Endo [a], Event [a]) -> Event a -> Either (Event [a]) (UTCTime, Int, Endo [a], Event [a])
+    go _ (Left _) Tick = Left Tick
+    go req (Left _) (Event a) = Right (addUTCTime (fromIntegral seconds) (requestTime req), 1, mempty, Event [a])
+    go req (Right (end, n, acc, _)) Tick
+      | requestTime req >= end = Left (Event $ appEndo acc [])
+      | otherwise = Right (end, n, acc, Tick)
+    go req (Right (end, n, acc, _)) (Event a)
+      | requestTime req >= end = Left (Event $ appEndo acc [a])
+      | n < limit = Right (end, n+1, acc, Event [a])
+      | otherwise = Right (end, n+1, acc <> e a, Tick)
+
+-- Sliding window into the events
+sliding :: Int -> Mealy eff (Event a) [a]
+sliding size = mapAccum go [] id
+  where
+    go :: [a] -> Event a -> [a]
+    go acc Tick = acc
+    go acc (Event a) = let xs = acc ++ [a] in drop (max 0 (length xs - size)) xs
+
+fixed :: Int -> Mealy eff (Event a) [a]
+fixed seconds = mapAccumRequest go Nothing (maybe [] ((`appEndo` []) . snd))
+  where
+    e a = Endo ([a] ++)
+    go :: Request -> Maybe (UTCTime, Endo [a]) -> Event a -> Maybe (UTCTime, Endo [a])
+    go req Nothing Tick = Just (addUTCTime (fromIntegral seconds) (requestTime req), mempty)
+    go req Nothing (Event a) = Just (addUTCTime (fromIntegral seconds) (requestTime req), e a)
+    go req (Just (end, acc)) ev =
+      case ev of
+           Tick | requestTime req >= end -> Just (addUTCTime (fromIntegral seconds) end, mempty)
+                | otherwise -> Just (end, acc)
+           Event a | requestTime req >= end -> Just (addUTCTime (fromIntegral seconds) end, e a)
+                   | otherwise -> Just (end, acc <> e a)
+
+
+currentTime :: Mealy eff a LocalTime
+currentTime = Mealy mempty $ \_ Request{requestTime, requestTimeZone} _ ->
+  pure (Pair (utcToLocalTime requestTimeZone requestTime) currentTime)
+
+
+onEvent :: Mealy eff a () -> Mealy eff (Event a) ()
+onEvent f = events >>> (arr (const ()) ||| f)

src/HomeAssistant/Controller.hs

@@ -14,34 +14,32 @@ module HomeAssistant.Controller
   , entityRead'
   , entityBool
   , entityBool'
-  , Ruuvi(..)
-  , ruuvi
-  , ruuviTemperatures
-  , ruuviPressures
   , DoorState(..)
-  , door
   , light
-  , lightController
   , Presence(..)
   , presence
   , debug
   , traceEvent
+  , traceValue
   , switch
   , Target(..)
+  , brightness
+  , Light(..)
   ) where
 
-import AFRP (Mealy (..), eff, Event(..), hold, events, changes, filterA, (>>|), toEvent, Request)
+import AFRP (Mealy (..), Pair (..), eff, Event(..), events, filterA, (>>|), toEvent, Request)
 import Control.Arrow (Arrow(..), returnA)
 import Control.Category ((>>>))
-import Data.Aeson (Value)
+import Data.Aeson (Value, object, (.=))
 import qualified Data.Text as T
+import qualified Data.Set as S
 import Control.Lens (has, only, (^?), to)
-import Data.Aeson.Lens (key, _String)
+import Data.Aeson.Lens (key, _String, _Integral)
 import qualified Data.Text.Lens as TL
 import Data.Bool (bool)
 
 data Target = EntityId !T.Text | AreaId !T.Text
-  deriving (Show,Eq)
+  deriving (Show,Eq,Ord)
 
 data Service = Service
   { serviceDomain :: T.Text
@@ -67,21 +65,16 @@ debug = proc x -> do
   returnA -< x
 
 traceEvent :: Show a => HASS (Event a) (Event a)
-traceEvent = Mealy $ \nt req -> \case
-  Event a -> nt (Trace req a) >>= \() -> pure (Event a, traceEvent)
-  Tick -> pure (Tick, traceEvent)
+traceEvent = m
+  where
+    m = Mealy mempty $ \nt req -> \case
+      Event a -> nt (Trace req a) >>= \() -> pure (Pair (Event a) m)
+      Tick -> pure (Pair Tick m)
 
-ruuviTemperatures :: Mealy eff (Event Value) Double
-ruuviTemperatures = entityRead @Double "sensor.ruuvitag_b168_temperature" >>> hold 0
-
-ruuviPressures :: Mealy eff (Event Value) Double
-ruuviPressures = entityRead "sensor.ruuvitag_b168_pressure" >>> hold 0
-
-data Ruuvi = Ruuvi { ruuviTemperature :: Double, ruuviPressure :: Double }
-            deriving (Show, Eq)
-
-ruuvi :: Mealy eff (Event Value) (Event Ruuvi)
-ruuvi = (Ruuvi <$> ruuviTemperatures <*> ruuviPressures) >>> changes
+traceValue :: Show a => HASS a a
+traceValue = proc x -> do
+  eff Trace -< x
+  returnA -< x
 
 data DoorState = Open | Closed
   deriving (Show, Eq)
@@ -95,17 +88,21 @@ presence entityId =entityBool entityId
 
 
 
-door :: HASS (Event Value) (Event DoorState)
-door = entityBool "binary_sensor.makuuhuone_ovi_contact"
-  >>> arr (fmap (bool Closed Open))
-  >>> hold Open
-  >>> changes
+data Light
+  = Off
+  | On { brightnessPercentage :: Maybe Double }
 
 -- Turn off lights when door is closed
-light :: [Target] -> Bool -> Service
-light targets b = Service
+light :: [Target] -> Light -> Service
+light targets (On {brightnessPercentage}) = Service
   { serviceDomain="light"
-  , serviceName= bool "turn_off" "turn_on" b
+  , serviceName= "turn_on"
+  , serviceData=fmap (\pct -> object ["brightness_pct" .= pct]) brightnessPercentage
+  , serviceTarget=targets
+  }
+light targets Off = Service
+  { serviceDomain="light"
+  , serviceName= "turn_off"
   , serviceData=Nothing
   , serviceTarget=targets
   }
@@ -119,33 +116,25 @@ switch targets b = Service
   , serviceTarget=targets
   }
 
-lightController :: HASS (Event Value) (Event DoorState)
-lightController = proc ev -> do
-  doorState <- door -< ev
-  case doorState of
-       Event Open -> callService (light [EntityId "light.bedroom_masse"] False) -< ()
-       Event Closed -> callService (light [EntityId "light.bedroom_masse"] True) -< ()
-       _ -> returnA -< ()
-  returnA -< doorState
 
 entityChangeEvent :: T.Text -> Mealy eff (Event Value) (Event Value)
 entityChangeEvent entityId = entityChangeEvent' entityId >>> toEvent
 
 entityChangeEvent' :: T.Text -> Mealy eff (Event Value) (Either () Value)
-entityChangeEvent' entityId = events >>| filterA isEntity
+entityChangeEvent' entityId = Mealy (S.singleton entityId) $ runMealy (events >>| filterA isEntity)
   where
     isEntity :: Value -> Bool
-    isEntity = has (key "event" . key "data" . key "entity_id" . _String . only entityId)
+    isEntity = has (key "event" . key "variables" . key "trigger" . key "entity_id" . _String . only entityId)
 
 entityRead' :: (Read a) => T.Text -> Mealy eff (Event Value) (Either () a)
 entityRead' entityId = entityChangeEvent' entityId >>| (arr state >>> arr (maybe (Left ()) Right))
   where
-    state v = v ^? key "event" . key "data" . key "new_state" . key "state" . _String . TL.unpacked . to read
+    state v = v ^? key "event" . key "variables" . key "trigger" . key "to_state" . key "state" . _String . TL.unpacked . to read
 
 entityBool' :: T.Text -> Mealy eff (Event Value) (Either () Bool)
 entityBool' entityId = entityChangeEvent' entityId >>| (arr state >>> arr (maybe (Left ()) Right))
   where
-    state v = v ^? key "event" . key "data" . key "new_state" . key "state" . _String . TL.unpacked . to toBool
+    state v = v ^? key "event" . key "variables" . key "trigger" . key "to_state" . key "state" . _String . TL.unpacked . to toBool
     toBool = \case
       "on" -> True
       "off" -> False
@@ -156,3 +145,11 @@ entityRead entityId = entityRead' entityId >>> toEvent
 
 entityBool :: T.Text -> Mealy eff (Event Value) (Event Bool)
 entityBool entityId = entityBool' entityId >>> toEvent
+
+brightness :: T.Text -> HASS (Event Value) (Event Int)
+brightness entityId =
+  entityChangeEvent' entityId
+  >>| arr (maybe (Left ()) Right . eventBrightness)
+  >>> AFRP.toEvent
+  where
+    eventBrightness v = v ^? key "event" . key "variables" . key "trigger" . key "to_state" . key "attributes" . key "brightness" . _Integral

src/HomeAssistant/Controller/Bedroom.hs

@@ -4,12 +4,14 @@ module HomeAssistant.Controller.Bedroom where
 
 import HomeAssistant.Controller
 import qualified Data.Text as T
-import AFRP (Event (..), (>>|), toEvent, lMerge)
+import AFRP (Event (..), (>>|), toEvent, lMerge, duration, edge)
 import Data.Aeson (Value, object, (.=))
 import Control.Arrow ((>>>), returnA, arr, Arrow (..))
 import Control.Lens ((^?), to, traversed)
 import Data.Aeson.Lens (key, _String)
 import Data.Bool (bool)
+import Data.Time (NominalDiffTime)
+import qualified AFRP
 
 
 
@@ -44,7 +46,7 @@ bedroomPresenceController :: HASS (Event Value) ()
 bedroomPresenceController = proc x -> do
   p <- bedroomPresence -< x
   case p of
-       Event Unoccupied -> callService createBedroomScene >>> callService (light bedroomLights False) -< ()
+       Event Unoccupied -> callService createBedroomScene >>> callService (light bedroomLights Off) -< ()
        Event Occupied -> callService (activateScene "makuuhuone_lights_snapshot") -< ()
        _ -> returnA -< ()
 
@@ -77,7 +79,7 @@ ikeaQuickButton entityId =
   >>| arr (maybe (Left ()) Right . eventType)
   >>> toEvent
   where
-    eventType v = v ^? key "event" . key "data" . key "new_state" . key "attributes" . key "event_type" . _String . to toIkeaQuickButton . traversed
+    eventType v = v ^? key "event" . key "variables" . key "trigger" . key "to_state" . key "attributes" . key "event_type" . _String . to toIkeaQuickButton . traversed
 
 
 data BedroomControls
@@ -96,14 +98,14 @@ bedroomButtonController :: HASS (Event Value) ()
 bedroomButtonController = proc x -> do
   ev <- bedroomButton >>> traceEvent -< x
   case ev of
-       Event (Masse (OnButton ShortRelease)) -> callService (activateScene "scene.makuuhuone_masse") -< () -- this should be on release
+       Event (Masse (OnButton ShortRelease)) -> callService (activateScene "scene.makuuhuone_masse") -< ()
        Event (Masse (OnButton DoubleClick)) -> callService (activateScene "scene.makuuhuone_keski") -< ()
        Event (Masse (OnButton LongClick)) -> callService (activateScene "scene.makuuhuone_kirkas") -< ()
-       Event (Masse (OffButton _)) -> callService (light [AreaId "makuuhuone"] False) -< ()
+       Event (Masse (OffButton _)) -> callService (light [AreaId "makuuhuone"] Off) -< ()
        Event (Enishen (OnButton ShortRelease)) -> callService (activateScene "scene.makuuhuone_jemina") -< ()
        Event (Enishen (OnButton DoubleClick)) -> callService (activateScene "scene.makuuhuone_keski") -< ()
        Event (Enishen (OnButton LongClick)) -> callService (activateScene "scene.makuuhuone_kirkas") -< ()
-       Event (Enishen (OffButton _)) -> callService (light [AreaId "makuuhuone"] False) -< ()
+       Event (Enishen (OffButton _)) -> callService (light [AreaId "makuuhuone"] Off) -< ()
        _ -> returnA -< ()
 
 
@@ -121,3 +123,34 @@ bedroomDrawerController = proc x -> do
        _ -> returnA -< ()
   where
     entity = EntityId "switch.bedroom_drawer_light_masse"
+
+door :: HASS (Event Value) (Event DoorState)
+door = entityBool "binary_sensor.makuuhuone_ovi_contact"
+  >>> arr (fmap (bool Closed Open))
+
+waitFor :: NominalDiffTime -> HASS a (Event ())
+waitFor n = duration >>> arr (> n) >>> edge
+
+delayedDoor :: HASS (Event Value) (Event DoorState)
+delayedDoor = door
+  >>> AFRP.debounce 15
+  >>> AFRP.hold Open
+  >>> AFRP.changes
+  >>> traceEvent
+
+humidifierController :: HASS (Event Value) ()
+humidifierController = proc x -> do
+  st <- delayedDoor -< x
+  case st of
+       -- Turn the humidifier on when the door is closed
+       -- the humidifier is low-powered, no point in losing all the humidity
+       Event Open -> callService (humidifier False) -< ()
+       Event Closed -> callService (humidifier True) -< ()
+       _ -> returnA -< ()
+  where
+    humidifier state = Service
+      { serviceDomain="humidifier"
+      , serviceName= bool "turn_off" "turn_on" state
+      , serviceData= Nothing
+      , serviceTarget= [EntityId "humidifier.makuuhuone_ilmankostutin"]
+      }

src/HomeAssistant/Controller/Children.hs

@@ -0,0 +1,82 @@
+{-# LANGUAGE Arrows #-}
+{-# LANGUAGE OverloadedStrings #-}
+module HomeAssistant.Controller.Children where
+import HomeAssistant.Controller (HASS, callService, Target (AreaId), light, Light(..))
+import AFRP (Event)
+import qualified AFRP
+import Control.Arrow (Arrow(..), (>>>))
+import Data.Time (Day, TimeOfDay (..), localDay, LocalTime (..))
+import Data.Time.Calendar.OrdinalDate (WeekOfYear, mondayStartWeek)
+import Data.Functor.Contravariant (Predicate (..), (>$<))
+
+-- Let's see building some reasonable interface for utctime
+dow :: Day -> (WeekOfYear, Int)
+dow = mondayStartWeek
+
+
+weekday :: Predicate Day
+weekday = Predicate (betweenInclusive 1 5 . snd . dow)
+  where
+    betweenInclusive a b c = c >= a && c <= b
+
+time :: (Int, Int) -> Predicate TimeOfDay
+time (h,m) = mconcat
+  [ Predicate (equals h . todHour)
+  , Predicate (equals m . todMin)
+  ]
+  where
+    equals a b = a == b
+
+
+atTime :: Predicate LocalTime -> HASS a (Event ())
+atTime p = AFRP.currentTime
+  >>> arr (getPredicate p)
+  >>> AFRP.edge
+
+-- I don't have any proper presence sensors in their bedroom
+-- and they are notoriously bad at changing clothes in complete darkness
+-- So I have set up an automation that attempts to turn on the lights sometime
+-- before they leave for school and turns them off a bit later
+
+-- Don't mconcat these predicates they have && behavior
+-- if you mconcat the actual arrows, they combine the behaviors of the separate branches
+-- essentially becoming || behavior
+timersOff :: [Predicate LocalTime]
+timersOff =
+  [ day 1 <> at (08,15)
+  , day 2 <> at (09,15)
+  , day 3 <> at (08,15)
+  , day 4 <> at (08,15)
+  , day 5 <> at (08,15)
+  , at (18,57) -- debug
+  ]
+  where
+    dayOfWeek = snd . mondayStartWeek . localDay
+    at (h,m) = localTimeOfDay >$< Predicate (\TimeOfDay{todHour, todMin} -> todHour == h && todMin == m)
+    day n = dayOfWeek >$< Predicate (== n)
+
+timersOn :: [Predicate LocalTime]
+timersOn =
+  [ day 1 <> at (07,30)
+  , day 2 <> at (08,30)
+  , day 3 <> at (07,30)
+  , day 4 <> at (07,30)
+  , day 5 <> at (07,30)
+  , at (18,55) -- debug
+  ]
+  where
+    dayOfWeek = snd . mondayStartWeek . localDay
+    at (h,m) = localTimeOfDay >$< Predicate (\TimeOfDay{todHour, todMin} -> todHour == h && todMin == m)
+    day n = dayOfWeek >$< Predicate (== n)
+
+schoolLightController :: HASS a ()
+schoolLightController = lightsOn <> lightsOff
+
+
+lightsOn :: HASS a ()
+lightsOn = foldMap atTime timersOn
+  >>> AFRP.onEvent (callService (light [AreaId "lasten_makuuhuone"] On{brightnessPercentage = Just 100}))
+
+lightsOff :: HASS a ()
+lightsOff = foldMap atTime timersOff
+  >>> AFRP.onEvent (callService (light [AreaId "lasten_makuuhuone"] Off))

src/HomeAssistant/Controller/Ruuvi.hs

@@ -0,0 +1,31 @@
+{-# LANGUAGE Arrows #-}
+{-# LANGUAGE OverloadedStrings #-}
+module HomeAssistant.Controller.Ruuvi
+  ( Ruuvi(..)
+  , ruuvi
+  , ruuviTemperatures
+  , ruuviPressures
+  , ruuviController
+  ) where
+
+import AFRP (Mealy, Event, hold, changes, rollup)
+import Control.Category ((>>>))
+import Data.Aeson (Value)
+import HomeAssistant.Controller (entityRead, traceEvent, HASS)
+import Control.Arrow (Arrow(..))
+
+ruuviTemperatures :: Mealy eff (Event Value) Double
+ruuviTemperatures = entityRead @Double "sensor.ruuvitag_b168_temperature" >>> hold 0
+
+ruuviPressures :: Mealy eff (Event Value) Double
+ruuviPressures = entityRead "sensor.ruuvitag_b168_pressure" >>> hold 0
+
+data Ruuvi = Ruuvi { ruuviTemperature :: Double, ruuviPressure :: Double }
+            deriving (Show, Eq)
+
+ruuvi :: Mealy eff (Event Value) (Event Ruuvi)
+ruuvi = (Ruuvi <$> ruuviTemperatures <*> ruuviPressures) >>> changes
+
+
+ruuviController :: HASS (Event Value) ()
+ruuviController = ruuvi >>> rollup 1 30 >>> traceEvent >>> arr (const ())

src/HomeAssistant/Runtime.hs

@@ -13,30 +13,33 @@ module HomeAssistant.Runtime
   , runController
   ) where
 
-import AFRP (Event (..), Mealy (..), Request (..))
+import AFRP (Event (..), Mealy (..), Pair (..), Request (..))
 import Control.Concurrent.Async (async, waitAny)
 import Control.Concurrent.STM (atomically, dupTChan, readTChan)
 import Data.Aeson (Value)
 import qualified Data.Text as T
-import Data.Time (getCurrentTime)
+import Data.Time (getCurrentTime, getCurrentTimeZone)
 import Data.Void (Void, absurd)
 import HomeAssistant.Controller (HASS, HASSEff (..))
 import HomeAssistant.Runtime.Bus
 import HomeAssistant.Runtime.Connection (readerAction, writerAction)
 import HomeAssistant.Runtime.Supervisor (defaultBackoff, supervised)
 import Network.Socket (withSocketsDo)
-import System.Environment (getEnv)
-import HomeAssistant.Controller.Bedroom (bedroomPresenceController, bedroomButtonController, bedroomDrawerController)
+import System.Environment (getEnv, lookupEnv)
+import HomeAssistant.Controller.Bedroom (bedroomPresenceController, bedroomButtonController, bedroomDrawerController, humidifierController)
 import Data.UUID (UUID, toText)
 import qualified Data.UUID.V4 as UUID.V4
 import Katip (runKatipT, logF, sl, Severity (..), ls, Namespace (Namespace), runKatipContextT)
 import Control.Monad.IO.Class (liftIO, MonadIO)
 import Control.Monad.Fix (MonadFix)
+import HomeAssistant.Controller.Ruuvi (ruuviController)
+import HomeAssistant.Controller.Children (schoolLightController)
 
-step :: (MonadFix m, MonadIO m) => (forall x. eff x -> m x) -> UUID -> Mealy eff a b -> a -> m (b, Mealy eff a b)
-step nt trace (Mealy f) a = do
+step :: (MonadFix m, MonadIO m) => (forall x. eff x -> m x) -> UUID -> Mealy eff a b -> a -> m (Pair b (Mealy eff a b))
+step nt trace (Mealy _ f) a = do
   now <- liftIO getCurrentTime
-  f nt (Request now trace) a
+  tz <- liftIO getCurrentTimeZone
+  f nt (Request now tz trace) a
 
 data Controller = forall b. Controller T.Text (HASS (Event Value) b) Bool
 
@@ -45,6 +48,9 @@ controllers =
   [ Controller "bedroom-presence" bedroomPresenceController False
   , Controller "bedroom-button" bedroomButtonController False -- This works but leaving for vacation
   , Controller "bedroom-drawer" bedroomDrawerController True
+  , Controller "bedroom-humidifier" humidifierController False
+  , Controller "ruuvi-controller" ruuviController False
+  , Controller "school-light-controller" schoolLightController True
   ]
 
 -- | Steps the machine for every inbound message; service calls go to the
@@ -59,14 +65,19 @@ runController bus (Controller name machine _enabled) = do
       msg <- atomically (readTChan inbound)
       uuid <- UUID.V4.nextRandom
       let ns = Namespace [name]
-      (_, f') <- step (runKatipContextT (busLogEnv bus) () ns . channelHassEval bus) uuid f (Event msg)
+      Pair _ f' <- step (runKatipContextT (busLogEnv bus) () ns . channelHassEval bus) uuid f msg
       go inbound f'
 
 defaultMain :: IO ()
-defaultMain = withSocketsDo $ withBus $ \bus -> do
+defaultMain = withSocketsDo $ do
+  severity <- maybe InfoS (const DebugS) <$> lookupEnv "HA_DEBUG"
+  withBus severity $ \bus -> do
     token <- getEnv "HA_TOKEN"
-  let workers =
-        [ ("reader", readerAction "last-resort-redux" 8123 token bus)
+    host <- getEnv "HA_HOST"
+    let active = [c | c@(Controller _ _ True) <- controllers]
+        ents = foldMap (\(Controller _ m _) -> entities m) active
+        workers =
+          [ ("reader", readerAction host 8123 token ents bus)
           , ("writer", writerAction bus)
           ] ++ [ (name, runController bus c) | c@(Controller name _ True) <- controllers ]
     as <- mapM (\(name, act) -> async (supervised name defaultBackoff act)) workers

src/HomeAssistant/Runtime/Bus.hs

@@ -26,23 +26,23 @@ import Katip (LogEnv, closeScribes, mkHandleScribe, ColorStrategy (..), permitIt
 import Control.Exception (bracket)
 import System.IO (stdout)
 import Data.UUID (toText)
-import AFRP (Request(..))
+import AFRP (Request(..), Event(..))
 import Control.Monad.IO.Class (MonadIO, liftIO)
 
 -- | Shared runtime state: inbound is a broadcast channel (controllers
 -- read from 'dupTChan' copies), outbound queues service calls for the
 -- writer, conn holds the current websocket (Nothing before first connect).
 data Bus = Bus
-  { busInbound  :: TChan Value
+  { busInbound  :: TChan (Event Value)
   , busOutbound :: TChan (Request, Service)
   , busConn     :: TVar (Maybe Connection)
   , busGen      :: CallIdGen
   , busLogEnv   :: LogEnv
   }
 
-withBus :: (Bus -> IO a) -> IO a
-withBus callback = do
-  handleScribe <- mkHandleScribe ColorIfTerminal stdout (permitItem DebugS) V2
+withBus :: Severity -> (Bus -> IO a) -> IO a
+withBus severity callback = do
+  handleScribe <- mkHandleScribe ColorIfTerminal stdout (permitItem severity) V2
   let makeLogEnv = registerScribe "stdout" handleScribe defaultScribeSettings =<< initLogEnv "hass-controller" "production"
   -- closeScribes will stop accepting new logs, flush existing ones and clean up resources
   bracket makeLogEnv closeScribes $ \le -> do

src/HomeAssistant/Runtime/Connection.hs

@@ -5,23 +5,30 @@ module HomeAssistant.Runtime.Connection
   ( readerAction
   , writerAction
   , encodeService
+  , dedupeBatch
   ) where
 
 import Control.Concurrent.STM
-  ( atomically
+  ( TChan
+  , atomically
   , readTChan
   , readTVar
   , retry
+  , tryReadTChan
   , writeTChan
   , writeTVar
   )
+import Control.Concurrent.Async (race)
+import Control.Concurrent (threadDelay)
 import Control.Exception (onException)
 import Control.Exception.Annotated (throw)
 import Control.Lens ((^?))
-import Control.Monad (forever)
+import Control.Monad (forever, forM_)
 import Data.Aeson (Value, eitherDecode, encode, object, (.=))
 import Data.Aeson.Lens (key, _String)
-import qualified Data.ByteString.Lazy as BL
+import Data.List (sort)
+import qualified Data.Map.Strict as M
+import qualified Data.Set as S
 import qualified Data.Text as T
 import Data.Void (Void)
 import HomeAssistant.Controller (Service (..), Target (..))
@@ -30,16 +37,16 @@ import HomeAssistant.Runtime.Supervisor (Fatal (..))
 import qualified Network.WebSockets as WS
 import Katip (runKatipContextT, sl, logFM, Severity (..), ls)
 import Data.UUID (toText)
-import AFRP (Request(..))
+import AFRP (Request(..), Event(..))
 
 -- | Connect, authenticate, subscribe, then receive and broadcast forever.
 -- Restarting this action reconnects. All setup sends happen before the
 -- connection is published in the bus, so only the writer sends afterwards.
-readerAction :: String -> Int -> String -> Bus -> IO Void
-readerAction host port token bus =
+readerAction :: String -> Int -> String -> S.Set T.Text -> Bus -> IO Void
+readerAction host port token ents bus =
   WS.runClient host port "/api/websocket" $ \conn -> do
     handshake conn token
-    subscribe bus conn
+    subscribe bus conn ents
     atomically $ writeTVar (busConn bus) (Just conn)
     putStrLn "[reader] connected"
     -- Unpublish on exit so the writer blocks and the backlog survives the outage.
@@ -62,23 +69,34 @@ expectType expected msg =
     Just t | t == expected -> pure ()
     _ -> throw (Fatal $ "expected " <> expected <> ", got: " <> T.pack (show msg))
 
-subscribe :: Bus -> WS.Connection -> IO ()
-subscribe bus conn = do
+subscribe :: Bus -> WS.Connection -> S.Set T.Text -> IO ()
+subscribe bus conn ents =
+  forM_ (S.toList ents) $ \entityId -> do
+    print entityId
     sid <- generateCallId (busGen bus)
     WS.sendTextData conn $ encode $ object
       [ "id"         .= sid
-    , "type"       .= ("subscribe_events" :: T.Text)
-    , "event_type" .= ("state_changed" :: T.Text)
+      , "type"       .= ("subscribe_trigger" :: T.Text)
+      , "trigger"    .= object
+        [ "platform" .= ("state" :: T.Text)
+        , "entity_id" .= entityId
+        ]
       ]
 
 -- | Undecodable messages are skipped: reconnecting cannot fix a decode
 -- problem, so crashing here would only produce a hot restart loop.
+--
+-- Each read races a one-second timeout: a timeout broadcasts 'Tick' so
+-- time-based primitives (debounce, rollup, fixed, ...) keep advancing
+-- even when no state changes arrive.
 receiveLoop :: Bus -> WS.Connection -> IO Void
 receiveLoop bus conn = forever $ do
-  msg <- WS.receiveData conn :: IO BL.ByteString
-  case eitherDecode msg of
+  winner <- race (threadDelay 1_000_000) (WS.receiveData conn)
+  case winner of
+    Left () -> atomically $ writeTChan (busInbound bus) Tick
+    Right msg -> case eitherDecode msg of
       Left err -> putStrLn $ "[reader] skipping undecodable message: " <> err
-    Right v  -> atomically $ writeTChan (busInbound bus) v
+      Right v  -> atomically $ writeTChan (busInbound bus) (Event v)
 
 receiveJSON :: WS.Connection -> IO Value
 receiveJSON conn = do
@@ -87,16 +105,41 @@ receiveJSON conn = do
     Left err -> throw (Fatal $ "Invalid JSON from Home Assistant: " <> T.pack err)
     Right x  -> pure x
 
-writerAction :: Bus -> IO Void
-writerAction bus = forever $ do
-  (request, svc) <- atomically $ readTChan (busOutbound bus)
-  conn <- atomically $ readTVar (busConn bus) >>= maybe retry pure
+-- | Floor between sends within a batch: 100ms, so a many-distinct-target
+-- flood still caps at ~10 sends/sec even after dedupe.
+minInterval :: Int
+minInterval = 100000
+
+-- | Non-blocking drain of everything queued on a channel. Returns items
+-- oldest-first (FIFO from the channel), so prepending the blocking
+-- `readTChan` item keeps the whole batch oldest-first for `dedupeBatch`.
+drainTry :: TChan a -> IO [a]
+drainTry chan = go []
+  where
+    go acc = do
+      m <- atomically $ tryReadTChan chan
+      case m of
+        Nothing -> pure (reverse acc)
+        Just x  -> go (x : acc)
+
+sendWithId :: Bus -> WS.Connection -> Request -> Service -> IO ()
+sendWithId bus conn request svc = do
   callId <- generateCallId (busGen bus)
   let textData = encode $ encodeService callId svc
   runKatipContextT (busLogEnv bus) (sl "traceId" (toText (requestTraceId request))) "connection" $
     logFM DebugS (ls textData)
   WS.sendTextData conn textData
 
+writerAction :: Bus -> IO Void
+writerAction bus = forever $ do
+  first <- atomically $ readTChan (busOutbound bus)
+  rest  <- drainTry (busOutbound bus)
+  let deduped = dedupeBatch (first : rest)
+  conn <- atomically $ readTVar (busConn bus) >>= maybe retry pure
+  forM_ deduped $ \(request, svc) -> do
+    sendWithId bus conn request svc
+    threadDelay minInterval
+
 encodeService :: Int -> Service -> Value
 encodeService callId Service{..} = object $
   [ "id"      .= callId
@@ -106,6 +149,18 @@ encodeService callId Service{..} = object $
   , "target"  .= targetObject serviceTarget
   ] <> maybe [] (\d -> ["service_data" .= d]) serviceData
 
+-- | Collapse a drained batch of outbound calls: the newest call per
+-- `(domain, service, sorted-targets)` survives; older duplicates are
+-- dropped. `serviceData` is not part of the key, so a newer `turn_on`
+-- with different brightness supersedes an older one to the same target.
+dedupeBatch :: [(Request, Service)] -> [(Request, Service)]
+dedupeBatch = M.elems . foldl' ins M.empty
+  where
+    ins m (req, svc) = M.insert (dedupeKey svc) (req, svc) m
+
+dedupeKey :: Service -> (T.Text, T.Text, [Target])
+dedupeKey Service{..} = (serviceDomain, serviceName, sort serviceTarget)
+
 -- | A single target encodes as a scalar; multiple encode as a list. Empty
 -- lists are omitted so Home Assistant receives only populated keys.
 targetObject :: [Target] -> Value

test/AFRPSpec.hs

@@ -0,0 +1,621 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+module AFRPSpec (spec) where
+
+import Control.Arrow (arr, (&&&), first, left)
+import Control.Category ((>>>))
+import Control.Monad.Fix (MonadFix (..))
+import Data.Foldable (for_)
+import AFRP
+import Data.Functor.Identity (Identity (..))
+import Data.List (sort)
+import qualified Data.Set as S
+import qualified Data.Text as T
+import Data.Time (NominalDiffTime, UTCTime (..), utc)
+import Data.UUID (nil)
+import Hedgehog
+import qualified Hedgehog.Gen as Gen
+import qualified Hedgehog.Range as Range
+import Test.Hspec
+import Test.Hspec.Hedgehog
+
+fakeRequest :: Request
+fakeRequest = Request (sec 0) utc nil
+
+sec :: Integer -> UTCTime
+sec n = UTCTime (toEnum 0) (fromIntegral n)
+
+runPure :: Mealy Identity a b -> [a] -> [b]
+runPure _ [] = []
+runPure m (a : as) = case runIdentity (AFRP.runMealy m id fakeRequest a) of
+  Pair b m' -> b : runPure m' as
+
+-- | Run a Mealy with a per-step wall clock (seconds since the day-0 epoch).
+runTimed :: Mealy Identity a b -> [(Integer, a)] -> [b]
+runTimed _ [] = []
+runTimed m ((s, a) : as) =
+  case runIdentity (AFRP.runMealy m id (Request (sec s) utc nil) a) of
+    Pair b m' -> b : runTimed m' as
+
+-- | A minimal State monad for observing effectful arrows (e.g. whenA gating).
+newtype St a = St { unSt :: Int -> (a, Int) }
+
+instance Functor St where
+  fmap f (St g) = St $ \s -> let (a, s') = g s in (f a, s')
+
+instance Applicative St where
+  pure a = St (\s -> (a, s))
+  St f <*> St x = St $ \s -> let (f', s') = f s; (a, s'') = x s' in (f' a, s'')
+
+instance Monad St where
+  St m >>= k = St $ \s -> let (a, s') = m s; (b, s'') = unSt (k a) s' in (b, s'')
+
+instance MonadFix St where
+  mfix f = St $ \s -> let (a, s') = unSt (f a) s in (a, s')
+
+runStEff :: Mealy St a b -> Int -> [a] -> ([b], Int)
+runStEff m s0 as = go m s0 as
+  where
+    go _ s [] = ([], s)
+    go m' s (a : rest) =
+      case unSt (AFRP.runMealy m' id fakeRequest a) s of
+        (Pair b m'', s') -> let (bs, s'') = go m'' s' rest in (b : bs, s'')
+
+spec :: Spec
+spec = describe "AFRP" $ do
+  entitiesSpec
+  holdSpec
+  eventsSpec
+  isEventSpec
+  tagSpec
+  toEventSpec
+  lMergeSpec
+  changesSpec
+  edgeSpec
+  dropFirstSpec
+  filterASpec
+  slidingSpec
+  mapAccumSpec
+  preMapAccumSpec
+  durationSpec
+  delayEventSpec
+  debounceSpec
+  rollupSpec
+  fixedSpec
+  effSpec
+  switchSpec
+  mapAccumRequestSpec
+  preMapAccumRequestSpec
+  whenASpec
+  thenASpec
+  sampleSpec
+
+holdSpec :: Spec
+holdSpec = describe "hold" $ do
+  it "holds initial value until an Event arrives" $
+    runPure (hold 'a') [Tick, Event 'b', Tick, Event 'c']
+      `shouldBe` ['a', 'b', 'b', 'c']
+
+  it "never changes on Tick" $
+    runPure (hold (0 :: Int)) (replicate 5 Tick) `shouldBe` replicate 5 (0 :: Int)
+
+eventsSpec :: Spec
+eventsSpec = describe "events" $ do
+  it "converts Tick to Left () and Event a to Right a" $
+    runPure events [Tick, Event 'a', Tick, Event 'b']
+      `shouldBe` [Left (), Right 'a', Left (), Right 'b']
+
+isEventSpec :: Spec
+isEventSpec = describe "isEvent" $ do
+  it "returns False for Tick" $
+    isEvent Tick `shouldBe` False
+
+  it "returns True for Event x" $
+    isEvent (Event ()) `shouldBe` True
+
+tagSpec :: Spec
+tagSpec = describe "tag" $ do
+  it "replaces value preserving structure" $ do
+    tag 'b' Tick `shouldBe` Tick
+    tag 'b' (Event 'a') `shouldBe` Event 'b'
+
+toEventSpec :: Spec
+toEventSpec = describe "toEvent" $ do
+  it "round-trips through events" $
+    runPure toEvent [Left (), Right 'a', Left ()]
+      `shouldBe` [Tick, Event 'a', Tick]
+
+  it "is inverse of events modulo Event/Either" $
+    runPure (events >>> toEvent) [Tick, Event 'a', Event 'b']
+      `shouldBe` [Tick, Event 'a', Event 'b']
+
+lMergeSpec :: Spec
+lMergeSpec = describe "lMerge" $ do
+  it "both Tick gives Tick" $
+    lMerge (Tick :: Event Int) (Tick :: Event Int) `shouldBe` Tick
+
+  it "prefers left Event" $
+    lMerge (Event (1 :: Int)) (Event (2 :: Int)) `shouldBe` Event (1 :: Int)
+
+  it "prefers right Event if left is Tick" $
+    lMerge Tick (Event (2 :: Int)) `shouldBe` Event (2 :: Int)
+
+  it "Tick is a left identity" $
+    hedgehog $ do
+      e <- forAll eventGen
+      lMerge Tick e === e
+
+  it "Tick is a right identity" $
+    hedgehog $ do
+      e <- forAll eventGen
+      lMerge e Tick === e
+
+  it "is associative" $
+    hedgehog $ do
+      a <- forAll eventGen
+      b <- forAll eventGen
+      c <- forAll eventGen
+      lMerge a (lMerge b c) === lMerge (lMerge a b) c
+
+changesSpec :: Spec
+changesSpec = describe "changes" $ do
+  it "first output is always Tick" $
+    runPure changes "hello" !! 0 `shouldBe` Tick
+
+  it "outputs Event only on value change" $
+    runPure changes "aaaabbbcca"
+      `shouldBe` [Tick, Tick, Tick, Tick
+                 , Event 'b', Tick, Tick
+                 , Event 'c', Tick
+                 , Event 'a'
+                 ]
+
+  it "first output is Tick, subsequent outputs are Event iff value changed" $
+    hedgehog $ do
+      xs <- forAll $ Gen.list (Range.linear 0 100) Gen.alpha
+      let out = runPure changes xs
+      length out === length xs
+      case out of
+        [] -> pure ()
+        (Tick : rest) -> do
+          let triples = zip3 xs (drop 1 xs) rest
+          for_ triples $ \(prev, curr, o) ->
+            if prev /= curr
+              then o === Event curr
+              else o === Tick
+        _ -> failure
+
+edgeSpec :: Spec
+edgeSpec = describe "edge" $ do
+  it "emits Event () only on rising edge" $
+    runPure edge [False, True, True, False, True]
+      `shouldBe` [Tick, Event (), Tick, Tick, Event ()]
+
+  it "starts from False, so first True is a rising edge" $
+    runPure edge [True, False, True]
+      `shouldBe` [Event (), Tick, Event ()]
+
+  it "Event () only on False -> True transition" $
+    hedgehog $ do
+      bs <- forAll $ Gen.list (Range.linear 0 50) Gen.bool
+      let out = runPure edge bs
+      length out === length bs
+      case (bs, out) of
+        ([], []) -> pure ()
+        (b : _, o : _) -> do
+          if b then o === Event () else o === Tick
+          let triples = zip3 bs (drop 1 bs) (drop 1 out)
+          for_ triples $ \(prev, curr, o') ->
+            if not prev && curr
+              then o' === Event ()
+              else o' === Tick
+        _ -> failure
+
+dropFirstSpec :: Spec
+dropFirstSpec = describe "dropFirst" $ do
+  it "drops the first Event, passes the rest" $
+    runPure dropFirst [Tick, Event 1, Event 2, Event 3]
+      `shouldBe` [Tick, Tick, Event 2, Event 3 :: Event Int]
+
+  it "passes Tick through untouched before first Event" $
+    runPure dropFirst [Tick, Tick, Tick :: Event Int]
+      `shouldBe` [Tick, Tick, Tick :: Event Int]
+
+  it "drops only the first Event, Ticks before it are inert" $
+    runPure dropFirst [Tick, Tick, Event 'a', Tick, Event 'b']
+      `shouldBe` [Tick, Tick, Tick, Tick, Event 'b' :: Event Char]
+
+filterASpec :: Spec
+filterASpec = describe "filterA" $ do
+  it "lets through values matching predicate" $
+    runPure (filterA (even @Int)) [1, 2, 3, 4]
+      `shouldBe` [Left (), Right 2, Left (), Right 4]
+
+  it "output is Right a iff predicate holds" $
+    hedgehog $ do
+      threshold <- forAll $ Gen.int (Range.linear (-10) 10)
+      let p = (> threshold)
+      xs <- forAll $ Gen.list (Range.linear 0 50) (Gen.int (Range.linear (-20) 20))
+      let out = runPure (filterA p) xs
+      length out === length xs
+      for_ (zip xs out) $ \(x, o) ->
+        if p x
+          then o === Right x
+          else o === Left ()
+
+slidingSpec :: Spec
+slidingSpec = describe "sliding" $ do
+  it "accumulates events up to the window size" $
+    runPure (sliding (3 :: Int)) [Event (1 :: Int), Event 2, Event 3, Event 4]
+      `shouldBe` [[1], [1, 2], [1, 2, 3], [2, 3, 4]]
+
+  it "Ticks don't change the accumulator" $
+    runPure (sliding 2) [Event (1 :: Int), Tick, Event 2]
+      `shouldBe` [[1], [1], [1, 2]]
+
+  it "empty list stays empty" $
+    runPure (sliding (5 :: Int)) ([] :: [Event Int]) `shouldBe` []
+
+  it "output length never exceeds window size" $
+    hedgehog $ do
+      n <- forAll $ Gen.int (Range.constant 1 10)
+      evs <- forAll $ Gen.list (Range.linear 0 20) (Gen.frequency
+        [(3, Event <$> Gen.alpha), (1, pure Tick)])
+      let out = runPure (sliding n) evs
+      for_ out $ \xs -> assert (length xs <= n)
+
+mapAccumSpec :: Spec
+mapAccumSpec = describe "mapAccum" $ do
+  it "running sum" $
+    runPure (mapAccum (+) (0 :: Int) id) [1, 2, 3]
+      `shouldBe` [1, 3, 6]
+
+  it "post-state extraction: output uses state after applying f" $
+    runPure (mapAccum (\s x -> s ++ [x]) ([] :: [Int]) id) [1, 2, 3]
+      `shouldBe` [[1], [1, 2], [1, 2, 3]]
+
+  it "output equals running sum of all inputs so far" $
+    hedgehog $ do
+      xs <- forAll $ Gen.list (Range.linear 0 50) (Gen.int (Range.linear (-100) 100))
+      let out = runPure (mapAccum (+) (0 :: Int) id) xs
+      length out === length xs
+      for_ (zip3 [0 ..] xs out) $ \(i, _x, cur) ->
+        cur === sum (take (i + 1) xs)
+
+preMapAccumSpec :: Spec
+preMapAccumSpec = describe "preMapAccum" $ do
+  it "running sum with pre-state extraction" $
+    runPure (preMapAccum (+) (0 :: Int) id) [1, 2, 3]
+      `shouldBe` [0, 1, 3]
+
+  it "pre-state extraction: output uses state before applying f" $
+    runPure (preMapAccum (\s x -> s ++ [x]) ([] :: [Int]) id) [1, 2, 3]
+      `shouldBe` [[], [1], [1, 2]]
+
+  it "output equals running sum before current input" $
+    hedgehog $ do
+      xs <- forAll $ Gen.list (Range.linear 0 50) (Gen.int (Range.linear (-100) 100))
+      let out = runPure (preMapAccum (+) (0 :: Int) id) xs
+      length out === length xs
+      for_ (zip3 [0 ..] xs out) $ \(i, _x, cur) ->
+        cur === sum (take i xs)
+
+durationSpec :: Spec
+durationSpec = describe "duration" $ do
+  it "first sample is 0, then elapsed time since first sample" $
+    runTimed duration [(0, 'a'), (5, 'b'), (10, 'c')]
+      `shouldBe` [0, 5, 10]
+
+  it "measures from the first observation, not the most recent" $
+    runTimed duration [(2, 'a'), (3, 'b'), (7, 'c')]
+      `shouldBe` [0, 1, 5]
+
+  it "output i equals times[i] - times[0]" $
+    hedgehog $ do
+      secs <- forAll $ Gen.list (Range.linear 0 50) (Gen.int (Range.linear 0 1000))
+      let out = runTimed duration [(fromIntegral s, ()) | s <- secs]
+      length out === length secs
+      case secs of
+        [] -> pure ()
+        (t0 : _) -> for_ (zip secs out) $ \(s, d) ->
+          d === fromIntegral (s - t0)
+
+delayEventSpec :: Spec
+delayEventSpec = describe "delayEvent" $ do
+  let delay = 5 :: NominalDiffTime
+
+  it "emits a queued event once the delay has elapsed" $
+    runTimed (delayEvent delay)
+             [(0, Event 'a'), (3, Tick), (6, Tick)]
+      `shouldBe` [Tick, Tick, Event 'a']
+
+  it "preserves order when multiple events are queued" $
+    runTimed (delayEvent delay)
+             [(0, Event 'a'), (1, Event 'b'), (10, Tick), (12, Tick)]
+      `shouldBe` [Tick, Tick, Event 'a', Event 'b']
+
+  it "emits nothing on pure Tick input" $
+    runTimed (delayEvent delay) [(0, Tick :: Event Char), (10, Tick)]
+      `shouldBe` [Tick, Tick]
+
+  it "emits exactly one output Event per input Event" $
+    hedgehog $ do
+      evs <- forAll $ Gen.list (Range.linear 0 30) eventGen
+      let evs' = evs ++ replicate 3 Tick
+          n = length [() | Event _ <- evs]
+          out = runTimed (delayEvent (1 :: NominalDiffTime))
+                         (zip [0, 2 ..] evs')
+      length [() | Event _ <- out] === n
+
+debounceSpec :: Spec
+debounceSpec = describe "debounce" $ do
+  let delay = 5 :: NominalDiffTime
+
+  it "fires the last event after the quiet period" $
+    runTimed (debounce delay)
+             [(0, Event 'a'), (3, Tick), (6, Tick)]
+      `shouldBe` [Tick, Tick, Event 'a']
+
+  it "a newer event before firing resets the timer" $
+    runTimed (debounce delay)
+             [(0, Event 'a'), (3, Event 'b'), (6, Tick), (8, Tick)]
+      `shouldBe` [Tick, Tick, Tick, Event 'b']
+
+  it "collapses a burst into a single emission" $
+    runTimed (debounce delay)
+             [(0, Event 'a'), (1, Event 'b'), (2, Event 'c'), (10, Tick)]
+      `shouldBe` [Tick, Tick, Tick, Event 'c']
+
+  it "emits no more output Events than input Events" $
+    hedgehog $ do
+      evs <- forAll $ Gen.list (Range.linear 0 30) eventGen
+      let nIn = length [() | Event _ <- evs]
+          out = runTimed (debounce (1 :: NominalDiffTime))
+                         (zip [0, 2 ..] evs)
+          nOut = length [() | Event _ <- out]
+      assert (nOut <= nIn)
+
+rollupSpec :: Spec
+rollupSpec = describe "rollup" $ do
+  it "passes the first `limit` events through immediately, then bursts" $
+    runTimed (rollup 2 10)
+             [ (0, Event 'a'), (1, Event 'b')
+             , (2, Event 'c'), (3, Event 'd')
+             , (12, Tick)
+             ]
+      `shouldBe` [ Event ['a'], Event ['b']
+                 , Tick, Tick
+                 , Event ['c', 'd']
+                 ]
+
+  it "flushes the accumulator when the window ends on a Tick" $
+    runTimed (rollup 1 10)
+             [(0, Event 'a'), (1, Event 'b'), (2, Tick), (12, Tick)]
+      `shouldBe` [Event ['a'], Tick, Tick, Event ['b']]
+
+  it "is idle (Tick) until the first Event" $
+    runTimed (rollup 2 10) [(0, Tick :: Event Char), (1, Tick)]
+      `shouldBe` [Tick, Tick]
+
+  it "every input Event appears exactly once across the outputs" $
+    hedgehog $ do
+      limit <- forAll $ Gen.int (Range.constant 1 5)
+      evs <- forAll $ Gen.list (Range.linear 0 30) eventGen
+      let evs' = evs ++ replicate 10 Tick
+          times = [0 ..]
+          out = runTimed (rollup limit 5) (zip times evs')
+          emitted = concat [xs | Event xs <- out]
+      sort emitted === sort [x | Event x <- evs]
+
+fixedSpec :: Spec
+fixedSpec = describe "fixed" $ do
+  it "accumulates events within a window and rolls over on expiry" $
+    runTimed (fixed 10)
+             [ (0, Event 'a'), (1, Event 'b'), (2, Tick)
+             , (12, Event 'c'), (13, Tick)
+             ]
+      `shouldBe` [ ['a'], ['a', 'b'], ['a', 'b']
+                 , ['c'], ['c']
+                 ]
+
+  it "starts a window even on a leading Tick" $
+    runTimed (fixed 10)
+             [ (0, Tick), (1, Event 'a')
+             , (12, Tick), (13, Tick)
+             , (25, Event 'b')
+             ]
+      `shouldBe` [ [], ['a'], [], [], ['b'] ]
+
+  it "output is always the current window's accumulated list" $
+    hedgehog $ do
+      w <- forAll $ Gen.int (Range.constant 1 10)
+      evs <- forAll $ Gen.list (Range.linear 0 30) eventGen
+      let out = runTimed (fixed w) (zip [0 ..] evs)
+          expected =
+            [ [ x | Event x <- take (i - lo + 1) (drop lo evs) ]
+            | i <- [0 .. length evs - 1]
+            , let lo = (i `div` w) * w
+            ]
+      out === expected
+
+eventGen :: Gen (Event Char)
+eventGen = Gen.frequency
+  [ (3, Event <$> Gen.alpha)
+  , (1, pure Tick)
+  ]
+
+effSpec :: Spec
+effSpec = describe "eff" $ do
+  it "lifts a pure effect function into a stateless Mealy" $
+    runPure (eff (\_ x -> Identity (x + 1))) [1, 2, 3]
+      `shouldBe` [2 :: Int, 3, 4]
+
+  it "output equals f(input) for every step" $
+    hedgehog $ do
+      xs <- forAll $ Gen.list (Range.linear 0 50) (Gen.int (Range.linear (-100) 100))
+      let out = runPure (eff (\_ x -> Identity (x * 2))) xs
+      out === map (* 2) xs
+
+switchSpec :: Spec
+switchSpec = describe "switch" $ do
+  it "switches to the continuation at the first Event" $
+    runPure (switch (arr (\x -> (x, if x >= (3 :: Int) then Event () else Tick)))
+                     (const (arr (const 99))))
+             [1, 2, 3, 4, 5]
+      `shouldBe` [1, 2, 99, 99, 99]
+
+  it "never switches if no Event is emitted" $
+    runPure (switch (arr (\x -> (x, Tick :: Event ())))
+                     (const (arr (const 99))))
+             [1, 2, 3]
+      `shouldBe` [1, 2, 3 :: Int]
+
+  it "prefix outputs come from the first arrow, suffix from the continuation" $
+    hedgehog $ do
+      threshold <- forAll $ Gen.int (Range.linear (-20) 20)
+      xs <- forAll $ Gen.list (Range.linear 0 30) (Gen.int (Range.linear (-20) 20))
+      let firstArr = arr (\x -> (x, if x >= threshold then Event () else Tick))
+          out = runPure (switch firstArr (const (arr (const 99)))) xs
+          (pre, _post) = break (>= threshold) xs
+      take (length pre) out === pre
+      drop (length pre) out === replicate (length xs - length pre) 99
+
+mapAccumRequestSpec :: Spec
+mapAccumRequestSpec = describe "mapAccumRequest" $ do
+  it "accumulates request times, post-state extraction" $
+    runTimed (mapAccumRequest (\req s _ -> s ++ [requestTime req]) [] id)
+             [(0, 'a'), (5, 'b'), (10, 'c')]
+      `shouldBe` [ [sec 0], [sec 0, sec 5], [sec 0, sec 5, sec 10] ]
+
+  it "output i is every request time seen so far" $
+    hedgehog $ do
+      secs' <- forAll $ Gen.list (Range.linear 0 30) (Gen.int (Range.linear 0 100))
+      let out = runTimed (mapAccumRequest (\req s _ -> s ++ [requestTime req]) [] id)
+                         [(fromIntegral s, ()) | s <- secs']
+          expected = [ map (sec . fromIntegral) (take (i + 1) secs') | i <- [0 .. length secs' - 1] ]
+      out === expected
+
+preMapAccumRequestSpec :: Spec
+preMapAccumRequestSpec = describe "preMapAccumRequest" $ do
+  it "accumulates request times, pre-state extraction" $
+    runTimed (preMapAccumRequest (\req s _ -> s ++ [requestTime req]) [] id)
+             [(0, 'a'), (5, 'b'), (10, 'c')]
+      `shouldBe` [ [], [sec 0], [sec 0, sec 5] ]
+
+  it "output i is every request time before the current step" $
+    hedgehog $ do
+      secs' <- forAll $ Gen.list (Range.linear 0 30) (Gen.int (Range.linear 0 100))
+      let out = runTimed (preMapAccumRequest (\req s _ -> s ++ [requestTime req]) [] id)
+                         [(fromIntegral s, ()) | s <- secs']
+          expected = [ map (sec . fromIntegral) (take i secs') | i <- [0 .. length secs' - 1] ]
+      out === expected
+
+whenASpec :: Spec
+whenASpec = describe "whenA" $ do
+  let counter = eff (\_ (_ :: Int) -> St (\s -> ((), s + 1)))
+
+  it "output is always () regardless of the predicate" $
+    fst (runStEff (whenA (> 5) counter) 0 [1, 6, 2, 7])
+      `shouldBe` [(), (), (), ()]
+
+  it "runs the inner arrow only when the predicate holds" $
+    snd (runStEff (whenA (> 5) counter) 0 [1, 6, 2, 7])
+      `shouldBe` 2
+
+  it "never runs the inner arrow when the predicate is always false" $
+    snd (runStEff (whenA (const False) counter) 0 [1, 6, 2, 7])
+      `shouldBe` 0
+
+  it "runs the inner arrow on every input when the predicate is always true" $
+    snd (runStEff (whenA (const True) counter) 0 [1, 6, 2, 7])
+      `shouldBe` 4
+
+thenASpec :: Spec
+thenASpec = describe "thenA" $ do
+  it "short-circuits on Left and continues on Right" $
+    runPure (filterA (even @Int) `thenA` filterA (> 3)) [1..6]
+      `shouldBe` [Left (), Left (), Left (), Right 4, Left (), Right 6]
+
+  it "(>>|) is an infix alias for thenA" $
+    runPure (filterA (even @Int) >>| filterA (> 3)) [1..6]
+      `shouldBe` [Left (), Left (), Left (), Right 4, Left (), Right 6]
+
+  it "second stage runs only when the first produces Right" $
+    hedgehog $ do
+      xs <- forAll $ Gen.list (Range.linear 0 30) (Gen.int (Range.linear (-20) 20))
+      let out = runPure (filterA (even @Int) >>| filterA (> 0)) xs
+          expected =
+            [ if odd x then Left ()
+              else if x > 0 then Right x
+              else Left ()
+            | x <- xs ]
+      out === expected
+
+sampleSpec :: Spec
+sampleSpec = describe "sample" $ do
+  it "tags the current value onto the Event structure" $
+    runPure sample [(1, Tick), (2, Event 'a'), (3, Tick)]
+      `shouldBe` [Tick, Event @Int 2, Tick]
+
+  it "output is Event a iff the input event is present" $
+    hedgehog $ do
+      vals <- forAll $ Gen.list (Range.linear 0 30) (Gen.int (Range.linear 0 100))
+      evs <- forAll $ Gen.list (Range.linear 0 30) eventGen
+      let n = min (length vals) (length evs)
+          ps = take n $ zip vals evs
+          out = runPure sample ps
+      for_ (zip ps out) $ \((v, ev), o) ->
+        o === tag v ev
+
+-- | A stateless arrow carrying a fixed entity set, for testing propagation.
+subscribed :: S.Set T.Text -> Mealy Identity Int Int
+subscribed ents = Mealy ents $ \_ _ a -> pure (Pair a (subscribed ents))
+
+-- | Same as 'subscribed' but yields a function, for testing '<*>'.
+subscribedF :: S.Set T.Text -> Mealy Identity Int (Int -> Int)
+subscribedF ents = Mealy ents $ \_ _ a -> pure (Pair (a +) (subscribedF ents))
+
+entitiesSpec :: Spec
+entitiesSpec = describe "entities" $ do
+  it "id carries no entities" $
+    entities (arr id :: Mealy Identity Int Int) `shouldBe` S.empty
+
+  it "arr carries no entities" $
+    entities (arr (+ 1) :: Mealy Identity Int Int) `shouldBe` S.empty
+
+  it "eff carries no entities" $
+    entities (eff (\_ x -> Identity (x + 1 :: Int))) `shouldBe` S.empty
+
+  it "primitive combinators carry no entities" $ do
+    entities (hold 'a') `shouldBe` S.empty
+    entities (changes @Int) `shouldBe` S.empty
+    entities edge `shouldBe` S.empty
+    entities (sliding (3 :: Int)) `shouldBe` S.empty
+
+  it "Category (.) unions entity sets" $
+    entities (subscribed (S.singleton "a") >>> subscribed (S.singleton "b"))
+      `shouldBe` S.fromList ["a", "b"]
+
+  it "Applicative (<*>) unions entity sets" $
+    entities (subscribedF (S.singleton "a") <*> subscribed (S.singleton "b"))
+      `shouldBe` S.fromList ["a", "b"]
+
+  it "Arrow (&&&) unions entity sets" $
+    entities (subscribed (S.singleton "a") &&& subscribed (S.singleton "b"))
+      `shouldBe` S.fromList ["a", "b"]
+
+  it "(>>>) unions entity sets" $
+    entities (subscribed (S.singleton "a") >>> arr id >>> subscribed (S.singleton "b"))
+      `shouldBe` S.fromList ["a", "b"]
+
+  it "left preserves the entity set" $
+    entities (left (subscribed (S.singleton "a")) :: Mealy Identity (Either Int Int) (Either Int Int))
+      `shouldBe` S.singleton "a"
+
+  it "first preserves the entity set" $
+    entities (first (subscribed (S.singleton "a")) :: Mealy Identity (Int, Int) (Int, Int))
+      `shouldBe` S.singleton "a"
+
+  it "fmap preserves the entity set" $
+    entities (fmap (+ 1) (subscribed (S.singleton "a")))
+      `shouldBe` S.singleton "a"

test/BedroomSpec.hs

@@ -0,0 +1,134 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+module BedroomSpec (spec) where
+
+import AFRP (Event (..), entities)
+import qualified Data.Set as S
+import qualified Data.Text as T
+import Data.Aeson (Value)
+import HomeAssistant.Controller
+import HomeAssistant.Controller.Bedroom
+import Support
+import Test.Hspec
+
+masseButton :: T.Text -> Event Value
+masseButton = Event . buttonEvent "event.bedroom_quick_remote_masse_action"
+
+enishenButton :: T.Text -> Event Value
+enishenButton = Event . buttonEvent "event.bedroom_quick_jemina_action"
+
+drawerState :: T.Text -> Event Value
+drawerState = Event . stateEvent "binary_sensor.bedroom_nightstand_drawer_sensor_masse_contact"
+
+spec :: Spec
+spec = describe "Bedroom" $ do
+  entitySpec
+  drawerSpec
+  buttonSpec
+
+entitySpec :: Spec
+entitySpec = describe "entities" $ do
+  it "entityChangeEvent' registers its entity id" $
+    entities (entityChangeEvent' "sensor.foo")
+      `shouldBe` S.singleton "sensor.foo"
+
+  it "entityRead / entityBool inherit the entity id" $ do
+    entities (entityRead @Double "sensor.bar") `shouldBe` S.singleton "sensor.bar"
+    entities (entityBool "binary_sensor.baz") `shouldBe` S.singleton "binary_sensor.baz"
+
+  it "bedroomDrawerController subscribes to the drawer sensor" $
+    entities bedroomDrawerController
+      `shouldBe` S.singleton "binary_sensor.bedroom_nightstand_drawer_sensor_masse_contact"
+
+  it "bedroomButtonController subscribes to both remote event entities" $
+    entities bedroomButtonController
+      `shouldBe` S.fromList
+        [ "event.bedroom_quick_remote_masse_action"
+        , "event.bedroom_quick_jemina_action"
+        ]
+
+  it "bedroomPresenceController subscribes to the presence sensor" $
+    entities bedroomPresenceController
+      `shouldBe` S.singleton "binary_sensor.presence_sensor_bedroom_occupancy"
+
+  it "humidifierController subscribes to the door sensor" $
+    entities humidifierController
+      `shouldBe` S.singleton "binary_sensor.makuuhuone_ovi_contact"
+
+drawerSpec :: Spec
+drawerSpec = describe "bedroomDrawerController" $ do
+  let entity = EntityId "switch.bedroom_drawer_light_masse"
+
+  it "turns the drawer light on when the drawer opens" $
+    services (runHASS bedroomDrawerController [drawerState "on"])
+      `shouldBe` [[switch [entity] True]]
+
+  it "turns the drawer light off when the drawer closes" $
+    services (runHASS bedroomDrawerController [drawerState "off"])
+      `shouldBe` [[switch [entity] False]]
+
+  it "toggles the light as the drawer opens and closes" $
+    services (runHASS bedroomDrawerController
+              [ drawerState "on", drawerState "off", drawerState "on" ])
+      `shouldBe` [ [switch [entity] True]
+                 , [switch [entity] False]
+                 , [switch [entity] True]
+                 ]
+
+  it "ignores state events for other entities" $
+    services (runHASS bedroomDrawerController
+              [Event (stateEvent "binary_sensor.some_other_contact" "on")])
+      `shouldBe` [[]]
+
+  it "does nothing on Tick" $
+    services (runHASS bedroomDrawerController [Tick])
+      `shouldBe` [[]]
+
+buttonSpec :: Spec
+buttonSpec = describe "bedroomButtonController" $ do
+  let area = AreaId "makuuhuone"
+
+  it "Masse single click turns on his nightstand scene (lowest)" $
+    services (runHASS bedroomButtonController [masseButton "1_short_release"])
+      `shouldBe` [[activateScene "scene.makuuhuone_masse"]]
+
+  it "Masse double click turns on the middle scene" $
+    services (runHASS bedroomButtonController [masseButton "1_double_press"])
+      `shouldBe` [[activateScene "scene.makuuhuone_keski"]]
+
+  it "Masse long click turns on the bright scene" $
+    services (runHASS bedroomButtonController [masseButton "1_long_press"])
+      `shouldBe` [[activateScene "scene.makuuhuone_kirkas"]]
+
+  it "Masse off click turns the bedroom lights off" $
+    services (runHASS bedroomButtonController [masseButton "2_short_release"])
+      `shouldBe` [[light [area] Off]]
+
+  it "Enishen single click turns on her nightstand scene (lowest)" $
+    services (runHASS bedroomButtonController [enishenButton "1_short_release"])
+      `shouldBe` [[activateScene "scene.makuuhuone_jemina"]]
+
+  it "Enishen double click turns on the middle scene" $
+    services (runHASS bedroomButtonController [enishenButton "1_double_press"])
+      `shouldBe` [[activateScene "scene.makuuhuone_keski"]]
+
+  it "Enishen long click turns on the bright scene" $
+    services (runHASS bedroomButtonController [enishenButton "1_long_press"])
+      `shouldBe` [[activateScene "scene.makuuhuone_kirkas"]]
+
+  it "Enishen off click turns the bedroom lights off" $
+    services (runHASS bedroomButtonController [enishenButton "2_short_release"])
+      `shouldBe` [[light [area] Off]]
+
+  it "ignores the initial press (scene only fires on release)" $
+    services (runHASS bedroomButtonController [masseButton "1_initial_press"])
+      `shouldBe` [[]]
+
+  it "ignores button events for other entities" $
+    services (runHASS bedroomButtonController
+              [Event (buttonEvent "event.some_other_action" "1_short_release")])
+      `shouldBe` [[]]
+
+  it "does nothing on Tick" $
+    services (runHASS bedroomButtonController [Tick])
+      `shouldBe` [[]]

test/BusSpec.hs

@@ -2,7 +2,7 @@
 
 module BusSpec (spec) where
 
-import AFRP (Request (..))
+import AFRP (Request (..), Event (..))
 import Control.Concurrent.STM
   ( atomically
   , dupTChan
@@ -10,31 +10,32 @@ import Control.Concurrent.STM
   , writeTChan
   )
 import Data.Aeson (Value (..))
-import Data.Time (UTCTime (..))
+import Data.Time (UTCTime (..), utc)
 import Data.UUID (nil)
 import HomeAssistant.Controller (HASSEff (..), Service (..), Target(..))
 import HomeAssistant.Runtime.Bus
-import Katip (Namespace (Namespace), runKatipContextT)
+import Katip (Namespace (Namespace), runKatipContextT, Severity (..))
 import Test.Hspec
 
 spec :: Spec
 spec = describe "Bus" $ do
-  it "broadcasts inbound messages to every dup'd channel in order" $ withBus $ \bus -> do
+  it "broadcasts inbound messages to every dup'd channel in order" $ withBus InfoS $ \bus -> do
     p1 <- atomically $ dupTChan (busInbound bus)
     p2 <- atomically $ dupTChan (busInbound bus)
-    atomically $ writeTChan (busInbound bus) (Number 1)
-    atomically $ writeTChan (busInbound bus) (Number 2)
+    atomically $ writeTChan (busInbound bus) (Event (Number 1))
+    atomically $ writeTChan (busInbound bus) (Event (Number 2))
     r1 <- atomically $ (,) <$> readTChan p1 <*> readTChan p1
     r2 <- atomically $ (,) <$> readTChan p2 <*> readTChan p2
-    r1 `shouldBe` (Number 1, Number 2)
-    r2 `shouldBe` (Number 1, Number 2)
+    r1 `shouldBe` (Event (Number 1), Event (Number 2))
+    r2 `shouldBe` (Event (Number 1), Event (Number 2))
 
-  it "channelHassEval writes CallService to the outbound channel" $ withBus $ \bus -> do
-    let req = Request (UTCTime (toEnum 0) 0) nil
+  it "channelHassEval writes CallService to the outbound channel" $ withBus InfoS $ \bus -> do
+    let req = Request (UTCTime (toEnum 0) 0) utc nil
         svc = Service "light" "turn_on" Nothing [EntityId "light.bedroom_masse"]
     runKatipContextT (busLogEnv bus) () (Namespace ["test"]) $
       channelHassEval bus (CallService req svc)
-    atomically (readTChan (busOutbound bus)) `shouldReturn` svc
+    (_, svc') <- atomically $ readTChan (busOutbound bus)
+    svc' `shouldBe` svc
 
 
   it "generates unique sequential call ids" $ do

test/ConnectionSpec.hs

@@ -2,14 +2,19 @@
 
 module ConnectionSpec (spec) where
 
+import AFRP (Request(..))
 import Data.Aeson (object, (.=))
+import Data.Maybe (fromJust)
 import Data.Text (Text)
+import Data.Time (UTCTime (..), utc)
+import Data.UUID (fromString)
 import HomeAssistant.Controller (Service (..), Target(..))
-import HomeAssistant.Runtime.Connection (encodeService)
+import HomeAssistant.Runtime.Connection (encodeService, dedupeBatch)
 import Test.Hspec
 
 spec :: Spec
-spec = describe "encodeService" $ do
+spec = do
+  describe "encodeService" $ do
     it "encodes a call_service message" $
       encodeService 7 (Service "light" "turn_on" Nothing [EntityId "light.bedroom_masse"])
         `shouldBe` object
@@ -30,3 +35,45 @@ spec = describe "encodeService" $ do
           , "target" .= object ["entity_id" .= ("light.bedroom_masse" :: Text)]
           , "service_data" .= object ["brightness" .= (200 :: Int)]
           ]
+
+  describe "dedupeBatch" $ do
+    it "collapses identical calls to one" $
+      let batch = [ (req 1, lightOn [AreaId "x"])
+                  , (req 2, lightOn [AreaId "x"])
+                  , (req 3, lightOn [AreaId "x"])
+                  ]
+      in dedupeBatch batch `shouldBe` [(req 3, lightOn [AreaId "x"])]
+
+    it "keeps same-target different-service calls separate" $
+      let batch = [ (req 1, lightOn  [AreaId "x"])
+                  , (req 2, lightOff [AreaId "x"])
+                  ]
+          result = dedupeBatch batch
+      in length result `shouldBe` 2
+
+    it "newest call wins for the same key" $
+      let batch = [ (req 1, lightOn [AreaId "x"])
+                  , (req 2, lightOn [AreaId "x"])
+                  , (req 3, lightOn [AreaId "x"])
+                  ]
+      in map requestTraceId (map fst (dedupeBatch batch)) `shouldBe`
+           [fromJust (fromString "00000000-0000-0000-0000-000000000003")]
+
+    it "treats target lists in different order as the same key" $
+      let batch = [ (req 1, lightOn [EntityId "a", EntityId "b"])
+                  , (req 2, lightOn [EntityId "b", EntityId "a"])
+                  ]
+      in length (dedupeBatch batch) `shouldBe` 1
+
+req :: Int -> Request
+req n = Request (UTCTime (toEnum 0) (fromIntegral (0 :: Int))) utc
+          (fromJust (fromString uuid))
+  where
+    pad i = replicate (12 - length (show i)) '0' <> show i
+    uuid = "00000000-0000-0000-0000-" <> pad n
+
+lightOn :: [Target] -> Service
+lightOn targets = Service "light" "turn_on" Nothing targets
+
+lightOff :: [Target] -> Service
+lightOff targets = Service "light" "turn_off" Nothing targets

test/Main.hs

@@ -1,7 +1,9 @@
 module Main (main) where
 
 import Test.Hspec (hspec)
+import qualified AFRPSpec
 import qualified BackoffProp
+import qualified BedroomSpec
 import qualified BusSpec
 import qualified ConnectionSpec
 import qualified RuntimeSpec
@@ -9,6 +11,8 @@ import qualified SupervisorSpec
 
 main :: IO ()
 main = hspec $ do
+  AFRPSpec.spec
+  BedroomSpec.spec
   BusSpec.spec
   ConnectionSpec.spec
   RuntimeSpec.spec

test/RuntimeSpec.hs

@@ -1,42 +1,42 @@
-{-# LANGUAGE OverloadedStrings #-}
 module RuntimeSpec (spec) where
 
-import Control.Concurrent (threadDelay)
-import Control.Concurrent.Async (async, race)
-import Control.Concurrent.STM (atomically, isEmptyTChan, readTChan, writeTChan)
-import Data.Aeson (Value, object, (.=))
-import Data.Text (Text)
-import HomeAssistant.Controller (light, lightController, Target (..))
-import HomeAssistant.Runtime (Controller (..), runController)
-import HomeAssistant.Runtime.Bus
 import Test.Hspec
 
 spec :: Spec
-spec = describe "runController" $ do
-  it "feeds inbound events through the machine and forwards service calls" $ withBus $ \bus -> do
-    _ <- async (runController bus (Controller "test" lightController True))
-    putStrLn "Before the delay"
-    threadDelay 100000 -- let the controller dup its inbound channel
-    putStrLn "After the delay"
-    atomically $ writeTChan (busInbound bus) (doorEvent "on")  -- initial value: no change event
-    atomically $ writeTChan (busInbound bus) (doorEvent "off") -- door closes: lights on
-    atomically $ writeTChan (busInbound bus) (doorEvent "on")  -- door opens: lights off
-    putStrLn "After the writes"
-    svc1 <- boundedRead (busOutbound bus)
-    svc2 <- boundedRead (busOutbound bus)
-    putStrLn "After the reads"
-    svc1 `shouldBe` Right (light [EntityId "light.bedroom_masse"] True)
-    svc2 `shouldBe` Right (light [EntityId "light.bedroom_masse"] False)
-    atomically (isEmptyTChan (busOutbound bus)) `shouldReturn` True
-  where
-    boundedRead channel = race (threadDelay 10000) (atomically (readTChan channel))
+spec = pure ()
 
-doorEvent :: Text -> Value
-doorEvent state = object
-  [ "event" .= object
-    [ "data" .= object
-      [ "entity_id" .= ("binary_sensor.makuuhuone_ovi_contact" :: Text)
-      , "new_state" .= object ["state" .= state]
-      ]
-    ]
-  ]
+-- I'm commenting out the test as that specific controller
+-- was just a dummy for seeing how the data flows (dead code)
+-- But leaving it commented as it retains some code sample on how and what to test
+--
+-- spec :: Spec
+-- spec = describe "runController" $ do
+--   it "feeds inbound events through the machine and forwards service calls" $ withBus InfoS $ \bus -> do
+--     _ <- async (runController bus (Controller "test" lightController True))
+--     putStrLn "Before the delay"
+--     threadDelay 100000 -- let the controller dup its inbound channel
+--     putStrLn "After the delay"
+--     atomically $ writeTChan (busInbound bus) (Event (doorEvent "on"))  -- initial value: no change event
+--     atomically $ writeTChan (busInbound bus) (Event (doorEvent "off")) -- door closes: lights on
+--     atomically $ writeTChan (busInbound bus) (Event (doorEvent "on"))  -- door opens: lights off
+--     putStrLn "After the writes"
+--     Right (_, svc1) <- boundedRead (busOutbound bus)
+--     Right (_, svc2) <- boundedRead (busOutbound bus)
+--     putStrLn "After the reads"
+--     svc1 `shouldBe` light [EntityId "light.bedroom_masse"] True
+--     svc2 `shouldBe` light [EntityId "light.bedroom_masse"] False
+--     atomically (isEmptyTChan (busOutbound bus)) `shouldReturn` True
+--   where
+--     boundedRead channel = race (threadDelay 10000) (atomically (readTChan channel))
+--
+-- doorEvent :: Text -> Value
+-- doorEvent state = object
+--   [ "event" .= object
+--     [ "variables" .= object
+--       [ "trigger" .= object
+--         [ "entity_id" .= ("binary_sensor.makuuhuone_ovi_contact" :: Text)
+--         , "to_state" .= object ["state" .= state]
+--         ]
+--       ]
+--     ]
+--   ]

test/Support.hs

@@ -0,0 +1,87 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+module Support
+  ( Acc(..)
+  , interp
+  , runHASS
+  , services
+  , fakeRequest
+  , sec
+  , stateEvent
+  , buttonEvent
+  ) where
+
+import Control.Monad.Fix (MonadFix (..))
+import Data.Aeson (Value, object, (.=))
+import qualified Data.Text as T
+import Data.Time (UTCTime (..), utc)
+import Data.UUID (nil)
+import AFRP (Mealy (..), Pair (..), Request (..))
+import HomeAssistant.Controller (HASSEff (..), Service)
+
+fakeRequest :: Request
+fakeRequest = Request (sec 0) utc nil
+
+sec :: Integer -> UTCTime
+sec n = UTCTime (toEnum 0) (fromIntegral n)
+
+-- | A pure Writer-like monad accumulating `Service` calls per step.
+newtype Acc a = Acc { runAcc :: [Service] -> (a, [Service]) }
+
+instance Functor Acc where
+  fmap f (Acc g) = Acc $ \s -> let (a, s') = g s in (f a, s')
+
+instance Applicative Acc where
+  pure a = Acc (\s -> (a, s))
+  Acc f <*> Acc x = Acc $ \s -> let (f', s') = f s; (a, s'') = x s' in (f' a, s'')
+
+instance Monad Acc where
+  Acc m >>= k = Acc $ \s -> let (a, s') = m s; (b, s'') = runAcc (k a) s' in (b, s'')
+
+instance MonadFix Acc where
+  mfix f = Acc $ \s -> let (a, s') = runAcc (f a) s in (a, s')
+
+-- | Interpret `HASSEff` in `Acc`: record `CallService`, drop tracing/debug.
+interp :: HASSEff a -> Acc a
+interp (CallService _ svc) = Acc $ \s -> ((), s ++ [svc])
+interp (Debug _) = pure ()
+interp (Trace _ _) = pure ()
+
+-- | Run a HASS arrow over a list of inputs, collecting per-step emitted services.
+runHASS :: Mealy HASSEff a b -> [a] -> [(b, [Service])]
+runHASS _ [] = []
+runHASS m (a : as) =
+  case runAcc (runMealy m interp fakeRequest a) [] of
+    (Pair b m', svcs) -> (b, svcs) : runHASS m' as
+
+services :: [(b, [Service])] -> [[Service]]
+services = map snd
+
+-- | Build a state-trigger payload matching `entityChangeEvent'` / `entityBool'`
+-- lenses. The subscribe_trigger websocket event wraps the trigger datum under
+-- `event.variables.trigger`, with `entity_id` and `to_state.state` fields.
+stateEvent :: T.Text -> T.Text -> Value
+stateEvent entityId state = object
+  [ "event" .= object
+    [ "variables" .= object
+      [ "trigger" .= object
+        [ "entity_id" .= entityId
+        , "to_state" .= object [ "state" .= state ]
+        ]
+      ]
+    ]
+  ]
+
+-- | Build an Ikea button trigger payload matching `ikeaQuickButton` lenses.
+buttonEvent :: T.Text -> T.Text -> Value
+buttonEvent entityId eventType = object
+  [ "event" .= object
+    [ "variables" .= object
+      [ "trigger" .= object
+        [ "entity_id" .= entityId
+        , "to_state" .= object
+          [ "attributes" .= object [ "event_type" .= eventType ] ]
+        ]
+      ]
+    ]
+  ]