Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import os
import io
import sys
import copy
import subprocess
import contextlib
import cppimport
import cppimport.build_module
import cppimport.templating
import cppimport.import_hook
cppimport.set_quiet(False)
@contextlib.contextmanager
def appended(filename, text):
orig = open(filename, 'r').read()
open(filename, 'a').write(text)
try:
yield
finally:
open(filename, 'w').write(orig)
def subprocess_check(test_code, returncode = 0):
p = subprocess.Popen([
'python', '-c', test_code
], cwd = os.path.dirname(__file__))
p.wait()
assert(p.returncode == returncode)
import cppimport
cppimport.force_rebuild()
cppimport.set_quiet(False)
snowdrift = cppimport.imp("snowdrift")
test_polymorphism = cppimport.imp("test_polymorphism")
import unittest
import fwdpy11.fitness as fp11w
import re
class testFitnessPolymorphism(unittest.TestCase):
@classmethod
def setUpClass(self):
self.w = [
fp11w.SlocusAdditive(),
fp11w.SlocusMult(),
snowdrift.SlocusSnowdrift(0.2, -0.2, 1, -2)]
self.v = test_polymorphism.test_callback_names(self.w)
import cppimport
cppimport.set_quiet(False)
wrapper = cppimport.imp("taskloaf.wrapper").wrapper
launch_local = wrapper.launch_local
task = wrapper.task
ready = wrapper.ready
Future = wrapper.Future
# def when_all(*args):
# def make_split_args(f):
# def split_args(x):
# return f(*x)
# return split_args
#
# def flatten_tuple(t):
# return sum(t, ())
#
parser = argparse.ArgumentParser(description='Run a C++ file with cppimport')
parser.add_argument('filename', help = 'The file to run.')
parser.add_argument(
'--add_main_caller', '-m',
action = 'store_true',
help = 'Add a pybind11 function that will call your main()'
)
parser.add_argument(
'--verbose', '-v',
action = 'store_true',
help = 'Tell me everything!'
)
args = parser.parse_args()
if args.verbose:
cppimport.set_quiet(False)
filename = args.filename
filedir = os.path.dirname(filename)
filebasename = os.path.basename(filename)
module_name, file_extension = os.path.splitext(filebasename)
if args.add_main_caller:
cpprun_dir = '.cpprunfiles'
if not os.path.exists(cpprun_dir):
os.makedirs(cpprun_dir)
src = os.path.join(cpprun_dir, filebasename)
open(src, 'w').write(open(filename, 'r').read() + footer)
sys.path.append(cpprun_dir)
else:
sys.path.append(filedir)
import argparse
import sys, time, os
# ONNX/Numpy
import onnx
import onnx.numpy_helper
import numpy as np
#gRPC for client-server communication
import grpc
# cpp
import cppimport
import cppimport.import_hook
cppimport.set_quiet(True)
import lib.minionn as minionn
#project imports
from common import minionn_onnx_pb2
from common import minionn_onnx_pb2_grpc
from common import onnx_helper, operation_handler, minionn_helper, config
# Logging
import logging
import logging.config
logging.config.fileConfig('common/logging.conf')
logger = logging.getLogger('minionn')
def main():
parser = argparse.ArgumentParser(description="MiniONN - ONNX compatible version")
parser.add_argument(