def __init__(self, mrgeo):
     self._mrgeo = mrgeo
     jvm = self._mrgeo._get_jvm()
     # Import the raster map op test support class and all other needed classes
     java_import(jvm, "org.mrgeo.mapalgebra.InlineCsvMapOp")
     self._jvm = jvm
     self._sparkContext = mrgeo.sparkContext

    def __init__(self, language, gateway, **kwargs):
        '''
        Constructor
        @param language: The language code for the proper initialization of this language-dependent tool
        @type language: string
        @param gateway: An already initialized Py4j java gateway
        @type gateway: py4j.java_gateway.JavaGateway
        '''
        self.language = language
        #self.jvm = JVM(java_classpath)
        #socket_no = self.jvm.socket_no
        #gatewayclient = GatewayClient('localhost', socket_no)
        #gateway = JavaGateway(gatewayclient, auto_convert=True, auto_field=True)
        #sys.stderr.write("Initialized local Java gateway with pid {} in socket {}\n".format(self.jvm.pid, socket_no))
    
        self.meteor_view = gateway.new_jvm_view()
        #import necessary java packages from meteor jar
        java_import(self.meteor_view, 'edu.cmu.meteor.scorer.*')
        java_import(self.meteor_view, 'edu.cmu.meteor.util.*')
#        java_import(self.meteor_view, '')
        
        #pass the language setting into the meteor configuration object
        config = self.meteor_view.MeteorConfiguration();
        config.setLanguage(language);
        #initialize object with the given config
        sys.stderr.write("If next line displays error, it is not critical, but METEOR language-specific transducer must be installed.")
        self.scorer = self.meteor_view.MeteorScorer(config)

 def _do_init(self, *args, **kwargs):
     # Modifies base _do_init to add a Java-Cassandra SparkContext (jcsc)
     # to the instance
     super(CassandraSparkContext, self)._do_init(*args, **kwargs)
     java_import(self._jvm, "com.datastax.spark.connector.CassandraJavaUtil")
     java_import(self._jvm, "com.datastax.spark.connector.RowConvertingIterator")
     self._jcsc = self._jvm.CassandraJavaUtil.javaFunctions(self._jsc)

def main():
    if len(sys.argv) != 3:
        print >> sys.stderr, "Usage: example <keyspace_name> <column_family_name>"
        sys.exit(-1)

    keyspace_name = sys.argv[1]
    column_family_name = sys.argv[2]

    # Valid config options here https://github.com/datastax/spark-cassandra-connector/blob/master/doc/1_connecting.md
    conf = SparkConf().set("spark.cassandra.connection.host", "127.0.0.1")

    sc = SparkContext(appName="Spark + Cassandra Example",
                      conf=conf)

    # import time; time.sleep(30)
    java_import(sc._gateway.jvm, "com.datastax.spark.connector.CassandraJavaUtil")
    print sc._jvm.CassandraJavaUtil

    users = (
        ["Mike", "Sukmanowsky"],
        ["Andrew", "Montalenti"],
        ["Keith", "Bourgoin"],
    )
    rdd = sc.parallelize(users)
    print rdd.collect()

def start_gateway_server():
    classPath = compute_classpath(DDF_HOME)
    # launch GatewayServer in a new process
    javaopts = os.getenv('JAVA_OPTS')
    if javaopts is not None:
        javaopts = javaopts.split()
    else:
        javaopts = []
    #command = ["java", "-classpath", classPath] + ["-Dlog4j.configuration=file:"+ DDF_HOME + "/core/conf/local/ddf-local-log4j.properties"] + ["py4j.GatewayServer", "--die-on-broken-pipe", "0"]
    command = ["java", "-classpath", classPath] + javaopts + ["py4j.GatewayServer", "--die-on-broken-pipe", "0"]
    
    proc = Popen(command, stdout = PIPE, stdin = PIPE, preexec_fn = preexec_func)
    # get the port of the GatewayServer
    port = int(proc.stdout.readline())

    class JavaOutputThread(Thread):
        def __init__(self, stream):
            Thread.__init__(self)
            self.daemon = True
            self.stream = stream

        def run(self):
            while True:
                line = self.stream.readline()
                sys.stderr.write(line)
    JavaOutputThread(proc.stdout).start()
    # connect to the gateway server
    gateway = JavaGateway(GatewayClient(port = port), auto_convert = False)
    java_import(gateway.jvm, "io.ddf.*")
    java_import(gateway.jvm, "io.ddf.spark.*")
    return gateway

def singlethread(java_classpath):
    print "Thread starting"
    
    jvm = JVM(java_classpath, dir_path)
    socket_no = self.jvm.socket_no
    gatewayclient = GatewayClient('localhost', socket_no)
    gateway = JavaGateway(gatewayclient, auto_convert=True, auto_field=True)
    sys.stderr.write("Initialized global Java gateway with pid {} in socket {}\n".format(self.jvm.pid, socket_no))

    
    gatewayclient = GatewayClient('localhost', socket_no)
    print "Gclient started"
    gateway = JavaGateway(gatewayclient, auto_convert=True, auto_field=True)
    print "Java Gateway started"
    #create a new view for the jvm
    meteor_view = gateway.new_jvm_view()
    #import required packages
    java_import(meteor_view, 'edu.cmu.meteor.scorer.*')
    #initialize the java object
    java_import(meteor_view, 'edu.cmu.meteor.util.*')
    print "Modules imported"
    #pass the language setting into the meteor configuration object
    config = meteor_view.MeteorConfiguration();
    config.setLanguage("en");
    scorer = meteor_view.MeteorScorer(config)
    print "object initialized"
    #run object function
    stats = scorer.getMeteorStats("Test sentence", "Test sentence !");
    print stats.score
    return 1

 def _connect(self, gateway, grammarfile):        
     module_view = gateway.new_jvm_view()      
     java_import(module_view, 'BParser')
     
     # get the application instance
     log.info("Grammar file: {}".format(grammarfile))
     self.bp_obj =  module_view.BParser(grammarfile)
     sys.stderr.write("got BParser object\n")

def scala_set_to_set(ctx, x):
    from py4j.java_gateway import java_import

    # import scala
    java_import(ctx._jvm, 'scala')

    # grab Scala's set converter and convert to a Python set
    return set(ctx._jvm.scala.collection.JavaConversions.setAsJavaSet(x))

    def get_smoothing_method(self, spark_context):
        java_import(spark_context._jvm, ClassNames.WEIGHTS)
        java_import(spark_context._jvm, ClassNames.WEIGHTED_MOVING_AVERAGE)
        java_weights = spark_context._jvm.Weights(self.__python_weights.limit())
        for index in range(self.__window_size):
            java_weights.add(self.__python_weights.get(index))

        return spark_context._jvm.WeightedMovingAverageMethod(self.__window_size, java_weights)

def launch_gateway():
    if "MRGEO_GATEWAY_PORT" in os.environ:
        gateway_port = int(os.environ["MRGEO_GATEWAY_PORT"])
    else:
        # Launch the Py4j gateway using the MrGeo command so that we pick up the proper classpath

        script = find_script()

        # Start a socket that will be used by PythonGatewayServer to communicate its port to us
        callback_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        callback_socket.bind(('127.0.0.1', 0))
        callback_socket.listen(1)
        callback_host, callback_port = callback_socket.getsockname()
        env = dict(os.environ)
        env['_MRGEO_DRIVER_CALLBACK_HOST'] = callback_host
        env['_MRGEO_DRIVER_CALLBACK_PORT'] = str(callback_port)

        command = [script, "python", "-v", "-h", callback_host, "-p", str(callback_port)]

        # Launch the Java gateway.
        # We open a pipe to stdin so that the Java gateway can die when the pipe is broken
        # Don't send ctrl-c / SIGINT to the Java gateway:
        def preexec_func():
            signal.signal(signal.SIGINT, signal.SIG_IGN)

        proc = Popen(command, stdin=PIPE, preexec_fn=preexec_func, env=env)

        gateway_port = None
        # We use select() here in order to avoid blocking indefinitely if the subprocess dies
        # before connecting
        while gateway_port is None and proc.poll() is None:
            timeout = 1  # (seconds)
            readable, _, _ = select.select([callback_socket], [], [], timeout)
            if callback_socket in readable:
                gateway_connection = callback_socket.accept()[0]
                # Determine which ephemeral port the server started on:
                gateway_port = read_int(gateway_connection.makefile(mode="rb"))
                gateway_connection.close()
                callback_socket.close()

        if gateway_port is None:
            raise Exception("Java gateway process exited before sending the driver its port number")

    print("Talking with MrGeo on port " + str(gateway_port))

    # Connect to the gateway
    gateway = JavaGateway(GatewayClient(port=gateway_port), auto_convert=True)

    # Import the classes used by MrGeo
    java_import(gateway.jvm, "org.mrgeo.python.*")

    # Import classes used by Spark
    java_import(gateway.jvm, "org.apache.spark.SparkConf")
    java_import(gateway.jvm, "org.apache.spark.api.java.*")
    java_import(gateway.jvm, "org.apache.spark.api.python.*")
    java_import(gateway.jvm, "org.apache.spark.mllib.api.python.*")

    return gateway

    def _getSome(self, value):
        java_import(self._jvm, "scala.Some")
        return self._jvm.Some(value)




# suite = TestLoader().loadTestsFromTestCase(MrGeoLocalIntegrationTests)
# TextTestRunner(verbosity=2).run(suite)

 def setUp(self):
     conf = SparkConf().setAppName('test').setMaster('local[*]')
     pwd = os.path.dirname(os.path.realpath(__file__))
     metastore_dir = os.path.abspath(os.path.join(pwd, '..',
                                                  'metastore_db'))
     silentremove(os.path.join(metastore_dir, "dbex.lck"))
     silentremove(os.path.join(metastore_dir, "db.lck"))
     self.sc = SparkContext(conf=conf)
     self.jvm = self.sc._gateway.jvm
     java_import(self.jvm, "org.apache.spark.sql.*")

def createColor(r, g, b):    

    global _gateway
    if _gateway is None:
        _gateway = JavaGateway()
        
    jvm = _gateway.jvm

    java_import(jvm, 'org.eclipse.swt.graphics.*')
    
    return jvm.Color(None, r, g, b)

def getPlottingSystem(plottingSystemName):
    
    global _gateway
    if _gateway is None:
        _gateway = JavaGateway()
        
    jvm = _gateway.jvm
    
    java_import(jvm, 'org.eclipse.dawnsci.plotting.api.*')
    
    return jvm.PlottingFactory.getPlottingSystem(plottingSystemName, True)

    def __init__(self, _jvm, smvconfig):
        self._jvm = _jvm

        self.smvconfig = smvconfig
        self.dsRepoFactories = []

        from py4j.java_gateway import java_import
        java_import(self._jvm, "org.tresamigos.smv.python.SmvPythonHelper")
        java_import(self._jvm, "org.tresamigos.smv.DataSetRepoFactoryPython")

        self.helper = self._jvm.SmvPythonHelper

def getService(serviceClass):
    
    global _gateway
    if _gateway is None:
        _gateway = JavaGateway()
        
    jvm = _gateway.jvm
    
    java_import(jvm, 'org.dawb.common.services.*')
    
    return jvm.Activator.getService(serviceClass)

def new_gateway_client():
    global __gateway_server_port

    if __gateway_server_port is None:
        __gateway_server_port = start_gateway_server()

    # connect to the gateway server
    gateway = JavaGateway(GatewayClient(port=__gateway_server_port), auto_convert=False)
    java_import(gateway.jvm, 'io.ddf.*')
    java_import(gateway.jvm, 'io.ddf.spark.*')
    return gateway

def createHistogramBound(position, r, g, b):    

    global _gateway
    if _gateway is None:
        _gateway = JavaGateway()
        
    jvm = _gateway.jvm

    java_import(jvm, 'org.eclipse.dawnsci.plotting.api.histogram.*')
    
    return jvm.HistogramBound(position, r, g, b)

    def _create_job(self):
        if not self._job:
            jvm = self._get_jvm()
            java_import(jvm, "org.mrgeo.job.*")

            appname = "PyMrGeo"

            self._job = jvm.JobArguments()
            java_gateway.set_field(self._job, "name", appname)

            # Yarn in the default
            self.useyarn()

 def __set_file_type(self, jvm, file_type):
     java_import(jvm, ClassNames.FileType)
     file_types = {
         'CSV': jvm.FileType.CSV,
         'TSV': jvm.FileType.TSV
     }
     if file_type in file_types.values():
         self.__file_type = file_type
     elif file_type.upper() in file_types:
         self.__file_type = file_types[file_type.upper()]
     else:
         raise ValueError('"%s" is not a valid file type\nValid file types are CSV and TSV' % file_type)