mfc中嵌入python_Python 中的 Hook 钩子函数
生活随笔
收集整理的這篇文章主要介紹了
mfc中嵌入python_Python 中的 Hook 钩子函数
小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.
1. 什么是Hook
經(jīng)常會聽到鉤子函數(shù)(hook function)這個概念，最近在看目標檢測開源框架mmdetection，里面也出現(xiàn)大量Hook的編程方式，那到底什么是hook？hook的作用是什么？
- what is hook ？鉤子hook，顧名思義，可以理解是一個掛鉤，作用是有需要的時候掛一個東西上去。具體的解釋是：鉤子函數(shù)是把我們自己實現(xiàn)的hook函數(shù)在某一時刻掛接到目標掛載點上。
- hook函數(shù)的作用 舉個例子，hook的概念在windows桌面軟件開發(fā)很常見，特別是各種事件觸發(fā)的機制; 比如C++的MFC程序中，要監(jiān)聽鼠標左鍵按下的時間，MFC提供了一個onLeftKeyDown的鉤子函數(shù)。很顯然，MFC框架并沒有為我們實現(xiàn)onLeftKeyDown具體的操作，只是為我們提供一個鉤子，當我們需要處理的時候，只要去重寫這個函數(shù)，把我們需要操作掛載在這個鉤子里，如果我們不掛載，MFC事件觸發(fā)機制中執(zhí)行的就是空的操作。
從上面可知
- hook函數(shù)是程序中預定義好的函數(shù)，這個函數(shù)處于原有程序流程當中（暴露一個鉤子出來）
- 我們需要再在有流程中鉤子定義的函數(shù)塊中實現(xiàn)某個具體的細節(jié)，需要把我們的實現(xiàn)，掛接或者注冊（register）到鉤子里，使得hook函數(shù)對目標可用
- hook 是一種編程機制，和具體的語言沒有直接的關系
- 如果從設計模式上看，hook模式是模板方法的擴展
- 鉤子只有注冊的時候，才會使用，所以原有程序的流程中，沒有注冊或掛載時，執(zhí)行的是空（即沒有執(zhí)行任何操作）
本文用python來解釋hook的實現(xiàn)方式，并展示在開源項目中hook的應用案例。hook函數(shù)和我們常聽到另外一個名稱：回調(diào)函數(shù)（callback function）功能是類似的，可以按照同種模式來理解。
2. hook實現(xiàn)例子
據(jù)我所知，hook函數(shù)最常使用在某種流程處理當中。這個流程往往有很多步驟。hook函數(shù)常常掛載在這些步驟中，為增加額外的一些操作，提供靈活性。
下面舉一個簡單的例子，這個例子的目的是實現(xiàn)一個通用往隊列中插入內(nèi)容的功能。流程步驟有2個
- 需要再插入隊列前，對數(shù)據(jù)進行篩選 input_filter_fn
- 插入隊列 insert_queue
class ContentStash(object):
"""
content stash for online operation
pipeline is
1. input_filter: filter some contents, no use to user
2. insert_queue(redis or other broker): insert useful content to queue
"""
def __init__(self):
self.input_filter_fn = None
self.broker = []
def register_input_filter_hook(self, input_filter_fn):
"""
register input filter function, parameter is content dict
Args:
input_filter_fn: input filter function
Returns:
"""
self.input_filter_fn = input_filter_fn
def insert_queue(self, content):
"""
insert content to queue
Args:
content: dict
Returns:
"""
self.broker.append(content)
def input_pipeline(self, content, use=False):
"""
pipeline of input for content stash
Args:
use: is use, defaul False
content: dict
Returns:
"""
if not use:
return
# input filter
if self.input_filter_fn:
_filter = self.input_filter_fn(content)
# insert to queue
if not _filter:
self.insert_queue(content)
# test
## 實現(xiàn)一個你所需要的鉤子實現(xiàn)：比如如果content 包含time就過濾掉，否則插入隊列
def input_filter_hook(content):
"""
test input filter hook
Args:
content: dict
Returns: None or content
"""
if content.get('time') is None:
return
else:
return content
# 原有程序
content = {'filename': 'test.jpg', 'b64_file': "#test", 'data': {"result": "cat", "probility": 0.9}}
content_stash = ContentStash('audit', work_dir='')
# 掛上鉤子函數(shù)， 可以有各種不同鉤子函數(shù)的實現(xiàn)，但是要主要函數(shù)輸入輸出必須保持原有程序中一致，比如這里是content
content_stash.register_input_filter_hook(input_filter_hook)
# 執(zhí)行流程
content_stash.input_pipeline(content)
3. hook在開源框架中的應用
3.1 keras
在深度學習訓練流程中，hook函數(shù)體現(xiàn)的淋漓盡致。
一個訓練過程（不包括數(shù)據(jù)準備），會輪詢多次訓練集，每次稱為一個epoch，每個epoch又分為多個batch來訓練。流程先后拆解成：
- 開始訓練
- 訓練一個epoch前
- 訓練一個batch前
- 訓練一個batch后
- 訓練一個epoch后
- 評估驗證集
- 結束訓練
這些步驟是穿插在訓練一個batch數(shù)據(jù)的過程中，這些可以理解成是鉤子函數(shù)，我們可能需要在這些鉤子函數(shù)中實現(xiàn)一些定制化的東西，比如在訓練一個epoch后我們要保存下訓練的模型，在結束訓練時用最好的模型執(zhí)行下測試集的效果等等。
keras中是通過各種回調(diào)函數(shù)來實現(xiàn)鉤子hook功能的。這里放一個callback的父類，定制時只要繼承這個父類，實現(xiàn)你過關注的鉤子就可以了。
@keras_export('keras.callbacks.Callback')
class Callback(object):
"""Abstract base class used to build new callbacks.
Attributes:
params: Dict. Training parameters
(eg. verbosity, batch size, number of epochs...).
model: Instance of `keras.models.Model`.
Reference of the model being trained.
The `logs` dictionary that callback methods
take as argument will contain keys for quantities relevant to
the current batch or epoch (see method-specific docstrings).
"""
def __init__(self):
self.validation_data = None # pylint: disable=g-missing-from-attributes
self.model = None
# Whether this Callback should only run on the chief worker in a
# Multi-Worker setting.
# TODO(omalleyt): Make this attr public once solution is stable.
self._chief_worker_only = None
self._supports_tf_logs = False
def set_params(self, params):
self.params = params
def set_model(self, model):
self.model = model
@doc_controls.for_subclass_implementers
@generic_utils.default
def on_batch_begin(self, batch, logs=None):
"""A backwards compatibility alias for `on_train_batch_begin`."""
@doc_controls.for_subclass_implementers
@generic_utils.default
def on_batch_end(self, batch, logs=None):
"""A backwards compatibility alias for `on_train_batch_end`."""
@doc_controls.for_subclass_implementers
def on_epoch_begin(self, epoch, logs=None):
"""Called at the start of an epoch.
Subclasses should override for any actions to run. This function should only
be called during TRAIN mode.
Arguments:
epoch: Integer, index of epoch.
logs: Dict. Currently no data is passed to this argument for this method
but that may change in the future.
"""
@doc_controls.for_subclass_implementers
def on_epoch_end(self, epoch, logs=None):
"""Called at the end of an epoch.
Subclasses should override for any actions to run. This function should only
be called during TRAIN mode.
Arguments:
epoch: Integer, index of epoch.
logs: Dict, metric results for this training epoch, and for the
validation epoch if validation is performed. Validation result keys
are prefixed with `val_`.
"""
@doc_controls.for_subclass_implementers
@generic_utils.default
def on_train_batch_begin(self, batch, logs=None):
"""Called at the beginning of a training batch in `fit` methods.
Subclasses should override for any actions to run.
Arguments:
batch: Integer, index of batch within the current epoch.
logs: Dict, contains the return value of `model.train_step`. Typically,
the values of the `Model`'s metrics are returned. Example:
`{'loss': 0.2, 'accuracy': 0.7}`.
"""
# For backwards compatibility.
self.on_batch_begin(batch, logs=logs)
@doc_controls.for_subclass_implementers
@generic_utils.default
def on_train_batch_end(self, batch, logs=None):
"""Called at the end of a training batch in `fit` methods.
Subclasses should override for any actions to run.
Arguments:
batch: Integer, index of batch within the current epoch.
logs: Dict. Aggregated metric results up until this batch.
"""
# For backwards compatibility.
self.on_batch_end(batch, logs=logs)
@doc_controls.for_subclass_implementers
@generic_utils.default
def on_test_batch_begin(self, batch, logs=None):
"""Called at the beginning of a batch in `evaluate` methods.
Also called at the beginning of a validation batch in the `fit`
methods, if validation data is provided.
Subclasses should override for any actions to run.
Arguments:
batch: Integer, index of batch within the current epoch.
logs: Dict, contains the return value of `model.test_step`. Typically,
the values of the `Model`'s metrics are returned. Example:
`{'loss': 0.2, 'accuracy': 0.7}`.
"""
@doc_controls.for_subclass_implementers
@generic_utils.default
def on_test_batch_end(self, batch, logs=None):
"""Called at the end of a batch in `evaluate` methods.
Also called at the end of a validation batch in the `fit`
methods, if validation data is provided.
Subclasses should override for any actions to run.
Arguments:
batch: Integer, index of batch within the current epoch.
logs: Dict. Aggregated metric results up until this batch.
"""
@doc_controls.for_subclass_implementers
@generic_utils.default
def on_predict_batch_begin(self, batch, logs=None):
"""Called at the beginning of a batch in `predict` methods.
Subclasses should override for any actions to run.
Arguments:
batch: Integer, index of batch within the current epoch.
logs: Dict, contains the return value of `model.predict_step`,
it typically returns a dict with a key 'outputs' containing
the model's outputs.
"""
@doc_controls.for_subclass_implementers
@generic_utils.default
def on_predict_batch_end(self, batch, logs=None):
"""Called at the end of a batch in `predict` methods.
Subclasses should override for any actions to run.
Arguments:
batch: Integer, index of batch within the current epoch.
logs: Dict. Aggregated metric results up until this batch.
"""
@doc_controls.for_subclass_implementers
def on_train_begin(self, logs=None):
"""Called at the beginning of training.
Subclasses should override for any actions to run.
Arguments:
logs: Dict. Currently no data is passed to this argument for this method
but that may change in the future.
"""
@doc_controls.for_subclass_implementers
def on_train_end(self, logs=None):
"""Called at the end of training.
Subclasses should override for any actions to run.
Arguments:
logs: Dict. Currently the output of the last call to `on_epoch_end()`
is passed to this argument for this method but that may change in
the future.
"""
@doc_controls.for_subclass_implementers
def on_test_begin(self, logs=None):
"""Called at the beginning of evaluation or validation.
Subclasses should override for any actions to run.
Arguments:
logs: Dict. Currently no data is passed to this argument for this method
but that may change in the future.
"""
@doc_controls.for_subclass_implementers
def on_test_end(self, logs=None):
"""Called at the end of evaluation or validation.
Subclasses should override for any actions to run.
Arguments:
logs: Dict. Currently the output of the last call to
`on_test_batch_end()` is passed to this argument for this method
but that may change in the future.
"""
@doc_controls.for_subclass_implementers
def on_predict_begin(self, logs=None):
"""Called at the beginning of prediction.
Subclasses should override for any actions to run.
Arguments:
logs: Dict. Currently no data is passed to this argument for this method
but that may change in the future.
"""
@doc_controls.for_subclass_implementers
def on_predict_end(self, logs=None):
"""Called at the end of prediction.
Subclasses should override for any actions to run.
Arguments:
logs: Dict. Currently no data is passed to this argument for this method
but that may change in the future.
"""
def _implements_train_batch_hooks(self):
"""Determines if this Callback should be called for each train batch."""
return (not generic_utils.is_default(self.on_batch_begin) or
not generic_utils.is_default(self.on_batch_end) or
not generic_utils.is_default(self.on_train_batch_begin) or
not generic_utils.is_default(self.on_train_batch_end))
這些鉤子的原始程序是在模型訓練流程中的
keras源碼位置： tensorflowpythonkerasenginetraining.py部分摘錄如下(## I am hook)：
# Container that configures and calls `tf.keras.Callback`s.
if not isinstance(callbacks, callbacks_module.CallbackList):
callbacks = callbacks_module.CallbackList(
callbacks,
add_history=True,
add_progbar=verbose != 0,
model=self,
verbose=verbose,
epochs=epochs,
steps=data_handler.inferred_steps)
## I am hook
callbacks.on_train_begin()
training_logs = None
# Handle fault-tolerance for multi-worker.
# TODO(omalleyt): Fix the ordering issues that mean this has to
# happen after `callbacks.on_train_begin`.
data_handler._initial_epoch = ( # pylint: disable=protected-access
self._maybe_load_initial_epoch_from_ckpt(initial_epoch))
for epoch, iterator in data_handler.enumerate_epochs():
self.reset_metrics()
callbacks.on_epoch_begin(epoch)
with data_handler.catch_stop_iteration():
for step in data_handler.steps():
with trace.Trace(
'TraceContext',
graph_type='train',
epoch_num=epoch,
step_num=step,
batch_size=batch_size):
## I am hook
callbacks.on_train_batch_begin(step)
tmp_logs = train_function(iterator)
if data_handler.should_sync:
context.async_wait()
logs = tmp_logs # No error, now safe to assign to logs.
end_step = step + data_handler.step_increment
callbacks.on_train_batch_end(end_step, logs)
epoch_logs = copy.copy(logs)
# Run validation.
## I am hook
callbacks.on_epoch_end(epoch, epoch_logs)
3.2 mmdetection
mmdetection是一個目標檢測的開源框架，集成了許多不同的目標檢測深度學習算法（pytorch版），如faster-rcnn, fpn, retianet等。里面也大量使用了hook，暴露給應用實現(xiàn)流程中具體部分。
詳見https://github.com/open-mmlab/mmdetection
這里看一個訓練的調(diào)用例子(摘錄)（https://github.com/open-mmlab/mmdetection/blob/5d592154cca589c5113e8aadc8798bbc73630d98/mmdet/apis/train.py）
def train_detector(model,
dataset,
cfg,
distributed=False,
validate=False,
timestamp=None,
meta=None):
logger = get_root_logger(cfg.log_level)
# prepare data loaders
# put model on gpus
# build runner
optimizer = build_optimizer(model, cfg.optimizer)
runner = EpochBasedRunner(
model,
optimizer=optimizer,
work_dir=cfg.work_dir,
logger=logger,
meta=meta)
# an ugly workaround to make .log and .log.json filenames the same
runner.timestamp = timestamp
# fp16 setting
# register hooks
runner.register_training_hooks(cfg.lr_config, optimizer_config,
cfg.checkpoint_config, cfg.log_config,
cfg.get('momentum_config', None))
if distributed:
runner.register_hook(DistSamplerSeedHook())
# register eval hooks
if validate:
# Support batch_size > 1 in validation
eval_cfg = cfg.get('evaluation', {})
eval_hook = DistEvalHook if distributed else EvalHook
runner.register_hook(eval_hook(val_dataloader, **eval_cfg))
# user-defined hooks
if cfg.get('custom_hooks', None):
custom_hooks = cfg.custom_hooks
assert isinstance(custom_hooks, list),
f'custom_hooks expect list type, but got {type(custom_hooks)}'
for hook_cfg in cfg.custom_hooks:
assert isinstance(hook_cfg, dict),
'Each item in custom_hooks expects dict type, but got '
f'{type(hook_cfg)}'
hook_cfg = hook_cfg.copy()
priority = hook_cfg.pop('priority', 'NORMAL')
hook = build_from_cfg(hook_cfg, HOOKS)
runner.register_hook(hook, priority=priority)
4. 總結
本文介紹了hook的概念和應用，并給出了python的實現(xiàn)細則。希望對比有幫助。
感謝閱讀！！！
多說一句，很多人學Python過程中會遇到各種煩惱問題，沒有人解答容易放棄。小編是一名python開發(fā)工程師，這里有我自己整理了一套最新的python系統(tǒng)學習教程，包括從基礎的python腳本到web開發(fā)、爬蟲、數(shù)據(jù)分析、數(shù)據(jù)可視化、機器學習等。想要這些資料的可以關注小編，并在后臺私信小編：“01”即可領取。
總結
以上是生活随笔為你收集整理的mfc中嵌入python_Python 中的 Hook 钩子函数的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： # 模糊图片边缘_PS大神给你变个魔术，
- 下一篇： python2与python3性能对比_