/
hooks.py
225 lines (172 loc) · 6.81 KB
/
hooks.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
# ../entities/hooks.py
"""Provides memory hooking functionality for entities."""
# =============================================================================
# >> IMPORTS
# =============================================================================
# Source.Python Imports
# Core
from core import AutoUnload
# Entities
# Memory
from memory.hooks import HookType
# Filters
from filters.entities import EntityIter
# Listeners
from listeners import OnNetworkedEntityCreated
# Entities
from entities.entity import Entity
# Players
from players.entity import Player
# =============================================================================
# >> ALL DECLARATION
# =============================================================================
__all__ = ('EntityCondition',
'EntityPostHook',
'EntityPreHook',
)
# =============================================================================
# >> CLASSES
# =============================================================================
class EntityCondition(object):
"""Store some default entity conditions."""
@staticmethod
def is_player(entity):
"""Return ``True`` if the entity is a player.
:rtype: bool
"""
return entity.is_player()
@staticmethod
def is_not_player(entity):
"""Return ``True`` if the entity is not a player.
:rtype: bool
"""
return not entity.is_player()
@staticmethod
def is_human_player(entity):
"""Return ``True`` if the entity is a human player.
:rtype: bool
"""
return entity.is_player() and Player(entity.index).steamid != 'BOT'
@staticmethod
def is_bot_player(entity):
"""Return ``True`` if the entity is a bot.
:rtype: bool
"""
return entity.is_player() and Player(entity.index).steamid == 'BOT'
@staticmethod
def equals_entity_classname(*classnames):
"""Return a function that requires an :class:`entities.entity.Entity``
instace. The returned function returns ``True`` if the entity's
classname equals one of the passed classnames.
:rtype: lambda
"""
return lambda entity: entity.classname in classnames
@staticmethod
def equals_datamap_classname(*classnames):
"""Return a function that requires an :class:`entities.entity.Entity``
instance. The returned function returns ``True`` if the entity's
datamap classname equals one of the passed classnames.
:rtype: lambda
"""
return lambda entity: entity.datamap.class_name in classnames
class _EntityHook(AutoUnload):
"""Create entity pre and post hooks that auto unload."""
def __init__(self, test_function, function):
"""Initialize the hook object.
:param callable test_function:
A callable object that accepts an :class:`entities.entity.Entity`
instance as a parameter. The function should return ``True`` if
the entity matches the required one.
:param str/callable function:
This is the function to hook. It can be either a string that
defines the name of a function of the entity or a callable object
that returns a :class:`memory.Function` instance.
"""
self.test_function = test_function
self.function = function
self.hooked_function = None
self.callback = None
def __call__(self, callback):
"""Store the callback and try initializing the hook.
:param callable callback:
The callback to store.
:return:
The passed callback.
:rtype: callable
"""
# Validate the given callback...
if not callable(callback):
raise TypeError('Given callback is not callable.')
self.callback = callback
# Try initializing the hook...
for entity in EntityIter():
if self.initialize(entity):
# Yay! The entity was the one we were looking for
return self.callback
# Initialization failed. There is currently no entity with the given
# class name. So, we need to wait until such an entity has been
# created.
_waiting_entity_hooks.append(self)
# Return the callback
return self.callback
@property
def hook_type(self):
"""Return the hook type of the decorator.
:rtype: HookType
"""
raise NotImplementedError('No hook_type defined for class.')
def initialize(self, entity):
"""Initialize the hook.
Return ``True`` if the initialization was successful.
:rtype: bool
"""
if not self.test_function(entity):
return False
if callable(self.function):
self.hooked_function = self.function(entity)
else:
self.hooked_function = getattr(entity, self.function)
self.hooked_function.add_hook(self.hook_type, self.callback)
return True
def _unload_instance(self):
"""Unload the hook."""
# Was a function hooked?
if self.hooked_function is not None:
# Was no callback registered?
if self.callback is None:
return
# Unregister the hook...
self.hooked_function.remove_hook(self.hook_type, self.callback)
# Otherwise, make sure the hook is still pending before removing it...
elif self in _waiting_entity_hooks:
_waiting_entity_hooks.remove(self)
class EntityPreHook(_EntityHook):
"""Decorator class used to create entity pre hooks that auto unload."""
hook_type = HookType.PRE
class EntityPostHook(_EntityHook):
"""Decorator class used to create entity post hooks that auto unload."""
hook_type = HookType.POST
class _WaitingEntityHooks(list):
"""A dictionary to store hooks waiting for intialization."""
def initialize(self, index):
"""Initialize all hooks waiting for the given entity.
:param int index:
Index of the entity that should be used to initialize the hooks.
"""
# There is nothing to do if no hook is waiting
if not self:
return
entity = Entity(index)
for hook in tuple(self):
# Try initializing the hook
if hook.initialize(entity):
# If it succeeded, remove the hook from the waiting list
self.remove(hook)
_waiting_entity_hooks = _WaitingEntityHooks()
# =============================================================================
# >> LISTENERS
# =============================================================================
@OnNetworkedEntityCreated
def on_networked_entity_created(entity):
"""Called when a new networked entity has been created."""
_waiting_entity_hooks.initialize(entity.index)