housekeeping: ya2 module

[pmachines.git] / pmachines / app.py

import argparse
import simplepbr
#import gltf
from json import loads
from sys import platform, exit, argv
from platform import node
from logging import info, debug
from os.path import exists
from os import makedirs
from multiprocessing import cpu_count
from panda3d.core import Filename, load_prc_file_data, AntialiasAttrib, \
    WindowProperties, LVector2i, TextNode, GraphicsBuffer
from panda3d.bullet import BulletWorld, BulletDebugNode
from direct.showbase.ShowBase import ShowBase
from direct.gui.OnscreenText import OnscreenText
from direct.fsm.FSM import FSM
from pmachines.audio.music import MusicMgr
from pmachines.items.background import Background
from pmachines.gui.menu import Menu
from pmachines.scene import Scene
from pmachines.posmgr import PositionMgr
from pmachines.persistent import Persistent
from ya2.utils.dictfile import DctFile
from ya2.utils.logics import LogicsTools
from ya2.utils.lang import LangMgr
from ya2.utils.log import LogMgr
from ya2.utils.functional import FunctionalTest
from ya2.utils.asserts import Assert


class MainFsm(FSM):

    def __init__(self, pmachines):
        super().__init__('Main FSM')
        self._pmachines = pmachines
        self.accept('new_scene', self.__on_new_scene)

    def enterMenu(self):
        self._pmachines.on_menu_enter()

    def exitMenu(self):
        self._pmachines.on_menu_exit()
        self.__do_asserts()

    def enterScene(self, cls):
        self._pmachines.on_scene_enter(cls)

    def exitScene(self):
        self._pmachines.on_scene_exit()
        self.__do_asserts()

    def __on_new_scene(self):
        self.demand('Scene', None)

    def __do_asserts(self):
        args = self._pmachines._args
        if not LogicsTools.runtime() or args.functional_test or args.functional_ref:
            Assert.assert_threads()
            Assert.assert_tasks()
            Assert.assert_render3d()
            Assert.assert_render2d()
            Assert.assert_aspect2d()
            Assert.assert_events()
            Assert.assert_buffers()

    def enterOff(self):
        self.ignore('new_scene')


class Pmachines:

    @staticmethod
    def scenes():
        with open('assets/scenes/index.json') as f:
            json = loads(f.read())
        return json['list']

    def __init__(self):
        info('platform: %s' % platform)
        info('exists main.py: %s' % exists('main.py'))
        self._args = args = self._parse_args()
        self._configure(args)
        self.base = ShowBase()
        self._pipeline = None
        self.is_update_run = args.update
        self.is_version_run = args.version
        self.log_mgr = LogMgr.init_cls()()
        self._pos_mgr = PositionMgr()
        self._prepare_window(args)
        self._fsm = MainFsm(self)
        self._fsm.demand('Start')  # otherwise it is Off and cleanup in tests won't work
        if args.update:
            return
        if args.functional_test:
            self._options['settings']['volume'] = 0
        self._music = MusicMgr(self._options['settings']['volume'])
        self.lang_mgr = LangMgr(self._options['settings']['language'],
                                'pmachines',
                                'assets/locale/')
        if args.functional_test or args.functional_ref:
            FunctionalTest(args.functional_ref, self._pos_mgr)
        if not LogicsTools.runtime() or args.functional_test or args.functional_ref:
            self.__fps_lst = []
            taskMgr.do_method_later(1.0, self.__assert_fps, 'assert_fps')

    def start(self):
        if self._args.screenshot:
            #cls = [cls for cls in self.scenes if cls.__name__ == self._args.screenshot][0]
            scene = Scene(BulletWorld(), None, True, False, lambda: None, self.scenes(), self._pos_mgr, None, None, None, self._args.screenshot, None, None)
            scene.screenshot()
            scene.destroy()
            exit()
        elif self._options['development']['auto_start']:
            # mod_name = 'pmachines.scenes.scene_' + self._options['development']['auto_start']
            # for member in import_module(mod_name).__dict__.values():
            #     if isclass(member) and issubclass(member, Scene) and \
            #             member != Scene:
            #         cls = member
            self._fsm.demand('Scene', self._options['development']['auto_start'])
        else:
            Scene.scenes_done = self.__persistent.scenes_done
            self._fsm.demand('Menu')

    def on_menu_enter(self):
        self._menu_bg = Background()
        self._menu = Menu(
            self._fsm, self.lang_mgr, self._options, self._music,
            self._pipeline, self.scenes(), self._args.functional_test or self._args.functional_ref,
            self._pos_mgr)

    def on_home(self):
        Scene.scenes_done = self.__persistent.scenes_done
        self._fsm.demand('Menu')

    def on_menu_exit(self):
        self._menu_bg.destroy()
        self._menu.destroy()

    def on_scene_enter(self, scene_name):
        self._set_physics()
        self._scene = Scene(
            self.world, self.on_home,
            self._options['development']['auto_close_instructions'],
            self._options['development']['debug_items'],
            self.reload,
            self.scenes(),
            self._pos_mgr,
            self._args.functional_test or self._args.functional_ref,
            self._options['development']['mouse_coords'],
            self.__persistent,
            scene_name,
            self._options['development']['editor'],
            self._options['development']['auto_start_editor'])

    def on_scene_exit(self):
        self._unset_physics()
        self._scene.destroy()

    def reload(self, cls):
        self._fsm.demand('Scene', cls)

    def _configure(self, args):
        load_prc_file_data('', 'window-title pmachines')
        load_prc_file_data('', 'framebuffer-srgb true')
        load_prc_file_data('', 'sync-video true')
        if args.functional_test or args.functional_ref:
            load_prc_file_data('', 'win-size 1360 768')
            # otherwise it is not centered in exwm
        # load_prc_file_data('', 'threading-model Cull/Draw')
        # it freezes when you go to the next scene
        if args.screenshot:
            load_prc_file_data('', 'window-type offscreen')
            load_prc_file_data('', 'audio-library-name null')

    def _parse_args(self):
        parser = argparse.ArgumentParser()
        parser.add_argument('--update', action='store_true')
        parser.add_argument('--version', action='store_true')
        parser.add_argument('--optfile')
        parser.add_argument('--screenshot')
        parser.add_argument('--functional-test', action='store_true')
        parser.add_argument('--functional-ref', action='store_true')
        cmd_line = [arg for arg in iter(argv[1:]) if not arg.startswith('-psn_')]
        args = parser.parse_args(cmd_line)
        return args

    def _prepare_window(self, args):
        data_path = ''
        if (platform.startswith('win') or platform.startswith('linux')) and (
                not exists('main.py') or __file__.startswith('/app/bin/')):
            # it is the deployed version for windows
            data_path = str(Filename.get_user_appdata_directory()) + '/pmachines'
            home = '/home/flavio'  # we must force this for wine
            if data_path.startswith('/c/users/') and exists(home + '/.wine/'):
                data_path = home + '/.wine/drive_' + data_path[1:]
            info('creating dirs: %s' % data_path)
            makedirs(data_path, exist_ok=True)
        optfile = args.optfile if args.optfile else 'options.ini'
        info('data path: %s' % data_path)
        info('option file: %s' % optfile)
        info('fixed path: %s' % LogicsTools.fixpath(data_path + '/' + optfile))
        default_opt = {
            'settings': {
                'volume': 1,
                'language': 'en',
                'fullscreen': 1,
                'resolution': '',
                'antialiasing': 1,
                'shadows': 1},
            'save': {
                'scenes_done': []
            },
            'development': {
                'simplepbr': 1,
                'verbose_log': 0,
                'physics_debug': 0,
                'auto_start': 0,
                'auto_close_instructions': 0,
                'show_buffers': 0,
                'debug_items': 0,
                'mouse_coords': 0,
                'fps': 0,
                'editor': 0,
                'auto_start_editor': 0}}
        opt_path = LogicsTools.fixpath(data_path + '/' + optfile) if data_path else optfile
        opt_exists = exists(opt_path)
        self._options = DctFile(
            LogicsTools.fixpath(data_path + '/' + optfile) if data_path else optfile,
            default_opt)
        if not opt_exists:
            self._options.store()
        self.__persistent = Persistent(self._options['save']['scenes_done'], self._options)
        Scene.scenes_done = self.__persistent.scenes_done
        res = self._options['settings']['resolution']
        if res:
            res = LVector2i(*[int(_res) for _res in res.split('x')])
        else:
            resolutions = []
            if not self.is_version_run:
                d_i = base.pipe.get_display_information()
                def _res(idx):
                    return d_i.get_display_mode_width(idx), \
                        d_i.get_display_mode_height(idx)
                resolutions = [
                    _res(idx) for idx in range(d_i.get_total_display_modes())]
                res = sorted(resolutions)[-1]
        fullscreen = self._options['settings']['fullscreen']
        props = WindowProperties()
        if args.functional_test or args.functional_ref:
            fullscreen = False
        elif not self.is_version_run:
            props.set_size(res)
        props.set_fullscreen(fullscreen)
        props.set_icon_filename('assets/images/icon/pmachines.ico')
        if not args.screenshot and not self.is_version_run and base.win and not isinstance(base.win, GraphicsBuffer):
            base.win.request_properties(props)
        #gltf.patch_loader(base.loader)
        if self._options['development']['simplepbr'] and not self.is_version_run and base.win:
            self._pipeline = simplepbr.init(
                use_normal_maps=True,
                use_emission_maps=False,
                use_occlusion_maps=True,
                msaa_samples=4 if self._options['settings']['antialiasing'] else 1,
                enable_shadows=int(self._options['settings']['shadows']))
            debug(f'msaa: {self._pipeline.msaa_samples}')
            debug(f'shadows: {self._pipeline.enable_shadows}')
        render.setAntialias(AntialiasAttrib.MAuto)
        self.base.set_background_color(0, 0, 0, 1)
        self.base.disable_mouse()
        if self._options['development']['show_buffers']:
            base.bufferViewer.toggleEnable()
        if self._options['development']['fps']:
            base.set_frame_rate_meter(True)
        #self.base.accept('window-event', self._on_win_evt)
        self.base.accept('aspectRatioChanged', self._on_aspect_ratio_changed)
        if self._options['development']['mouse_coords']:
            coords_txt = OnscreenText(
                '', parent=base.a2dTopRight, scale=0.04,
                pos=(-.03, -.06), fg=(.9, .9, .9, 1), align=TextNode.A_right)
            def update_coords(task):
                txt = '%s %s' % (int(base.win.get_pointer(0).x),
                                 int(base.win.get_pointer(0).y))
                coords_txt['text'] = txt
                return task.cont
            taskMgr.add(update_coords, 'update_coords')

    def _set_physics(self):
        if self._options['development']['physics_debug']:
            debug_node = BulletDebugNode('Debug')
            debug_node.show_wireframe(True)
            debug_node.show_constraints(True)
            debug_node.show_bounding_boxes(True)
            debug_node.show_normals(True)
            self._debug_np = render.attach_new_node(debug_node)
            self._debug_np.show()
        self.world = BulletWorld()
        self.world.set_gravity((0, 0, -9.81))
        if self._options['development']['physics_debug']:
            self.world.set_debug_node(self._debug_np.node())
        def update(task):
            dt = globalClock.get_dt()
            self.world.do_physics(dt, 10, 1/180)
            return task.cont
        self._phys_tsk = taskMgr.add(update, 'update')

    def _unset_physics(self):
        if self._options['development']['physics_debug']:
            self._debug_np.remove_node()
        self.world = None
        taskMgr.remove(self._phys_tsk)

    def _on_aspect_ratio_changed(self):
        if self._fsm.state == 'Scene':
            self._scene.on_aspect_ratio_changed()

    def __assert_fps(self, task):
        if len(self.__fps_lst) > 3:
            self.__fps_lst.pop(0)
        self.__fps_lst += [globalClock.average_frame_rate]
        if len(self.__fps_lst) == 4:
            fps_threshold = 55 if cpu_count() >= 4 and node() != 'localhost.localdomain' else 10  # i.e. it is the builder machine
            assert not all(fps < fps_threshold for fps in self.__fps_lst), 'low fps %s' % self.__fps_lst
        return task.again

    def destroy(self):
        self._fsm.cleanup()
        self.base.destroy()

    def run(self):
        self.start()
        self.base.run()