TickDispatcher

Artemis Cosmos runs the python script at a set interval (currently once a second). This is done via the HandleSimulationTick method.

Usage

The TickDispatcher is used to create tasks that are called when the HandleSimulationTick is called. The function dispatch_tick should be called in HandleSimulationTick, and ideally this is the only code that is needed.

Python

   def  HandleSimulationTick(sim):
      TickDispatcher.dispatch_tick(sim)

Importing the hookhandlers module it does by default.

Python

   from sbs_utils.handlerhooks import *
   # no longer need to implement handlers in script.py

One could implement their own way of handling time and tick related code, the goal of the TickDispatcher is to have a common way to organize this logic.

A TickTask is similar to Artemis XML's timer events and are ideally more efficient since they only live while they are relevant.

To create a TickTask you can use one of two functions.

• do_once
• do_interval

do_once will schedule a task that is run only one time.

do_interval will schedule to run multiple times or continuously.

Both methods take the simulation used to track the start time, a callback function, a delay time. do_interval additionally optionally takes the number of times it should run. The default is None, which is meant to indicate run continuously.

The callback functions will receive the simulation and the task as arguments. Class methods used as a callback will receive 'self' as the first argument.

Example: run a function once

The following will run the function after 5 seconds.

Python

from sbs_utils.tickdispatcher import TickDispatcher

def call_me_later(sim, task):
   print('Hello, Tick Tasks')

# Some other place
TickDispatcher.do_once(sim, call_me_later, 5)   

The following will run a class method on an object after 5 seconds.

Python

from sbs_utils.tickdispatcher import TickDispatcher

class MyPlayer(PlayerShip):
   def call_me_later(self, sim, task):
      print('Hello, Tick Tasks')

# Some other place
player_one = MyPlayer()
TickDispatcher.do_once(sim, player_one.call_me_later, 5)   

The following will run a class method on an object after 5 seconds. This one show using self.

Python

from sbs_utils.tickdispatcher import TickDispatcher

class MyPlayer(PlayerShip):
   def call_me_later(self, sim, task):
      print('Hello, Tick Tasks')

   def some_other_method(self):      
      TickDispatcher.do_once(sim, self.call_me_later, 5)   

Example: run a function multiple times

The following will run the function every 5 seconds, 4 times

Python

from sbs_utils.tickdispatcher import TickDispatcher

def call_me_later(sim, task):
   print('Hello, Tick Tasks')

# Some other place
TickDispatcher.do_interval(sim, call_me_later, 5, 4)   

The following will run a class method on an object after 5 seconds four times. This one show using self.

Python

from sbs_utils.tickdispatcher import TickDispatcher

class MyPlayer(PlayerShip):
   def call_me_later(self, sim, task):
      print('Hello, Tick Tasks')

   def some_other_method(self):      
      TickDispatcher.do_interval(sim, self.call_me_later, 5,4)   

Example: run continuously

The following will run the function every tick. This can be done with classes as well, those examples will be similar to above.

Python

from sbs_utils.tickdispatcher import TickDispatcher

def call_me_later(sim, task):
   print('Hello, Tick Tasks')

# Some other place
TickDispatcher.do_interval(sim, call_me_later, 0)   

Example: Stopping a task

The following will run as task and stop it when a condition is met.

Python

from sbs_utils.tickdispatcher import TickDispatcher

class MyPlayer(PlayerShip):
   def call_me_later(self, sim, task):
      print('Hello, Tick Tasks')
      if some_thing_that_stops_it:
         task.stop()

   def some_other_method(self):      
      TickDispatcher.do_interval(sim, self.call_me_later, 5)   

Example: passing data

The following will run will pass data to the callback.

Python

from sbs_utils.tickdispatcher import TickDispatcher

def call_me_later(sim, task):
   print('Hello, Tick Tasks')
   # use the data attached to the task
   print(task.data)

# Some other place
thetask = TickDispatcher.do_interval(sim, call_me_later, 0)
# attach data to the task
thetask.data = 42

For completeness: using the object 'self' data

Python

from sbs_utils.tickdispatcher import TickDispatcher

class MyPlayer(PlayerShip):
   data = 42
   def call_me_later(self, sim, task):
      print('Hello, Tick Tasks')
      print(self.data)

   def some_other_method(self):      
      TickDispatcher.do_interval(sim, self.call_me_later, 5)   

API: tickdispatcher module

DripQueue

Drain a FINITE work list over a few seconds instead of in one frame.

RollingSlicer's sibling. That one paces recurring work over a live set (brains, objectives, urges); this paces one-shot creation, so a burst -- a map's whole terrain field, a fleet, a prefab scatter -- becomes a drip. The engine measurement behind this: LM's terrain block is ~280 ms of work in a single frame, a ~5x hitch, with no sync tail afterwards. The cost is all in the frame that creates the objects, so spreading that frame out is the whole fix.

Two properties make a queued call equivalent to the inline one it replaced:

• each item captures random.getstate() when queued and runs under that state, so it creates exactly what it would have created inline -- whenever it runs, in whatever order (the same trick terrain_spawn_field_keyed already uses per cell);
• items run NEAREST-FIRST from a focus point, so the space around the focus is correct first and the fill-in lands where nobody is looking.

That equivalence is PER ITEM. Deferring still moves when the caller's own RNG stream is consumed, so a sequence of queued calls whose plans read that stream (terrain's cluster loops do) produces a deterministic but different result from running them inline. Queue whole units of work, not halves of one.

Draining is deadline-driven: everything queued has run by over sim-seconds after it was queued, and items added mid-drain push the deadline out rather than crowding the remaining ticks.

Usage

q = DripQueue(over=6, focus=Vec3(0, 0, 0), name="terrain") q.add(spawn_a_cluster, (points, height), pos=points[0]) ... q.flush() # or let it drain itself, a slice per tick

add(fn, args=(), kwargs=None, pos=None)

Queue one unit of work, with the RNG state it would have run under.

clear()

Drop queued work without running it (mission reset).

flush()

Run everything still queued, right now. Returns how many ran.

run_slice()

Run this tick's share. Returns how many items ran.

set_focus(focus)

Move the point work is ordered nearest-first from.

RollingSlicer

Spread per-tick work over a set of ids across ticks (anti-spike).

A large per-tick batch (e.g. running every brain or objective in one frame) causes a periodic hitch. A RollingSlicer instead hands back a small slice each tick, sized by a fractional accumulator so a full pass over every id completes in exactly pass_seconds of sim time -- regardless of set size or tick rate (no over-run on small sets, no spike on large ones). The set's sorted order is cached and only rebuilt when membership changes, so the cursor advances predictably as ids are added/removed.

Usage

_slicer = RollingSlicer() for id in _slicer.slice(id_set, pass_seconds=3): ...work one item...

TickDispatcher

The Tick Dispatcher is used to manager timed items via the HandleSimulationTick

clear() classmethod

Drop all scheduled ticks (fresh mission / in-process recompile).

dispatch_tick()

Process all the tasks The task is updated to see if it should be triggered, and if it is completed

do_interval(cb, delay, count=None)

Create and return a task that executes more than once

:param ctx: The Artemis Cosmos simulation :param cb: call back function :param delay: the time in seconds for the task to delay :type delay: int :param count: The number of times to run None mean infinite :type count: int or None :return: The task is returned and can be used to attach data for future use. :rtype: TickFTask

example:

.. code-block:: python

def some_use():
    t = TickDispatcher.do_interval(the_callback, 5)
    t.data = some_data

def the_callback(t):
    print(t.some_data)
    if t.some_data.some_condition:
        t.stop()

do_once(cb, delay)

Create and return a task that executes once

:param delay: the time in seconds for the task to delay :type delay: int :return: The task is returned and can be used to attach data for future use. :rtype: TickTask

example

def some_use(): t = TickDispatcher.do_once(the_callback, 5) t.data = some_data

def the_callback(t): print(t.some_data)

TickTask

Bases: Agent

A task that is managed by the TickDispatcher

done property

returns if this is the task will not run in the future

__init__(cb, delay, count)

new TickTask

:param sim: The Artemis Cosmos simulation :param cb: call back function :param delay: the time in seconds for the task to delay :type delay: int :param count: The number of times to run None mean infinite :type count: int or None

stop()

Stop a tasks The task is removed