Merge branch 'main' of https://github.com/biaslab/RxEnvironments.jl

benchmark/benchmarks.jl

@@ -1,5 +1,5 @@
 using BenchmarkTools
-
+import RxEnvironments: __send!
 const SUITE = BenchmarkGroup()
 
 include("mockenv.jl")
@@ -9,8 +9,8 @@ function simple_discrete_benchmarks()
     SUITE = BenchmarkGroup(["Simple Discrete Environment"])
     env = RxEnvironment(MockEnvironment(); is_discrete=true)
     agent = add!(env, MockEntity())
-    SUITE["send to agent"] = @benchmarkable send!($agent, $env, 10.0)
-    SUITE["send to environment"] = @benchmarkable send!($env, $agent, 10.0)
+    SUITE["send to agent"] = @benchmarkable __send!($agent, $env, 10.0)
+    SUITE["send to environment"] = @benchmarkable __send!($env, $agent, 10.0)
     return SUITE
 end
 
@@ -25,17 +25,17 @@ function complex_discrete_benchmarks()
     add!(hearing_aid, agent)
     add!(hearing_aid, user)
 
-    SUITE["send to agent"] = @benchmarkable send!($hearing_aid, $world, view([1.0, 2.0, 3.0], 1:2))
-    SUITE["send to environment"] = @benchmarkable send!($world, $hearing_aid, 10.0)
+    SUITE["send to agent"] = @benchmarkable __send!($hearing_aid, $world, view([1.0, 2.0, 3.0], 1:2))
+    SUITE["send to environment"] = @benchmarkable __send!($world, $hearing_aid, 10.0)
     return SUITE
 end
 
 function simple_continuous_benchmarks()
     SUITE = BenchmarkGroup(["Simple Continuous Environment"])
     env = RxEnvironment(MockEnvironment(); is_discrete=false)
     agent = add!(env, MockEntity())
-    SUITE["send to agent"] = @benchmarkable send!($agent, $env, 10.0)
-    SUITE["send to environment"] = @benchmarkable send!($env, $agent, 10.0)
+    SUITE["send to agent"] = @benchmarkable __send!($agent, $env, 10.0)
+    SUITE["send to environment"] = @benchmarkable __send!($env, $agent, 10.0)
     return SUITE
 end
 
@@ -50,8 +50,8 @@ function complex_continuous_benchmarks()
     add!(hearing_aid, agent)
     add!(hearing_aid, user)
 
-    SUITE["send to agent"] = @benchmarkable send!($hearing_aid, $world, view([1.0, 2.0, 3.0], 1:2))
-    SUITE["send to environment"] = @benchmarkable send!($world, $hearing_aid, 10.0)
+    SUITE["send to agent"] = @benchmarkable __send!($hearing_aid, $world, view([1.0, 2.0, 3.0], 1:2))
+    SUITE["send to environment"] = @benchmarkable __send!($world, $hearing_aid, 10.0)
     return SUITE
 end
 

src/abstractentity.jl

@@ -231,6 +231,23 @@ function send!(
     recipient::Union{AbstractEntity,Rocket.Actor},
     emitter::AbstractEntity,
     action::Any,
+)
+    if recipient ∈ subscribers(emitter)
+        __send!(recipient, emitter, action)
+    else
+        throw(NotSubscribedException(emitter, recipient))
+    end
+end
+
+"""
+    __send!(recipient::AbstractEntity, emitter::AbstractEntity, action::Any)
+
+Send an action from `emitter` to `recipient`.
+"""
+function __send!(
+    recipient::Union{AbstractEntity,Rocket.Actor},
+    emitter::AbstractEntity,
+    action::Any,
 )
     actuator = get_actuator(emitter, recipient)::Actuator{action_type(decorated(emitter))}
     send_action!(actuator, action)
@@ -272,7 +289,6 @@ receive!(recipient::AbstractEntity, observation::Observation) =
 receive!(recipient::AbstractEntity, emitter::AbstractEntity, observation::Any) =
     receive!(decorated(recipient), decorated(emitter), observation)
 receive!(recipient::AbstractEntity, obs::TimerMessage) = receive!(decorated(recipient), obs)
-receive!(recipient::AbstractEntity, observation::Any) = nothing
 receive!(recipient, emitter, observation) = nothing
 receive!(recipient, observation) = nothing
 

src/exceptions.jl

@@ -33,6 +33,5 @@ Base.showerror(io::IO, e::SelfSubscriptionException) =
 
 mutable struct NotPausedException <: Exception end
 
-
 Base.showerror(io::IO, e::NotPausedException) =
     print(io, "Trying to access paused time for unpaused entity")

src/markovblanket.jl

@@ -67,8 +67,6 @@ end
 
 Rocket.subscribe!(observations::Observations, actor::Rocket.Actor{T} where {T}) =
     subscribe!(target(observations), actor)
-Rocket.subscribe!(observations::Observations, actor::F where {F<:AbstractActorFactory}) =
-    subscribe!(target(observations) |> map(Any, (x) -> data(x)), actor)
 
 Rocket.next!(
     observations::Observations{ContinuousEntity,<:T},

src/rxentity.jl

@@ -28,7 +28,7 @@ function Rocket.on_next!(
     foreach(subscribers(subject)) do listener
         if emits(subject, listener, observation)
             action = what_to_send(listener, subject, observation)
-            send!(listener, subject, action)
+            __send!(listener, subject, action)
         end
     end
 end

test/entity_tests.jl

@@ -7,6 +7,7 @@
         e = create_entity(MockEntity())
         @test !is_active(e)
         @test e isa RxEnvironments.RxEntity{MockEntity}
+        @test occursin(r"Continuous", repr(e))
     end
 
     @testset "decorated" begin
@@ -190,7 +191,7 @@
 
                 unsubscribe!(first_entity, second_entity)
                 @test !is_subscribed(second_entity, first_entity)
-                @test_throws KeyError send!(
+                @test_throws RxEnvironments.NotSubscribedException send!(
                     second_entity,
                     first_entity,
                     1,
@@ -372,6 +373,7 @@ end
     @testset "constructor" begin
         e = create_entity(MockEntity(); is_discrete=true)
         @test e isa RxEnvironments.RxEntity{MockEntity}
+        @test occursin(r"Discrete", repr(e))
     end
 
     @testset "markov blanket functionality" begin
@@ -398,6 +400,8 @@ end
                 @test last(obs) === RxEnvironments.ObservationCollection((
                     RxEnvironments.Observation(second_entity, nothing),
                 ))
+                @test length(last(obs)) == 1
+                @test data(last(obs)) === nothing
                 @test data.(obs.values) == [nothing]
 
                 next!(
@@ -497,7 +501,7 @@ end
 
                 unsubscribe!(first_entity, second_entity)
                 @test !is_subscribed(second_entity, first_entity)
-                @test_throws KeyError send!(
+                @test_throws RxEnvironments.NotSubscribedException send!(
                     second_entity,
                     first_entity,
                     1,

test/environment_tests.jl

@@ -60,6 +60,20 @@
             terminate!(env)
         end
     end
+
+    @testset "keep actor" begin
+        using Rocket
+        using RxEnvironments
+
+        let e = RxEnvironment(MockEntity())
+            other = create_entity(MockEntity())
+            add!(e, other)
+            actor = keep(Any)
+            subscribe!(e, actor)
+            send!(e, other, 1)
+            @test last(actor) == nothing
+        end
+    end
 end
 
 @testitem "discrete environment" begin

test/exceptions_tests.jl

@@ -1,4 +1,3 @@
-
 @testitem "NotSubscribedException" begin
     using RxEnvironments
 
@@ -12,6 +11,9 @@
         @test origin(exception) === env
         @test recipient(exception) === agent
         @test_throws NotSubscribedException throw(exception)
+        buf = IOBuffer()
+        showerror(buf, exception)
+        @test occursin(r"Entity (.*?) is not subscribed to (.*?)", String(take!(buf)))
     end
 
 end
@@ -29,6 +31,9 @@ end
         @test first_entity(exception) === env
         @test second_entity(exception) === agent
         @test_throws MixedStateSpaceException throw(exception)
+        buf = IOBuffer()
+        showerror(buf, exception)
+        @test occursin(r"Entities (.*?) and (.*?) have different state spaces", String(take!(buf)))
     end
 end
 
@@ -43,5 +48,21 @@ end
         @test exception isa SelfSubscriptionException
         @test entity(exception) === env
         @test_throws SelfSubscriptionException throw(exception)
+        buf = IOBuffer()
+        showerror(buf, exception)
+        @test occursin(r"Entity cannot subscribe to itself, attempted in (.*?)", String(take!(buf)))
     end
 end
+
+@testitem "NotPausedException" begin
+    using RxEnvironments
+
+    import RxEnvironments: NotPausedException
+    let exception = NotPausedException()
+        @test exception isa NotPausedException
+        @test_throws NotPausedException throw(exception)
+        buf = IOBuffer()
+        showerror(buf, exception)
+        @test occursin(r"Trying to access paused time for unpaused entity", String(take!(buf)))
+    end
+end

test/markovblanket_tests.jl

@@ -76,7 +76,7 @@ end
             @test !is_subscribed(agent, env)
             @test is_subscribed(second_agent, env)
             @test length(subscribers(env)) == 1
-            @test_throws KeyError send!(env, agent, 10)
+            @test_throws RxEnvironments.NotSubscribedException send!(env, agent, 10)
 
             unsubscribe!(env, second_agent)
             @test !is_subscribed(agent, env)
@@ -139,16 +139,13 @@ end
     end
 end
 
-# @testitem "send branching" begin
-#     using RxEnvironments
-#     import RxEnvironments: DiscreteEntity, create_entity
-#     include("../mockenvironment.jl")
-#     let env = RxEnvironment(MockEnvironment(0.0); discrete = true)
-#         agent = add!(env, MockAgent())
-#         obs = subscribe_to_observations!(agent, RxEnvironments.keep(Any))
-#         send!(env, agent, 10)
-#         send!(env, agent, 10.0)
-#         @test obs.values[1] == "Integer"
-
-#     end
-# end
+@testitem "show" begin
+    using RxEnvironments
+
+    include("mockenvironment.jl")
+
+    let e = RxEnvironment(MockEntity())
+        markov_blanket = RxEnvironments.markov_blanket(e)
+        @test occursin(r"MarkovBlanket{RxEnvironments.ContinuousEntity}", repr(markov_blanket))
+    end
+end