So I thought I might just post it here.
Imagine you have a VMF-like part of an output connection that connects output of one entity (caller) to the input of another entity (target) and applies some restriction to such connection (delay and maximum allowed times to fire).
The whole connection would look like
"OnPressed" "my_door01,Toggle,,0,-1"
In this case when something is pressed (func_button probably), entity with a targetname "my_door01" recieves an input Toggle with no parameter, zero delay (immediately) and this connection can be fired as many times as you want (-1).
And you need to execute the my_door01,Toggle,,0,-1 part of it.
More samples:
func_door,Open,,0,-1 - opens all func_door's on the map
func_door*,Open,,0,-1 - opens all regular doors (func_door) and rotating doors (func_door_rotating) on the map
!player,SetHealth,42,5,-1 - sets the health of the first found player to 42 with a delay of 5 seconds
*,SetHealth,1,0,1 - sets the health of everything (actually of those entities that support that) to 1, but this particular instance of a connection should only fire once a round.
There're several existing approaches:
1. Strip 'cheat' flag from ent_fire console command, execute it on a client, then put the flag back.
Pros: Easy, behaves the same exact way as any other usual entity IO call
Cons: Your activator and caller is the player you're executing this command on, you can't explicitly change it. Also, you can't set a maximum allowed times to fire, you'll need to track it somehow by yourself.
Notes: ent_fire uses a bit different format where input name is separated with a space from a rest of the connection part, so you must reformat the string.
2. Spawn a dummy cross-game entity like info_target, add output OnUser1 with the needed content (func_door,Open,,0,-1) and then fire FireUser1 on it. Don't forget to remove the dummy on the next tick.
Pros: Easy, behaves the same exact way as any other usual entity IO call
Cons: Your caller is this dummy. I'm not sure about activator though.
3. Take a journey of finding those entities by yourself. Then you are trying to actually call the input on each of those.
Pros: Customizable, full control, ability to exclude/include any entities, you can also define Gaben as your caller or activator.
Cons: You basically reinvent what Source engine does for you. No guarantee that it will find the same entities as the previous two approaches would.
So, for the third approach you need to...
Import:
Syntax: Select all
from entities.classes import server_classes
from entities.datamaps import FieldType
from events import Event
from filters.entities import BaseEntityIter
from filters.players import PlayerIter
from listeners.tick import tick_delays
from memory import make_object
Map expected input types to python classes (do it somewhere in a global namespace of your module):
Syntax: Select all
_input_types = {
FieldType.BOOLEAN: lambda arg: arg == '1', # So that '0' won't become True
FieldType.FLOAT: float,
FieldType.INTEGER: int,
FieldType.STRING: str,
FieldType.VOID: None,
}Define [B]Fire class:[/B]
Syntax: Select all
class Fire:
def __call__(self, target_pattern, input_name, parameter=None, caller=None, activator=None):
"""Find target entities using the given pattern and try to call an input on each of them"""
targets = self._get_targets(target_pattern, caller, activator)
for target in targets:
self._call_input(target, input_name, parameter, caller, activator)
def _get_targets(self, target_pattern, caller, activator):
"""Return iterable of targets depending on given pattern, caller and activator."""
if target_pattern.startswith('!'):
return self._get_special_name_target(target_pattern, caller, activator)
filter_ = self._get_entity_filter(target_pattern, caller, activator)
return filter(filter_, BaseEntityIter())
def _get_special_name_target(self, target_pattern, caller, activator):
"""Find target by a special (starting with '!') target name."""
if target_pattern == "!self":
return (caller, )
if target_pattern == "!player":
for player in PlayerIter():
return (player, )
return ()
if target_pattern in ("!caller", "!activator"):
return (activator, )
def _get_entity_filter(self, target_pattern, caller, activator):
"""Return a filter that will be applied to all entities on the server."""
if target_pattern.endswith('*'):
def filter_(entity):
targetname = entity.get_key_value_string('targetname')
return (targetname.startswith(target_pattern[:-1]) or
entity.classname.startswith(target_pattern[:-1]))
return filter_
if not target_pattern:
return lambda entity: False
def filter_(entity):
targetname = entity.get_key_value_string('targetname')
return target_pattern in (targetname, entity.classname)
return filter_
def _get_input(self, target, input_name):
"""Return input function based on target and input name."""
for server_class in server_classes.get_entity_server_classes(target):
if input_name in server_class.inputs:
return getattr(
make_object(server_class._inputs, target.pointer), input_name)
return None
def _call_input(self, target, input_name, parameter, caller, activator):
"""Fire an input of a particular entity."""
input_function = self._get_input(target, input_name)
# If entity doesn't support the input, we don't work with this entity
if input_function is None:
return
caller_index = None if caller is None else caller.index
activator_index = None if activator is None else activator.index
# Check if type is unsupported, but we actually support all types that can possibly
# be passed as a string to input: int, float, bool, str
# TODO: Implement support for entity arguments (passed as a special name like !activator, !caller etc)
if input_function._argument_type not in _input_types:
return
type_ = _input_types[input_function._argument_type]
# Case: input does not require parameter
if type_ is None:
parameter = None
# Case: input does require parameter
else:
# Try to cast the parameter to the given type
try:
parameter = type_(parameter)
# We don't give up the target if the value can't be casted;
# Instead, we fire its input with a default value just like ent_fire does
except ValueError:
parameter = type_()
# Fire an input
input_function(parameter, caller_index, activator_index)
Define OutputConnection class:
Syntax: Select all
class OutputConnection:
def __init__(self, fire_func, destroy_func, string, caller=None, activator=None):
try:
target_pattern, input_name, parameter, delay, times_to_fire = string.split(',')
except ValueError:
raise ValueError("Invalid output connection string")
delay = max(0.0, float(delay))
times_to_fire = max(-1, int(times_to_fire))
self._fire_func = fire_func
self._destroy_func = destroy_func
self.target_pattern = target_pattern
self.input_name = input_name
self.parameter = parameter or None
self.delay = delay
self.times_to_fire = times_to_fire
self.caller = caller
self.activator = activator
self._delayed_callbacks = []
self._times_fired = 0
def reset(self):
"""Cancel all pending callbacks and set fire count to zero."""
for delayed_callback in self._delayed_callbacks:
try:
delayed_callback.cancel()
except KeyError:
continue
self._delayed_callbacks = []
self._times_fired = 0
def fire(self):
"""Fire this output connection."""
if self.times_to_fire > -1 and self._times_fired >= self.times_to_fire:
return
def callback():
self._fire_func(self.target_pattern, self.input_name, self.parameter, self.caller, self.activator)
if self.delay == 0.0:
callback()
else:
self._delayed_callbacks.append(tick_delays.delay(self.delay, callback))
def destroy(self):
"""Remove a reference to the connection from this lib and stop resetting connection on every round start."""
self._destroy_func(self)
def __str__(self):
return "OutputConnection('{0},{1},{2},{3},{4}')".format(
self.target_pattern, self.input_name, self.parameter or "", self.delay, self.times_to_fire
)
Prepare your methods:
Syntax: Select all
output_connections = [] # We will store connections here
fire = Fire() # callable that is used to call inputs when they're ready
Define your methods:
Syntax: Select all
def new_output_connection(string, caller=None, activator=None):
"""Create and register a new OutputConnection instance using given values."""
output_connection = OutputConnection(fire, destroy_output_connection, string, caller, activator)
output_connections.append(output_connection)
return output_connection
def destroy_output_connection(output_connection):
"""Remove connection reference from this lib thus stopping resetting the connection every round."""
output_connections.remove(output_connection)
Attach an event handler to round_start event (you may use Event decorator):
Syntax: Select all
@Event('round_start')
def on_round_start(game_event):
for output_connection in output_connections:
output_connection.reset()Use it like that:
Syntax: Select all
connection = new_output_connection("my_door01,Toggle,,0,-1")
connection.fire()
connection.destroy()Once you destroy your connection, you can still use it, but it won't reset when a round starts anymore.
If you don't need your connection, don't forget to destroy it, otherwise it won't be garbage collected.
