Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
}
logger = DummyLogger()
# Start this test at the top of the hour, at least an hour before the current time
# Top of the hour gives some repeatability.
# It's based on the current time because otherwise the radio will infer the wrong year on message timestamps.
# TODO Design a better solution: Get year from context? Get time from context? Get time from radio?
t = int((time.time() - 60 * 60) / (60 * 60)) * 60 * 60
si4707args = "--hardware-context RPiNWR.sources.radio.Si4707.mock.MockContext --mute-after 0 --transmitter WXL58".split()
injector = ScriptInjector()
clock = MockClock(epoch=t)
self.box = box = Radio_Component(si4707args, clock=clock.time) + Radio_Squelch() + \
TextPull(location=location, url='http://127.0.0.1:17/') + \
AlertTimer(continuation_reminder_interval_sec=.05, clock=clock.time) + \
MessageCache(location, clock=clock.time) + \
Debugger(logger=logger) + injector
box.start()
logger.wait_for_n_events(1, re.compile('
def test_Logger_error():
app = App()
logger = Logger()
debugger = Debugger(logger=logger)
debugger.register(app)
while len(app):
app.flush()
e = test(raiseException=True)
app.fire(e)
while len(app):
app.flush()
assert logger.error_msg.startswith("ERROR (")
def __init__(self, *args, **kwargs):
super(Test, self).__init__(*args, **kwargs)
pygame.init()
self += PyGameDriver() + Debugger()
screen = pygame.display.set_mode((640, 480), DOUBLEBUF | HWSURFACE)
def main():
opts, args = parse_options()
if opts.speed and psyco:
psyco.full()
manager = Manager()
monitor = Monitor(opts)
manager += monitor
state = State(opts)
manager += state
if opts.debug:
manager += Debugger()
if opts.listen or args:
nodes = []
if args:
for node in args:
if ":" in node:
host, port = node.split(":")
port = int(port)
else:
host = node
port = 8000
nodes.append((host, port))
if opts.bind is not None:
if ":" in opts.bind:
address, port = opts.bind.split(":")
def main():
manager = Manager()
debugger = Debugger()
bridge = Bridge(8001, nodes=[("127.0.0.1", 8000)])
debugger.IgnoreEvents = ["Read", "Write"]
manager += bridge
manager += debugger
manager += Client()
manager.start()
sTime = time()
while True:
try:
if (time() - sTime) > 5:
manager.push(Event("Hello World"), "hello")
def main():
opts, args = parse_options()
# Configure and "run" the System.
app = Telnet(*args, **opts.__dict__)
if opts.verbose:
from circuits import Debugger
Debugger().register(app)
stdin.register(app)
app.run()
self.fire(task(factorial, 11)) # async
self.fire(task(factorial, 12)) # async
self.fire(task(factorial, 14)) # async
Timer(1, Event.create("foo"), persist=True).register(self)
def task_success(self, function_called, factorial_result):
func, argument = function_called
print("factorial({0}) = {1:d}".format(str(argument), factorial_result))
# Stop after the last and longest running task
if argument == 14:
self.stop()
if __name__ == '__main__':
app = App()
Debugger().register(app)
app.run()
def start(self):
# Create a new circuits Manager
#ignevents = [Update, MouseMove]
ignchannames = ['update', 'started', 'on_mousemove', 'on_mousedrag', 'on_keydown', 'on_input', 'on_mouseclick', 'on_entityupdated', 'on_exit', 'on_keyup', 'on_login', 'on_inboundnetwork', 'on_genericmessage', 'on_scene', 'on_entity_visuals_modified', 'on_logout', 'on_worldstreamready', 'on_sceneadded']
ignchannels = [('*', n) for n in ignchannames]
# Note: instantiating Manager with debugger causes severe lag when running as a true windowed app (no console), so instantiate without debugger
# Fix attempt: give the circuits Debugger a logger which uses Naali logging system, instead of the default which writes to sys.stderr
# Todo: if the default stdout is hidden, the py stdout should be changed
# to something that shows e.g. in console, so prints from scripts show
# (people commonly use those for debugging so they should show somewhere)
d = Debugger(IgnoreChannels = ignchannels, logger=NaaliLogger()) #IgnoreEvents = ignored)
self.m = Manager() + d
#self.m = Manager()
#or __all__ in pymodules __init__ ? (i.e. import pymodules would do that)
if self.firstrun:
import autoload
self.firstrun = False
else: #reload may cause something strange sometimes, so better not do it always in production use, is just a dev tool.
#print "reloading autoload"
import autoload
autoload = reload(autoload)
#print "Autoload module:", autoload
autoload.load(self.m)
self.m.push(Started(self.m, None)) #webserver requires this now, temporarily XXX
def test2():
import gobject
from circuits import Manager, Debugger
from circuits.net.sockets import TCPClient, Connect, Write
m = Manager()
m += GtkDriver()
m += Debugger()
def on_delete(window, event):
m.stop()
w = gtk.Window()
w.connect('delete-event', on_delete)
vb = gtk.VBox()
b = gtk.Button('Click me')
tv = gtk.TextView()
sw = gtk.ScrolledWindow()
sw.add(tv)
vb.pack_start(sw)