cengal.parallel_execution.coroutines.coro_tools.log_viewer.versions.v_0.log_viewer

Module Docstring Docstrings: http://www.python.org/dev/peps/pep-0257/

View Source

 1#!/usr/bin/env python
 2# coding=utf-8
 3
 4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>
 5# 
 6# Licensed under the Apache License, Version 2.0 (the "License");
 7# you may not use this file except in compliance with the License.
 8# You may obtain a copy of the License at
 9# 
10#     http://www.apache.org/licenses/LICENSE-2.0
11# 
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS,
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
17
18
19"""
20Module Docstring
21Docstrings: http://www.python.org/dev/peps/pep-0257/
22"""
23
24
25__author__ = "ButenkoMS <gtalk@butenkoms.space>"
26__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>"
27__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ]
28__license__ = "Apache License, Version 2.0"
29__version__ = "4.4.1"
30__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>"
31__email__ = "gtalk@butenkoms.space"
32# __status__ = "Prototype"
33__status__ = "Development"
34# __status__ = "Production"
35
36
37from cengal.parallel_execution.coroutines.coro_scheduler import *
38from cengal.parallel_execution.coroutines.coro_tools.run_in_loop import run_in_loop
39from cengal.parallel_execution.coroutines.coro_tools.loop_administration.admin_tk import start_admin
40from cengal.parallel_execution.coroutines.coro_standard_services.shutdown_on_keyboard_interrupt import ShutdownOnKeyboardInterrupt
41from cengal.parallel_execution.coroutines.coro_standard_services.shutdown_loop import ShutdownLoop
42from cengal.parallel_execution.coroutines.coro_standard_services.instance import InstanceRequest
43from cengal.parallel_execution.coroutines.coro_standard_services.log import LogRequest, Log
44from cengal.parallel_execution.coroutines.coro_standard_services.run_coro import RunCoro
45from cengal.file_system.app_fs_structure.app_dir_path import AppDirPath, AppDirectoryType
46from cengal.code_flow_control.args_manager import AK
47from cengal.system import current_cengal_module_import_str
48from cengal.introspection.inspect import pdi
49import os
50import tkinter as tk
51from tkinter import filedialog
52
53
54async def on_close(i: Interface):
55    await i(ShutdownLoop)
56
57
58def request_and_set_log_env_dir(i: Interface, admin_tk_app: tk.Tk):
59    app_name_for_fs = i(InstanceRequest().wait('app_name_for_fs'))
60    app_data_dir_path_type: AppDirectoryType = i(InstanceRequest().wait('app_data_dir_path_type'))
61    app_dir_path: AppDirPath = i(InstanceRequest().wait(AppDirPath))
62    app_data_dir_path: str = app_dir_path.cached(app_data_dir_path_type, app_name_for_fs, False)
63    initialdir = os.path.dirname(app_data_dir_path)
64    log_db_env_path = filedialog.askdirectory(parent=admin_tk_app,
65                                                initialdir=initialdir,
66                                                title="Please select a folder:")
67    if not isinstance(log_db_env_path, str):
68        i(LogRequest().sync())
69        i.log.warning('Path to Log DB Env was not provided! Using own Log DB Env instead.')
70        return
71    
72    log_db_env_path = log_db_env_path.strip()
73    log_db_env_path = os.path.normpath(log_db_env_path)
74    if not (os.path.exists(log_db_env_path) and os.path.isdir(log_db_env_path)):
75        i(LogRequest().sync())
76        i.log.warning('Path does not exist! Using own Log DB Env instead.')
77        return
78
79    i(LogRequest().set_db_environment_path(log_db_env_path))
80
81
82async def main(i: Interface):
83    await i(ShutdownOnKeyboardInterrupt)
84    await i(InstanceRequest().set('app_name_for_fs', current_cengal_module_import_str()))
85    await start_admin(i, on_close, AK(current_children_pack_type = 2))
86    admin_tk_app: tk.Tk = await i(InstanceRequest().wait('admin_tk_app'))
87    await i(RunCoro, request_and_set_log_env_dir, admin_tk_app)
88
89
90if '__main__' == __name__:
91    run_in_loop(main)

async def on_close( i: cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface): View Source

55async def on_close(i: Interface):
56    await i(ShutdownLoop)

def request_and_set_log_env_dir( i: cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, admin_tk_app: tkinter.Tk): View Source

59def request_and_set_log_env_dir(i: Interface, admin_tk_app: tk.Tk):
60    app_name_for_fs = i(InstanceRequest().wait('app_name_for_fs'))
61    app_data_dir_path_type: AppDirectoryType = i(InstanceRequest().wait('app_data_dir_path_type'))
62    app_dir_path: AppDirPath = i(InstanceRequest().wait(AppDirPath))
63    app_data_dir_path: str = app_dir_path.cached(app_data_dir_path_type, app_name_for_fs, False)
64    initialdir = os.path.dirname(app_data_dir_path)
65    log_db_env_path = filedialog.askdirectory(parent=admin_tk_app,
66                                                initialdir=initialdir,
67                                                title="Please select a folder:")
68    if not isinstance(log_db_env_path, str):
69        i(LogRequest().sync())
70        i.log.warning('Path to Log DB Env was not provided! Using own Log DB Env instead.')
71        return
72    
73    log_db_env_path = log_db_env_path.strip()
74    log_db_env_path = os.path.normpath(log_db_env_path)
75    if not (os.path.exists(log_db_env_path) and os.path.isdir(log_db_env_path)):
76        i(LogRequest().sync())
77        i.log.warning('Path does not exist! Using own Log DB Env instead.')
78        return
79
80    i(LogRequest().set_db_environment_path(log_db_env_path))

async def main( i: cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface): View Source

83async def main(i: Interface):
84    await i(ShutdownOnKeyboardInterrupt)
85    await i(InstanceRequest().set('app_name_for_fs', current_cengal_module_import_str()))
86    await start_admin(i, on_close, AK(current_children_pack_type = 2))
87    admin_tk_app: tk.Tk = await i(InstanceRequest().wait('admin_tk_app'))
88    await i(RunCoro, request_and_set_log_env_dir, admin_tk_app)