Search code examples
pythontransitionstate-machinefsmpytransitions

Handling multiple objects in pytransitions


https://github.com/pytransitions/transitions

I was trying with the batman = NarcolepticSuperhero("Batman") example in the documentation. But let's say if we have 100 batmans to handle, My approach is to create a list of every batman object and track their states individually using a loop. IMO this is very naive.

Can some-one please suggest some effective solution.

Thanks


Solution

  • Assuming all NarcolepticSuperheros should behave the same, I recommend to use one Machine for all 'batmen' instead of instantiating NarcolepticSuperhero a hundred times. This way, you can use Machine.dispatch to trigger an event on ALL models instead of looping through a list of superheroes (under the hood Machine also just loops through the model list though). I condensed the mentioned example a bit (all transitions in one list), removed the machine that was bound to NarcolepticSuperheros and introduced a SuperheroDen. Concerning state checking, you could use a dictionary to keep track of how many heroes are currently in what state. I added a hero_log to NarcolepticSuperhero and every time a hero changes its state, it will first log out of its current state and then log in with its new state.

    from transitions import Machine
    from collections import  defaultdict
    import random
    
    
    class NarcolepticSuperhero(object):
    
        hero_log = defaultdict(list)
    
        def __init__(self, name):
            self.name = name
            self.kittens_rescued = 0
    
        def update_journal(self):
            self.kittens_rescued += 1
    
        @property
        def is_exhausted(self):
            return random.random() < 0.5
    
        @staticmethod
        def change_into_super_secret_costume():
            print("Beauty, eh?")
    
        def log_out(self):
            self.hero_log[self.state].remove(self)
    
        def log_in(self):
            self.hero_log[self.state].append(self)
    
    
    class SuperheroDen(Machine):
        states = ['asleep', 'hanging out', 'hungry', 'sweaty', 'saving the world']
        transitions = [
            ['wake_up', 'asleep', 'hanging out'],
            ['work_out', 'hanging out', 'hungry'],
            ['eat', 'hungry', 'hanging out'],
            ['nap', '*', 'asleep'],
            dict(trigger='distress_call', source='*', dest='saving the world', before='change_into_super_secret_costume'),
            dict(trigger='complete_mission', source='saving the world', dest='sweaty', after='update_journal'),
            dict(trigger='clean_up', source='sweaty', dest='asleep', conditions=['is_exhausted']),
            ['clean_up', 'sweaty', 'hanging out'],
        ]
    
        def __init__(self):
            super().__init__(model=[NarcolepticSuperhero('Clone warrior') for i in range(100)],
                             states=self.states,
                             transitions=self.transitions,
                             # tell each model to 'log_out' right before state change
                             before_state_change='log_out',
                             # and to 'log_in' right after
                             after_state_change='log_in',
                             initial='asleep')
            # since our super heroes are asleep (and 'spawn' in their state), we have to log them in the first time
            for model in self.models:
                NarcolepticSuperhero.hero_log[self.initial].append(model)
    
    
    machine = SuperheroDen()
    # trigger event 'wake_up' on all models
    machine.dispatch('wake_up')
    assert len(NarcolepticSuperhero.hero_log['asleep']) == 0
    assert len(NarcolepticSuperhero.hero_log['hanging out']) == 100
    for i in range(10):
        machine.models[i].work_out()
    assert len(NarcolepticSuperhero.hero_log['hanging out']) == 90
    assert len(NarcolepticSuperhero.hero_log['hungry']) == 10
    assert machine.models[0] in NarcolepticSuperhero.hero_log['hungry']