Memoru

System Engineering and Programming and IT

watchdog

tree /f

PS G:\workspace\py\sandbox\watchdog_sample> tree /f
Folder PATH listing for volume Users
Volume serial number is 04A1-7D2F
G:.
│  app.py
│  apps.log
│  configbase.py
│  logging_debug.conf
│  watchdog_sample.py
│
├─configs
│  │  config.py
│  │  config_debug.py
│  │  default.py
│  │
│  └─__pycache__
│
├─plugins
│  │  a.py
│  │  b.py
│  │  c.py
│  │
│  └─__pycache__
│
└─__pycache__

watchdog

#! /usr/bin/env python
"""{fname}
# sample code for module : watchdog
Usage:
"""

__author__ = 'Memoru'
__version__ = '0.0.1'
__date__ = '2018-06-23'

import sys
import os
__doc__ = __doc__.format(fname=os.path.basename(__file__))

import logging.config
import configbase as base
CONF = base.CONF
ENV = base.ENV


import sys
import time
from importlib import import_module, reload
from pathlib import Path

from watchdog.events import FileSystemEvent, PatternMatchingEventHandler
from watchdog.observers import Observer


class PluginManager:

    class Handler(PatternMatchingEventHandler):
        def __init__(self, manager: 'PluginManager', *args, **kwargs):
            super().__init__(*args, **kwargs)
            self.manager = manager

        def on_created(self, event: FileSystemEvent):
            print(event)
            if event.src_path.endswith('.py'):
                self.manager.load_plugin(Path(event.src_path))

        def on_modified(self, event):
            print(event)

    def __init__(self, path: str):
        self.plugins = {}
        self.path = path
        self.observer = Observer()
        sys.path.append(self.path)

    def start(self):
        self.scan_plugin()
        self.observer.schedule(self.Handler(self, patterns='*.py'), self.path)
        self.observer.start()

    def stop(self):
        self.observer.stop()
        self.observer.join()

    def scan_plugin(self):
        for file_path in Path(self.path).glob('*.py'):
            self.load_plugin(file_path)

    def load_plugin(self, file_path):
        module_name = file_path.stem
        if module_name not in self.plugins:
            self.plugins[module_name] = import_module(module_name)
            print('{} loaded.'.format(module_name))
        else:
            self.plugins[module_name] = reload(self.plugins[module_name])
            print('{} reloaded.'.format(module_name))


def main():

    plugin_manager = PluginManager('plugins/')
    plugin_manager.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        plugin_manager.stop()


#  Debug ---------------------------------------  #

def process():
    try:
        logger = logging.getLogger(__name__)
        logger.debug('◇◇--Begin sample -----------------◇◇')
        logger.debug(__doc__)
        main()
        logger.debug('◆◆--End   sample -----------------◆◆')
    except Exception as ex:
        logger.exception(ex)
        raise ex
        

if __name__ == '__main__':
    logging.config.fileConfig('logging_debug.conf')
    process()

execution

(sandbox) PS G:\workspace\py\sandbox\watchdog_sample> python .\watchdog_sample.py
[2018/06/27 07:10:43][DEBUG](watchdog_sample.py:91) ◇◇--Begin sample -----------------◇◇
[2018/06/27 07:10:43][DEBUG](watchdog_sample.py:92) watchdog_sample.py
# sample code for module : watchdog
Usage:

a loaded.
b loaded.
c loaded.