aboutsummaryrefslogtreecommitdiff
path: root/src_py
diff options
context:
space:
mode:
Diffstat (limited to 'src_py')
-rw-r--r--src_py/opcut/common.py2
-rw-r--r--src_py/opcut/doit/_common.py28
-rw-r--r--src_py/opcut/doit/jsopcut.py55
-rw-r--r--src_py/opcut/doit/main.py100
-rw-r--r--src_py/opcut/doit/pyopcut.py106
-rw-r--r--src_py/opcut/main.py168
-rw-r--r--src_py/opcut/server.py83
-rw-r--r--src_py/opcut/util.py107
8 files changed, 130 insertions, 519 deletions
diff --git a/src_py/opcut/common.py b/src_py/opcut/common.py
index 5749a92..d2a56b1 100644
--- a/src_py/opcut/common.py
+++ b/src_py/opcut/common.py
@@ -1,6 +1,6 @@
import enum
-from opcut import util
+from hat import util
mm = 72 / 25.4
diff --git a/src_py/opcut/doit/_common.py b/src_py/opcut/doit/_common.py
deleted file mode 100644
index 8e33791..0000000
--- a/src_py/opcut/doit/_common.py
+++ /dev/null
@@ -1,28 +0,0 @@
-import os
-import shutil
-from pathlib import Path
-
-
-def mkdir_p(*paths):
- for path in paths:
- os.makedirs(str(Path(path)), exist_ok=True)
-
-
-def rm_rf(*paths):
- for path in paths:
- p = Path(path)
- if not p.exists():
- continue
- if p.is_dir():
- shutil.rmtree(str(p), ignore_errors=True)
- else:
- p.unlink()
-
-
-def cp_r(src, dest):
- src = Path(src)
- dest = Path(dest)
- if src.is_dir():
- shutil.copytree(str(src), str(dest))
- else:
- shutil.copy2(str(src), str(dest))
diff --git a/src_py/opcut/doit/jsopcut.py b/src_py/opcut/doit/jsopcut.py
deleted file mode 100644
index 0525aa6..0000000
--- a/src_py/opcut/doit/jsopcut.py
+++ /dev/null
@@ -1,55 +0,0 @@
-import subprocess
-
-from opcut.doit import _common
-
-
-__all__ = ['task_jsopcut_clean', 'task_jsopcut_install_deps',
- 'task_jsopcut_remove_deps', 'task_jsopcut_build',
- 'task_jsopcut_watch', 'task_jsopcut_check']
-
-
-def task_jsopcut_clean():
- """JsOpcut - clean"""
-
- return {'actions': [(_common.rm_rf, ['build/jsopcut',
- 'src_js/opcut/validator.js'])]}
-
-
-def task_jsopcut_install_deps():
- """JsOpcut - install dependencies"""
-
- def patch():
- subprocess.Popen(['patch', '-r', '/dev/null', '--forward', '-p0',
- '-i', 'node_modules.patch'],
- stdout=subprocess.DEVNULL,
- stderr=subprocess.DEVNULL).wait()
-
- return {'actions': ['yarn install --silent',
- patch]}
-
-
-def task_jsopcut_remove_deps():
- """JsOpcut - remove dependencies"""
-
- return {'actions': [(_common.rm_rf, ['node_modules'])]}
-
-
-def task_jsopcut_build():
- """JsOpcut - build"""
-
- return {'actions': ['yarn run build'],
- 'task_dep': ['jsopcut_install_deps']}
-
-
-def task_jsopcut_watch():
- """JsOpcut - build on change"""
-
- return {'actions': ['yarn run watch'],
- 'task_dep': ['jsopcut_install_deps']}
-
-
-def task_jsopcut_check():
- """JsOpcut - check"""
-
- return {'actions': ['yarn run check'],
- 'task_dep': ['jsopcut_install_deps']}
diff --git a/src_py/opcut/doit/main.py b/src_py/opcut/doit/main.py
deleted file mode 100644
index b26cadd..0000000
--- a/src_py/opcut/doit/main.py
+++ /dev/null
@@ -1,100 +0,0 @@
-from doit.action import CmdAction
-
-from opcut.doit import _common
-
-import opcut.doit.pyopcut
-import opcut.doit.jsopcut
-from opcut.doit.pyopcut import * # NOQA
-from opcut.doit.jsopcut import * # NOQA
-
-
-__all__ = (['task_clean_all', 'task_gen_all', 'task_dist_build',
- 'task_dist_clean'] +
- opcut.doit.pyopcut.__all__ +
- opcut.doit.jsopcut.__all__)
-
-
-def task_clean_all():
- """Clean all"""
-
- return {'actions': [(_common.rm_rf, ['build', 'dist'])],
- 'task_dep': ['pyopcut_clean',
- 'jsopcut_clean',
- 'dist_clean']}
-
-
-def task_gen_all():
- """Generate all"""
-
- return {'actions': None,
- 'task_dep': ['pyopcut_gen']}
-
-
-def task_check_all():
- """Check all"""
-
- return {'actions': None,
- 'task_dep': ['pyopcut_check']}
-
-
-def task_dist_clean():
- """Distribution - clean"""
-
- return {'actions': [(_common.rm_rf, ['dist'])]}
-
-
-def task_dist_build():
- """Distribution - build (DEFAULT)"""
-
- def generate_setup_py():
- with open('VERSION', encoding='utf-8') as f:
- version = f.read().strip()
- with open('README.rst', encoding='utf-8') as f:
- readme = f.read().strip()
- with open('requirements.txt', encoding='utf-8') as f:
- dependencies = [i.strip() for i in f.readlines() if i.strip()]
- with open('build/dist/setup.py', 'w', encoding='utf-8') as f:
- f.write(_setup_py.format(
- version=repr(version),
- long_description=repr(readme),
- dependencies=repr(dependencies)))
-
- return {'actions': [
- (_common.rm_rf, ['dist', 'build/dist']),
- (_common.cp_r, ['build/pyopcut', 'build/dist']),
- (_common.cp_r, ['build/jsopcut', 'build/dist/opcut/web']),
- (_common.cp_r, ['README.rst', 'build/dist/README.rst']),
- generate_setup_py,
- CmdAction('python setup.py -q bdist_wheel -d ../../dist',
- cwd='build/dist')],
- 'task_dep': [
- 'gen_all',
- 'pyopcut_build',
- 'jsopcut_build']}
-
-
-_setup_py = r"""
-from setuptools import setup
-setup(
- name='opcut',
- version={version},
- description='Cutting stock problem optimizer',
- long_description={long_description},
- url='https://github.com/bozokopic/opcut',
- author='Bozo Kopic',
- author_email='bozo.kopic@gmail.com',
- license='GPLv3',
- python_requires='>=3.5',
- zip_safe=False,
- packages=['opcut'],
- package_data={{
- 'opcut': ['web/*', 'web/fonts/*']
- }},
- install_requires={dependencies},
- entry_points={{
- 'console_scripts': [
- 'opcut = opcut.main:main'
- ]
- }}
-)
-"""
diff --git a/src_py/opcut/doit/pyopcut.py b/src_py/opcut/doit/pyopcut.py
deleted file mode 100644
index 0aecb30..0000000
--- a/src_py/opcut/doit/pyopcut.py
+++ /dev/null
@@ -1,106 +0,0 @@
-import os
-import yaml
-from pathlib import Path
-from doit.action import CmdAction
-
-from opcut.doit import _common
-
-
-__all__ = ['task_pyopcut_clean', 'task_pyopcut_build', 'task_pyopcut_check',
- 'task_pyopcut_gen', 'task_pyopcut_gen_json_validator']
-
-
-def task_pyopcut_clean():
- """PyOpcut - clean"""
-
- return {'actions': [(_common.rm_rf, ['build/pyopcut',
- 'src_py/opcut/json_validator.py'])]}
-
-
-def task_pyopcut_build():
- """PyOpcut - build"""
-
- generated_files = {Path('src_py/opcut/json_validator.py')}
-
- def compile(src_path, dst_path):
- _common.mkdir_p(dst_path.parent)
- _common.cp_r(src_path, dst_path)
-
- def create_subtask(src_path):
- dst_path = Path('build/pyopcut') / src_path.relative_to('src_py')
- return {'name': str(src_path),
- 'actions': [(compile, [src_path, dst_path])],
- 'file_dep': [src_path],
- 'targets': [dst_path]}
-
- for src_path in generated_files:
- yield create_subtask(src_path)
-
- for dirpath, dirnames, filenames in os.walk('src_py'):
- for i in ['__pycache__', 'doit']:
- if i in dirnames:
- dirnames.remove(i)
- for i in filenames:
- src_path = Path(dirpath) / i
- if src_path not in generated_files:
- yield create_subtask(src_path)
-
-
-def task_pyopcut_check():
- """PyOpcut - run flake8"""
-
- return {'actions': [CmdAction('python -m flake8 .', cwd='src_py')]}
-
-
-def task_pyopcut_gen():
- """PyOpcut - generate additional python modules"""
-
- return {'actions': None,
- 'task_dep': ['pyopcut_gen_json_validator']}
-
-
-def task_pyopcut_gen_json_validator():
- """PyOpcut - generate json validator"""
-
- schema_files = list(Path('schemas_json').glob('**/*.yaml'))
- output_file = Path('src_py/opcut/json_validator.py')
-
- def parse_schemas():
- schemas = {}
- for schema_file in schema_files:
- with open(schema_file, encoding='utf-8') as f:
- data = yaml.safe_load(f)
- if data['id'] in schemas:
- raise Exception("duplicate schema id " + data['id'])
- schemas[data['id']] = data
- return schemas
-
- def generate_output():
- schemas = parse_schemas()
- with open(output_file, 'w', encoding='utf-8') as f:
- f.write(
- '# pylint: skip-file\n'
- 'import jsonschema\n\n\n'
- '_schemas = {schemas} # NOQA\n\n\n'
- 'def validate(data, schema_id):\n'
- ' """ Validate data with JSON schema\n\n'
- ' Args:\n'
- ' data: validated data\n'
- ' schema_id (str): JSON schema identificator\n\n'
- ' Raises:\n'
- ' Exception: validation fails\n\n'
- ' """\n'
- ' base_uri = schema_id.split("#")[0] + "#"\n'
- ' resolver = jsonschema.RefResolver(\n'
- ' base_uri=base_uri,\n'
- ' referrer=_schemas[base_uri],\n'
- ' store=_schemas,\n'
- ' cache_remote=False)\n'
- ' jsonschema.validate(\n'
- ' instance=data,\n'
- ' schema=resolver.resolve(schema_id)[1],\n'
- ' resolver=resolver)\n'.format(schemas=schemas))
-
- return {'actions': [generate_output],
- 'file_dep': schema_files,
- 'targets': [output_file]}
diff --git a/src_py/opcut/main.py b/src_py/opcut/main.py
index b5abdfa..02ce856 100644
--- a/src_py/opcut/main.py
+++ b/src_py/opcut/main.py
@@ -1,92 +1,82 @@
-import sys
+from pathlib import Path
import argparse
-import yaml
-import logging.config
-import urllib.parse
import asyncio
-import os.path
+import contextlib
+import logging.config
+import sys
+
+from hat import util
+from hat.util import aio
+from hat.util import json
-from opcut import util
from opcut import common
-import opcut.json_validator
import opcut.csp
-import opcut.server
import opcut.output
+import opcut.server
+
+
+package_path = Path(__file__).parent
+
+default_ui_path = package_path / 'ui'
+default_schemas_json_path = package_path / 'schemas_json'
def main():
"""Application main entry point"""
-
args = _create_parser().parse_args()
+ json_schema_repo = json.SchemaRepository(json.default_schemas_json_path,
+ args.schemas_json_path)
+
if args.log_conf_path:
- with open(args.log_conf_path, encoding='utf-8') as log_conf_file:
- log_conf = yaml.safe_load(log_conf_file)
- opcut.json_validator.validate(log_conf, 'opcut://logging.yaml#')
+ log_conf = json.decode_file(args.log_conf_path)
+ json_schema_repo.validate('hat://logging.yaml#', log_conf)
logging.config.dictConfig(log_conf)
- action_fn = {
- 'server': server,
- 'calculate': calculate,
- 'output': output}.get(args.action, server)
- action_fn(args)
-
+ if args.action == 'calculate':
+ calculate(json_schema_repo, args.params_path, args.method,
+ args.result_path, args.output_path, args.output_type,
+ args.output_panel_id, args.output_path)
-def server(args):
- """Server main entry point
+ elif args.action == 'output':
+ output(json_schema_repo, args.output_path, args.result_path,
+ args.output_type, args.output_panel_id, args.output_path)
- Args:
- args: command line argument
+ else:
+ server(json_schema_repo, args.addr, args.pem_path, args.ui_path)
- """
- if sys.platform == 'win32':
- asyncio.set_event_loop(asyncio.ProactorEventLoop())
- addr = urllib.parse.urlparse(args.ui_addr)
- pem_path = args.ui_pem_path
- ui_path = args.ui_path or os.path.join(os.path.dirname(__file__), 'web')
+def server(json_schema_repo, addr, ui_pem_path, ui_path):
+ """Server main entry point"""
+ aio.init_asyncio()
+ with contextlib.suppress(asyncio.CancelledError):
+ aio.run_asyncio(
+ opcut.server.run(json_schema_repo, addr, ui_pem_path, ui_path))
- util.run_until_complete_without_interrupt(
- opcut.server.run(addr, pem_path, ui_path))
-
-def calculate(args):
- """Calculate result and generate outputs
-
- Args:
- args: command line argument
-
- """
- with open(args.params_path, 'r', encoding='utf-8') as f:
- params_json_data = yaml.safe_load(f)
- opcut.json_validator.validate(params_json_data, 'opcut://params.yaml#')
+def calculate(json_schema_repo, params_path, method, result_path, output_path,
+ output_type, output_panel_id):
+ """Calculate result and generate outputs"""
+ params_json_data = json.decode_file(params_path)
+ opcut.json_validator.validate('opcut://params.yaml#', params_json_data)
params = common.json_data_to_params(params_json_data)
- result = opcut.csp.calculate(params, args.method)
+ result = opcut.csp.calculate(params, method)
result_json_data = common.result_to_json_data(result)
- with open(args.result_path, 'w', encoding='utf-8') as f:
- yaml.safe_dump(result_json_data, f,
- indent=4, default_flow_style=False,
- explicit_start=True, explicit_end=True)
- output(args)
+ json.encode_file(result_json_data, result_path)
+ if output_path:
+ output(json_schema_repo, output_path, result_path, output_type,
+ output_panel_id)
-def output(args):
- """Generate outputs based on calculation result
-
- Args:
- args: command line argument
-
- """
- if not args.output_path:
- return
- with open(args.result_path, 'r', encoding='utf-8') as f:
- result_json_data = yaml.safe_load(f)
- opcut.json_validator.validate(result_json_data, 'opcut://result.yaml#')
+def output(json_schema_repo, output_path, result_path, output_type,
+ output_panel_id):
+ """Generate outputs based on calculation result"""
+ result_json_data = json.decode_file(result_path)
+ json_schema_repo.validate('opcut://result.yaml#', result_json_data)
result = common.json_data_to_result(result_json_data)
- output_type = common.OutputType[args.output_type]
output_bytes = opcut.output.generate_output(result, output_type,
- args.output_panel_id)
- with open(args.output_path, 'wb') as f:
+ output_panel_id)
+ with open(output_path, 'wb') as f:
f.write(output_bytes)
@@ -94,55 +84,85 @@ def _create_parser():
parser = argparse.ArgumentParser(prog='opcut')
parser.add_argument(
'--log', default=None, metavar='path', dest='log_conf_path',
+ action=util.EnvPathArgParseAction,
help="logging configuration")
subparsers = parser.add_subparsers(title='actions', dest='action')
server = subparsers.add_parser('server', help='run web server')
server.add_argument(
- '--ui-addr', default='http://0.0.0.0:8080',
- metavar='addr', dest='ui_addr',
+ '--addr', default='http://0.0.0.0:8080', metavar='addr', dest='addr',
help="address of listening web ui socket formated as "
"'<type>://<host>:<port>' - <type> is 'http' or 'https'; "
"<host> is hostname; <port> is tcp port number "
"(default http://0.0.0.0:8080)")
server.add_argument(
- '--ui-pem', default=None, metavar='path', dest='ui_pem_path',
+ '--pem', default=None, metavar='path', dest='pem_path',
+ action=util.EnvPathArgParseAction,
help="web front-end pem file path - required for https")
- server.add_argument(
- '--ui-path', default=None, metavar='path', dest='ui_path',
- help="override path to front-end web app directory")
calculate = subparsers.add_parser('calculate', help='calculate result')
calculate.add_argument(
'--params', required=True, metavar='path', dest='params_path',
+ action=util.EnvPathArgParseAction,
help="calculate parameters file path "
"(specified by opcut://params.yaml#)")
calculate.add_argument(
- '--method', dest='method', type=common.Method,
+ '--method', dest='method',
default=common.Method.FORWARD_GREEDY,
- choices=list(map(lambda x: x.name, common.Method)),
+ choices=[method.name for method in common.Method],
+ action=_MethodParseAction,
help="calculate method (default FORWARD_GREEDY)")
+ calculate.add_argument(
+ '--output', default=None, metavar='path', dest='output_path',
+ action=util.EnvPathArgParseAction,
+ help="optional output file path")
output = subparsers.add_parser('output', help='generate output')
+ output.add_argument(
+ '--output', required=True, metavar='path', dest='output_path',
+ action=util.EnvPathArgParseAction,
+ help="output file path")
for p in [calculate, output]:
p.add_argument(
'--result', required=True, metavar='path', dest='result_path',
+ action=util.EnvPathArgParseAction,
help="calculate result file path "
"(specified by opcut://result.yaml#)")
p.add_argument(
'--output-type', dest='output_type', default='PDF',
- choices=list(map(lambda x: x.name, common.OutputType)),
+ choices=[output_type.name for output_type in common.OutputType],
help="output type (default PDF)")
p.add_argument(
'--output-panel', default=None, metavar='panel_id',
dest='output_panel_id', help="output panel id")
- p.add_argument(
- '--output', default=None, metavar='path', dest='output_path',
- help="optional output file path")
+
+ dev_args = parser.add_argument_group('development arguments')
+ dev_args.add_argument(
+ '--json-schemas-path', metavar='path', dest='schemas_json_path',
+ default=default_schemas_json_path,
+ action=util.EnvPathArgParseAction,
+ help="override json schemas directory path")
+ dev_args.add_argument(
+ '--ui-path', metavar='path', dest='ui_path',
+ default=default_ui_path,
+ action=util.EnvPathArgParseAction,
+ help="override web ui directory path")
return parser
+class _MethodParseAction(argparse.Action):
+
+ def __call__(self, parser, namespace, values, option_string=None):
+ ret = []
+ for value in (values if self.nargs else [values]):
+ try:
+ ret.append(common.Method[value])
+ except Exception as e:
+ parser.error(str(e))
+ setattr(namespace, self.dest, ret if self.nargs else ret[0])
+
+
if __name__ == '__main__':
sys.exit(main())
diff --git a/src_py/opcut/server.py b/src_py/opcut/server.py
index 0dd607d..d4b169d 100644
--- a/src_py/opcut/server.py
+++ b/src_py/opcut/server.py
@@ -1,91 +1,78 @@
-import ssl
-import contextlib
import asyncio
-import functools
import base64
+import functools
+import ssl
+import urllib.parse
+from hat.util import aio
import aiohttp.web
-from opcut import util
from opcut import common
-import opcut.json_validator
import opcut.csp
import opcut.output
-async def run(addr, pem_path, ui_path):
+async def run(json_schema_repo, addr, pem_path, ui_path):
- executor = util.create_async_executor()
+ executor = aio.create_executor()
+ addr = urllib.parse.urlparse(addr)
if addr.scheme == 'https':
ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
ssl_ctx.load_cert_chain(pem_path)
else:
ssl_ctx = None
+ async def root_handler(request):
+ raise aiohttp.web.HTTPFound('/index.html')
+
app = aiohttp.web.Application()
- app.add_routes([
- aiohttp.web.get('/', lambda req: aiohttp.web.HTTPFound('/index.html')),
- aiohttp.web.post(
- '/calculate',
- functools.partial(_calculate_handler, executor)),
- aiohttp.web.post(
- '/generate_output',
- functools.partial(_generate_output_handler, executor)),
- aiohttp.web.static('/', ui_path)])
+ app.add_routes([aiohttp.web.get('/', root_handler),
+ aiohttp.web.post('/calculate', functools.partial(
+ _calculate_handler, json_schema_repo, executor)),
+ aiohttp.web.post('/generate_output', functools.partial(
+ _generate_output_handler, json_schema_repo, executor)),
+ aiohttp.web.static('/', ui_path)])
runner = aiohttp.web.AppRunner(app)
- await runner.setup()
- site = aiohttp.web.TCPSite(runner,
- host=addr.hostname,
- port=addr.port,
- ssl_context=ssl_ctx,
- shutdown_timeout=0.1)
- await site.start()
-
- with contextlib.suppress(asyncio.CancelledError):
+ try:
+ await runner.setup()
+ site = aiohttp.web.TCPSite(runner,
+ host=addr.hostname,
+ port=addr.port,
+ ssl_context=ssl_ctx,
+ shutdown_timeout=0.1)
+ await site.start()
await asyncio.Future()
-
- await runner.cleanup()
+ finally:
+ await runner.cleanup()
-async def _calculate_handler(executor, request):
+async def _calculate_handler(json_schema_repo, executor, request):
try:
msg = await request.json()
- opcut.json_validator.validate(
- msg, 'opcut://messages.yaml#/definitions/calculate/request')
+ json_schema_repo.validate(
+ 'opcut://messages.yaml#/definitions/calculate/request', msg)
params = common.json_data_to_params(msg['params'])
method = common.Method[msg['method']]
- result = await executor(_ext_calculate, params, method)
+ result = await executor(opcut.csp.calculate, params, method)
result_json_data = common.result_to_json_data(result)
- except asyncio.CancelledError:
- raise
except Exception:
result_json_data = None
return aiohttp.web.json_response({'result': result_json_data})
-async def _generate_output_handler(executor, request):
+async def _generate_output_handler(json_schema_repo, executor, request):
try:
msg = await request.json()
- opcut.json_validator.validate(
- msg, 'opcut://messages.yaml#/definitions/generate_output/request')
+ json_schema_repo.validate(
+ 'opcut://messages.yaml#/definitions/generate_output/request', msg)
result = common.json_data_to_result(msg['result'])
output_type = common.OutputType[msg['output_type']]
panel = msg['panel']
- output = await executor(_ext_generate_output, result, output_type,
- panel)
+ output = await executor(opcut.output.generate_output, result,
+ output_type, panel)
output_json_data = base64.b64encode(output).decode('utf-8')
- except asyncio.CancelledError:
- raise
except Exception:
output_json_data = None
return aiohttp.web.json_response({'data': output_json_data})
-
-
-def _ext_calculate(params, method):
- return opcut.csp.calculate(params, method)
-
-
-def _ext_generate_output(result, output_type, panel):
- return opcut.output.generate_output(result, output_type, panel)
diff --git a/src_py/opcut/util.py b/src_py/opcut/util.py
deleted file mode 100644
index 38aaf41..0000000
--- a/src_py/opcut/util.py
+++ /dev/null
@@ -1,107 +0,0 @@
-import contextlib
-import asyncio
-import sys
-import collections
-import concurrent
-
-
-def namedtuple(name, *props):
- """Create documented namedtuple
-
- Args:
- name (Union[str,Tuple[str,str]]):
- named tuple's name or named tuple's name with documentation
- props (Iterable[Union[str,Tuple[str,str],Tuple[str,str,Any]]]):
- named tuple' properties with optional documentation and
- optional default value
-
- Returns:
- class implementing collections.namedtuple
-
- """
- props = [(i, None) if isinstance(i, str) else list(i) for i in props]
- cls = collections.namedtuple(name if isinstance(name, str) else name[0],
- [i[0] for i in props])
- default_values = []
- for i in props:
- if default_values and len(i) < 3:
- raise Exception("property with default value not at end")
- if len(i) > 2:
- default_values.append(i[2])
- if default_values:
- cls.__new__.__defaults__ = tuple(default_values)
- if not isinstance(name, str) and name[1]:
- cls.__doc__ = name[1]
- for i in props:
- if i[1]:
- getattr(cls, i[0]).__doc__ = i[1]
- try:
- cls.__module__ = sys._getframe(1).f_globals.get('__name__', '__main__')
- except (AttributeError, ValueError):
- pass
- return cls
-
-
-def run_until_complete_without_interrupt(future, loop=None):
- """Run event loop until future or coroutine is done
-
- Args:
- future (Awaitable): future or coroutine
- loop (Optional[asyncio.AbstractEventLoop]): asyncio loop
-
- Returns:
- Any: provided future's result
-
- KeyboardInterrupt is suppressed (while event loop is running) and is mapped
- to single cancelation of running task. If multipple KeyboardInterrupts
- occur, task is cancelled only once.
-
- """
- if not loop:
- loop = asyncio.get_event_loop()
-
- async def ping_loop():
- with contextlib.suppress(asyncio.CancelledError):
- while True:
- await asyncio.sleep(1, loop=loop)
-
- task = asyncio.ensure_future(future, loop=loop)
- if sys.platform == 'win32':
- ping_loop_task = asyncio.ensure_future(ping_loop(), loop=loop)
- with contextlib.suppress(KeyboardInterrupt):
- loop.run_until_complete(task)
- loop.call_soon(task.cancel)
- if sys.platform == 'win32':
- loop.call_soon(ping_loop_task.cancel)
- while not task.done():
- with contextlib.suppress(KeyboardInterrupt):
- loop.run_until_complete(task)
- if sys.platform == 'win32':
- while not ping_loop_task.done():
- with contextlib.suppress(KeyboardInterrupt):
- loop.run_until_complete(ping_loop_task)
- return task.result()
-
-
-def create_async_executor(*args,
- executor_cls=concurrent.futures.ThreadPoolExecutor,
- loop=None):
- """Create run_in_executor wrapper
-
- Args:
- args (Any): executor init args
- executor_cls (Type): executor class
- loop (Optional[asyncio.AbstractEventLoop]): asyncio loop
-
- Returns:
- Callable[[Callable,...],Any]: coroutine accepting function and it's
- arguments and returning function call result
-
- """
- executor = executor_cls(*args)
-
- async def executor_wrapper(fn, *fn_args):
- _loop = loop if loop else asyncio.get_event_loop()
- return await _loop.run_in_executor(executor, fn, *fn_args)
-
- return executor_wrapper
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-02-11 14:51:35 +0000