Attachable Test

attachable_test.py

from ignite.engine import Events, Engine

class Attachable:
    """
    This class contains an internal dictionary `attach_map` associating events with the bound methods to be triggered
    on those events. This dictionary can be populated by the contructor or by external clients after construction and 
    before `attach` is called. 
    """
    def __init__(self, attach_map):
        self.attach_map = dict(attach_map)
        
    def attach(self, engine: Engine, name: str) -> None:
        for event,method in self.attach_map.items():
            if not engine.has_event_handler(method, event):
                # used to ensure that only the epoch complete event has the `name` argument passed to it
                args=(name,) if event == Events.EPOCH_COMPLETED else ()
                
                engine.add_event_handler(event, method, *args)
            
    def detach(self, engine: Engine) -> None:
        for event,method in self.attach_map.items():
            if engine.has_event_handler(method, event):
                engine.remove_event_handler(method, event)
                
    def is_attached(self, engine: Engine) -> bool:
        return any(engine.has_event_handler(method, event) for method, event in self.attach_map.items())
    
    
class AttachableTest(Attachable):
    """
    Example use case of Attachable, the constructor creates two attachments with defined methods.
    """
    def __init__(self):
        attachments={
            Events.EPOCH_COMPLETED : self.completed,
            Events.EPOCH_STARTED : self.started,
        }
        super().__init__(attachments)
        
    def completed(self, engine: Engine, name: str) -> None:
        print(f'epoch completed, name = {repr(name)}')
        
    def started(self, engine: Engine) -> None:
        print('epoch started')

    def iteration_completed(self, engine: Engine) -> None:
        print('iteration completed')
        
        
engine=Engine(process_function=lambda *_:None)

at=AttachableTest()

# add another attachment by adding a dictionary entry
at.attach_map[Events.ITERATION_COMPLETED] = at.iteration_completed

at.attach(engine, 'test')

engine.run([None]) # silly run of the engine