Python Asteroids+Invaders on GitHub

Let’s do some tasks, now that we have the Tasks object. We’ll see how we like it.

The core idea—thanks again, Tomas—is to stop using the task reminders in Fleets and have objects manage tasks on their own. That will be a tiny bit more work than with Fleets, but in the end we’ll b able to simplify Fleets, and some might argue that it was weird that objects could just say “remind_me” and then things just kind of randomly happened later.

No one has mentioned that objection but perhaps they were looking the other way.

We’ll just look for calls to remind_me and see what we find. When they all refer to Tasks, we can remove the task stuff from Fleets. At that point there may still be places where the Tasks idea will be useful. We’ll find out.

PyCharm tells me that there are three real uses, and five test uses. There’s a single use in InvaderShot. This will be the worst case scenario for changing over:

class InvaderShot(InvadersFlyer):
    def die_on_collision(self, flyer, fleets):
        if self.colliding(flyer):
            fleets.remind_me(lambda: self.die(fleets))

To use Tasks, we’ll need an instance to talk to, and for proper usage, we should clear it on begin and finish the tasks on end. Like this:

class InvaderShot(InvadersFlyer):
    def __init__(self, position, maps):
        self.maps = maps
        self.masks = [pygame.mask.from_surface(bitmap) for bitmap in self.maps]
        self._map = maps[0]
        self.map_index = 0
        self._rect = self._map.get_rect()
        self.rect.center = position
        self.count = 0
        self.moves = 0
        self._available = True
        self._tasks = Tasks()

    def begin_interactions(self, fleets):
        self._tasks.clear()


    def end_interactions(self, fleets):
        self._tasks.finish()

    def die_on_collision(self, flyer, fleets):
        if self.colliding(flyer):
            self._tasks.remind_me(lambda: self.die(fleets))

This breaks a couple of tests, doubtless ones that need to call end_interactions to help out. For example:

    def test_dies_on_shield(self):
        fleets = Fleets()
        fi = FI(fleets)
        shield = Shield(Vector2(100, 100))
        maker = BitmapMaker.instance()
        shot = InvaderShot(Vector2(100, 100), maker.rollers)
        assert shot.colliding(shield)
        fleets.append(shot)
        assert fi.invader_shots
        shot.interact_with_shield(shield, fleets)
        fleets._execute_reminders()
        assert not fi.invader_shots

Sure enough, calling the secret execute method in fleets isn’t going to help.

We’ll call end_interactions instead. Test runs. Same in the other one? Yes. I also find one for layer shots and change it preemptively. It continues to work, of course, because Fleets is still doing its job.

We’re green. Commit: InvaderShot uses Tasks, not Fleet, for its deferred tasks.

So let’s reflect.

Reflection

There’s absolutely no doubt that this took more code. We had to add a line to init, we had to add two new methods (in this case), one clearing and one finishing, and we had to modify the actual one line that defined a task. Is this a losing proposition? YMMV.

If we really cared about it, we could put creation of Tasks, the clearing, and the finishing up in the superclass and inherit it all. That would bother our friend Hill, but it wouldn’t bother us here chez Ron so much. It would bother us a little, though, because while we’re OK with inheriting “do nothing” for an interface method, putting actual code up there is a bit more than we’d usually do.

That said, setting things up in begin, making notes during interact, and cleaning things up in end is exactly why that sequence exists. So it is righteous in that regard.

It’s a bit more explicit, a bit less mysterious pay no attention to the man behind the curtain. That’s a good thing.

I think it’ll be a net win, by a narrow margin. We’ll see. For now, we have another object to look at, the PlayerShot:

class PlayerShot(InvadersFlyer):

    def hit_invader(self, fleets):
        self.remind_me_to_die(fleets)

    def interact_with_invadershot(self, shot, fleets):
        if self.colliding(shot):
            self.remind_me_to_explode_and_die(fleets)

    def interact_with_shield(self, shield, fleets):
        if self.colliding(shield):
            self.remind_me_to_die(fleets)

    def remind_me_to_explode_and_die(self, fleets):
        self.remind_me_to_explode(fleets)
        self.remind_me_to_die(fleets)

    def remind_me_to_explode(self, fleets):
        fleets.remind_me(lambda: self.actually_explode(fleets))

    def remind_me_to_die(self, fleets):
        fleets.remind_me(lambda: self.actually_die(fleets))

We proceed just as before:

class PlayerShot(InvadersFlyer):
    def __init__(self, position=u.CENTER):
        offset = Vector2(2, -8*4)
        self.velocity = Vector2(0, -4*4)
        maker = BitmapMaker.instance()
        self.bits = maker.player_shot
        self._mask = pygame.mask.from_surface(self.bits)
        self._rect = self.bits.get_rect()
        self.position = position + offset
        self.should_die = False
        self._tasks = Tasks()

    def begin_interactions(self, fleets):
        self._tasks.clear()

    def end_interactions(self, fleets):
        self._tasks.finish()

    def remind_me_to_explode(self, fleets):
        self._tasks.remind_me(lambda: self.actually_explode(fleets))

    def remind_me_to_die(self, fleets):
        self._tasks.remind_me(lambda: self.actually_die(fleets))

A test has broken. Let’s hope it’s another of those easy ones. Yes, another one using Fleets’ private method:

    def test_top_bumper(self):
        fleets = Fleets()
        fi = FI(fleets)
        bumper = TopBumper()
        shot = PlayerShot(u.CENTER)
        fleets.append(bumper)
        fleets.append(shot)
        shot.interact_with_topbumper(bumper, fleets)
        assert fi.player_shots
        # position is virtual, can't just set its y
        pos = shot.position
        pos.y = bumper.y
        shot.position = pos
        shot.interact_with_topbumper(bumper, fleets)
        fleets._execute_reminders()
        assert not fi.player_shots

Same fix, replace that penultimate line:

        fleets.end_interactions()
        assert not fi.player_shots

Green. Commit: PlayerShot uses Tasks, not Fleets reminders. No more game users of the fleets facility.

I play a little celebratory game. Reflect again:

Reflection

Same pattern. This time the overhead wasn’t so high, but there’s no doubt that using tasks requires us at least to use begin_interactions and end_interactions, and generally speaking I think creating one instance of Tasks in __init__ is the way to go. And then thereafter it’s the same just referring to our own tasks list, not Fleets.

So far nothing untoward has happened, all this was foretold in the ancient scrolls of my mind. Let’s remove the capability from Fleets, unless we think Fleets itself would benefit.

A quick review of Fleets shows us that it’s about the most linear class you can imagine and has no need of tasks of its own:

class Fleets:
    def __init__(self):
        self.flyers = list()
        self.tasks = Tasks()
    def begin_interactions(self):
        self.tasks = Tasks()
        for flyer in self.all_objects:
            flyer.begin_interactions(self)

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

    def remind_me(self, reminder: Callable):
        self.tasks.remind_me(reminder)

    def _execute_reminders(self):
        self.tasks.finish()

We’ll just go ahead and remove those tasks references. Tests fail, presumably all of them testing this thing we just removed. With the code removed, PyCHarm throws squiggles all over the offending test methods, making them easy to spot, and I obligingly remove the methods.

I notice the references to Remindable, my test class, and remove that as well. We are quickly green again. Commit: Remove Fleets tasks and tests thereof.

And there we are, converted from Fleets carrying that extra burden to using Tasks instead. We only have two invaders objects using it so far but we will surely have more. One, for example, will be the object that keeps track of whether the evil little bug invaders have killed the player, coming soon to a computer near you.

What should we do now? Are there other nails to which we could apply this hammer? Certainly there are plenty over on the Asteroids side of the house, but that’s those other guys and they can pick up using Tasks or not, as they see fit. We know they’re not using the Fleets feature yet, and now it’s gone and they can’t.

What about the invaders side? What have we got so far? Invaders, InvaderFleet, InvaderGroup, Player and Invader Shots, Bumpers side and top, and Shields. Ah, Shields. They do that weird stuff with their masks and appearance. Let’s have a look at that!

In Shield, the oddness is that we cannot update the shield’s mask until end_interactions, because when we do, colliding objects tend to fall right through them. What happens is that the shield eliminates the bits where it is hit, and if the colliding object gets its interaction after that, it sees no common pixels and doe not die. Rinse, repeat. So Shield keeps a copy of the mask, updates that, and replaces it:

class Shield(InvadersFlyer):
    def __init__(self, position):
        map = BitmapMaker.instance().shield
        explo = BitmapMaker.instance().invader_shot_explosion
        self._explosion_mask = pygame.mask.from_surface(explo)
        player_explo = BitmapMaker.instance().player_shot_explosion
        self._player_explosion_mask = pygame.mask.from_surface(player_explo)
        self._map = map.copy()
        self._map.set_colorkey("black")
        self._mask = pygame.mask.from_surface(map)
        self._mask_copy = self.mask.copy()
        self._rect = self._map.get_rect()
        self._rect.center = position

    def end_interactions(self, fleets):
        self._mask = self._mask_copy.copy()
        pass

    def process_shot_collision(self, shot, explosion_mask):
        collider = Collider(self, shot)
        if collider.colliding():
            mask: Mask = collider.overlap_mask()
            self.erase_shot_and_explosion_from_mask(shot, collider, mask, explosion_mask)
            self.erase_visible_pixels(mask.get_rect())

    def erase_shot_and_explosion_from_mask(self, shot, collider, shot_overlap_mask, explosion_mask):
        self._mask_copy.erase(shot_overlap_mask, (0, 0))
        self.erase_explosion_from_mask(collider, explosion_mask, shot)

    def erase_explosion_from_mask(self, collider, explosion_mask, shot):
        expl_rect = explosion_mask.get_rect()
        offset_x = (shot.rect.w - expl_rect.w) // 2
        adjust_image_to_center = collider.offset() + Vector2(offset_x, 0)
        self._mask_copy.erase(explosion_mask, adjust_image_to_center)

    def erase_visible_pixels(self, shot_rect):
        surf = self._mask_copy.to_surface()
        self._map.blit(surf, shot_rect)

The thing we’d like to clean up is the references to _mask_copy. I think we can almost do it by simply deferring the lines after the if, here:

    def process_shot_collision(self, shot, explosion_mask):
        collider = Collider(self, shot)
        if collider.colliding():
            mask: Mask = collider.overlap_mask()
            self.erase_shot_and_explosion_from_mask(shot, collider, mask, explosion_mask)
            self.erase_visible_pixels(mask.get_rect())

But what would we do about the reference to _mask_copy in erase_visible_pixels? Let’s do a rename, and a change of signature. First rename that temp:

    def process_shot_collision(self, shot, explosion_mask):
        collider = Collider(self, shot)
        if collider.colliding():
            overlap_mask: Mask = collider.overlap_mask()
            self.erase_shot_and_explosion_from_mask(shot, collider, overlap_mask, explosion_mask)
            self.erase_visible_pixels(overlap_mask.get_rect())

Then let’s pass the mask down to erase_visible_pixels, not just the rectangle. We’ll have to check that in game to be sure it’s OK.

    def process_shot_collision(self, shot, explosion_mask):
        collider = Collider(self, shot)
        if collider.colliding():
            overlap_mask: Mask = collider.overlap_mask()
            self.erase_shot_and_explosion_from_mask(shot, collider, overlap_mask, explosion_mask)
            self.erase_visible_pixels(overlap_mask)

    def erase_visible_pixels(self, shot_mask):
        rect = shot_mask.get_rect()
        surf = self._mask_copy.to_surface()
        self._map.blit(surf, rect)

OK. That continues to work. Let’s commit it: refactor to use shot overlap mask in erase visible pixels, in preparation for deferred tasks.

Now I think we can extract those last two lines from the if and defer them. This step just continues to work:

    def process_shot_collision(self, shot, explosion_mask):
        collider = Collider(self, shot)
        if collider.colliding():
            overlap_mask: Mask = collider.overlap_mask()
            self.update_mask_and_visible_pixels(collider, explosion_mask, overlap_mask, shot)

    def update_mask_and_visible_pixels(self, collider, explosion_mask, overlap_mask, shot):
        self.erase_shot_and_explosion_from_mask(shot, collider, overlap_mask, explosion_mask)
        self.erase_visible_pixels(overlap_mask)

Commit: pull out method to be deferred.

And now: Tasks.

class Shield(InvadersFlyer):
    def __init__(self, position):
        map = BitmapMaker.instance().shield
        ...
        self._tasks = Tasks()

    def begin_interactions(self, fleets):
        self._tasks.clear()

    def end_interactions(self, fleets):
        self._tasks.finish()

    def process_shot_collision(self, shot, explosion_mask):
        collider = Collider(self, shot)
        if collider.colliding():
            overlap_mask: Mask = collider.overlap_mask()
            self._tasks.remind_me(self.update_mask_and_visible_pixels(collider, explosion_mask, overlap_mask, shot))

Does it work? I have to run the game to find out. Let’s remember to think about a test for it though.

Oh. Forgot the lambda. Put it in. Still doesn’t work. I think I forgot to change the code to refer always to _masks rather than the copy. True but still not right. There’s a reference to the collider offset which will change. If this doesn’t work, we’re going back to the drawing board.

Still no go. I think I see it but it’s too much. Roll back.

Reflection

For deferral to work, the lambda must capture all the things that might change between when we call remind_me and when we do the work. The way the code is written, it refers to things that can change. We need to capture all of them.

Let’s review the code again:

    def process_shot_collision(self, shot, explosion_mask):
        collider = Collider(self, shot)
        if collider.colliding():
            overlap_mask: Mask = collider.overlap_mask()
            self.update_mask_and_visible_pixels(collider, explosion_mask, overlap_mask, shot)

    def update_mask_and_visible_pixels(self, collider, explosion_mask, overlap_mask, shot):
        self.erase_shot_and_explosion_from_mask(shot, collider, overlap_mask, explosion_mask)
        self.erase_visible_pixels(overlap_mask)

    def erase_shot_and_explosion_from_mask(self, shot, collider, shot_overlap_mask, explosion_mask):
        self._mask_copy.erase(shot_overlap_mask, (0, 0))
        self.erase_explosion_from_mask(collider, explosion_mask, shot)

    def erase_explosion_from_mask(self, collider, explosion_mask, shot):
        expl_rect = explosion_mask.get_rect()
        offset_x = (shot.rect.w - expl_rect.w) // 2
        adjust_image_to_center = collider.offset() + Vector2(offset_x, 0)
        self._mask_copy.erase(explosion_mask, adjust_image_to_center)

    def erase_visible_pixels(self, shot_mask):
        rect = shot_mask.get_rect()
        surf = self._mask_copy.to_surface()
        self._map.blit(surf, rect)

It’s the call to update_mask_and_visible_pixels that we want to defer. Everything from there on down needs to be nailed down. Let’s work bottom up, passing down to each lower level function, the exact things it needs. Start here:

    def erase_visible_pixels(self, shot_mask):
        rect = shot_mask.get_rect()
        surf = self._mask_copy.to_surface()
        self._map.blit(surf, rect)

It’s OK for it to blit to _map, that’s its job. And when we’re done, it could refer to _mask, which will have been updated by then, but it would be better if it didn’t.

I think we want to do this in two phases, first updating _mask and then using it on the visible pixels. No, then things are temporally coupled. We know that deferred tasks are done in the order received but let’s not rely on it quite that much.

Let’s change erase_visible_pixels to receive the updated shield mask and use it to make the surface.

    def update_mask_and_visible_pixels(self, collider, explosion_mask, overlap_mask, shot):
        self.erase_shot_and_explosion_from_mask(shot, collider, overlap_mask, explosion_mask)
        self.erase_visible_pixels(overlap_mask, self._mask)

    def erase_visible_pixels(self, shot_mask, shield_mask):
        rect = shot_mask.get_rect()
        surf = shield_mask.to_surface()
        self._map.blit(surf, rect)

Does this work? Not remotely. The marks appear off to the left. Why? Well, partly because we need to pass in _mask_copy for now, I think.

    def update_mask_and_visible_pixels(self, collider, explosion_mask, overlap_mask, shot):
        self.erase_shot_and_explosion_from_mask(shot, collider, overlap_mask, explosion_mask)
        self.erase_visible_pixels(overlap_mask, self._mask_copy)

That works. I wish I had a test for this but I can’t think of one now. I’m running a bit scared, things have not gone well. Must try to keep calm.

Now the erase_visible_pixels has no live parameters, no parameters that are going to change between deferral and updating.

Does that mean we could defer it separately, right now? I think it does. I put that much in, including all the frills:

class Shield(InvadersFlyer):
    def __init__(self, position):
        map = BitmapMaker.instance().shield
        ...
        self._tasks = Tasks()

    def begin_interactions(self, fleets):
        self._tasks.clear()

    def end_interactions(self, fleets):
        self._tasks.finish()
        self._mask = self._mask_copy.copy()
        pass

    def update_mask_and_visible_pixels(self, collider, explosion_mask, overlap_mask, shot):
        self.erase_shot_and_explosion_from_mask(shot, collider, overlap_mask, explosion_mask)
        self._tasks.remind_me(lambda: self.erase_visible_pixels(overlap_mask, self._mask_copy))

That works a treat. Commit: Use Tasks to update visible pixels at end of interactions.

Let’s reflect.

Reflection

We could go home at this point. We have a foothold for Tasks in Shield, deferring the update of the visible bits until after interactions are over. This is a smaller bite than I tried before. I can’t say that I was looking for a smaller bite, but when I spotted the possibility I certainly went for it. And it paid off.

As it stands, I think it’s questionable. What will be the value of _mask_copy by the time we get to erase_visible_bits, on the off chance that two shots are hitting at the same time? It will be the same instance, which I think means it will have accumulated all the shots thus far, even though we may defer two calls. (This is a very low-probability event which could occur. If we don’t get it right, it will occur and create a heisenbug. We do not want a heisenbug.)

Accumulating the changes is OK. In a sense that’s what we’re trying to do anyway.

I think we can move forward, looking at the erase_shot_and_explosion to see how to defer them. The issue, I think, is that we’ll need more constant inputs.

    def erase_shot_and_explosion_from_mask(self, shot, collider, shot_overlap_mask, explosion_mask):
        self._mask_copy.erase(shot_overlap_mask, (0, 0))
        self.erase_explosion_from_mask(collider, explosion_mask, shot)

    def erase_explosion_from_mask(self, collider, explosion_mask, shot):
        expl_rect = explosion_mask.get_rect()
        offset_x = (shot.rect.w - expl_rect.w) // 2
        adjust_image_to_center = collider.offset() + Vector2(offset_x, 0)
        self._mask_copy.erase(explosion_mask, adjust_image_to_center)

What’s variable here? I think it’s just the collider. The only reference to it seems to be getting its offset, which is certainly dynamic, changing on each occurrence. And it will change, if there are multiple missiles in flight, for example. Let’s back it up the chain:

    def erase_explosion_from_mask(self, collider_offset, explosion_mask, shot):
        expl_rect = explosion_mask.get_rect()
        offset_x = (shot.rect.w - expl_rect.w) // 2
        adjust_image_to_center = collider_offset + Vector2(offset_x, 0)
        self._mask_copy.erase(explosion_mask, adjust_image_to_center)

    def erase_shot_and_explosion_from_mask(self, shot, collider_offset, shot_overlap_mask, explosion_mask):
        self._mask_copy.erase(shot_overlap_mask, (0, 0))
        self.erase_explosion_from_mask(collider_offset, explosion_mask, shot)

    def update_mask_and_visible_pixels(self, collider, explosion_mask, overlap_mask, shot):
        self.erase_shot_and_explosion_from_mask(shot, collider.offset(), overlap_mask, explosion_mask)
        self._tasks.remind_me(lambda: self.erase_visible_pixels(overlap_mask, self._mask_copy))

Does this work? It does. Then will deferring work? Only if we update the real mask, not the copy … although if we leave the end_interactions alone, we might be able to do that after deferral.

This does work:

    def update_mask_and_visible_pixels(self, collider, explosion_mask, overlap_mask, shot):
        self._tasks.remind_me(lambda: self.erase_shot_and_explosion_from_mask(shot, collider.offset(), overlap_mask, explosion_mask))
        self._tasks.remind_me(lambda: self.erase_visible_pixels(overlap_mask, self._mask_copy))

    def end_interactions(self, fleets):
        self._tasks.finish()
        self._mask = self._mask_copy.copy()

Now can I change all the references to _mask_copy to _mask?

    def end_interactions(self, fleets):
        self._tasks.finish()

    def erase_shot_and_explosion_from_mask(self, shot, collider_offset, shot_overlap_mask, explosion_mask):
        self._mask.erase(shot_overlap_mask, (0, 0))
        self.erase_explosion_from_mask(collider_offset, explosion_mask, shot)

    def erase_explosion_from_mask(self, collider_offset, explosion_mask, shot):
        expl_rect = explosion_mask.get_rect()
        offset_x = (shot.rect.w - expl_rect.w) // 2
        adjust_image_to_center = collider_offset + Vector2(offset_x, 0)
        self._mask.erase(explosion_mask, adjust_image_to_center)

And does this work? In fact it does. Remove _mask_copy member entirely. Commit: Shield mask and visible bits use deferred Task updating, no longer copy and restore masks.

Reflection and Summary

Well! That was interesting. I don’t know why that didn’t work the first time, and I don’t care. It seems clear that I must have missed a reference to the copy, or possibly something else that needed to remain constant between remind_me and finish. But it does not matter. “Fail again. Fail better.” We roll back, we do it over, better if we see how, just the same if not, trusting that we’ll either see our mistake, accidentally not make our mistake, or, in rare cases, find that it’s failing in the same way and realize we need a deeper think. Generally we do not need the deeper debugging thought process, because most times we’ve made a simple mistake that we will see, or just not make again.

But there is a lesson here, to keep in mind with future uses of the Tasks facility:

Whatever objects we refer to in our lambda must remain constant over the course of all interactions.

If they do not remain constant, then they’ll be changed1, and the deferred operation will not do what it would have done before deferral. And that will be bad.

This session has been just over two hours, but it has generated more text than we usually get in that period of time. I think I’ll attribute it to much bigger chunks of code than we usually provide.

I think another big learning worth underlining is the power, when something doesn’t work and we really think it should have, of rolling back and just doing it over. I still remember the first time I saw Kent Beck do that, while working on the screen in front of a class. He wrote a test, saw it fail, put in the code to make it work. It still failed. Some in the audience saw the bug, a typo or something. They started to tell him what it was. He just rolled back and did it over and it worked.

Sometimes, sure, we do see the problem. But if we’re doing small steps and we don’t see what we did wrong, it’s a very strong move to just roll back and repeat.

And then there’s the smaller step. Second time around, after rolling back, I saw a smaller step that I could take, just deferring the visible part, and letting the mask calculation stay undeferred. The visible part worked, and I could then ignore it and work on the other part. Divide and conquer.

So three good things:

  • Consider rolling back when we think it “should” have worked;
  • Take smaller steps when things have gone wrong;
  • Make sure everything passed to your task lambda is invariant between then and finishing.

I am pleased, and welcome your observations. See you next time!

  1. Funny how that works. If things aren’t constant, they change. I wonder what’s up with that? ↩