Python Asteroids on GitHub

Let’s put the full interaction protocol in place, so that more of our special Flyers can operate.

Imagine objects in space whose job is to detect when there are no more asteroids, and when that situation has gone on long enough, to create more asteroids. Such an object needs a way to know whether asteroids exist or not. The rules of the design, however, are that the object can only see what it interacts with.

The object, therefore, might choose to implement interact_with_asteroid, and if a call to that “event” occurs, the object can relax until next time events happen. But how does that object know it’s time to reset its flag? And if all the interacts go by without an asteroid showing up, how does it know it’s time to take action?

We answer those questions by providing not just one interact_with_ event, but three events:

  1. begin_interactions, which is sent before we start going through all the interactions;
  2. interact_with_, which is sent whenever a particular type of object is paired with our object;
  3. end_interactions, which is sent after all the pairs have been processed.

Our asteroid detection object could set its flag on begin, clear its flag if it is ever sent interact_with_asteroid, and when end arrives, if the flag is still set, take whatever next action is required.

Our first activity this morning will be to implement those two new event calls, begin_interactions and end_interactions. They will be available to all Flyers, and they should be optional. No one who doesn’t care needs to implement them.

Subscriptions
I am thinking of the various messages that the Flyers can be sent as if they were events, and I’m using the word to emphasize that thinking. In a publish-subscribe scheme, an object would explicitly say something like subscribe("begin_interactions"). In our scheme, you signify that you are subscribed by implementing the method, because if you don’t implement it, the system will not send it to you — because it is defaulted to pass in the Flyer superclass. So let’s begin there:
class Flyer(ABC):

    # concrete methods, inheritable
    # so sue me

    def begin_interactions(self, fleets):
        pass

    def end_interactions(self, fleets):
        pass

Good so far. Not much to test there, so I didn’t. Now to send them. Currently, in Game, we have:

class Game:
    def process_interactions(self):
        interactor = Interactor(self.fleets)
        self.score = interactor.perform_interactions()

That looks like the place to send the other messages. I think Fleets should do it:

class Game:
    def process_interactions(self):
        self.fleets.begin_interactions()
        interactor = Interactor(self.fleets)
        self.score = interactor.perform_interactions()
        self.fleets.end_interactions()

And in Fleets:

class Fleets:
    def begin_interactions(self):
        for flyer in self.all_objects:
            flyer.begin_interactions(self)

    def end_interactions(self):
        for flyer in self.all_objects:
            flyer.end_interactions(self)

Is there anything to test here? There’s certainly no branching logic to check. I suppose we could do a simple test just for drill. Let’s find a place, probably in test_interactions.

Looking at the tests, I see that I don’t like what I’ve done. I want those messages sent from Interactor, which I can test more readily than Game. Roll back Game.

Now write the test.

    def test_begin_end(self):
        begin = BeginChecker()
        end = EndChecker()
        assert begin.triggered == False
        assert end.triggered == False
        asteroids = [begin, end]
        interactor = Interactor(Fleets(asteroids))
        interactor.perform_interactions()
        assert begin.triggered == True
        assert end.triggered == True

I need these two new testing-only classes BeginChecker and EndChecker. PyCharm helps me by implementing the required methods:

class BeginChecker(Flyer):
    def __init__(self):
        self.triggered = False

    def tick(self, delta_time, fleet, fleets):
        pass

    def draw(self, screen):
        pass

    def interact_with(self, other, fleets):
        pass

    def begin_interactions(self, fleets):
        self.triggered = True


class EndChecker(Flyer):
    def __init__(self):
        self.triggered = False

    def tick(self, delta_time, fleet, fleets):
        pass

    def draw(self, screen):
        pass

    def interact_with(self, other, fleets):
        pass

    def end_interactions(self, fleets):
        self.triggered = True

The test is still red, because no one is calling begin and end … yet.

class Interactor:
    def perform_interactions(self):
        self.fleets.begin_interactions()
        for target, attacker in itertools.combinations(self.fleets.all_objects, 2):
            self.interact_one_pair(target, attacker)
        self.fleets.end_interactions()
        return self.score

The test is green. I am happy to have written this test. It told me of a better place for the capability, and gives me confidence that the right things are happening.

Commit: Interactor now sends begin_interactions and end_interactions messages to all Flyers, via Fleets.

Reflection

After a change like this, I like to relax a bit, look over what I’ve done, think about what’s next. I also have a battery problem in my mouse, so let’s plug it in and see what I can do without mousing for a while.

We now have our begin and end events, which means that we can surely start creating our special objects that cause the Asteroids game to emerge from their cooperation. One such issue is the creation of waves of asteroids when there are none on the screen. Let’s see how we do that now.

Ah, yes, that is presently done in AsteroidFleet:

class AsteroidFleet(Fleet):
    def __init__(self, asteroids):
        super().__init__(asteroids)
        self.timer = Timer(u.ASTEROID_DELAY, self.create_wave)
        self.asteroids_in_this_wave = 2

    def create_wave(self):
        self.extend([Asteroid() for _ in range(0, self.next_wave_size())])

    def next_wave_size(self):
        self.asteroids_in_this_wave += 2
        if self.asteroids_in_this_wave > 10:
            self.asteroids_in_this_wave = 11
        return self.asteroids_in_this_wave

    def tick(self, delta_time, fleets):
        super().tick(delta_time, fleets)
        if not self.flyers:
            self.timer.tick(delta_time)
        return True

We want to change things so that there are one or more objects in the mix that detect the need for a new wave, and create one. Should we use one object, or two? I think one unless building it suggests it should be split. (And I can think of a reason why that might happen.)

Let’s think how it might work.

  1. On creation, build a timer like the one above;
  2. On begin, set a no_asteroids flag;
  3. On interact_with_asteroid, clear the flag;
  4. On end_interactions, tick the timer. If it triggers, it will create the asteroids.

Let’s TDD a WaveMaker.

class TestWaveMaker:
    def test_detects_asteroids(self):
        fleets = Fleets()
        maker = WaveMaker()
        maker.begin_interactions(fleets)
        assert maker.no_asteroids
        maker.interact_with_asteroid(None, fleets)
        assert not maker.no_asteroids

And a simple enough class:

class WaveMaker(Flyer):
    def __init__(self):
        self.no_asteroids = None

    def begin_interactions(self, fleets):
        self.no_asteroids = True

    def interact_with_asteroid(self, asteroid, fleets):
        self.no_asteroids = False

The first test passes. The no_asteroids flag is odd. Maybe we should make it more positive.

class TestWaveMaker:
    def test_detects_asteroids(self):
        fleets = Fleets()
        maker = WaveMaker()
        maker.begin_interactions(fleets)
        assert not maker.saw_asteroids
        maker.interact_with_asteroid(None, fleets)
        assert maker.saw_asteroids

class WaveMaker(Flyer):
    def __init__(self):
        self.saw_asteroids = None

    def begin_interactions(self, fleets):
        self.saw_asteroids = False

    def interact_with_asteroid(self, asteroid, fleets):
        self.saw_asteroids = True

OK, fine. Now a test for the timer. As I envision this it is going to seem odd.

Ah. If we’re going to tick the timer … we’ll need delta_time, won’t we?

We do have a required tick method, and we could do the timing there. I think that’s best. Let’s try it and see how it works out.

    def test_timer(self):
        fleets = Fleets()
        maker = WaveMaker()
        maker.begin_interactions(fleets)
        assert not maker.saw_asteroids
        assert not fleets.asteroid_count
        maker.tick(u.ASTEROID_DELAY, None, fleets)
        assert fleets.asteroid_count

That seems direct and straightforward. Let’s make it happen, first trivially:

class WaveMaker:
    def tick(self, delta_time, fleet, fleets):
        if not self.saw_asteroids:
            fleets.add_asteroid(Asteroid())

Test passes. Make it harder.

    def test_timer(self):
        fleets = Fleets()
        maker = WaveMaker()
        maker.begin_interactions(fleets)
        assert not maker.saw_asteroids
        assert not fleets.asteroid_count
        maker.tick(0.1, None, fleets)
        assert not fleets.asteroid_count
        maker.tick(u.ASTEROID_DELAY, None, fleets)
        assert fleets.asteroid_count

Now we have to be more clever in WaveMaker. Let’s make a timer.

class WaveMaker(Flyer):
    def __init__(self):
        self.saw_asteroids = None
        self.timer = Timer(u.ASTEROID_DELAY, self.create_asteroids)

    def create_asteroids(self, fleets):
        fleets.add_asteroid(Asteroid())

    def tick(self, delta_time, fleet, fleets):
        if not self.saw_asteroids:
            self.timer.tick(delta_time, fleets)

So that’s triggering correctly now. I could be committing this class but since it’s not ready for use, I won’t, not yet.

We need to get the timer reset. Let’s extend this last test.

I expected this to fail:

    def test_timer(self):
        fleets = Fleets()
        maker = WaveMaker()
        maker.begin_interactions(fleets)
        assert not maker.saw_asteroids
        assert not fleets.asteroid_count
        maker.tick(0.1, None, fleets)
        assert not fleets.asteroid_count
        maker.tick(u.ASTEROID_DELAY, None, fleets)
        assert fleets.asteroid_count
        for asteroid in fleets.asteroids:
            fleets.remove_asteroid(asteroid)
        maker.begin_interactions(fleets)
        maker.tick(0.1, None, fleets)
        assert not fleets.asteroid_count

I guess that if Timers trigger, they reset their time. That would be good. Check the code.

class Timer:
    def tick(self, delta_time, *tick_args):
        self.elapsed += delta_time
        if self.elapsed >= self.delay:
            action_complete = self.action(*self.args, *tick_args)
            if action_complete is None or action_complete:
                self.elapsed = 0

Perfect.

I think that now all we need to implement is the number of asteroids to be created.

I accidentally coded this before testing it.

class WaveMaker(Flyer):
    def __init__(self):
        self.saw_asteroids = None
        self.timer = Timer(u.ASTEROID_DELAY, self.create_asteroids)
        self.asteroid_count = 2

    def create_asteroids(self, fleets):
        self.asteroid_count += 2
        if self.asteroid_count > 11:
            self.asteroid_count = 11
        for i in range(self.asteroid_count):
            fleets.add_asteroid(Asteroid())

Let’s test at least a bit:

    def test_timer(self):
        fleets = Fleets()
        maker = WaveMaker()
        maker.begin_interactions(fleets)
        assert not maker.saw_asteroids
        assert not fleets.asteroid_count
        maker.tick(0.1, None, fleets)
        assert not fleets.asteroid_count
        maker.tick(u.ASTEROID_DELAY, None, fleets)
        assert fleets.asteroid_count == 4
        for asteroid in fleets.asteroids:
            fleets.remove_asteroid(asteroid)
        maker.begin_interactions(fleets)
        maker.tick(0.1, None, fleets)
        assert not fleets.asteroid_count
        maker.tick(u.ASTEROID_DELAY, None, fleets)
        assert fleets.asteroid_count == 6

Sweet. I think this may actually work.

Let’s commit this: WaveMaker passes all tests.

Oh, we should move it to its own file. Commit: moved to own file.

Now let’s disable the code in AsteroidFleet, and add a WaveMaker to see what happens.

    def __init__(self, testing=False):
        self.delta_time = 0
        self.score = 0
        self.init_pygame_and_display(testing)
        self.fleets = Fleets()
        self.fleets.add_scorekeeper(ScoreKeeper(testing))
        self.fleets.add_wavemaker(WaveMaker())
        self.running = not testing

And in Fleets:

    def add_wavemaker(self, wavemaker):
        self.others.append(wavemaker)

Hm some test failed. Ah it’s the test for the AsteroidFleet. No problem. I want to run the game and see if it actually works.

My lords and ladies and others, it works! Voila, etc.!

OK, remove that test for AsteroidFleet and I think we can move most of the code from AsteroidFleet. Yes, it is unneeded. Remove it and change Fleets:

class Fleets:
    def __init__(self, asteroids=None, missiles=None, saucers=None, saucer_missiles=None, ships=None):
        asteroids = asteroids if asteroids is not None else []
        missiles = missiles if missiles is not None else []
        saucers = saucers if saucers is not None else []
        saucer_missiles = saucer_missiles if saucer_missiles is not None else []
        ships = ships if ships is not None else []
        self.fleets = (
            Fleet(asteroids), # no longer need AsteroidFleet
            MissileFleet(missiles, u.MISSILE_LIMIT),
            SaucerFleet(saucers),
            MissileFleet(saucer_missiles, u.SAUCER_MISSILE_LIMIT),
            ShipFleet(ships),
            ExplosionFleet(),
            Fleet([]))
        self.thumper = Thumper(self.beat1, self.beat2)
        self.score = 0

We’re green and good. Commit: WaveMaker in use, permits removal of AsteroidFleet class and its tests.

Let’s sum up.

Summary

With zero trouble at all, we implemented (and then tested) the begin_interactions and end_interactions event calls, then test-drove a WaveMaker object that enabled us to remove the AsteroidsFleet class and all the wave-making logic from the upper levels of the game.

There may will be some debris left over, but if there is, it’s not having any impact on the game. All the wave-making is in the WaveMaker.

It’s not absolutely entirely tested yet. I haven’t tested the upper limit of the wave size. OK, here ya go:

    def test_timer(self):
        fleets = Fleets()
        maker = WaveMaker()
        maker.begin_interactions(fleets)
        assert not maker.saw_asteroids
        assert not fleets.asteroid_count
        maker.tick(0.1, None, fleets)
        assert not fleets.asteroid_count
        maker.tick(u.ASTEROID_DELAY, None, fleets)
        assert fleets.asteroid_count == 4

        self.clear_and_tick(fleets, maker)
        assert fleets.asteroid_count == 6

        self.clear_and_tick(fleets, maker)
        assert fleets.asteroid_count == 8

        self.clear_and_tick(fleets, maker)
        assert fleets.asteroid_count == 10

        self.clear_and_tick(fleets, maker)
        assert fleets.asteroid_count == 11

        self.clear_and_tick(fleets, maker)
        assert fleets.asteroid_count == 11

    @staticmethod
    def clear_and_tick(fleets, maker):
        for asteroid in fleets.asteroids:
            fleets.remove_asteroid(asteroid)
        maker.begin_interactions(fleets)
        maker.tick(0.1, None, fleets)
        assert not fleets.asteroid_count
        maker.tick(u.ASTEROID_DELAY, None, fleets)

I did this the not-very-bright way, first duplicating the code, then extracting the method, at which point PyCharm offered to do the other replacements. Commit: improve testing of wave size limits.

We’ve accomplished today’s mission, which was to prepare the begin and end functions, and we’ve also implemented our first object that uses them. I am a bit surprised that we didn’t need the end_interactions method. At this point the method is there speculatively, since we have no one actually using it. Will that change? We’ll find out. What we’ve “observed” here is that we can set flags and then use tick to take necessary action.

This object is stateful

I predict that quite a few of our special objects will be stateful, as is WaveMaker, which only takes action when it has detected no asteroids, and is otherwise just quietly waiting. It also embodies the number of asteroids that should be created. We could, in principle, get rid of the incrementing of asteroid_count by removing the old WaveMaker and adding a new one with a larger counter, but we’d still have the boolean saw_asteroids.

I am not as into immutable objects as perhaps I should be, so I am not driven to resolve this, but I do note the possibility of at least reducing the mutability here. I’m not seeing the value. After all, the point of this thing is to detect a state and upon detecting it, take an action.

I could be wrong. Immutable might be better, but I don’t see how to get rid of the boolean.

A good thing!

Mutable or not, our WaveMaker is a very small object that manages an important single function, creating new waves of asteroids. I am pleased with it, and here’s another look at it.

See you next time!


class WaveMaker(Flyer):
    def __init__(self):
        self.saw_asteroids = None
        self.timer = Timer(u.ASTEROID_DELAY, self.create_asteroids)
        self.asteroid_count = 2

    def create_asteroids(self, fleets):
        self.asteroid_count += 2
        if self.asteroid_count > 11:
            self.asteroid_count = 11
        for i in range(self.asteroid_count):
            fleets.add_asteroid(Asteroid())

    def begin_interactions(self, fleets):
        self.saw_asteroids = False

    def interact_with_asteroid(self, asteroid, fleets):
        self.saw_asteroids = True

    def tick(self, delta_time, fleet, fleets):
        if not self.saw_asteroids:
            self.timer.tick(delta_time, fleets)

    def interact_with(self, other, fleets):
        pass

    def draw(self, screen):
        pass