public Object year(Object time_ms) throws ScriptException {
    if (nullCheck(time_ms)) {
        return time_ms;
    }
    PyLong time = new PyLong(((Number) time_ms).longValue() / 1000);
    PyTuple timeTuple = Time.gmtime(time);
    return timeTuple.__getitem__(0);
}
 

public Object month(Object time_ms) throws ScriptException {
    if (nullCheck(time_ms)) {
        return time_ms;
    }
    PyLong time = new PyLong(((Number) time_ms).longValue() / 1000);
    PyTuple timeTuple = Time.gmtime(time);
    return timeTuple.__getitem__(1);
}
 

public Object day(Object time_ms) throws ScriptException {
    if (nullCheck(time_ms)) {
        return time_ms;
    }
    PyLong time = new PyLong(((Number) time_ms).longValue() / 1000);
    PyTuple timeTuple = Time.gmtime(time);
    return timeTuple.__getitem__(2);
}
 

public Object hour(Object time_ms) throws ScriptException {
    if (nullCheck(time_ms)) {
        return time_ms;
    }
    PyLong time = new PyLong(((Number) time_ms).longValue() / 1000);
    PyTuple timeTuple = Time.gmtime(time);
    return timeTuple.__getitem__(3);
}
 

public Object minute(Object time_ms) throws ScriptException {
    if (nullCheck(time_ms)) {
        return time_ms;
    }
    PyLong time = new PyLong(((Number) time_ms).longValue() / 1000);
    PyTuple timeTuple = Time.gmtime(time);
    return timeTuple.__getitem__(4);
}
 

public Object second(Object time_ms) throws ScriptException {
    if (nullCheck(time_ms)) {
        return time_ms;
    }
    PyLong time = new PyLong(((Number) time_ms).longValue() / 1000);
    PyTuple timeTuple = Time.gmtime(time);
    return timeTuple.__getitem__(5);
}
 

public Object weekday(Object time_ms) throws ScriptException {
    if (nullCheck(time_ms)) {
        return time_ms;
    }
    PyLong time = new PyLong(((Number) time_ms).longValue() / 1000);
    PyTuple timeTuple = Time.gmtime(time);
    return timeTuple.__getitem__(6);
}
 

public Object month_name(Object time_ms) throws ScriptException {
    if (nullCheck(time_ms)) {
        return time_ms;
    }
    PyLong time = new PyLong(((Number) time_ms).longValue() / 1000);
    PyTuple timeTuple = Time.gmtime(time);
    return Time.strftime("%B", timeTuple);
}
 

public Object weekday_name(Object time_ms) throws ScriptException {
    if (nullCheck(time_ms)) {
        return time_ms;
    }
    PyLong time = new PyLong(((Number) time_ms).longValue() / 1000);
    PyTuple timeTuple = Time.gmtime(time);
    return Time.strftime("%A", timeTuple);
}
 

public Object time_from_date(Object time_ms) throws ScriptException {
    if (nullCheck(time_ms)) {
        return time_ms;
    }
    PyLong time = new PyLong(((Number) time_ms).longValue() / 1000);
    PyTuple timeTuple = Time.gmtime(time);
    return (timeTuple.__getitem__(3).asInt() * MINS_IN_HOUR * SECS_IN_MIN * MILLISECS_IN_SEC)
            + (timeTuple.__getitem__(4).asInt() * SECS_IN_MIN * MILLISECS_IN_SEC)
            + (timeTuple.__getitem__(5).asInt() * MILLISECS_IN_SEC);
}
 

public Object time_string_from_date(Object time_ms) throws ScriptException {
    if (nullCheck(time_ms)) {
        return time_ms;
    }
    PyLong time = new PyLong(((Number) time_ms).longValue() / 1000);
    PyTuple timeTuple = Time.gmtime(time);
    return Time.strftime("%H:%M:%S", timeTuple);
}
 

public Object date_as_string(Object time_ms) throws ScriptException {
    if (nullCheck(time_ms)) {
        return time_ms;
    }
    PyLong time = new PyLong(((Number) time_ms).longValue() / 1000);
    PyTuple timeTuple = Time.gmtime(time);
    return Time.strftime("%Y-%m-%d %H:%M:%S", timeTuple);
}
 

private static void registerJythonSerializers(StreamExecutionEnvironment env) {
	env.registerTypeWithKryoSerializer(PyBoolean.class, PyBooleanSerializer.class);
	env.registerTypeWithKryoSerializer(PyFloat.class, PyFloatSerializer.class);
	env.registerTypeWithKryoSerializer(PyInteger.class, PyIntegerSerializer.class);
	env.registerTypeWithKryoSerializer(PyLong.class, PyLongSerializer.class);

	env.registerTypeWithKryoSerializer(PyString.class, PyStringSerializer.class);
	env.registerTypeWithKryoSerializer(PyUnicode.class, PyObjectSerializer.class);

	env.registerTypeWithKryoSerializer(PyTuple.class, PyObjectSerializer.class);
	env.registerTypeWithKryoSerializer(PyObjectDerived.class, PyObjectSerializer.class);
	env.registerTypeWithKryoSerializer(PyInstance.class, PyObjectSerializer.class);
}
 

@Override
public void write(Kryo kryo, Output output, PyLong object) {
	byte[] data = object.getValue().toByteArray();
	output.writeShort(data.length);
	output.writeBytes(data);
}
 

@Override
public PyLong read(Kryo kryo, Input input, Class<PyLong> type) {
	int length = input.readShort();
	return new PyLong(new BigInteger(input.readBytes(length)));
}
 

/**
 * A wrapper static method for Python user-defined function
 *
 * @param args if the registered stored procedure has n parameters, the args[0] to args[n-1] are these parameters.
 * The args[n] must be a String which contains the Python script.
 */
public static Object pyFunctionWrapper(Object... args)
        throws Exception
{
    PyInterpreterPool pool = null;
    PythonInterpreter interpreter = null;
    String setFacFuncName = "setFactory";
    String funcName = "execute";
    PyObject pyResult = null;
    Object javaResult = null;

    try {
        byte[] compiledCode;
        int nargs = args.length;
        int pyScriptIdx = args.length - 1;

        // set pyScript
        compiledCode = (byte[]) args[pyScriptIdx];

        // set the Object[] to pass in
        Object[] pyArgs;
        if(nargs - 1 ==0){
            pyArgs = null;
        }
        else{
            pyArgs = new Object[nargs - 1];
            System.arraycopy(args, 0, pyArgs, 0, nargs - 1);
        }

        pool = PyInterpreterPool.getInstance();
        interpreter = pool.acquire();
        PyCodeUtil.exec(compiledCode, interpreter);
        // add global variable factory, so that the user can use it to construct JDBC ResultSet
        Object[] factoryArg = {new PyStoredProcedureResultSetFactory()};
        PyFunction addFacFunc = interpreter.get(setFacFuncName, PyFunction.class);
        addFacFunc._jcall(factoryArg);

        // execute the user defined function, the user needs to fill the ResultSet himself,
        // just like the original Java Stored Procedure
        PyFunction userFunc = interpreter.get(funcName, PyFunction.class);
        if(pyArgs == null){
            pyResult = userFunc.__call__();
        }else{
            pyResult = userFunc._jcall(pyArgs);
        }
        javaResult = pyResult.__tojava__(Object.class);
        if(pyResult instanceof PyLong){
            // When the result has type PyLong, the result's corresponding
            // sql type should be BigInt.
            javaResult = ((BigInteger) javaResult).longValue();
        }
        return javaResult;
    }
    catch (Exception e){
        throw StandardException.plainWrapException(e);
    }
    finally{
        if(pool != null && interpreter != null){
            pool.release(interpreter);
        }
    }
}