#725

Committed 21 Oct 2025 07:59AM UTC coverage: 75.387% (-2.9%) from 78.302%

Build # #725

Build Type

Pull #113

github

Committed by

sceredi

Commit Message

chore: generalize action proposals to dynamic entity

Pull Request Pull Request #113: feat: create reinforcement learning controller

Run Details

84 of 179 new or added lines in 20 files covered. (46.93%)

7 existing lines in 4 files now uncovered.

1461 of 1938 relevant lines covered (75.39%)

8.29 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.77

/src/main/scala/io/github/srs/controller/ControllerModule.scala

package io.github.srs.controller

import scala.concurrent.duration.{ DurationInt, FiniteDuration, MILLISECONDS }
import scala.language.postfixOps

import cats.effect.std.Queue
import cats.effect.{ Clock, IO }
import io.github.srs.controller.message.DynamicEntityProposal
import io.github.srs.controller.protocol.Event
import io.github.srs.model.*
import io.github.srs.model.SimulationConfig.SimulationStatus.*
import io.github.srs.model.entity.dynamicentity.sensor.Sensor.senseAll
import io.github.srs.model.logic.*
import io.github.srs.utils.EqualityGivenInstances.given
import cats.implicits.*
import io.github.srs.utils.random.RNG
import com.typesafe.scalalogging.Logger
import io.github.srs.protos.ping.PongerFs2Grpc
import io.github.srs.controller.protobuf.ping.PongerService
import io.github.srs.model.entity.dynamicentity.robot.Robot
import io.github.srs.model.entity.dynamicentity.robot.behavior.BehaviorContext
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder

/**
 * Module that defines the controller logic for the Scala Robotics Simulator.
 */
object ControllerModule:

  /**
   * Controller trait that defines the interface for the controller.
   *
   * @tparam S
   *   the type of the state, which must extend [[io.github.srs.model.ModelModule.State]].
   */
  trait Controller[S <: ModelModule.State]:
    /**
     * Starts the controller with the initial state.
     *
     * @param initialState
     *   the initial state of the simulation.
     */
    def start(initialState: S): IO[S]

    /**
     * Runs the simulation loop, processing events from the queue and updating the state.
     *
     * @param s
     *   the current state of the simulation.
     * @param queue
     *   a concurrent queue that holds events to be processed.
     * @return
     *   an [[cats.effect.IO]] task that completes when the simulation loop ends.
     */
    def simulationLoop(s: S, queue: Queue[IO, Event]): IO[S]

  end Controller

  /**
   * Provider trait that defines the interface for providing a controller.
   *
   * @tparam S
   *   the type of the state, which must extend [[io.github.srs.model.ModelModule.State]].
   */
  trait Provider[S <: ModelModule.State]:
    val controller: Controller[S]

  /**
   * Defines the dependencies required by the controller module. In particular, it requires a
   * [[io.github.srs.view.ViewModule.Provider]] and a [[io.github.srs.model.ModelModule.Provider]].
   */
  type Requirements[S <: ModelModule.State] =
    io.github.srs.view.ViewModule.Provider[S] & io.github.srs.model.ModelModule.Provider[S]

  /**
   * Component trait that defines the interface for creating a controller.
   *
   * @tparam S
   *   the type of the simulation state, which must extend [[io.github.srs.model.ModelModule.State]].
   */
  trait Component[S <: ModelModule.State]:
    context: Requirements[S] =>

    object Controller:

      /**
       * Creates a controller instance.
       *
       * @return
       *   a [[Controller]] instance.
       */
      def apply()(using bundle: LogicsBundle[S]): Controller[S] = new ControllerImpl

      /**
       * Private controller implementation that delegates the simulation loop to the provided model and view.
       */
      private class ControllerImpl(using bundle: LogicsBundle[S]) extends Controller[S]:

        private val logger = Logger(getClass.getName)

        /**
         * Starts the controller with the initial state.
         * @param initialState
         *   the initial state of the simulation.
         * @return
         *   an [[IO]] task that completes when the controller is started.
         */
        override def start(initialState: S): IO[S] =
          for
            queueSim <- Queue.unbounded[IO, Event]
            service <- PongerFs2Grpc.bindServiceResource(new PongerService).allocated
            (svc, release) = service
            server = NettyServerBuilder.forPort(50051).addService(svc).build()
            _ <- IO.println("Starting PongerService") *> IO(server.start())
            _ <- context.view.init(queueSim)
            _ <- runBehavior(queueSim, initialState)
            result <- simulationLoop(initialState, queueSim)
            _ <- IO(server.shutdownNow().awaitTermination()) *> release
          yield result

        /**
         * Runs the simulation loop, processing events from the queue and updating the state.
         * @param s
         *   the current state of the simulation.
         * @param queue
         *   a concurrent queue that holds events to be processed.
         * @return
         *   an [[IO]] task that completes when the simulation loop ends.
         */
        override def simulationLoop(s: S, queue: Queue[IO, Event]): IO[S] =
          def loop(state: S): IO[S] =
            for
              startTime <- Clock[IO].realTime.map(_.toMillis)
              _ <- runBehavior(queue, state).whenA(state.simulationStatus == RUNNING)
              events <- queue.tryTakeN(None)
              newState <- handleEvents(state, events)
              _ <- context.view.render(newState)
              result <- handleStopCondition(newState) match
                case Some(io) => io
                case None =>
                  for
                    nextState <- nextStep(newState, startTime)
                    endTime <- Clock[IO].realTime.map(_.toMillis)
                    _ <- IO.blocking(logger.debug(s"Simulation loop took ${endTime - startTime} ms"))
                    res <- loop(nextState)
                  yield res
            yield result

          loop(s)

        end simulationLoop

        private def handleStopCondition(state: S): Option[IO[S]] =
          state.simulationStatus match
            case STOPPED =>
              Some(context.view.close() *> IO.pure(state))
            case ELAPSED_TIME =>
              Some(context.view.timeElapsed(state) *> IO.pure(state))
            case _ =>
              None

        /**
         * Processes the next step in the simulation based on the current state and start time.
         * @param state
         *   the current state of the simulation.
         * @param startTime
         *   the start time of the current simulation step in milliseconds.
         * @return
         *   the next state of the simulation wrapped in an [[IO]] task.
         */
        private def nextStep(state: S, startTime: Long): IO[S] =
          state.simulationStatus match
            case RUNNING =>
              tickEvents(startTime, state.simulationSpeed.tickSpeed, state)

            case PAUSED =>
              IO.sleep(50.millis).as(state)

            case _ =>
              IO.pure(state)

        /**
         * Runs the behavior of all robots in the environment and collects their action proposals.
         * @param queue
         *   the queue to which the proposals will be offered through the [[Event.DynamicEntityActionProposals]] event.
         * @param state
         *   the current state of the simulation.
         * @return
         *   an [[IO]] task that completes when the behavior has been run.
         */
        private def runBehavior(queue: Queue[IO, Event], state: S): IO[Unit] =
          val robots = state.environment.entities.collect { case r: Robot => r }.sortBy(_.id.toString)

          def process(
              remaining: List[Robot],
              rng: RNG,
              acc: List[DynamicEntityProposal],
          ): IO[(List[DynamicEntityProposal], RNG)] =
            remaining match
              case Nil => IO.pure((acc.reverse, rng))
              case robot :: tail =>
                for
                  sensorReadings <- robot.senseAll[IO](state.environment)
                  ctx = BehaviorContext(sensorReadings, rng)
                  (action, newRng) = robot.behavior.run[IO](ctx)
                  _ <- queue.offer(Event.Random(newRng))
                  next <- process(tail, newRng, DynamicEntityProposal(robot, action) :: acc)
                yield next

          for
            (proposals, _) <- process(robots, state.simulationRNG, Nil)
            _ <- queue.offer(Event.DynamicEntityActionProposals(proposals))
          yield ()

        end runBehavior

        /**
         * Processes tick events, adjusting the tick speed based on the elapsed time since the last tick.
         * @param start
         *   the start time of the current tick in milliseconds.
         * @param tickSpeed
         *   the speed of the tick in [[FiniteDuration]].
         * @param state
         *   the current state of the simulation.
         * @return
         *   the next state of the simulation wrapped in an [[IO]] task.
         */
        private def tickEvents(start: Long, tickSpeed: FiniteDuration, state: S): IO[S] =
          for
            now <- Clock[IO].realTime.map(_.toMillis)
            timeToNextTick = tickSpeed.toMillis - (now - start)
            adjustedTickSpeed = if timeToNextTick > 0 then timeToNextTick else 0L
            sleepTime = FiniteDuration(adjustedTickSpeed, MILLISECONDS)
            _ <- IO.sleep(sleepTime)
            tick <- handleEvent(state, Event.Tick(state.dt))
          yield tick

        /**
         * Handles a sequence of events, processing them in the order they were received.
         * @param state
         *   the current state of the simulation.
         * @param events
         *   the sequence of events to be processed.
         * @return
         *   the final state of the simulation after processing all events, wrapped in an [[IO]] task.
         */
        private def handleEvents(state: S, events: Seq[Event]): IO[S] =
          for finalState <- events.foldLeft(IO.pure(state)) { (taskState, event) =>
              for
                currentState <- taskState
                newState <- handleEvent(currentState, event)
              yield newState
            }
          yield finalState

        /**
         * Handles a single event and updates the state accordingly.
         * @param state
         *   the current state of the simulation.
         * @param event
         *   the event to be processed.
         * @return
         *   the updated state of the simulation after processing the event, wrapped in an [[IO]] task.
         */
        private def handleEvent(state: S, event: Event): IO[S] =
          event match
            case Event.Tick(deltaTime) =>
              context.model.update(state)(using s => bundle.tickLogic.tick(s, deltaTime))
            case Event.TickSpeed(speed) =>
              context.model.update(state)(using s => bundle.tickLogic.tickSpeed(s, speed))
            case Event.Random(rng) =>
              context.model.update(state)(using s => bundle.randomLogic.random(s, rng))
            case Event.Pause =>
              context.model.update(state)(using s => bundle.pauseLogic.pause(s))
            case Event.Resume =>
              context.model.update(state)(using s => bundle.resumeLogic.resume(s))
            case Event.Stop =>
              context.model.update(state)(using s => bundle.stopLogic.stop(s))
            case Event.DynamicEntityActionProposals(proposals) =>
              context.model.update(state)(using
                s => bundle.dynamicEntityActionsLogic.handleDynamicEntityActionsProposals(s, proposals),
              )

      end ControllerImpl

    end Controller

  end Component

  /**
   * Interface trait that combines the provider and component traits for the controller module.
   *
   * @tparam S
   *   the type of the simulation state, which must extend [[io.github.srs.model.ModelModule.State]].
   */
  trait Interface[S <: ModelModule.State] extends Provider[S] with Component[S]:
    self: Requirements[S] =>
end ControllerModule

1	package io.github.srs.controller
2
3	import scala.concurrent.duration.{ DurationInt, FiniteDuration, MILLISECONDS }
4	import scala.language.postfixOps
5
6	import cats.effect.std.Queue
7	import cats.effect.{ Clock, IO }
8	import io.github.srs.controller.message.DynamicEntityProposal
9	import io.github.srs.controller.protocol.Event
10	import io.github.srs.model.*
11	import io.github.srs.model.SimulationConfig.SimulationStatus.*
12	import io.github.srs.model.entity.dynamicentity.sensor.Sensor.senseAll
13	import io.github.srs.model.logic.*
14	import io.github.srs.utils.EqualityGivenInstances.given
15	import cats.implicits.*
16	import io.github.srs.utils.random.RNG
17	import com.typesafe.scalalogging.Logger
18	import io.github.srs.protos.ping.PongerFs2Grpc
19	import io.github.srs.controller.protobuf.ping.PongerService
20	import io.github.srs.model.entity.dynamicentity.robot.Robot
21	import io.github.srs.model.entity.dynamicentity.robot.behavior.BehaviorContext
22	import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder
23
24	/**
25	* Module that defines the controller logic for the Scala Robotics Simulator.
26	*/
27	object ControllerModule:
28		×
29	/**
30	* Controller trait that defines the interface for the controller.
31	*
32	* @tparam S
33	* the type of the state, which must extend [[io.github.srs.model.ModelModule.State]].
34	*/
35	trait Controller[S <: ModelModule.State]:
36	/**	5✔
37	* Starts the controller with the initial state.
38	*
39	* @param initialState
40	* the initial state of the simulation.
41	*/
42	def start(initialState: S): IO[S]
43
44	/**
45	* Runs the simulation loop, processing events from the queue and updating the state.
46	*
47	* @param s
48	* the current state of the simulation.
49	* @param queue
50	* a concurrent queue that holds events to be processed.
51	* @return
52	* an [[cats.effect.IO]] task that completes when the simulation loop ends.
53	*/
54	def simulationLoop(s: S, queue: Queue[IO, Event]): IO[S]
55
56	end Controller
57
58	/**
59	* Provider trait that defines the interface for providing a controller.
60	*
61	* @tparam S
62	* the type of the state, which must extend [[io.github.srs.model.ModelModule.State]].
63	*/
64	trait Provider[S <: ModelModule.State]:
65	val controller: Controller[S]
66
67	/**
68	* Defines the dependencies required by the controller module. In particular, it requires a
69	* [[io.github.srs.view.ViewModule.Provider]] and a [[io.github.srs.model.ModelModule.Provider]].
70	*/
71	type Requirements[S <: ModelModule.State] =
72	io.github.srs.view.ViewModule.Provider[S] & io.github.srs.model.ModelModule.Provider[S]
73
74	/**
75	* Component trait that defines the interface for creating a controller.
76	*
77	* @tparam S
78	* the type of the simulation state, which must extend [[io.github.srs.model.ModelModule.State]].
79	*/
80	trait Component[S <: ModelModule.State]:
81	context: Requirements[S] =>	1✔
82
83	object Controller:
84		5✔
85	/**
86	* Creates a controller instance.
87	*
88	* @return
89	* a [[Controller]] instance.
90	*/
91	def apply()(using bundle: LogicsBundle[S]): Controller[S] = new ControllerImpl
92		9✔
93	/**
94	* Private controller implementation that delegates the simulation loop to the provided model and view.
95	*/
96	private class ControllerImpl(using bundle: LogicsBundle[S]) extends Controller[S]:
97		8✔
98	private val logger = Logger(getClass.getName)
99		8✔
100	/**
101	* Starts the controller with the initial state.
102	* @param initialState
103	* the initial state of the simulation.
104	* @return
105	* an [[IO]] task that completes when the controller is started.
106	*/
107	override def start(initialState: S): IO[S] =
108	for
109	queueSim <- Queue.unbounded[IO, Event]
110	service <- PongerFs2Grpc.bindServiceResource(new PongerService).allocated	10✔
111	(svc, release) = service
112	server = NettyServerBuilder.forPort(50051).addService(svc).build()
113	_ <- IO.println("Starting PongerService") *> IO(server.start())
114	_ <- context.view.init(queueSim)	3✔
115	_ <- runBehavior(queueSim, initialState)
116	result <- simulationLoop(initialState, queueSim)
117	_ <- IO(server.shutdownNow().awaitTermination()) *> release
118	yield result	4✔
119
120	/**
121	* Runs the simulation loop, processing events from the queue and updating the state.
122	* @param s
123	* the current state of the simulation.
124	* @param queue
125	* a concurrent queue that holds events to be processed.
126	* @return
127	* an [[IO]] task that completes when the simulation loop ends.
128	*/
129	override def simulationLoop(s: S, queue: Queue[IO, Event]): IO[S] =
130	def loop(state: S): IO[S] =
131	for
132	startTime <- Clock[IO].realTime.map(_.toMillis)
133	_ <- runBehavior(queue, state).whenA(state.simulationStatus == RUNNING)	15✔
134	events <- queue.tryTakeN(None)	5✔
135	newState <- handleEvents(state, events)
136	_ <- context.view.render(newState)
137	result <- handleStopCondition(newState) match
138	case Some(io) => io
139	case None =>
140	for
141	nextState <- nextStep(newState, startTime)
142	endTime <- Clock[IO].realTime.map(_.toMillis)
143	_ <- IO.blocking(logger.debug(s"Simulation loop took ${endTime - startTime} ms"))
144	res <- loop(nextState)	8✔
145	yield res
146	yield result
147
148	loop(s)
149		5✔
150	end simulationLoop
151
152	private def handleStopCondition(state: S): Option[IO[S]] =
153	state.simulationStatus match
154	case STOPPED =>	3✔
155	Some(context.view.close() *> IO.pure(state))	8✔
156	case ELAPSED_TIME =>	×
157	Some(context.view.timeElapsed(state) *> IO.pure(state))	8✔
158	case _ =>	14✔
159	None
160		2✔
161	/**
162	* Processes the next step in the simulation based on the current state and start time.
163	* @param state
164	* the current state of the simulation.
165	* @param startTime
166	* the start time of the current simulation step in milliseconds.
167	* @return
168	* the next state of the simulation wrapped in an [[IO]] task.
169	*/
170	private def nextStep(state: S, startTime: Long): IO[S] =
171	state.simulationStatus match
172	case RUNNING =>	3✔
173	tickEvents(startTime, state.simulationSpeed.tickSpeed, state)	8✔
174		8✔
175	case PAUSED =>
176	IO.sleep(50.millis).as(state)	×
177		×
178	case _ =>
179	IO.pure(state)
180		×
181	/**
182	* Runs the behavior of all robots in the environment and collects their action proposals.
183	* @param queue
184	* the queue to which the proposals will be offered through the [[Event.DynamicEntityActionProposals]] event.
185	* @param state
186	* the current state of the simulation.
187	* @return
188	* an [[IO]] task that completes when the behavior has been run.
189	*/
190	private def runBehavior(queue: Queue[IO, Event], state: S): IO[Unit] =
191	val robots = state.environment.entities.collect { case r: Robot => r }.sortBy(_.id.toString)
192		35✔
193	def process(
194	remaining: List[Robot],
195	rng: RNG,
196	acc: List[DynamicEntityProposal],
197	): IO[(List[DynamicEntityProposal], RNG)] =
198	remaining match
199	case Nil => IO.pure((acc.reverse, rng))	2✔
200	case robot :: tail =>	17✔
201	for	15✔
202	sensorReadings <- robot.senseAll[IO](state.environment)
203	ctx = BehaviorContext(sensorReadings, rng)	24✔
204	(action, newRng) = robot.behavior.run[IO](ctx)
205	_ <- queue.offer(Event.Random(newRng))
206	next <- process(tail, newRng, DynamicEntityProposal(robot, action) :: acc)
207	yield next
208		×
209	for
210	(proposals, _) <- process(robots, state.simulationRNG, Nil)
211	_ <- queue.offer(Event.DynamicEntityActionProposals(proposals))	12✔
212	yield ()
213
214	end runBehavior
215
216	/**
217	* Processes tick events, adjusting the tick speed based on the elapsed time since the last tick.
218	* @param start
219	* the start time of the current tick in milliseconds.
220	* @param tickSpeed
221	* the speed of the tick in [[FiniteDuration]].
222	* @param state
223	* the current state of the simulation.
224	* @return
225	* the next state of the simulation wrapped in an [[IO]] task.
226	*/
227	private def tickEvents(start: Long, tickSpeed: FiniteDuration, state: S): IO[S] =
228	for
229	now <- Clock[IO].realTime.map(_.toMillis)
230	timeToNextTick = tickSpeed.toMillis - (now - start)	18✔
231	adjustedTickSpeed = if timeToNextTick > 0 then timeToNextTick else 0L
232	sleepTime = FiniteDuration(adjustedTickSpeed, MILLISECONDS)
233	_ <- IO.sleep(sleepTime)
234	tick <- handleEvent(state, Event.Tick(state.dt))
235	yield tick
236
237	/**
238	* Handles a sequence of events, processing them in the order they were received.
239	* @param state
240	* the current state of the simulation.
241	* @param events
242	* the sequence of events to be processed.
243	* @return
244	* the final state of the simulation after processing all events, wrapped in an [[IO]] task.
245	*/
246	private def handleEvents(state: S, events: Seq[Event]): IO[S] =
247	for finalState <- events.foldLeft(IO.pure(state)) { (taskState, event) =>
248	for	11✔
249	currentState <- taskState
250	newState <- handleEvent(currentState, event)
251	yield newState
252	}
253	yield finalState
254
255	/**
256	* Handles a single event and updates the state accordingly.
257	* @param state
258	* the current state of the simulation.
259	* @param event
260	* the event to be processed.
261	* @return
262	* the updated state of the simulation after processing the event, wrapped in an [[IO]] task.
263	*/
264	private def handleEvent(state: S, event: Event): IO[S] =
265	event match
266	case Event.Tick(deltaTime) =>	2✔
267	context.model.update(state)(using s => bundle.tickLogic.tick(s, deltaTime))	15✔
268	case Event.TickSpeed(speed) =>	11✔
269	context.model.update(state)(using s => bundle.tickLogic.tickSpeed(s, speed))	3✔
270	case Event.Random(rng) =>	×
271	context.model.update(state)(using s => bundle.randomLogic.random(s, rng))	15✔
272	case Event.Pause =>	11✔
273	context.model.update(state)(using s => bundle.pauseLogic.pause(s))	8✔
274	case Event.Resume =>	×
275	context.model.update(state)(using s => bundle.resumeLogic.resume(s))	8✔
276	case Event.Stop =>	×
277	context.model.update(state)(using s => bundle.stopLogic.stop(s))	8✔
NEW 278	case Event.DynamicEntityActionProposals(proposals) =>	×
279	context.model.update(state)(using	15✔
280	s => bundle.dynamicEntityActionsLogic.handleDynamicEntityActionsProposals(s, proposals),	6✔
281	)	5✔
282		2✔
283	end ControllerImpl
284
285	end Controller
286
287	end Component
288
289	/**
290	* Interface trait that combines the provider and component traits for the controller module.
291	*
292	* @tparam S
293	* the type of the simulation state, which must extend [[io.github.srs.model.ModelModule.State]].
294	*/
295	trait Interface[S <: ModelModule.State] extends Provider[S] with Component[S]:
296	self: Requirements[S] =>
297	end ControllerModule

Scala-Robotics-Simulator / PPS-22-srs / #725

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous