ya2/p3d/asserts.py

   1 import threading
   2
   3
   4 class Assert:
   5
   6     @staticmethod
   7     def assert_render3d():
   8         preserve = ['camera', 'DIRECT']
   9         unexpected_nodes = [c for c in render.children if c.name not in preserve]
  10         for node in unexpected_nodes: Assert.__process_render3d_exception(node)
  11
  12     @staticmethod
  13     def __process_render3d_exception(node):
  14         render.ls()
  15         message = f'unexpected render3d node: {node.name}'
  16         raise Exception(message)
  17
  18     @staticmethod
  19     def assert_aspect2d():
  20         preserve = [
  21             'a2dBackground', 'a2dTopCenter', 'a2dTopCenterNS',
  22             'a2dBottomCenter', 'a2dBottomCenterNS', 'a2dLeftCenter',
  23             'a2dLeftCenterNS', 'a2dRightCenter', 'a2dRightCenterNS',
  24             'a2dTopLeft', 'a2dTopLeftNS', 'a2dTopRight',
  25             'a2dTopRightNS', 'a2dBottomLeft', 'a2dBottomLeftNS',
  26             'a2dBottomRight', 'a2dBottomRightNS', 'test_txt']
  27         unexpected_nodes = [c for c in aspect2d.children
  28                             if c.name not in preserve and not c.has_python_tag('preserve')]
  29         for node in unexpected_nodes: Assert.__process_aspect2d_exception(node)
  30
  31     @staticmethod
  32     def __process_aspect2d_exception(node):
  33         aspect2d.ls()
  34         message = f'unexpected aspect2d node: {node.name}'
  35         raise Exception(message)
  36
  37     @staticmethod
  38     def assert_render2d():
  39         preserve = ['aspect2d', 'pixel2d', 'camera2d']
  40         unexpected_nodes = [c for c in render2d.children if c.name not in preserve and not c.has_python_tag('preserve')]
  41         for node in unexpected_nodes: Assert.__process_render2d_exception(node)
  42
  43     @staticmethod
  44     def __process_render2d_exception(self, node):
  45         render2d.ls()
  46         message = f'unexpected render2d node: {node.name}'
  47         raise Exception(message)
  48
  49     @staticmethod
  50     def assert_events():
  51         preserve = ['window-event', 'window-closed', 'async_loader_0', 'async_loader_1',
  52                     'render-texture-targets-changed', 'aspectRatioChanged']
  53         unexpected_events = [e for e in messenger.getEvents() if e not in preserve]
  54         for e in unexpected_events: Assert.__process_event_exception(e)
  55
  56     @staticmethod
  57     def __process_event_exception(event):
  58         if (acc := messenger.who_accepts(event)):
  59             Assert.__process_event_acceptors(acc, event)
  60
  61     @staticmethod
  62     def __process_event_acceptors(acceptors, event):
  63         for key, a in acceptors.items(): Assert.__process_event_acceptor(key, a, event)
  64
  65     @staticmethod
  66     def __process_event_acceptor(key, acceptor, event):
  67         #if acceptor[2]:
  68         #    print(f'the event {event} accepted by <{key}, {acceptor[0]}> is persistent')
  69         #else:
  70             Assert.__actually_process_event_exception(acceptor, event)
  71
  72     @staticmethod
  73     def __actually_process_event_exception(acceptor, event):
  74         message = f'unexpected event: {event}, {str(acceptor)}'
  75         raise Exception(message)
  76
  77     @staticmethod
  78     def assert_tasks():
  79         preserve = [
  80             'ivalLoop', 'garbageCollectStates', 'collisionLoop',
  81             'igLoop', 'audioLoop', 'resetPrevTransform', 'dataLoop',
  82             'eventManager', 'simplepbr update', 'on frame music',
  83             'assert_fps', 'DIRECTContextTask']
  84         unexpected_tasks = [t for t in taskMgr.getTasks() + taskMgr.getDoLaters()
  85                             if t.name not in preserve and not hasattr(t, 'preserve')]
  86         for t in unexpected_tasks: Assert.__process_task_exception(t)
  87
  88     @staticmethod
  89     def __process_task_exception(task):
  90         message = f'unexpected task: {task.name}'
  91         raise Exception(message)
  92
  93     @staticmethod
  94     def assert_buffers():
  95         pass
  96         #if RenderToTexture.buffers:
  97         #    raise Error()
  98
  99     @staticmethod
 100     def assert_threads():
 101         thread_names = [thread.name for thread in threading.enumerate()]
 102         preserve = ['MainThread', 'rpc_server']
 103         unexpected_tasks = [t for t in thread_names if t not in preserve]
 104         for t in unexpected_tasks: Assert.__process_unexpected_thread(t)
 105
 106     @staticmethod
 107     def __process_unexpected_thread(thread_name):
 108         message = f'unexpected thread: {thread_name}'
 109         raise Exception(message)