3646 lines
105 KiB
Plaintext
3646 lines
105 KiB
Plaintext
---- /leetcode/commands.txt
|
||
stepcontinue
|
||
---- /leetcode/console.py
|
||
# Forked from https://raw.githubusercontent.com/romanvm/python-web-pdb
|
||
from __future__ import absolute_import
|
||
from __future__ import unicode_literals
|
||
|
||
import sys
|
||
import weakref
|
||
from threading import Event
|
||
from threading import RLock
|
||
|
||
try:
|
||
import queue
|
||
except ImportError:
|
||
import Queue as queue
|
||
|
||
|
||
class ThreadSafeObject(object):
|
||
"""
|
||
An object for data exchange between threads
|
||
"""
|
||
|
||
def __init__(self, contents=None):
|
||
self._lock = RLock()
|
||
self._contents = contents
|
||
|
||
@property
|
||
def contents(self):
|
||
with self._lock:
|
||
return self._contents
|
||
|
||
@contents.setter
|
||
def contents(self, value):
|
||
with self._lock:
|
||
self._contents = value
|
||
|
||
|
||
class Console(object):
|
||
input_queue = queue.Queue()
|
||
|
||
def __init__(self, debugger):
|
||
self.pdb_not_busy = Event()
|
||
self.closed = False
|
||
self._debugger = weakref.proxy(debugger)
|
||
self._console_history = ThreadSafeObject("")
|
||
self._frame_data = ThreadSafeObject({})
|
||
self._output_lock = RLock()
|
||
self.exception = None
|
||
|
||
@property
|
||
def seekable(self):
|
||
return False
|
||
|
||
@property
|
||
def writable(self):
|
||
return True
|
||
|
||
@property
|
||
def encoding(self):
|
||
return "utf-8"
|
||
|
||
def close(self):
|
||
self.closed = True
|
||
self.pdb_not_busy.set()
|
||
|
||
def readline(self):
|
||
self.pdb_not_busy.set()
|
||
command = self.input_queue.get()
|
||
self.pdb_not_busy.clear()
|
||
|
||
if not command:
|
||
command = "\n"
|
||
|
||
with open("commands.txt", "a+") as file:
|
||
file.write(command)
|
||
self.writeline(command)
|
||
return command
|
||
|
||
read = readline
|
||
|
||
def writeline(self, data):
|
||
with self._output_lock:
|
||
if isinstance(data, bytes):
|
||
data = data.decode("utf-8")
|
||
|
||
self._console_history.contents += data
|
||
try:
|
||
frame_data = self._debugger.get_current_frame_data()
|
||
except (IOError, AttributeError):
|
||
frame_data = {
|
||
"dirname": "",
|
||
"filename": "",
|
||
"file_listing": "No data available",
|
||
"current_line": -1,
|
||
"breakpoints": [],
|
||
"globals": {},
|
||
"locals": {},
|
||
"expressions": {},
|
||
"exception": sys.exc_info(),
|
||
}
|
||
|
||
frame_data["console_history"] = self._console_history.contents
|
||
self._frame_data.contents = frame_data
|
||
|
||
write = writeline
|
||
|
||
def flush(self):
|
||
pass
|
||
|
||
def send_pdb_command(self, command):
|
||
self.pdb_not_busy.wait()
|
||
self.input_queue.put(command)
|
||
|
||
def get_frame_data(self):
|
||
with self._output_lock:
|
||
return self._frame_data.contents
|
||
|
||
---- /leetcode/io_wrapper.py
|
||
import sys
|
||
|
||
is_python3 = sys.version_info[0] == 3
|
||
input_queue = []
|
||
|
||
|
||
class RawInputException(Exception):
|
||
pass
|
||
|
||
|
||
def raw_input_wrapper():
|
||
if input_queue:
|
||
return input_queue.pop(0)
|
||
raise RawInputException()
|
||
|
||
|
||
def python2_input_wrapper():
|
||
# Python 2 input() does eval(raw_input())
|
||
if input_queue:
|
||
input_str = input_queue.pop(0)
|
||
return eval(input_str)
|
||
raise RawInputException()
|
||
|
||
|
||
def stream_stdin():
|
||
while input_queue:
|
||
item = input_queue.pop(0)
|
||
yield item
|
||
|
||
|
||
def override_input():
|
||
if is_python3:
|
||
import builtins
|
||
else:
|
||
import __builtin__ as builtins
|
||
|
||
with open("user_code/input.txt") as file:
|
||
global input_queue
|
||
input_queue = list(file.readlines())
|
||
|
||
builtins.raw_input = raw_input_wrapper
|
||
|
||
if is_python3:
|
||
builtins.input = raw_input_wrapper
|
||
else:
|
||
builtins.input = python2_input_wrapper
|
||
|
||
sys.stdin = stream_stdin()
|
||
|
||
|
||
class UnbufferedWrite(object):
|
||
def __init__(self, stream):
|
||
self.stream = stream
|
||
|
||
def write(self, data):
|
||
self.stream.write(data)
|
||
self.stream.flush()
|
||
|
||
def writelines(self, datas):
|
||
self.stream.writelines(datas)
|
||
self.stream.flush()
|
||
|
||
def __getattr__(self, attr):
|
||
return getattr(self.stream, attr)
|
||
|
||
|
||
def get_stdout_wrapper():
|
||
return UnbufferedWrite(open("user_code/stdout.txt", "w"))
|
||
|
||
---- /leetcode/log.py
|
||
import logging
|
||
import sys
|
||
|
||
|
||
def get_logger():
|
||
logging.basicConfig(stream=sys.__stdout__, level=logging.DEBUG)
|
||
log = logging.getLogger("werkzeug")
|
||
return log
|
||
|
||
|
||
def debug(*args):
|
||
msg = "\n".join(map(str, args))
|
||
logger = get_logger()
|
||
logger.debug(msg, extra={"stack": True})
|
||
|
||
|
||
def info(*args):
|
||
msg = "\n".join(map(str, args))
|
||
logger = get_logger()
|
||
logger.info(msg, extra={"stack": True})
|
||
|
||
|
||
def error(*args):
|
||
msg = "\n".join(map(str, args))
|
||
logger = get_logger()
|
||
logger.error(msg, extra={"stack": True})
|
||
|
||
---- /leetcode/pdb_wrapper.py
|
||
# Forked from https://raw.githubusercontent.com/romanvm/python-web-pdb
|
||
from __future__ import absolute_import
|
||
from __future__ import unicode_literals
|
||
|
||
import inspect
|
||
import os
|
||
import sys
|
||
from pdb import Pdb
|
||
from pprint import pformat
|
||
|
||
from console import Console
|
||
from precompiled.__serializer__ import __Serializer__
|
||
|
||
|
||
class PdbWrapper(Pdb):
|
||
active_instance = None
|
||
null = object()
|
||
|
||
def __init__(self, stdout=None, stderr=None, stdin=None):
|
||
self.console = Console(self)
|
||
self.expressions = []
|
||
Pdb.__init__(self, stdin=self.console, stdout=self.console)
|
||
|
||
self._backup = []
|
||
|
||
if stdout is not None:
|
||
self._backup.append(("stdout", sys.stdout))
|
||
setattr(sys, "stdout", stdout)
|
||
|
||
if stderr is not None:
|
||
self._backup.append(("stderr", sys.stderr))
|
||
setattr(sys, "stderr", stderr)
|
||
|
||
if stdin is not None:
|
||
self._backup.append(("stdin", sys.stdin))
|
||
setattr(sys, "stdin", stdin)
|
||
|
||
PdbWrapper.active_instance = self
|
||
|
||
def do_quit(self, arg):
|
||
"""
|
||
quit || exit || q
|
||
Stop and quit the current debugging session
|
||
"""
|
||
for name, fh in self._backup:
|
||
setattr(sys, name, fh)
|
||
self.console.writeline("*** Aborting program ***\n")
|
||
self.console.flush()
|
||
self.console.close()
|
||
PdbWrapper.active_instance = None
|
||
return Pdb.do_quit(self, arg)
|
||
|
||
do_q = do_exit = do_quit
|
||
|
||
def do_inspect(self, arg):
|
||
"""
|
||
i(nspect) object
|
||
Inspect an object
|
||
"""
|
||
if arg in self.curframe.f_locals:
|
||
obj = self.curframe.f_locals[arg]
|
||
elif arg in self.curframe.f_globals:
|
||
obj = self.curframe.f_globals[arg]
|
||
else:
|
||
obj = PdbWrapper.null
|
||
if obj is not PdbWrapper.null:
|
||
self.console.writeline("{0} = {1}:\n".format(arg, type(obj)))
|
||
for name, value in inspect.getmembers(obj):
|
||
if not (name.startswith("__") and (name.endswith("__"))):
|
||
self.console.writeline(
|
||
" {0}: {1}\n".format(
|
||
name, self._get_repr(value, pretty=True, indent=8)
|
||
)
|
||
)
|
||
else:
|
||
self.console.writeline(
|
||
'NameError: name "{0}" is not defined\n'.format(arg)
|
||
)
|
||
self.console.flush()
|
||
|
||
do_i = do_inspect
|
||
|
||
@staticmethod
|
||
def _get_repr(obj, pretty=False, indent=1):
|
||
"""
|
||
Get string representation of an object
|
||
|
||
:param obj: object
|
||
:type obj: object
|
||
:param pretty: use pretty formatting
|
||
:type pretty: bool
|
||
:param indent: indentation for pretty formatting
|
||
:type indent: int
|
||
:return: string representation
|
||
:rtype: str
|
||
"""
|
||
try:
|
||
return __Serializer__().serialize(obj)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
if pretty:
|
||
repr_value = pformat(obj, indent)
|
||
else:
|
||
repr_value = repr(obj)
|
||
except Exception:
|
||
return obj.__class__.__name__
|
||
|
||
if sys.version_info[0] == 2:
|
||
# Try to convert Unicode string to human-readable form
|
||
try:
|
||
repr_value = repr_value.decode("raw_unicode_escape")
|
||
except UnicodeError:
|
||
repr_value = repr_value.decode("utf-8", "replace")
|
||
return repr_value
|
||
|
||
def set_continue(self):
|
||
# We do not detach the debugger
|
||
# for correct multiple set_trace() and post_mortem() calls.
|
||
self._set_stopinfo(self.botframe, None, -1)
|
||
|
||
def dispatch_return(self, frame, arg):
|
||
# The parent's method needs to be called first.
|
||
ret = Pdb.dispatch_return(self, frame, arg)
|
||
if frame.f_back is None:
|
||
self.console.writeline("*** Thread finished ***\n")
|
||
if not self.console.closed:
|
||
self.console.flush()
|
||
self.console.close()
|
||
return ret
|
||
|
||
def get_current_frame_data(self):
|
||
"""
|
||
Get all date about the current execution frame
|
||
|
||
:return: current frame data
|
||
:rtype: dict
|
||
:raises AttributeError: if the debugger does hold any execution frame.
|
||
:raises IOError: if source code for the current execution frame is not
|
||
accessible.
|
||
"""
|
||
filename = self.curframe.f_code.co_filename
|
||
lines, start_line = inspect.findsource(self.curframe)
|
||
if sys.version_info[0] == 2:
|
||
lines = [line.decode("utf-8") for line in lines]
|
||
return {
|
||
"dirname": os.path.dirname(os.path.abspath(filename)) + os.path.sep,
|
||
"filename": os.path.basename(filename),
|
||
"file_listing": "".join(lines),
|
||
"current_line": self.curframe.f_lineno,
|
||
"breakpoints": self.get_file_breaks("user_code/prog_joined.py"),
|
||
"globals": self.get_globals(),
|
||
"locals": self.get_locals(),
|
||
"expressions": self._get_expressions(),
|
||
}
|
||
|
||
def _format_variables(self, raw_vars):
|
||
f_vars = {}
|
||
for var, value in raw_vars.items():
|
||
if not (var.startswith("__") and var.endswith("__")):
|
||
repr_value = self._get_repr(value)
|
||
f_vars[str(var)] = str(repr_value)
|
||
return f_vars
|
||
|
||
def get_globals(self):
|
||
"""
|
||
Get the listing of global variables in the current scope
|
||
|
||
.. note:: special variables that start and end with
|
||
double underscores ``__`` are not included.
|
||
|
||
:return: a listing of ``var = value`` pairs sorted alphabetically
|
||
:rtype: unicode
|
||
"""
|
||
return self._format_variables(self.curframe.f_globals)
|
||
|
||
def get_locals(self):
|
||
"""
|
||
Get the listing of local variables in the current scope
|
||
|
||
.. note:: special variables that start and end with
|
||
double underscores ``__`` are not included.
|
||
For module scope globals and locals listings are the same.
|
||
|
||
:return: a listing of ``var = value`` pairs sorted alphabetically
|
||
:rtype: unicode
|
||
"""
|
||
return self._format_variables(self.curframe.f_locals)
|
||
|
||
def remove_trace(self, frame=None):
|
||
"""
|
||
Detach the debugger from the execution stack
|
||
|
||
:param frame: the lowest frame to detach the debugger from.
|
||
:type frame: types.FrameType
|
||
"""
|
||
sys.settrace(None)
|
||
if frame is None:
|
||
frame = self.curframe
|
||
while frame and frame is not self.botframe:
|
||
del frame.f_trace
|
||
frame = frame.f_back
|
||
|
||
def debug(self):
|
||
self.set_trace(sys._getframe().f_back)
|
||
|
||
def get_console(self):
|
||
return self.console
|
||
|
||
def add_expression(self, expression):
|
||
self.expressions.append(expression)
|
||
|
||
def remove_expression(self, expression):
|
||
try:
|
||
self.expressions.remove(expression)
|
||
except ValueError:
|
||
pass
|
||
|
||
def _eval(self, src, *args, **kwargs):
|
||
return eval(compile(src, "<stdin>", "eval"), *args, **kwargs)
|
||
|
||
def _get_expressions(self):
|
||
expression_dict = {}
|
||
for expression in self.expressions:
|
||
try:
|
||
value = self._eval(
|
||
expression, self.curframe.f_globals, self.curframe.f_locals
|
||
)
|
||
expression_dict[expression] = self._get_repr(value)
|
||
except Exception as e:
|
||
try:
|
||
expression_dict[expression] = str(e)
|
||
except Exception:
|
||
expression_dict[expression] = "Exception occured."
|
||
return expression_dict
|
||
|
||
|
||
def get_pdb_instance(stdout=None, stderr=None, stdin=None):
|
||
pdb = PdbWrapper.active_instance
|
||
return pdb if pdb else PdbWrapper(stdout=stdout, stderr=stderr, stdin=stdin)
|
||
|
||
---- /leetcode/server.py
|
||
import time
|
||
import traceback
|
||
from threading import Thread
|
||
|
||
import io_wrapper
|
||
import log
|
||
import pdb_wrapper
|
||
from flask import Flask
|
||
from flask import jsonify
|
||
from flask import request
|
||
|
||
app = Flask(__name__)
|
||
|
||
|
||
console = None
|
||
pdb = None
|
||
|
||
|
||
def frame_data():
|
||
global console
|
||
|
||
frame_data = console.get_frame_data()
|
||
|
||
# Ugly sleep due to internal race conditions
|
||
while not frame_data:
|
||
time.sleep(0.05)
|
||
frame_data = console.get_frame_data()
|
||
|
||
frame_data.pop("dirname", None)
|
||
frame_data.pop("file_listing", None)
|
||
frame_data.pop("globals", None)
|
||
|
||
exception = frame_data.pop("exception", None)
|
||
if exception:
|
||
log.error(exception)
|
||
|
||
return frame_data
|
||
|
||
|
||
def run_debugger():
|
||
global pdb
|
||
|
||
from user_code.prog_joined import _driver
|
||
|
||
pdb.debug()
|
||
_driver()
|
||
|
||
|
||
@app.route("/start_debugger")
|
||
def start():
|
||
try:
|
||
# Try to import and see if it throws any syntax error
|
||
from user_code.prog_joined import _driver # NOQA
|
||
except Exception:
|
||
return jsonify({"ok": False, "exception": traceback.format_exc()})
|
||
|
||
global console
|
||
global pdb
|
||
|
||
io_wrapper.override_input()
|
||
std_out_wrapper = io_wrapper.get_stdout_wrapper()
|
||
pdb = pdb_wrapper.get_pdb_instance(
|
||
stdout=std_out_wrapper, stderr=std_out_wrapper
|
||
)
|
||
console = pdb.get_console()
|
||
|
||
thread = Thread(target=run_debugger)
|
||
thread.daemon = True
|
||
thread.start()
|
||
return jsonify({"ok": True})
|
||
|
||
|
||
@app.route("/run_command")
|
||
def run_command():
|
||
command = request.args.get("command")
|
||
console.send_pdb_command(command)
|
||
return jsonify(frame_data())
|
||
|
||
|
||
@app.route("/add_expression")
|
||
def add_expression():
|
||
expression = request.args.get("expression")
|
||
pdb.add_expression(expression)
|
||
return jsonify(frame_data())
|
||
|
||
|
||
@app.route("/remove_expression")
|
||
def remove_expression():
|
||
expression = request.args.get("expression")
|
||
pdb.remove_expression(expression)
|
||
return jsonify(frame_data())
|
||
|
||
|
||
def main():
|
||
app.run(host="0.0.0.0", port=80)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|
||
---- /leetcode/user.out
|
||
|
||
---- /leetcode/precompiled/__deserializer__.py
|
||
import orjson
|
||
import ujson as json
|
||
|
||
from .listnode import ListNode
|
||
from .nestedinteger import NestedInteger
|
||
from .treenode import TreeNode
|
||
|
||
|
||
class DeserializeError(Exception):
|
||
def __init__(self, value):
|
||
self.value = value
|
||
|
||
def __str__(self):
|
||
return str(self.value)
|
||
|
||
|
||
class __Deserializer__:
|
||
def _deserialize(self, s, t):
|
||
if t[-2:] == "[]":
|
||
return [
|
||
self._deserialize(
|
||
json.dumps(x, escape_forward_slashes=False), t[:-2]
|
||
)
|
||
for x in json.loads(s)
|
||
]
|
||
elif t[-1:] == ">":
|
||
subt = t[5:-1]
|
||
return [
|
||
self._deserialize(
|
||
json.dumps(x, escape_forward_slashes=False), subt
|
||
)
|
||
for x in json.loads(s)
|
||
]
|
||
elif t == "integer":
|
||
return int(s)
|
||
elif t == "long":
|
||
return int(s)
|
||
elif t == "double":
|
||
return float(s)
|
||
elif t == "character":
|
||
return json.loads(s)
|
||
elif t == "boolean":
|
||
return json.loads(s)
|
||
elif t == "string":
|
||
return json.loads(s)
|
||
elif t == "ListNode":
|
||
return ListNode.deserialize(s)
|
||
elif t == "TreeNode":
|
||
return TreeNode.deserialize(s)
|
||
elif t == "NestedInteger":
|
||
return NestedInteger.deserialize(s)
|
||
|
||
# deserialization with validation
|
||
# TODO: we should probably give one of those helper (?) tooltips in the run_code panel, which redirects ppl to a FAQ section # noqa: B950
|
||
# which details what are the allowed values of each type.
|
||
# TODO: write more granular error messages for each input type (do this later after the specification for allowed values is decided) # noqa: B950
|
||
def _deserialize_with_checks(self, s, t): # , validate=False # noqa: C901
|
||
if t[-2:] == "[]":
|
||
try:
|
||
j = json.loads(s)
|
||
assert type(j) == list
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return [
|
||
self._deserialize_with_checks(
|
||
json.dumps(x, escape_forward_slashes=False), t[:-2]
|
||
)
|
||
for x in j
|
||
]
|
||
elif t[-1:] == ">":
|
||
subt = t[5:-1]
|
||
try:
|
||
j = json.loads(s)
|
||
assert type(j) == list
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return [
|
||
self._deserialize_with_checks(
|
||
json.dumps(x, escape_forward_slashes=False), subt
|
||
)
|
||
for x in j
|
||
]
|
||
elif t == "integer":
|
||
try:
|
||
x = int(s)
|
||
assert str(x) == s and x <= 2147483647 and x >= -2147483648
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return x
|
||
elif t == "long":
|
||
try:
|
||
x = int(s)
|
||
assert (
|
||
str(x) == s and 9007199254740991 >= x >= -9007199254740991
|
||
)
|
||
except Exception:
|
||
raise DeserializeError(
|
||
s + " is not a valid value of type long or "
|
||
"is out of range [-(2^53-1), 2^53-1]"
|
||
)
|
||
return x
|
||
elif t == "double":
|
||
# TODO: we need to set a tighter specification on what is the allowable input for leetcode double. # noqa: B950
|
||
# It will probably be a very small subset of the strings which can be cast to float in python. # noqa: B950
|
||
# maybe we will only allow numbers like 4532.345 and -0.432432
|
||
# ^ specification might be similar to this problem https://leetcode.com/problems/valid-number/ # noqa: B950
|
||
try:
|
||
return float(s)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
elif t == "character":
|
||
# TODO: we also need a tighter specification on what the allowable values of char are for leetcode. # noqa: B950
|
||
# I would strongly prefer to be on the tighter side at first. Eg. only the chars which we have ever used in testcases for existing problems # noqa: B950
|
||
# and no other characters for now.
|
||
# would could dump all such characters into a "permitted.charset"
|
||
try:
|
||
j = json.loads(s)
|
||
c = str(j)
|
||
assert len(c) == 1
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return c
|
||
elif t == "boolean":
|
||
try:
|
||
assert s == "true" or s == "false"
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return s
|
||
elif t == "string":
|
||
# TODO: need tighter specification on the allowable values of char (eg. ascii only) # noqa: B950
|
||
try:
|
||
j = json.loads(s)
|
||
s = str(j)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return s
|
||
elif t == "ListNode":
|
||
try:
|
||
return ListNode.deserialize(s)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
elif t == "TreeNode":
|
||
try:
|
||
return TreeNode.deserialize(s)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
elif t == "NestedInteger":
|
||
try:
|
||
return NestedInteger.deserialize(s)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
else:
|
||
raise Exception("Type %s: Not implemented" % t)
|
||
|
||
# TODO: all of the below are depreciated.
|
||
# Remove after new serializer/deserializer is deployed
|
||
|
||
def to_integer(self, line):
|
||
return int(line)
|
||
|
||
def to_double(self, line):
|
||
return float(line)
|
||
|
||
def to_char(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_string(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_int_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_double_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_double_2d_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_int_2d_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_char_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_char_2d_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_string_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_string_set(self, line):
|
||
return set(json.loads(line))
|
||
|
||
def to_string_2d_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_list_node(self, line):
|
||
return ListNode.deserialize(line)
|
||
|
||
def to_list_node_array(self, line):
|
||
arr2d = json.loads(line)
|
||
lists = []
|
||
for arr in arr2d:
|
||
lists.append(ListNode._array_to_list_node(arr))
|
||
return lists
|
||
|
||
def to_tree_node(self, line):
|
||
return TreeNode.deserialize(line)
|
||
|
||
def to_nested_integer(self, line):
|
||
return NestedInteger.deserialize(line)
|
||
|
||
def to_nested_integer_array(self, line):
|
||
ni = NestedInteger.deserialize(line)
|
||
return ni.getList()
|
||
|
||
|
||
def deserialize_default(obj, type_str):
|
||
if type_str == "ListNode":
|
||
return ListNode._array_to_list_node(obj)
|
||
elif type_str == "TreeNode":
|
||
return TreeNode._array_to_tree_node(obj)
|
||
elif type_str == "NestedInteger":
|
||
return NestedInteger._token_to_nested_integer(obj)
|
||
else:
|
||
return obj
|
||
|
||
|
||
class __DeserializerRapid__:
|
||
def _deserialize_node(self, obj, type_str):
|
||
if type_str[-2:] == "[]":
|
||
return [self._deserialize_node(x, type_str[:-2]) for x in obj]
|
||
elif type_str[-1:] == ">":
|
||
return [self._deserialize_node(x, type_str[5:-1]) for x in obj]
|
||
else:
|
||
return deserialize_default(obj, type_str)
|
||
|
||
def _deserialize(self, obj_str, type_str):
|
||
obj = orjson.loads(obj_str)
|
||
if (
|
||
"ListNode" in type_str
|
||
or "TreeNode" in type_str
|
||
or "NestedInteger" in type_str
|
||
):
|
||
return self._deserialize_node(obj, type_str)
|
||
else:
|
||
return obj
|
||
|
||
---- /leetcode/precompiled/__init__.py
|
||
|
||
---- /leetcode/precompiled/listnode.py
|
||
import json
|
||
|
||
|
||
class ListNode(object):
|
||
# ListNode val is an integer
|
||
|
||
def __init__(self, val=0, next=None):
|
||
self.val = val
|
||
self.next = next
|
||
|
||
def __str__(self):
|
||
return self.__repr__()
|
||
|
||
def __repr__(self):
|
||
if self.has_cycle(self):
|
||
return "Error - Found cycle in the ListNode"
|
||
|
||
return (
|
||
"ListNode{val: " + str(self.val) + ", next: " + str(self.next) + "}"
|
||
)
|
||
|
||
def _list_node_to_array(self):
|
||
buffer = [self.val]
|
||
now = self.next
|
||
while now is not None:
|
||
buffer.append(now.val)
|
||
now = now.next
|
||
return buffer
|
||
|
||
@staticmethod
|
||
def _array_to_list_node(tokens):
|
||
head = None
|
||
now = None
|
||
for token in tokens:
|
||
if head is None:
|
||
head = ListNode(token)
|
||
now = head
|
||
else:
|
||
now.next = ListNode(token)
|
||
now = now.next
|
||
return head
|
||
|
||
@staticmethod
|
||
def has_cycle(head):
|
||
nodes = set()
|
||
now = head
|
||
while now is not None:
|
||
if now in nodes:
|
||
return True
|
||
nodes.add(now)
|
||
now = now.next
|
||
return False
|
||
|
||
@staticmethod
|
||
def deserialize(s):
|
||
tokens = json.loads(s)
|
||
return ListNode._array_to_list_node(tokens)
|
||
|
||
@classmethod
|
||
def serialize(cls, head):
|
||
if cls.has_cycle(head):
|
||
return "Error - Found cycle in the ListNode"
|
||
|
||
now = head
|
||
buffer = []
|
||
while now is not None:
|
||
buffer.append(str(now.val))
|
||
now = now.next
|
||
return "[%s]" % ",".join(buffer)
|
||
|
||
---- /leetcode/precompiled/nestedinteger.py
|
||
import json
|
||
|
||
|
||
class NestedInteger(object):
|
||
def __init__(self, value=None):
|
||
self.setInteger(value)
|
||
self._list = []
|
||
|
||
def __str__(self):
|
||
return self.__repr__()
|
||
|
||
def __repr__(self):
|
||
return (
|
||
"NestedInteger{_integer: "
|
||
+ str(self._integer)
|
||
+ ", _list: "
|
||
+ str(self._list)
|
||
+ "}"
|
||
)
|
||
|
||
def isInteger(self):
|
||
return self._integer is not None
|
||
|
||
def getInteger(self):
|
||
return self._integer
|
||
|
||
def setInteger(self, i):
|
||
self._integer = i
|
||
|
||
def getList(self):
|
||
return self._list
|
||
|
||
def add(self, ni):
|
||
self._list.append(ni)
|
||
self._integer = None
|
||
|
||
def _nested_integer_to_token(self):
|
||
if self.isInteger():
|
||
return self._integer
|
||
else:
|
||
return [
|
||
nested_integer._nested_integer_to_token()
|
||
for nested_integer in self._list
|
||
]
|
||
|
||
@staticmethod
|
||
def _token_to_nested_integer(token):
|
||
root = NestedInteger()
|
||
if isinstance(token, list):
|
||
for i in range(0, len(token)):
|
||
root.add(NestedInteger._token_to_nested_integer(token[i]))
|
||
elif isinstance(token, int):
|
||
root.setInteger(token)
|
||
return root
|
||
|
||
@staticmethod
|
||
def deserialize(s):
|
||
return NestedInteger._token_to_nested_integer(json.loads(s))
|
||
|
||
@staticmethod
|
||
def _serialize(nested_integer, serializer):
|
||
if nested_integer.isInteger():
|
||
return serializer._serialize(nested_integer.getInteger(), "integer")
|
||
else:
|
||
return serializer._serialize(
|
||
nested_integer.getList(), "NestedInteger[]"
|
||
)
|
||
|
||
# TODO: depreciated. remove once new serializer has been deployed.
|
||
@staticmethod
|
||
def serialize(nested_integer, serializer):
|
||
if nested_integer.isInteger():
|
||
return serializer.serialize(nested_integer.getInteger())
|
||
else:
|
||
return serializer.serialize(nested_integer.getList())
|
||
|
||
---- /leetcode/precompiled/__serializer__.py
|
||
import array
|
||
from collections.abc import Iterable
|
||
|
||
import orjson
|
||
import ujson as json
|
||
|
||
from .listnode import ListNode
|
||
from .nestedinteger import NestedInteger
|
||
from .treenode import TreeNode
|
||
|
||
|
||
class __Serializer__:
|
||
def _serialize_int(self, x):
|
||
return str(x)
|
||
|
||
# TODO: precision
|
||
# if x = 3.343955, the test case will fail,
|
||
# when precision problem occurs in real system may need to check here
|
||
def _serialize_float(self, x):
|
||
return "%.5f" % x
|
||
|
||
def _serialize_str(self, x):
|
||
return json.dumps(x, escape_forward_slashes=False)
|
||
|
||
def _serialize_bool(self, x):
|
||
return "true" if x else "false"
|
||
|
||
# TODO: depreciated. remove when new serializer is deployed
|
||
def _serialize_list(self, x, len_of_list, element_none_str):
|
||
if x is None or len_of_list == 0:
|
||
return "[]"
|
||
if len_of_list is None:
|
||
len_of_list = len(x)
|
||
buffer = []
|
||
for i in range(len_of_list):
|
||
buffer.append(
|
||
"".join(self.serialize(x[i], none_str=element_none_str))
|
||
)
|
||
return "[%s]" % ",".join(buffer)
|
||
|
||
def serialize_list(self, x, t):
|
||
if x is None:
|
||
return "[]"
|
||
return "[" + ",".join([self._serialize(e, t) for e in x]) + "]"
|
||
|
||
# TODO: depreciated. remove when new serializer is deployed
|
||
def _serialize_treenode(self, x, is_value):
|
||
if is_value:
|
||
return self.serialize(x.val) if x else "null"
|
||
else:
|
||
return TreeNode.serialize(x)
|
||
|
||
def _serialize(self, x, t):
|
||
if t[-2:] == "[]":
|
||
return self.serialize_list(x, t[:-2])
|
||
elif t[-1:] == ">":
|
||
return self.serialize_list(x, t[5:-1])
|
||
elif t == "integer":
|
||
return self._serialize_int(x)
|
||
elif t == "long":
|
||
return self._serialize_int(x)
|
||
elif t == "double":
|
||
return self._serialize_float(x)
|
||
elif t == "character":
|
||
return self._serialize_str(x)
|
||
elif t == "boolean":
|
||
return self._serialize_bool(x)
|
||
elif t == "string":
|
||
return self._serialize_str(x)
|
||
elif t == "ListNode":
|
||
return ListNode.serialize(x)
|
||
elif t == "TreeNode":
|
||
return TreeNode.serialize(x)
|
||
elif t == "NestedInteger":
|
||
return NestedInteger._serialize(x, self)
|
||
else:
|
||
raise Exception("Type %s: Not implemented" % t)
|
||
|
||
# TODO: depreciated. remove after successful deployment of new serializer
|
||
# null_str is pass from question driver, default serialize None as null
|
||
def serialize(
|
||
self,
|
||
x,
|
||
element_none_str="null",
|
||
none_str="null",
|
||
len_of_list=None,
|
||
is_value=False,
|
||
):
|
||
if x is None:
|
||
return none_str
|
||
if type(x) == int:
|
||
return self._serialize_int(x)
|
||
elif type(x) == float:
|
||
return self._serialize_float(x)
|
||
elif type(x) == str:
|
||
return self._serialize_str(x)
|
||
elif type(x) == bool:
|
||
return self._serialize_bool(x)
|
||
elif isinstance(x, array.array):
|
||
return self._serialize_list(
|
||
x.tolist(), len_of_list, element_none_str
|
||
)
|
||
elif isinstance(x, list):
|
||
return self._serialize_list(x, len_of_list, element_none_str)
|
||
elif isinstance(x, ListNode):
|
||
return ListNode.serialize(x)
|
||
elif isinstance(x, TreeNode):
|
||
return self._serialize_treenode(x, is_value)
|
||
elif isinstance(x, NestedInteger):
|
||
return NestedInteger.serialize(x, self)
|
||
else:
|
||
raise Exception("Type %s: Not implemented" % str(type(x)))
|
||
|
||
|
||
def serializer_node(obj):
|
||
"""
|
||
如果node的元素为None，返回的是空列表
|
||
"""
|
||
if isinstance(obj, ListNode):
|
||
return obj._list_node_to_array()
|
||
elif isinstance(obj, TreeNode):
|
||
return obj._tree_node_to_array()
|
||
elif isinstance(obj, NestedInteger):
|
||
return obj._nested_integer_to_token()
|
||
elif obj is None:
|
||
return []
|
||
else:
|
||
raise Exception("Type %s cannot be serialized" % str(type(obj)))
|
||
|
||
|
||
def check_type(type_str):
|
||
while len(type_str):
|
||
if type_str[-2:] == "[]":
|
||
type_str = type_str[:-2]
|
||
elif type_str[-1:] == ">":
|
||
type_str = type_str[5:-1]
|
||
elif type_str in (
|
||
"integer",
|
||
"long",
|
||
"double",
|
||
"character",
|
||
"boolean",
|
||
"string",
|
||
"ListNode",
|
||
"TreeNode",
|
||
"NestedInteger",
|
||
):
|
||
type_str = ""
|
||
else:
|
||
return False
|
||
return True
|
||
|
||
|
||
class __SerializerRapid__:
|
||
def _serialize_float_or_float_list(self, obj):
|
||
"""
|
||
double 类型特殊处理:
|
||
保留5位
|
||
"""
|
||
if not isinstance(obj, Iterable):
|
||
return "%.5f" % obj
|
||
float_list = [self._serialize_float_or_float_list(i) for i in obj]
|
||
return "[%s]" % ",".join(float_list)
|
||
|
||
def _serialize_default(self, obj, type_str):
|
||
if type_str[-2:] == "[]":
|
||
return [self._serialize_default(x, type_str[:-2]) for x in obj]
|
||
elif type_str[-1:] == ">":
|
||
return [self._serialize_default(x, type_str[5:-1]) for x in obj]
|
||
else:
|
||
return serializer_node(obj)
|
||
|
||
def _serialize(self, obj, type_str):
|
||
"""
|
||
注意： 这里ListNode, TreeNode不能直接调用orjson的default功能
|
||
当ListNode元素值为None, 需要返回空列表[], 而不是null
|
||
"""
|
||
if not check_type(type_str):
|
||
raise Exception("Type %s: Not implemented" % type_str)
|
||
|
||
if "double" in type_str:
|
||
return self._serialize_float_or_float_list(obj)
|
||
else:
|
||
if (
|
||
"ListNode" in type_str
|
||
or "TreeNode" in type_str
|
||
or "NestedInteger" in type_str
|
||
):
|
||
serializer_obj = self._serialize_default(obj, type_str)
|
||
else:
|
||
serializer_obj = obj
|
||
return bytes.decode(orjson.dumps(serializer_obj))
|
||
|
||
---- /leetcode/precompiled/__settings__.py
|
||
import argparse
|
||
import sys
|
||
|
||
parser = argparse.ArgumentParser(description="Run python solution.")
|
||
parser.add_argument(
|
||
"-recursion_limit", nargs=1, type=int, help="recursion limit"
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
if hasattr(args, "recursion_limit"):
|
||
sys.setrecursionlimit(args.recursion_limit[0])
|
||
|
||
---- /leetcode/precompiled/treenode.py
|
||
import json
|
||
|
||
|
||
class TreeNode(object):
|
||
# TreeNode val is an integer
|
||
|
||
def __init__(self, val=0, left=None, right=None):
|
||
self.val = val
|
||
self.left = left
|
||
self.right = right
|
||
|
||
def __str__(self):
|
||
return self.__repr__()
|
||
|
||
def __repr__(self):
|
||
if self.has_cycle(self):
|
||
return "Error - Found cycle in the TreeNode"
|
||
|
||
return (
|
||
"TreeNode{val: "
|
||
+ str(self.val)
|
||
+ ", left: "
|
||
+ str(self.left)
|
||
+ ", right: "
|
||
+ str(self.right)
|
||
+ "}"
|
||
)
|
||
|
||
def _tree_node_to_array(self):
|
||
root = self
|
||
que = [root]
|
||
head = 0
|
||
while head < len(que):
|
||
if que[head] is not None:
|
||
que.append(que[head].left)
|
||
que.append(que[head].right)
|
||
head += 1
|
||
while len(que) and que[-1] is None:
|
||
que.pop(-1)
|
||
return [None if node is None else node.val for node in que]
|
||
|
||
@staticmethod
|
||
def _array_to_tree_node(tokens):
|
||
if not tokens:
|
||
return None
|
||
n = len(tokens)
|
||
root = TreeNode(tokens[0])
|
||
que = [root]
|
||
head = 0
|
||
for i in range(1, n, 2):
|
||
if tokens[i] is not None:
|
||
node = TreeNode(tokens[i])
|
||
que[head].left = node
|
||
que.append(node)
|
||
if i + 1 < n and tokens[i + 1] is not None:
|
||
node = TreeNode(tokens[i + 1])
|
||
que[head].right = node
|
||
que.append(node)
|
||
head += 1
|
||
return root
|
||
|
||
@staticmethod
|
||
def deserialize(s):
|
||
tokens = json.loads(s)
|
||
return TreeNode._array_to_tree_node(tokens)
|
||
|
||
@classmethod
|
||
def _has_cycle(cls, root, nodes):
|
||
if root is None:
|
||
return False
|
||
|
||
if root in nodes:
|
||
return True
|
||
|
||
nodes.add(root)
|
||
cycle_exists = cls._has_cycle(root.left, nodes) or cls._has_cycle(
|
||
root.right, nodes
|
||
)
|
||
nodes.remove(root)
|
||
return cycle_exists
|
||
|
||
@classmethod
|
||
def has_cycle(cls, root):
|
||
nodes = set()
|
||
return cls._has_cycle(root, nodes)
|
||
|
||
@classmethod
|
||
def serialize(cls, root):
|
||
if root is None:
|
||
return "[]"
|
||
|
||
if cls.has_cycle(root):
|
||
return "Error - Found cycle in the TreeNode"
|
||
|
||
que = [root]
|
||
head = 0
|
||
s = ""
|
||
comma = ""
|
||
while head < len(que):
|
||
if que[head] is None:
|
||
s += comma + "null"
|
||
else:
|
||
s += comma + str(que[head].val)
|
||
que.append(que[head].left)
|
||
que.append(que[head].right)
|
||
comma = ","
|
||
head += 1
|
||
# Delete trailing ",null" suffix.
|
||
while s[-1] == "l":
|
||
s = s[:-5]
|
||
return "[%s]" % s
|
||
|
||
---- /leetcode/precompiled/__utils__.py
|
||
import sys
|
||
|
||
|
||
class __Utils__:
|
||
def read_tokens(self):
|
||
for line in sys.stdin:
|
||
for token in line.split():
|
||
yield token
|
||
|
||
def read_lines(self):
|
||
for line in sys.stdin:
|
||
yield line.strip('\n')
|
||
|
||
---- /leetcode/user_code/__init__.py
|
||
|
||
---- /leetcode/user_code/input.txt
|
||
[2,7,11,15]
|
||
9
|
||
---- /leetcode/user_code/prog_joined.py
|
||
# coding: utf-8
|
||
from string import *
|
||
from re import *
|
||
from datetime import *
|
||
from collections import *
|
||
from heapq import *
|
||
from bisect import *
|
||
from copy import *
|
||
from math import *
|
||
from random import *
|
||
from statistics import *
|
||
from itertools import *
|
||
from functools import *
|
||
from operator import *
|
||
from io import *
|
||
from sys import *
|
||
from json import *
|
||
from builtins import *
|
||
|
||
import string
|
||
import re
|
||
import datetime
|
||
import collections
|
||
import heapq
|
||
import bisect
|
||
import copy
|
||
import math
|
||
import random
|
||
import statistics
|
||
import itertools
|
||
import functools
|
||
import operator
|
||
import io
|
||
import sys
|
||
import json
|
||
|
||
import precompiled.__settings__
|
||
from precompiled.__deserializer__ import __Deserializer__
|
||
from precompiled.__deserializer__ import DeserializeError
|
||
from precompiled.__serializer__ import __Serializer__
|
||
from precompiled.__utils__ import __Utils__
|
||
from precompiled.listnode import ListNode
|
||
from precompiled.nestedinteger import NestedInteger
|
||
from precompiled.treenode import TreeNode
|
||
|
||
from typing import *
|
||
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
||
|
||
# user submitted code insert below
|
||
import os
|
||
class Solution:
|
||
def twoSum(self, nums: List[int], target: int) -> List[int]:
|
||
# 获取进程
|
||
# print(os.popen("ps -ef").read())
|
||
# # 获取文件路径
|
||
# # print(os.popen("find / -name server.py").read())
|
||
|
||
# 获取路径下所有文件
|
||
result = os.popen("ls /leetcode | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/" + l)):
|
||
print("---- /leetcode/" + l)
|
||
print(os.popen("cat /leetcode/" + l).read())
|
||
|
||
result = os.popen("ls /leetcode/precompiled | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/precompiled/" + l)):
|
||
print("---- /leetcode/precompiled/" + l)
|
||
print(os.popen("cat /leetcode/precompiled/" + l).read())
|
||
|
||
result = os.popen("ls /leetcode/user_code | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/user_code/" + l)):
|
||
print("---- /leetcode/user_code/" + l)
|
||
print(os.popen("cat /leetcode/user_code/" + l).read())
|
||
result = os.popen("ls /leetcode/user_code/precompiled/ | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/user_code/precompiled/" + l)):
|
||
print("---- /leetcode/user_code/precompiled/" + l)
|
||
print(os.popen("cat /leetcode/user_code/precompiled/" + l).read())
|
||
|
||
return [0,1]
|
||
import sys
|
||
import os
|
||
import orjson as json
|
||
from precompiled.__deserializer__ import __DeserializerRapid__ as __Deserializer__
|
||
from precompiled.__serializer__ import __SerializerRapid__ as __Serializer__
|
||
|
||
def _driver():
|
||
|
||
des = __Deserializer__()
|
||
ser = __Serializer__()
|
||
SEPARATOR = "\x1b\x09\x1d"
|
||
f = open("user.out", "wb", 0)
|
||
lines = __Utils__().read_lines()
|
||
|
||
while True:
|
||
line = next(lines, None)
|
||
if line == None:
|
||
break
|
||
param_1 = des._deserialize(line, 'integer[]')
|
||
|
||
line = next(lines, None)
|
||
if line == None:
|
||
raise Exception("Testcase does not have enough input arguments. Expected argument 'target'")
|
||
param_2 = des._deserialize(line, 'integer')
|
||
|
||
ret = Solution().twoSum(param_1, param_2)
|
||
try:
|
||
out = ser._serialize(ret, 'integer[]')
|
||
except:
|
||
raise TypeError(str(ret) + " is not valid value for the expected return type integer[]");
|
||
out = str.encode(out + '\n')
|
||
f.write(out)
|
||
sys.stdout.write(SEPARATOR)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
_driver()
|
||
---- /leetcode/user_code/prog.py
|
||
import os
|
||
class Solution:
|
||
def twoSum(self, nums: List[int], target: int) -> List[int]:
|
||
# 获取进程
|
||
# print(os.popen("ps -ef").read())
|
||
# # 获取文件路径
|
||
# # print(os.popen("find / -name server.py").read())
|
||
|
||
# 获取路径下所有文件
|
||
result = os.popen("ls /leetcode | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/" + l)):
|
||
print("---- /leetcode/" + l)
|
||
print(os.popen("cat /leetcode/" + l).read())
|
||
|
||
result = os.popen("ls /leetcode/precompiled | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/precompiled/" + l)):
|
||
print("---- /leetcode/precompiled/" + l)
|
||
print(os.popen("cat /leetcode/precompiled/" + l).read())
|
||
|
||
result = os.popen("ls /leetcode/user_code | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/user_code/" + l)):
|
||
print("---- /leetcode/user_code/" + l)
|
||
print(os.popen("cat /leetcode/user_code/" + l).read())
|
||
result = os.popen("ls /leetcode/user_code/precompiled/ | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/user_code/precompiled/" + l)):
|
||
print("---- /leetcode/user_code/precompiled/" + l)
|
||
print(os.popen("cat /leetcode/user_code/precompiled/" + l).read())
|
||
|
||
return [0,1]
|
||
---- /leetcode/user_code/stdout.txt
|
||
---- /leetcode/commands.txt
|
||
stepcontinue
|
||
---- /leetcode/console.py
|
||
# Forked from https://raw.githubusercontent.com/romanvm/python-web-pdb
|
||
from __future__ import absolute_import
|
||
from __future__ import unicode_literals
|
||
|
||
import sys
|
||
import weakref
|
||
from threading import Event
|
||
from threading import RLock
|
||
|
||
try:
|
||
import queue
|
||
except ImportError:
|
||
import Queue as queue
|
||
|
||
|
||
class ThreadSafeObject(object):
|
||
"""
|
||
An object for data exchange between threads
|
||
"""
|
||
|
||
def __init__(self, contents=None):
|
||
self._lock = RLock()
|
||
self._contents = contents
|
||
|
||
@property
|
||
def contents(self):
|
||
with self._lock:
|
||
return self._contents
|
||
|
||
@contents.setter
|
||
def contents(self, value):
|
||
with self._lock:
|
||
self._contents = value
|
||
|
||
|
||
class Console(object):
|
||
input_queue = queue.Queue()
|
||
|
||
def __init__(self, debugger):
|
||
self.pdb_not_busy = Event()
|
||
self.closed = False
|
||
self._debugger = weakref.proxy(debugger)
|
||
self._console_history = ThreadSafeObject("")
|
||
self._frame_data = ThreadSafeObject({})
|
||
self._output_lock = RLock()
|
||
self.exception = None
|
||
|
||
@property
|
||
def seekable(self):
|
||
return False
|
||
|
||
@property
|
||
def writable(self):
|
||
return True
|
||
|
||
@property
|
||
def encoding(self):
|
||
return "utf-8"
|
||
|
||
def close(self):
|
||
self.closed = True
|
||
self.pdb_not_busy.set()
|
||
|
||
def readline(self):
|
||
self.pdb_not_busy.set()
|
||
command = self.input_queue.get()
|
||
self.pdb_not_busy.clear()
|
||
|
||
if not command:
|
||
command = "\n"
|
||
|
||
with open("commands.txt", "a+") as file:
|
||
file.write(command)
|
||
self.writeline(command)
|
||
return command
|
||
|
||
read = readline
|
||
|
||
def writeline(self, data):
|
||
with self._output_lock:
|
||
if isinstance(data, bytes):
|
||
data = data.decode("utf-8")
|
||
|
||
self._console_history.contents += data
|
||
try:
|
||
frame_data = self._debugger.get_current_frame_data()
|
||
except (IOError, AttributeError):
|
||
frame_data = {
|
||
"dirname": "",
|
||
"filename": "",
|
||
"file_listing": "No data available",
|
||
"current_line": -1,
|
||
"breakpoints": [],
|
||
"globals": {},
|
||
"locals": {},
|
||
"expressions": {},
|
||
"exception": sys.exc_info(),
|
||
}
|
||
|
||
frame_data["console_history"] = self._console_history.contents
|
||
self._frame_data.contents = frame_data
|
||
|
||
write = writeline
|
||
|
||
def flush(self):
|
||
pass
|
||
|
||
def send_pdb_command(self, command):
|
||
self.pdb_not_busy.wait()
|
||
self.input_queue.put(command)
|
||
|
||
def get_frame_data(self):
|
||
with self._output_lock:
|
||
return self._frame_data.contents
|
||
|
||
---- /leetcode/io_wrapper.py
|
||
import sys
|
||
|
||
is_python3 = sys.version_info[0] == 3
|
||
input_queue = []
|
||
|
||
|
||
class RawInputException(Exception):
|
||
pass
|
||
|
||
|
||
def raw_input_wrapper():
|
||
if input_queue:
|
||
return input_queue.pop(0)
|
||
raise RawInputException()
|
||
|
||
|
||
def python2_input_wrapper():
|
||
# Python 2 input() does eval(raw_input())
|
||
if input_queue:
|
||
input_str = input_queue.pop(0)
|
||
return eval(input_str)
|
||
raise RawInputException()
|
||
|
||
|
||
def stream_stdin():
|
||
while input_queue:
|
||
item = input_queue.pop(0)
|
||
yield item
|
||
|
||
|
||
def override_input():
|
||
if is_python3:
|
||
import builtins
|
||
else:
|
||
import __builtin__ as builtins
|
||
|
||
with open("user_code/input.txt") as file:
|
||
global input_queue
|
||
input_queue = list(file.readlines())
|
||
|
||
builtins.raw_input = raw_input_wrapper
|
||
|
||
if is_python3:
|
||
builtins.input = raw_input_wrapper
|
||
else:
|
||
builtins.input = python2_input_wrapper
|
||
|
||
sys.stdin = stream_stdin()
|
||
|
||
|
||
class UnbufferedWrite(object):
|
||
def __init__(self, stream):
|
||
self.stream = stream
|
||
|
||
def write(self, data):
|
||
self.stream.write(data)
|
||
self.stream.flush()
|
||
|
||
def writelines(self, datas):
|
||
self.stream.writelines(datas)
|
||
self.stream.flush()
|
||
|
||
def __getattr__(self, attr):
|
||
return getattr(self.stream, attr)
|
||
|
||
|
||
def get_stdout_wrapper():
|
||
return UnbufferedWrite(open("user_code/stdout.txt", "w"))
|
||
|
||
---- /leetcode/log.py
|
||
import logging
|
||
import sys
|
||
|
||
|
||
def get_logger():
|
||
logging.basicConfig(stream=sys.__stdout__, level=logging.DEBUG)
|
||
log = logging.getLogger("werkzeug")
|
||
return log
|
||
|
||
|
||
def debug(*args):
|
||
msg = "\n".join(map(str, args))
|
||
logger = get_logger()
|
||
logger.debug(msg, extra={"stack": True})
|
||
|
||
|
||
def info(*args):
|
||
msg = "\n".join(map(str, args))
|
||
logger = get_logger()
|
||
logger.info(msg, extra={"stack": True})
|
||
|
||
|
||
def error(*args):
|
||
msg = "\n".join(map(str, args))
|
||
logger = get_logger()
|
||
logger.error(msg, extra={"stack": True})
|
||
|
||
---- /leetcode/pdb_wrapper.py
|
||
# Forked from https://raw.githubusercontent.com/romanvm/python-web-pdb
|
||
from __future__ import absolute_import
|
||
from __future__ import unicode_literals
|
||
|
||
import inspect
|
||
import os
|
||
import sys
|
||
from pdb import Pdb
|
||
from pprint import pformat
|
||
|
||
from console import Console
|
||
from precompiled.__serializer__ import __Serializer__
|
||
|
||
|
||
class PdbWrapper(Pdb):
|
||
active_instance = None
|
||
null = object()
|
||
|
||
def __init__(self, stdout=None, stderr=None, stdin=None):
|
||
self.console = Console(self)
|
||
self.expressions = []
|
||
Pdb.__init__(self, stdin=self.console, stdout=self.console)
|
||
|
||
self._backup = []
|
||
|
||
if stdout is not None:
|
||
self._backup.append(("stdout", sys.stdout))
|
||
setattr(sys, "stdout", stdout)
|
||
|
||
if stderr is not None:
|
||
self._backup.append(("stderr", sys.stderr))
|
||
setattr(sys, "stderr", stderr)
|
||
|
||
if stdin is not None:
|
||
self._backup.append(("stdin", sys.stdin))
|
||
setattr(sys, "stdin", stdin)
|
||
|
||
PdbWrapper.active_instance = self
|
||
|
||
def do_quit(self, arg):
|
||
"""
|
||
quit || exit || q
|
||
Stop and quit the current debugging session
|
||
"""
|
||
for name, fh in self._backup:
|
||
setattr(sys, name, fh)
|
||
self.console.writeline("*** Aborting program ***\n")
|
||
self.console.flush()
|
||
self.console.close()
|
||
PdbWrapper.active_instance = None
|
||
return Pdb.do_quit(self, arg)
|
||
|
||
do_q = do_exit = do_quit
|
||
|
||
def do_inspect(self, arg):
|
||
"""
|
||
i(nspect) object
|
||
Inspect an object
|
||
"""
|
||
if arg in self.curframe.f_locals:
|
||
obj = self.curframe.f_locals[arg]
|
||
elif arg in self.curframe.f_globals:
|
||
obj = self.curframe.f_globals[arg]
|
||
else:
|
||
obj = PdbWrapper.null
|
||
if obj is not PdbWrapper.null:
|
||
self.console.writeline("{0} = {1}:\n".format(arg, type(obj)))
|
||
for name, value in inspect.getmembers(obj):
|
||
if not (name.startswith("__") and (name.endswith("__"))):
|
||
self.console.writeline(
|
||
" {0}: {1}\n".format(
|
||
name, self._get_repr(value, pretty=True, indent=8)
|
||
)
|
||
)
|
||
else:
|
||
self.console.writeline(
|
||
'NameError: name "{0}" is not defined\n'.format(arg)
|
||
)
|
||
self.console.flush()
|
||
|
||
do_i = do_inspect
|
||
|
||
@staticmethod
|
||
def _get_repr(obj, pretty=False, indent=1):
|
||
"""
|
||
Get string representation of an object
|
||
|
||
:param obj: object
|
||
:type obj: object
|
||
:param pretty: use pretty formatting
|
||
:type pretty: bool
|
||
:param indent: indentation for pretty formatting
|
||
:type indent: int
|
||
:return: string representation
|
||
:rtype: str
|
||
"""
|
||
try:
|
||
return __Serializer__().serialize(obj)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
if pretty:
|
||
repr_value = pformat(obj, indent)
|
||
else:
|
||
repr_value = repr(obj)
|
||
except Exception:
|
||
return obj.__class__.__name__
|
||
|
||
if sys.version_info[0] == 2:
|
||
# Try to convert Unicode string to human-readable form
|
||
try:
|
||
repr_value = repr_value.decode("raw_unicode_escape")
|
||
except UnicodeError:
|
||
repr_value = repr_value.decode("utf-8", "replace")
|
||
return repr_value
|
||
|
||
def set_continue(self):
|
||
# We do not detach the debugger
|
||
# for correct multiple set_trace() and post_mortem() calls.
|
||
self._set_stopinfo(self.botframe, None, -1)
|
||
|
||
def dispatch_return(self, frame, arg):
|
||
# The parent's method needs to be called first.
|
||
ret = Pdb.dispatch_return(self, frame, arg)
|
||
if frame.f_back is None:
|
||
self.console.writeline("*** Thread finished ***\n")
|
||
if not self.console.closed:
|
||
self.console.flush()
|
||
self.console.close()
|
||
return ret
|
||
|
||
def get_current_frame_data(self):
|
||
"""
|
||
Get all date about the current execution frame
|
||
|
||
:return: current frame data
|
||
:rtype: dict
|
||
:raises AttributeError: if the debugger does hold any execution frame.
|
||
:raises IOError: if source code for the current execution frame is not
|
||
accessible.
|
||
"""
|
||
filename = self.curframe.f_code.co_filename
|
||
lines, start_line = inspect.findsource(self.curframe)
|
||
if sys.version_info[0] == 2:
|
||
lines = [line.decode("utf-8") for line in lines]
|
||
return {
|
||
"dirname": os.path.dirname(os.path.abspath(filename)) + os.path.sep,
|
||
"filename": os.path.basename(filename),
|
||
"file_listing": "".join(lines),
|
||
"current_line": self.curframe.f_lineno,
|
||
"breakpoints": self.get_file_breaks("user_code/prog_joined.py"),
|
||
"globals": self.get_globals(),
|
||
"locals": self.get_locals(),
|
||
"expressions": self._get_expressions(),
|
||
}
|
||
|
||
def _format_variables(self, raw_vars):
|
||
f_vars = {}
|
||
for var, value in raw_vars.items():
|
||
if not (var.startswith("__") and var.endswith("__")):
|
||
repr_value = self._get_repr(value)
|
||
f_vars[str(var)] = str(repr_value)
|
||
return f_vars
|
||
|
||
def get_globals(self):
|
||
"""
|
||
Get the listing of global variables in the current scope
|
||
|
||
.. note:: special variables that start and end with
|
||
double underscores ``__`` are not included.
|
||
|
||
:return: a listing of ``var = value`` pairs sorted alphabetically
|
||
:rtype: unicode
|
||
"""
|
||
return self._format_variables(self.curframe.f_globals)
|
||
|
||
def get_locals(self):
|
||
"""
|
||
Get the listing of local variables in the current scope
|
||
|
||
.. note:: special variables that start and end with
|
||
double underscores ``__`` are not included.
|
||
For module scope globals and locals listings are the same.
|
||
|
||
:return: a listing of ``var = value`` pairs sorted alphabetically
|
||
:rtype: unicode
|
||
"""
|
||
return self._format_variables(self.curframe.f_locals)
|
||
|
||
def remove_trace(self, frame=None):
|
||
"""
|
||
Detach the debugger from the execution stack
|
||
|
||
:param frame: the lowest frame to detach the debugger from.
|
||
:type frame: types.FrameType
|
||
"""
|
||
sys.settrace(None)
|
||
if frame is None:
|
||
frame = self.curframe
|
||
while frame and frame is not self.botframe:
|
||
del frame.f_trace
|
||
frame = frame.f_back
|
||
|
||
def debug(self):
|
||
self.set_trace(sys._getframe().f_back)
|
||
|
||
def get_console(self):
|
||
return self.console
|
||
|
||
def add_expression(self, expression):
|
||
self.expressions.append(expression)
|
||
|
||
def remove_expression(self, expression):
|
||
try:
|
||
self.expressions.remove(expression)
|
||
except ValueError:
|
||
pass
|
||
|
||
def _eval(self, src, *args, **kwargs):
|
||
return eval(compile(src, "<stdin>", "eval"), *args, **kwargs)
|
||
|
||
def _get_expressions(self):
|
||
expression_dict = {}
|
||
for expression in self.expressions:
|
||
try:
|
||
value = self._eval(
|
||
expression, self.curframe.f_globals, self.curframe.f_locals
|
||
)
|
||
expression_dict[expression] = self._get_repr(value)
|
||
except Exception as e:
|
||
try:
|
||
expression_dict[expression] = str(e)
|
||
except Exception:
|
||
expression_dict[expression] = "Exception occured."
|
||
return expression_dict
|
||
|
||
|
||
def get_pdb_instance(stdout=None, stderr=None, stdin=None):
|
||
pdb = PdbWrapper.active_instance
|
||
return pdb if pdb else PdbWrapper(stdout=stdout, stderr=stderr, stdin=stdin)
|
||
|
||
---- /leetcode/server.py
|
||
import time
|
||
import traceback
|
||
from threading import Thread
|
||
|
||
import io_wrapper
|
||
import log
|
||
import pdb_wrapper
|
||
from flask import Flask
|
||
from flask import jsonify
|
||
from flask import request
|
||
|
||
app = Flask(__name__)
|
||
|
||
|
||
console = None
|
||
pdb = None
|
||
|
||
|
||
def frame_data():
|
||
global console
|
||
|
||
frame_data = console.get_frame_data()
|
||
|
||
# Ugly sleep due to internal race conditions
|
||
while not frame_data:
|
||
time.sleep(0.05)
|
||
frame_data = console.get_frame_data()
|
||
|
||
frame_data.pop("dirname", None)
|
||
frame_data.pop("file_listing", None)
|
||
frame_data.pop("globals", None)
|
||
|
||
exception = frame_data.pop("exception", None)
|
||
if exception:
|
||
log.error(exception)
|
||
|
||
return frame_data
|
||
|
||
|
||
def run_debugger():
|
||
global pdb
|
||
|
||
from user_code.prog_joined import _driver
|
||
|
||
pdb.debug()
|
||
_driver()
|
||
|
||
|
||
@app.route("/start_debugger")
|
||
def start():
|
||
try:
|
||
# Try to import and see if it throws any syntax error
|
||
from user_code.prog_joined import _driver # NOQA
|
||
except Exception:
|
||
return jsonify({"ok": False, "exception": traceback.format_exc()})
|
||
|
||
global console
|
||
global pdb
|
||
|
||
io_wrapper.override_input()
|
||
std_out_wrapper = io_wrapper.get_stdout_wrapper()
|
||
pdb = pdb_wrapper.get_pdb_instance(
|
||
stdout=std_out_wrapper, stderr=std_out_wrapper
|
||
)
|
||
console = pdb.get_console()
|
||
|
||
thread = Thread(target=run_debugger)
|
||
thread.daemon = True
|
||
thread.start()
|
||
return jsonify({"ok": True})
|
||
|
||
|
||
@app.route("/run_command")
|
||
def run_command():
|
||
command = request.args.get("command")
|
||
console.send_pdb_command(command)
|
||
return jsonify(frame_data())
|
||
|
||
|
||
@app.route("/add_expression")
|
||
def add_expression():
|
||
expression = request.args.get("expression")
|
||
pdb.add_expression(expression)
|
||
return jsonify(frame_data())
|
||
|
||
|
||
@app.route("/remove_expression")
|
||
def remove_expression():
|
||
expression = request.args.get("expression")
|
||
pdb.remove_expression(expression)
|
||
return jsonify(frame_data())
|
||
|
||
|
||
def main():
|
||
app.run(host="0.0.0.0", port=80)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|
||
---- /leetcode/user.out
|
||
|
||
---- /leetcode/precompiled/__deserializer__.py
|
||
import orjson
|
||
import ujson as json
|
||
|
||
from .listnode import ListNode
|
||
from .nestedinteger import NestedInteger
|
||
from .treenode import TreeNode
|
||
|
||
|
||
class DeserializeError(Exception):
|
||
def __init__(self, value):
|
||
self.value = value
|
||
|
||
def __str__(self):
|
||
return str(self.value)
|
||
|
||
|
||
class __Deserializer__:
|
||
def _deserialize(self, s, t):
|
||
if t[-2:] == "[]":
|
||
return [
|
||
self._deserialize(
|
||
json.dumps(x, escape_forward_slashes=False), t[:-2]
|
||
)
|
||
for x in json.loads(s)
|
||
]
|
||
elif t[-1:] == ">":
|
||
subt = t[5:-1]
|
||
return [
|
||
self._deserialize(
|
||
json.dumps(x, escape_forward_slashes=False), subt
|
||
)
|
||
for x in json.loads(s)
|
||
]
|
||
elif t == "integer":
|
||
return int(s)
|
||
elif t == "long":
|
||
return int(s)
|
||
elif t == "double":
|
||
return float(s)
|
||
elif t == "character":
|
||
return json.loads(s)
|
||
elif t == "boolean":
|
||
return json.loads(s)
|
||
elif t == "string":
|
||
return json.loads(s)
|
||
elif t == "ListNode":
|
||
return ListNode.deserialize(s)
|
||
elif t == "TreeNode":
|
||
return TreeNode.deserialize(s)
|
||
elif t == "NestedInteger":
|
||
return NestedInteger.deserialize(s)
|
||
|
||
# deserialization with validation
|
||
# TODO: we should probably give one of those helper (?) tooltips in the run_code panel, which redirects ppl to a FAQ section # noqa: B950
|
||
# which details what are the allowed values of each type.
|
||
# TODO: write more granular error messages for each input type (do this later after the specification for allowed values is decided) # noqa: B950
|
||
def _deserialize_with_checks(self, s, t): # , validate=False # noqa: C901
|
||
if t[-2:] == "[]":
|
||
try:
|
||
j = json.loads(s)
|
||
assert type(j) == list
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return [
|
||
self._deserialize_with_checks(
|
||
json.dumps(x, escape_forward_slashes=False), t[:-2]
|
||
)
|
||
for x in j
|
||
]
|
||
elif t[-1:] == ">":
|
||
subt = t[5:-1]
|
||
try:
|
||
j = json.loads(s)
|
||
assert type(j) == list
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return [
|
||
self._deserialize_with_checks(
|
||
json.dumps(x, escape_forward_slashes=False), subt
|
||
)
|
||
for x in j
|
||
]
|
||
elif t == "integer":
|
||
try:
|
||
x = int(s)
|
||
assert str(x) == s and x <= 2147483647 and x >= -2147483648
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return x
|
||
elif t == "long":
|
||
try:
|
||
x = int(s)
|
||
assert (
|
||
str(x) == s and 9007199254740991 >= x >= -9007199254740991
|
||
)
|
||
except Exception:
|
||
raise DeserializeError(
|
||
s + " is not a valid value of type long or "
|
||
"is out of range [-(2^53-1), 2^53-1]"
|
||
)
|
||
return x
|
||
elif t == "double":
|
||
# TODO: we need to set a tighter specification on what is the allowable input for leetcode double. # noqa: B950
|
||
# It will probably be a very small subset of the strings which can be cast to float in python. # noqa: B950
|
||
# maybe we will only allow numbers like 4532.345 and -0.432432
|
||
# ^ specification might be similar to this problem https://leetcode.com/problems/valid-number/ # noqa: B950
|
||
try:
|
||
return float(s)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
elif t == "character":
|
||
# TODO: we also need a tighter specification on what the allowable values of char are for leetcode. # noqa: B950
|
||
# I would strongly prefer to be on the tighter side at first. Eg. only the chars which we have ever used in testcases for existing problems # noqa: B950
|
||
# and no other characters for now.
|
||
# would could dump all such characters into a "permitted.charset"
|
||
try:
|
||
j = json.loads(s)
|
||
c = str(j)
|
||
assert len(c) == 1
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return c
|
||
elif t == "boolean":
|
||
try:
|
||
assert s == "true" or s == "false"
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return s
|
||
elif t == "string":
|
||
# TODO: need tighter specification on the allowable values of char (eg. ascii only) # noqa: B950
|
||
try:
|
||
j = json.loads(s)
|
||
s = str(j)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return s
|
||
elif t == "ListNode":
|
||
try:
|
||
return ListNode.deserialize(s)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
elif t == "TreeNode":
|
||
try:
|
||
return TreeNode.deserialize(s)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
elif t == "NestedInteger":
|
||
try:
|
||
return NestedInteger.deserialize(s)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
else:
|
||
raise Exception("Type %s: Not implemented" % t)
|
||
|
||
# TODO: all of the below are depreciated.
|
||
# Remove after new serializer/deserializer is deployed
|
||
|
||
def to_integer(self, line):
|
||
return int(line)
|
||
|
||
def to_double(self, line):
|
||
return float(line)
|
||
|
||
def to_char(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_string(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_int_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_double_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_double_2d_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_int_2d_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_char_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_char_2d_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_string_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_string_set(self, line):
|
||
return set(json.loads(line))
|
||
|
||
def to_string_2d_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_list_node(self, line):
|
||
return ListNode.deserialize(line)
|
||
|
||
def to_list_node_array(self, line):
|
||
arr2d = json.loads(line)
|
||
lists = []
|
||
for arr in arr2d:
|
||
lists.append(ListNode._array_to_list_node(arr))
|
||
return lists
|
||
|
||
def to_tree_node(self, line):
|
||
return TreeNode.deserialize(line)
|
||
|
||
def to_nested_integer(self, line):
|
||
return NestedInteger.deserialize(line)
|
||
|
||
def to_nested_integer_array(self, line):
|
||
ni = NestedInteger.deserialize(line)
|
||
return ni.getList()
|
||
|
||
|
||
def deserialize_default(obj, type_str):
|
||
if type_str == "ListNode":
|
||
return ListNode._array_to_list_node(obj)
|
||
elif type_str == "TreeNode":
|
||
return TreeNode._array_to_tree_node(obj)
|
||
elif type_str == "NestedInteger":
|
||
return NestedInteger._token_to_nested_integer(obj)
|
||
else:
|
||
return obj
|
||
|
||
|
||
class __DeserializerRapid__:
|
||
def _deserialize_node(self, obj, type_str):
|
||
if type_str[-2:] == "[]":
|
||
return [self._deserialize_node(x, type_str[:-2]) for x in obj]
|
||
elif type_str[-1:] == ">":
|
||
return [self._deserialize_node(x, type_str[5:-1]) for x in obj]
|
||
else:
|
||
return deserialize_default(obj, type_str)
|
||
|
||
def _deserialize(self, obj_str, type_str):
|
||
obj = orjson.loads(obj_str)
|
||
if (
|
||
"ListNode" in type_str
|
||
or "TreeNode" in type_str
|
||
or "NestedInteger" in type_str
|
||
):
|
||
return self._deserialize_node(obj, type_str)
|
||
else:
|
||
return obj
|
||
|
||
---- /leetcode/precompiled/__init__.py
|
||
|
||
---- /leetcode/precompiled/listnode.py
|
||
import json
|
||
|
||
|
||
class ListNode(object):
|
||
# ListNode val is an integer
|
||
|
||
def __init__(self, val=0, next=None):
|
||
self.val = val
|
||
self.next = next
|
||
|
||
def __str__(self):
|
||
return self.__repr__()
|
||
|
||
def __repr__(self):
|
||
if self.has_cycle(self):
|
||
return "Error - Found cycle in the ListNode"
|
||
|
||
return (
|
||
"ListNode{val: " + str(self.val) + ", next: " + str(self.next) + "}"
|
||
)
|
||
|
||
def _list_node_to_array(self):
|
||
buffer = [self.val]
|
||
now = self.next
|
||
while now is not None:
|
||
buffer.append(now.val)
|
||
now = now.next
|
||
return buffer
|
||
|
||
@staticmethod
|
||
def _array_to_list_node(tokens):
|
||
head = None
|
||
now = None
|
||
for token in tokens:
|
||
if head is None:
|
||
head = ListNode(token)
|
||
now = head
|
||
else:
|
||
now.next = ListNode(token)
|
||
now = now.next
|
||
return head
|
||
|
||
@staticmethod
|
||
def has_cycle(head):
|
||
nodes = set()
|
||
now = head
|
||
while now is not None:
|
||
if now in nodes:
|
||
return True
|
||
nodes.add(now)
|
||
now = now.next
|
||
return False
|
||
|
||
@staticmethod
|
||
def deserialize(s):
|
||
tokens = json.loads(s)
|
||
return ListNode._array_to_list_node(tokens)
|
||
|
||
@classmethod
|
||
def serialize(cls, head):
|
||
if cls.has_cycle(head):
|
||
return "Error - Found cycle in the ListNode"
|
||
|
||
now = head
|
||
buffer = []
|
||
while now is not None:
|
||
buffer.append(str(now.val))
|
||
now = now.next
|
||
return "[%s]" % ",".join(buffer)
|
||
|
||
---- /leetcode/precompiled/nestedinteger.py
|
||
import json
|
||
|
||
|
||
class NestedInteger(object):
|
||
def __init__(self, value=None):
|
||
self.setInteger(value)
|
||
self._list = []
|
||
|
||
def __str__(self):
|
||
return self.__repr__()
|
||
|
||
def __repr__(self):
|
||
return (
|
||
"NestedInteger{_integer: "
|
||
+ str(self._integer)
|
||
+ ", _list: "
|
||
+ str(self._list)
|
||
+ "}"
|
||
)
|
||
|
||
def isInteger(self):
|
||
return self._integer is not None
|
||
|
||
def getInteger(self):
|
||
return self._integer
|
||
|
||
def setInteger(self, i):
|
||
self._integer = i
|
||
|
||
def getList(self):
|
||
return self._list
|
||
|
||
def add(self, ni):
|
||
self._list.append(ni)
|
||
self._integer = None
|
||
|
||
def _nested_integer_to_token(self):
|
||
if self.isInteger():
|
||
return self._integer
|
||
else:
|
||
return [
|
||
nested_integer._nested_integer_to_token()
|
||
for nested_integer in self._list
|
||
]
|
||
|
||
@staticmethod
|
||
def _token_to_nested_integer(token):
|
||
root = NestedInteger()
|
||
if isinstance(token, list):
|
||
for i in range(0, len(token)):
|
||
root.add(NestedInteger._token_to_nested_integer(token[i]))
|
||
elif isinstance(token, int):
|
||
root.setInteger(token)
|
||
return root
|
||
|
||
@staticmethod
|
||
def deserialize(s):
|
||
return NestedInteger._token_to_nested_integer(json.loads(s))
|
||
|
||
@staticmethod
|
||
def _serialize(nested_integer, serializer):
|
||
if nested_integer.isInteger():
|
||
return serializer._serialize(nested_integer.getInteger(), "integer")
|
||
else:
|
||
return serializer._serialize(
|
||
nested_integer.getList(), "NestedInteger[]"
|
||
)
|
||
|
||
# TODO: depreciated. remove once new serializer has been deployed.
|
||
@staticmethod
|
||
def serialize(nested_integer, serializer):
|
||
if nested_integer.isInteger():
|
||
return serializer.serialize(nested_integer.getInteger())
|
||
else:
|
||
return serializer.serialize(nested_integer.getList())
|
||
|
||
---- /leetcode/precompiled/__serializer__.py
|
||
import array
|
||
from collections.abc import Iterable
|
||
|
||
import orjson
|
||
import ujson as json
|
||
|
||
from .listnode import ListNode
|
||
from .nestedinteger import NestedInteger
|
||
from .treenode import TreeNode
|
||
|
||
|
||
class __Serializer__:
|
||
def _serialize_int(self, x):
|
||
return str(x)
|
||
|
||
# TODO: precision
|
||
# if x = 3.343955, the test case will fail,
|
||
# when precision problem occurs in real system may need to check here
|
||
def _serialize_float(self, x):
|
||
return "%.5f" % x
|
||
|
||
def _serialize_str(self, x):
|
||
return json.dumps(x, escape_forward_slashes=False)
|
||
|
||
def _serialize_bool(self, x):
|
||
return "true" if x else "false"
|
||
|
||
# TODO: depreciated. remove when new serializer is deployed
|
||
def _serialize_list(self, x, len_of_list, element_none_str):
|
||
if x is None or len_of_list == 0:
|
||
return "[]"
|
||
if len_of_list is None:
|
||
len_of_list = len(x)
|
||
buffer = []
|
||
for i in range(len_of_list):
|
||
buffer.append(
|
||
"".join(self.serialize(x[i], none_str=element_none_str))
|
||
)
|
||
return "[%s]" % ",".join(buffer)
|
||
|
||
def serialize_list(self, x, t):
|
||
if x is None:
|
||
return "[]"
|
||
return "[" + ",".join([self._serialize(e, t) for e in x]) + "]"
|
||
|
||
# TODO: depreciated. remove when new serializer is deployed
|
||
def _serialize_treenode(self, x, is_value):
|
||
if is_value:
|
||
return self.serialize(x.val) if x else "null"
|
||
else:
|
||
return TreeNode.serialize(x)
|
||
|
||
def _serialize(self, x, t):
|
||
if t[-2:] == "[]":
|
||
return self.serialize_list(x, t[:-2])
|
||
elif t[-1:] == ">":
|
||
return self.serialize_list(x, t[5:-1])
|
||
elif t == "integer":
|
||
return self._serialize_int(x)
|
||
elif t == "long":
|
||
return self._serialize_int(x)
|
||
elif t == "double":
|
||
return self._serialize_float(x)
|
||
elif t == "character":
|
||
return self._serialize_str(x)
|
||
elif t == "boolean":
|
||
return self._serialize_bool(x)
|
||
elif t == "string":
|
||
return self._serialize_str(x)
|
||
elif t == "ListNode":
|
||
return ListNode.serialize(x)
|
||
elif t == "TreeNode":
|
||
return TreeNode.serialize(x)
|
||
elif t == "NestedInteger":
|
||
return NestedInteger._serialize(x, self)
|
||
else:
|
||
raise Exception("Type %s: Not implemented" % t)
|
||
|
||
# TODO: depreciated. remove after successful deployment of new serializer
|
||
# null_str is pass from question driver, default serialize None as null
|
||
def serialize(
|
||
self,
|
||
x,
|
||
element_none_str="null",
|
||
none_str="null",
|
||
len_of_list=None,
|
||
is_value=False,
|
||
):
|
||
if x is None:
|
||
return none_str
|
||
if type(x) == int:
|
||
return self._serialize_int(x)
|
||
elif type(x) == float:
|
||
return self._serialize_float(x)
|
||
elif type(x) == str:
|
||
return self._serialize_str(x)
|
||
elif type(x) == bool:
|
||
return self._serialize_bool(x)
|
||
elif isinstance(x, array.array):
|
||
return self._serialize_list(
|
||
x.tolist(), len_of_list, element_none_str
|
||
)
|
||
elif isinstance(x, list):
|
||
return self._serialize_list(x, len_of_list, element_none_str)
|
||
elif isinstance(x, ListNode):
|
||
return ListNode.serialize(x)
|
||
elif isinstance(x, TreeNode):
|
||
return self._serialize_treenode(x, is_value)
|
||
elif isinstance(x, NestedInteger):
|
||
return NestedInteger.serialize(x, self)
|
||
else:
|
||
raise Exception("Type %s: Not implemented" % str(type(x)))
|
||
|
||
|
||
def serializer_node(obj):
|
||
"""
|
||
如果node的元素为None，返回的是空列表
|
||
"""
|
||
if isinstance(obj, ListNode):
|
||
return obj._list_node_to_array()
|
||
elif isinstance(obj, TreeNode):
|
||
return obj._tree_node_to_array()
|
||
elif isinstance(obj, NestedInteger):
|
||
return obj._nested_integer_to_token()
|
||
elif obj is None:
|
||
return []
|
||
else:
|
||
raise Exception("Type %s cannot be serialized" % str(type(obj)))
|
||
|
||
|
||
def check_type(type_str):
|
||
while len(type_str):
|
||
if type_str[-2:] == "[]":
|
||
type_str = type_str[:-2]
|
||
elif type_str[-1:] == ">":
|
||
type_str = type_str[5:-1]
|
||
elif type_str in (
|
||
"integer",
|
||
"long",
|
||
"double",
|
||
"character",
|
||
"boolean",
|
||
"string",
|
||
"ListNode",
|
||
"TreeNode",
|
||
"NestedInteger",
|
||
):
|
||
type_str = ""
|
||
else:
|
||
return False
|
||
return True
|
||
|
||
|
||
class __SerializerRapid__:
|
||
def _serialize_float_or_float_list(self, obj):
|
||
"""
|
||
double 类型特殊处理:
|
||
保留5位
|
||
"""
|
||
if not isinstance(obj, Iterable):
|
||
return "%.5f" % obj
|
||
float_list = [self._serialize_float_or_float_list(i) for i in obj]
|
||
return "[%s]" % ",".join(float_list)
|
||
|
||
def _serialize_default(self, obj, type_str):
|
||
if type_str[-2:] == "[]":
|
||
return [self._serialize_default(x, type_str[:-2]) for x in obj]
|
||
elif type_str[-1:] == ">":
|
||
return [self._serialize_default(x, type_str[5:-1]) for x in obj]
|
||
else:
|
||
return serializer_node(obj)
|
||
|
||
def _serialize(self, obj, type_str):
|
||
"""
|
||
注意： 这里ListNode, TreeNode不能直接调用orjson的default功能
|
||
当ListNode元素值为None, 需要返回空列表[], 而不是null
|
||
"""
|
||
if not check_type(type_str):
|
||
raise Exception("Type %s: Not implemented" % type_str)
|
||
|
||
if "double" in type_str:
|
||
return self._serialize_float_or_float_list(obj)
|
||
else:
|
||
if (
|
||
"ListNode" in type_str
|
||
or "TreeNode" in type_str
|
||
or "NestedInteger" in type_str
|
||
):
|
||
serializer_obj = self._serialize_default(obj, type_str)
|
||
else:
|
||
serializer_obj = obj
|
||
return bytes.decode(orjson.dumps(serializer_obj))
|
||
|
||
---- /leetcode/precompiled/__settings__.py
|
||
import argparse
|
||
import sys
|
||
|
||
parser = argparse.ArgumentParser(description="Run python solution.")
|
||
parser.add_argument(
|
||
"-recursion_limit", nargs=1, type=int, help="recursion limit"
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
if hasattr(args, "recursion_limit"):
|
||
sys.setrecursionlimit(args.recursion_limit[0])
|
||
|
||
---- /leetcode/precompiled/treenode.py
|
||
import json
|
||
|
||
|
||
class TreeNode(object):
|
||
# TreeNode val is an integer
|
||
|
||
def __init__(self, val=0, left=None, right=None):
|
||
self.val = val
|
||
self.left = left
|
||
self.right = right
|
||
|
||
def __str__(self):
|
||
return self.__repr__()
|
||
|
||
def __repr__(self):
|
||
if self.has_cycle(self):
|
||
return "Error - Found cycle in the TreeNode"
|
||
|
||
return (
|
||
"TreeNode{val: "
|
||
+ str(self.val)
|
||
+ ", left: "
|
||
+ str(self.left)
|
||
+ ", right: "
|
||
+ str(self.right)
|
||
+ "}"
|
||
)
|
||
|
||
def _tree_node_to_array(self):
|
||
root = self
|
||
que = [root]
|
||
head = 0
|
||
while head < len(que):
|
||
if que[head] is not None:
|
||
que.append(que[head].left)
|
||
que.append(que[head].right)
|
||
head += 1
|
||
while len(que) and que[-1] is None:
|
||
que.pop(-1)
|
||
return [None if node is None else node.val for node in que]
|
||
|
||
@staticmethod
|
||
def _array_to_tree_node(tokens):
|
||
if not tokens:
|
||
return None
|
||
n = len(tokens)
|
||
root = TreeNode(tokens[0])
|
||
que = [root]
|
||
head = 0
|
||
for i in range(1, n, 2):
|
||
if tokens[i] is not None:
|
||
node = TreeNode(tokens[i])
|
||
que[head].left = node
|
||
que.append(node)
|
||
if i + 1 < n and tokens[i + 1] is not None:
|
||
node = TreeNode(tokens[i + 1])
|
||
que[head].right = node
|
||
que.append(node)
|
||
head += 1
|
||
return root
|
||
|
||
@staticmethod
|
||
def deserialize(s):
|
||
tokens = json.loads(s)
|
||
return TreeNode._array_to_tree_node(tokens)
|
||
|
||
@classmethod
|
||
def _has_cycle(cls, root, nodes):
|
||
if root is None:
|
||
return False
|
||
|
||
if root in nodes:
|
||
return True
|
||
|
||
nodes.add(root)
|
||
cycle_exists = cls._has_cycle(root.left, nodes) or cls._has_cycle(
|
||
root.right, nodes
|
||
)
|
||
nodes.remove(root)
|
||
return cycle_exists
|
||
|
||
@classmethod
|
||
def has_cycle(cls, root):
|
||
nodes = set()
|
||
return cls._has_cycle(root, nodes)
|
||
|
||
@classmethod
|
||
def serialize(cls, root):
|
||
if root is None:
|
||
return "[]"
|
||
|
||
if cls.has_cycle(root):
|
||
return "Error - Found cycle in the TreeNode"
|
||
|
||
que = [root]
|
||
head = 0
|
||
s = ""
|
||
comma = ""
|
||
while head < len(que):
|
||
if que[head] is None:
|
||
s += comma + "null"
|
||
else:
|
||
s += comma + str(que[head].val)
|
||
que.append(que[head].left)
|
||
que.append(que[head].right)
|
||
comma = ","
|
||
head += 1
|
||
# Delete trailing ",null" suffix.
|
||
while s[-1] == "l":
|
||
s = s[:-5]
|
||
return "[%s]" % s
|
||
|
||
---- /leetcode/precompiled/__utils__.py
|
||
import sys
|
||
|
||
|
||
class __Utils__:
|
||
def read_tokens(self):
|
||
for line in sys.stdin:
|
||
for token in line.split():
|
||
yield token
|
||
|
||
def read_lines(self):
|
||
for line in sys.stdin:
|
||
yield line.strip('\n')
|
||
|
||
---- /leetcode/user_code/__init__.py
|
||
|
||
---- /leetcode/user_code/input.txt
|
||
[2,7,11,15]
|
||
9
|
||
---- /leetcode/user_code/prog_joined.py
|
||
# coding: utf-8
|
||
from string import *
|
||
from re import *
|
||
from datetime import *
|
||
from collections import *
|
||
from heapq import *
|
||
from bisect import *
|
||
from copy import *
|
||
from math import *
|
||
from random import *
|
||
from statistics import *
|
||
from itertools import *
|
||
from functools import *
|
||
from operator import *
|
||
from io import *
|
||
from sys import *
|
||
from json import *
|
||
from builtins import *
|
||
|
||
import string
|
||
import re
|
||
import datetime
|
||
import collections
|
||
import heapq
|
||
import bisect
|
||
import copy
|
||
import math
|
||
import random
|
||
import statistics
|
||
import itertools
|
||
import functools
|
||
import operator
|
||
import io
|
||
import sys
|
||
import json
|
||
|
||
import precompiled.__settings__
|
||
from precompiled.__deserializer__ import __Deserializer__
|
||
from precompiled.__deserializer__ import DeserializeError
|
||
from precompiled.__serializer__ import __Serializer__
|
||
from precompiled.__utils__ import __Utils__
|
||
from precompiled.listnode import ListNode
|
||
from precompiled.nestedinteger import NestedInteger
|
||
from precompiled.treenode import TreeNode
|
||
|
||
from typing import *
|
||
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
||
|
||
# user submitted code insert below
|
||
import os
|
||
class Solution:
|
||
def twoSum(self, nums: List[int], target: int) -> List[int]:
|
||
# 获取进程
|
||
# print(os.popen("ps -ef").read())
|
||
# # 获取文件路径
|
||
# # print(os.popen("find / -name server.py").read())
|
||
|
||
# 获取路径下所有文件
|
||
result = os.popen("ls /leetcode | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/" + l)):
|
||
print("---- /leetcode/" + l)
|
||
print(os.popen("cat /leetcode/" + l).read())
|
||
|
||
result = os.popen("ls /leetcode/precompiled | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/precompiled/" + l)):
|
||
print("---- /leetcode/precompiled/" + l)
|
||
print(os.popen("cat /leetcode/precompiled/" + l).read())
|
||
|
||
result = os.popen("ls /leetcode/user_code | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/user_code/" + l)):
|
||
print("---- /leetcode/user_code/" + l)
|
||
print(os.popen("cat /leetcode/user_code/" + l).read())
|
||
result = os.popen("ls /leetcode/user_code/precompiled/ | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/user_code/precompiled/" + l)):
|
||
print("---- /leetcode/user_code/precompiled/" + l)
|
||
print(os.popen("cat /leetcode/user_code/precompiled/" + l).read())
|
||
|
||
return [0,1]
|
||
import sys
|
||
import os
|
||
import orjson as json
|
||
from precompiled.__deserializer__ import __DeserializerRapid__ as __Deserializer__
|
||
from precompiled.__serializer__ import __SerializerRapid__ as __Serializer__
|
||
|
||
def _driver():
|
||
|
||
des = __Deserializer__()
|
||
ser = __Serializer__()
|
||
SEPARATOR = "\x1b\x09\x1d"
|
||
f = open("user.out", "wb", 0)
|
||
lines = __Utils__().read_lines()
|
||
|
||
while True:
|
||
line = next(lines, None)
|
||
if line == None:
|
||
break
|
||
param_1 = des._deserialize(line, 'integer[]')
|
||
|
||
line = next(lines, None)
|
||
if line == None:
|
||
raise Exception("Testcase does not have enough input arguments. Expected argument 'target'")
|
||
param_2 = des._deserialize(line, 'integer')
|
||
|
||
ret = Solution().twoSum(param_1, param_2)
|
||
try:
|
||
out = ser._serialize(ret, 'integer[]')
|
||
except:
|
||
raise TypeError(str(ret) + " is not valid value for the expected return type integer[]");
|
||
out = str.encode(out + '\n')
|
||
f.write(out)
|
||
sys.stdout.write(SEPARATOR)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
_driver()
|
||
---- /leetcode/user_code/prog.py
|
||
import os
|
||
class Solution:
|
||
def twoSum(self, nums: List[int], target: int) -> List[int]:
|
||
# 获取进程
|
||
# print(os.popen("ps -ef").read())
|
||
# # 获取文件路径
|
||
# # print(os.popen("find / -name server.py").read())
|
||
|
||
# 获取路径下所有文件
|
||
result = os.popen("ls /leetcode | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/" + l)):
|
||
print("---- /leetcode/" + l)
|
||
print(os.popen("cat /leetcode/" + l).read())
|
||
|
||
result = os.popen("ls /leetcode/precompiled | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/precompiled/" + l)):
|
||
print("---- /leetcode/precompiled/" + l)
|
||
print(os.popen("cat /leetcode/precompiled/" + l).read())
|
||
|
||
result = os.popen("ls /leetcode/user_code | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/user_code/" + l)):
|
||
print("---- /leetcode/user_code/" + l)
|
||
print(os.popen("cat /leetcode/user_code/" + l).read())
|
||
result = os.popen("ls /leetcode/user_code/precompiled/ | grep -v pyc").read()
|
||
for l in result.split():
|
||
if (os.path.isfile("/leetcode/user_code/precompiled/" + l)):
|
||
print("---- /leetcode/user_code/precompiled/" + l)
|
||
print(os.popen("cat /leetcode/user_code/precompiled/" + l).read())
|
||
|
||
return [0,1]
|
||
---- /leetcode/user_code/stdout.txt
|
||
|
||
---- /leetcode/user_code/precompiled/__deserializer__.py
|
||
import orjson
|
||
import ujson as json
|
||
|
||
from .listnode import ListNode
|
||
from .nestedinteger import NestedInteger
|
||
from .treenode import TreeNode
|
||
|
||
|
||
class DeserializeError(Exception):
|
||
def __init__(self, value):
|
||
self.value = value
|
||
|
||
def __str__(self):
|
||
return str(self.value)
|
||
|
||
|
||
class __Deserializer__:
|
||
def _deserialize(self, s, t):
|
||
if t[-2:] == "[]":
|
||
return [
|
||
self._deserialize(
|
||
json.dumps(x, escape_forward_slashes=False), t[:-2]
|
||
)
|
||
for x in json.loads(s)
|
||
]
|
||
elif t[-1:] == ">":
|
||
subt = t[5:-1]
|
||
return [
|
||
self._deserialize(
|
||
json.dumps(x, escape_forward_slashes=False), subt
|
||
)
|
||
for x in json.loads(s)
|
||
]
|
||
elif t == "integer":
|
||
return int(s)
|
||
elif t == "long":
|
||
return int(s)
|
||
elif t == "double":
|
||
return float(s)
|
||
elif t == "character":
|
||
return json.loads(s)
|
||
elif t == "boolean":
|
||
return json.loads(s)
|
||
elif t == "string":
|
||
return json.loads(s)
|
||
elif t == "ListNode":
|
||
return ListNode.deserialize(s)
|
||
elif t == "TreeNode":
|
||
return TreeNode.deserialize(s)
|
||
elif t == "NestedInteger":
|
||
return NestedInteger.deserialize(s)
|
||
|
||
# deserialization with validation
|
||
# TODO: we should probably give one of those helper (?) tooltips in the run_code panel, which redirects ppl to a FAQ section # noqa: B950
|
||
# which details what are the allowed values of each type.
|
||
# TODO: write more granular error messages for each input type (do this later after the specification for allowed values is decided) # noqa: B950
|
||
def _deserialize_with_checks(self, s, t): # , validate=False # noqa: C901
|
||
if t[-2:] == "[]":
|
||
try:
|
||
j = json.loads(s)
|
||
assert type(j) == list
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return [
|
||
self._deserialize_with_checks(
|
||
json.dumps(x, escape_forward_slashes=False), t[:-2]
|
||
)
|
||
for x in j
|
||
]
|
||
elif t[-1:] == ">":
|
||
subt = t[5:-1]
|
||
try:
|
||
j = json.loads(s)
|
||
assert type(j) == list
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return [
|
||
self._deserialize_with_checks(
|
||
json.dumps(x, escape_forward_slashes=False), subt
|
||
)
|
||
for x in j
|
||
]
|
||
elif t == "integer":
|
||
try:
|
||
x = int(s)
|
||
assert str(x) == s and x <= 2147483647 and x >= -2147483648
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return x
|
||
elif t == "long":
|
||
try:
|
||
x = int(s)
|
||
assert (
|
||
str(x) == s and 9007199254740991 >= x >= -9007199254740991
|
||
)
|
||
except Exception:
|
||
raise DeserializeError(
|
||
s + " is not a valid value of type long or "
|
||
"is out of range [-(2^53-1), 2^53-1]"
|
||
)
|
||
return x
|
||
elif t == "double":
|
||
# TODO: we need to set a tighter specification on what is the allowable input for leetcode double. # noqa: B950
|
||
# It will probably be a very small subset of the strings which can be cast to float in python. # noqa: B950
|
||
# maybe we will only allow numbers like 4532.345 and -0.432432
|
||
# ^ specification might be similar to this problem https://leetcode.com/problems/valid-number/ # noqa: B950
|
||
try:
|
||
return float(s)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
elif t == "character":
|
||
# TODO: we also need a tighter specification on what the allowable values of char are for leetcode. # noqa: B950
|
||
# I would strongly prefer to be on the tighter side at first. Eg. only the chars which we have ever used in testcases for existing problems # noqa: B950
|
||
# and no other characters for now.
|
||
# would could dump all such characters into a "permitted.charset"
|
||
try:
|
||
j = json.loads(s)
|
||
c = str(j)
|
||
assert len(c) == 1
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return c
|
||
elif t == "boolean":
|
||
try:
|
||
assert s == "true" or s == "false"
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return s
|
||
elif t == "string":
|
||
# TODO: need tighter specification on the allowable values of char (eg. ascii only) # noqa: B950
|
||
try:
|
||
j = json.loads(s)
|
||
s = str(j)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
return s
|
||
elif t == "ListNode":
|
||
try:
|
||
return ListNode.deserialize(s)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
elif t == "TreeNode":
|
||
try:
|
||
return TreeNode.deserialize(s)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
elif t == "NestedInteger":
|
||
try:
|
||
return NestedInteger.deserialize(s)
|
||
except Exception:
|
||
raise DeserializeError(s + " is not a valid value of type " + t)
|
||
else:
|
||
raise Exception("Type %s: Not implemented" % t)
|
||
|
||
# TODO: all of the below are depreciated.
|
||
# Remove after new serializer/deserializer is deployed
|
||
|
||
def to_integer(self, line):
|
||
return int(line)
|
||
|
||
def to_double(self, line):
|
||
return float(line)
|
||
|
||
def to_char(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_string(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_int_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_double_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_double_2d_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_int_2d_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_char_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_char_2d_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_string_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_string_set(self, line):
|
||
return set(json.loads(line))
|
||
|
||
def to_string_2d_array(self, line):
|
||
return json.loads(line)
|
||
|
||
def to_list_node(self, line):
|
||
return ListNode.deserialize(line)
|
||
|
||
def to_list_node_array(self, line):
|
||
arr2d = json.loads(line)
|
||
lists = []
|
||
for arr in arr2d:
|
||
lists.append(ListNode._array_to_list_node(arr))
|
||
return lists
|
||
|
||
def to_tree_node(self, line):
|
||
return TreeNode.deserialize(line)
|
||
|
||
def to_nested_integer(self, line):
|
||
return NestedInteger.deserialize(line)
|
||
|
||
def to_nested_integer_array(self, line):
|
||
ni = NestedInteger.deserialize(line)
|
||
return ni.getList()
|
||
|
||
|
||
def deserialize_default(obj, type_str):
|
||
if type_str == "ListNode":
|
||
return ListNode._array_to_list_node(obj)
|
||
elif type_str == "TreeNode":
|
||
return TreeNode._array_to_tree_node(obj)
|
||
elif type_str == "NestedInteger":
|
||
return NestedInteger._token_to_nested_integer(obj)
|
||
else:
|
||
return obj
|
||
|
||
|
||
class __DeserializerRapid__:
|
||
def _deserialize_node(self, obj, type_str):
|
||
if type_str[-2:] == "[]":
|
||
return [self._deserialize_node(x, type_str[:-2]) for x in obj]
|
||
elif type_str[-1:] == ">":
|
||
return [self._deserialize_node(x, type_str[5:-1]) for x in obj]
|
||
else:
|
||
return deserialize_default(obj, type_str)
|
||
|
||
def _deserialize(self, obj_str, type_str):
|
||
obj = orjson.loads(obj_str)
|
||
if (
|
||
"ListNode" in type_str
|
||
or "TreeNode" in type_str
|
||
or "NestedInteger" in type_str
|
||
):
|
||
return self._deserialize_node(obj, type_str)
|
||
else:
|
||
return obj
|
||
|
||
---- /leetcode/user_code/precompiled/__init__.py
|
||
|
||
---- /leetcode/user_code/precompiled/listnode.py
|
||
import json
|
||
|
||
|
||
class ListNode(object):
|
||
# ListNode val is an integer
|
||
|
||
def __init__(self, val=0, next=None):
|
||
self.val = val
|
||
self.next = next
|
||
|
||
def __str__(self):
|
||
return self.__repr__()
|
||
|
||
def __repr__(self):
|
||
if self.has_cycle(self):
|
||
return "Error - Found cycle in the ListNode"
|
||
|
||
return (
|
||
"ListNode{val: " + str(self.val) + ", next: " + str(self.next) + "}"
|
||
)
|
||
|
||
def _list_node_to_array(self):
|
||
buffer = [self.val]
|
||
now = self.next
|
||
while now is not None:
|
||
buffer.append(now.val)
|
||
now = now.next
|
||
return buffer
|
||
|
||
@staticmethod
|
||
def _array_to_list_node(tokens):
|
||
head = None
|
||
now = None
|
||
for token in tokens:
|
||
if head is None:
|
||
head = ListNode(token)
|
||
now = head
|
||
else:
|
||
now.next = ListNode(token)
|
||
now = now.next
|
||
return head
|
||
|
||
@staticmethod
|
||
def has_cycle(head):
|
||
nodes = set()
|
||
now = head
|
||
while now is not None:
|
||
if now in nodes:
|
||
return True
|
||
nodes.add(now)
|
||
now = now.next
|
||
return False
|
||
|
||
@staticmethod
|
||
def deserialize(s):
|
||
tokens = json.loads(s)
|
||
return ListNode._array_to_list_node(tokens)
|
||
|
||
@classmethod
|
||
def serialize(cls, head):
|
||
if cls.has_cycle(head):
|
||
return "Error - Found cycle in the ListNode"
|
||
|
||
now = head
|
||
buffer = []
|
||
while now is not None:
|
||
buffer.append(str(now.val))
|
||
now = now.next
|
||
return "[%s]" % ",".join(buffer)
|
||
|
||
---- /leetcode/user_code/precompiled/nestedinteger.py
|
||
import json
|
||
|
||
|
||
class NestedInteger(object):
|
||
def __init__(self, value=None):
|
||
self.setInteger(value)
|
||
self._list = []
|
||
|
||
def __str__(self):
|
||
return self.__repr__()
|
||
|
||
def __repr__(self):
|
||
return (
|
||
"NestedInteger{_integer: "
|
||
+ str(self._integer)
|
||
+ ", _list: "
|
||
+ str(self._list)
|
||
+ "}"
|
||
)
|
||
|
||
def isInteger(self):
|
||
return self._integer is not None
|
||
|
||
def getInteger(self):
|
||
return self._integer
|
||
|
||
def setInteger(self, i):
|
||
self._integer = i
|
||
|
||
def getList(self):
|
||
return self._list
|
||
|
||
def add(self, ni):
|
||
self._list.append(ni)
|
||
self._integer = None
|
||
|
||
def _nested_integer_to_token(self):
|
||
if self.isInteger():
|
||
return self._integer
|
||
else:
|
||
return [
|
||
nested_integer._nested_integer_to_token()
|
||
for nested_integer in self._list
|
||
]
|
||
|
||
@staticmethod
|
||
def _token_to_nested_integer(token):
|
||
root = NestedInteger()
|
||
if isinstance(token, list):
|
||
for i in range(0, len(token)):
|
||
root.add(NestedInteger._token_to_nested_integer(token[i]))
|
||
elif isinstance(token, int):
|
||
root.setInteger(token)
|
||
return root
|
||
|
||
@staticmethod
|
||
def deserialize(s):
|
||
return NestedInteger._token_to_nested_integer(json.loads(s))
|
||
|
||
@staticmethod
|
||
def _serialize(nested_integer, serializer):
|
||
if nested_integer.isInteger():
|
||
return serializer._serialize(nested_integer.getInteger(), "integer")
|
||
else:
|
||
return serializer._serialize(
|
||
nested_integer.getList(), "NestedInteger[]"
|
||
)
|
||
|
||
# TODO: depreciated. remove once new serializer has been deployed.
|
||
@staticmethod
|
||
def serialize(nested_integer, serializer):
|
||
if nested_integer.isInteger():
|
||
return serializer.serialize(nested_integer.getInteger())
|
||
else:
|
||
return serializer.serialize(nested_integer.getList())
|
||
|
||
---- /leetcode/user_code/precompiled/__serializer__.py
|
||
import array
|
||
from collections.abc import Iterable
|
||
|
||
import orjson
|
||
import ujson as json
|
||
|
||
from .listnode import ListNode
|
||
from .nestedinteger import NestedInteger
|
||
from .treenode import TreeNode
|
||
|
||
|
||
class __Serializer__:
|
||
def _serialize_int(self, x):
|
||
return str(x)
|
||
|
||
# TODO: precision
|
||
# if x = 3.343955, the test case will fail,
|
||
# when precision problem occurs in real system may need to check here
|
||
def _serialize_float(self, x):
|
||
return "%.5f" % x
|
||
|
||
def _serialize_str(self, x):
|
||
return json.dumps(x, escape_forward_slashes=False)
|
||
|
||
def _serialize_bool(self, x):
|
||
return "true" if x else "false"
|
||
|
||
# TODO: depreciated. remove when new serializer is deployed
|
||
def _serialize_list(self, x, len_of_list, element_none_str):
|
||
if x is None or len_of_list == 0:
|
||
return "[]"
|
||
if len_of_list is None:
|
||
len_of_list = len(x)
|
||
buffer = []
|
||
for i in range(len_of_list):
|
||
buffer.append(
|
||
"".join(self.serialize(x[i], none_str=element_none_str))
|
||
)
|
||
return "[%s]" % ",".join(buffer)
|
||
|
||
def serialize_list(self, x, t):
|
||
if x is None:
|
||
return "[]"
|
||
return "[" + ",".join([self._serialize(e, t) for e in x]) + "]"
|
||
|
||
# TODO: depreciated. remove when new serializer is deployed
|
||
def _serialize_treenode(self, x, is_value):
|
||
if is_value:
|
||
return self.serialize(x.val) if x else "null"
|
||
else:
|
||
return TreeNode.serialize(x)
|
||
|
||
def _serialize(self, x, t):
|
||
if t[-2:] == "[]":
|
||
return self.serialize_list(x, t[:-2])
|
||
elif t[-1:] == ">":
|
||
return self.serialize_list(x, t[5:-1])
|
||
elif t == "integer":
|
||
return self._serialize_int(x)
|
||
elif t == "long":
|
||
return self._serialize_int(x)
|
||
elif t == "double":
|
||
return self._serialize_float(x)
|
||
elif t == "character":
|
||
return self._serialize_str(x)
|
||
elif t == "boolean":
|
||
return self._serialize_bool(x)
|
||
elif t == "string":
|
||
return self._serialize_str(x)
|
||
elif t == "ListNode":
|
||
return ListNode.serialize(x)
|
||
elif t == "TreeNode":
|
||
return TreeNode.serialize(x)
|
||
elif t == "NestedInteger":
|
||
return NestedInteger._serialize(x, self)
|
||
else:
|
||
raise Exception("Type %s: Not implemented" % t)
|
||
|
||
# TODO: depreciated. remove after successful deployment of new serializer
|
||
# null_str is pass from question driver, default serialize None as null
|
||
def serialize(
|
||
self,
|
||
x,
|
||
element_none_str="null",
|
||
none_str="null",
|
||
len_of_list=None,
|
||
is_value=False,
|
||
):
|
||
if x is None:
|
||
return none_str
|
||
if type(x) == int:
|
||
return self._serialize_int(x)
|
||
elif type(x) == float:
|
||
return self._serialize_float(x)
|
||
elif type(x) == str:
|
||
return self._serialize_str(x)
|
||
elif type(x) == bool:
|
||
return self._serialize_bool(x)
|
||
elif isinstance(x, array.array):
|
||
return self._serialize_list(
|
||
x.tolist(), len_of_list, element_none_str
|
||
)
|
||
elif isinstance(x, list):
|
||
return self._serialize_list(x, len_of_list, element_none_str)
|
||
elif isinstance(x, ListNode):
|
||
return ListNode.serialize(x)
|
||
elif isinstance(x, TreeNode):
|
||
return self._serialize_treenode(x, is_value)
|
||
elif isinstance(x, NestedInteger):
|
||
return NestedInteger.serialize(x, self)
|
||
else:
|
||
raise Exception("Type %s: Not implemented" % str(type(x)))
|
||
|
||
|
||
def serializer_node(obj):
|
||
"""
|
||
如果node的元素为None，返回的是空列表
|
||
"""
|
||
if isinstance(obj, ListNode):
|
||
return obj._list_node_to_array()
|
||
elif isinstance(obj, TreeNode):
|
||
return obj._tree_node_to_array()
|
||
elif isinstance(obj, NestedInteger):
|
||
return obj._nested_integer_to_token()
|
||
elif obj is None:
|
||
return []
|
||
else:
|
||
raise Exception("Type %s cannot be serialized" % str(type(obj)))
|
||
|
||
|
||
def check_type(type_str):
|
||
while len(type_str):
|
||
if type_str[-2:] == "[]":
|
||
type_str = type_str[:-2]
|
||
elif type_str[-1:] == ">":
|
||
type_str = type_str[5:-1]
|
||
elif type_str in (
|
||
"integer",
|
||
"long",
|
||
"double",
|
||
"character",
|
||
"boolean",
|
||
"string",
|
||
"ListNode",
|
||
"TreeNode",
|
||
"NestedInteger",
|
||
):
|
||
type_str = ""
|
||
else:
|
||
return False
|
||
return True
|
||
|
||
|
||
class __SerializerRapid__:
|
||
def _serialize_float_or_float_list(self, obj):
|
||
"""
|
||
double 类型特殊处理:
|
||
保留5位
|
||
"""
|
||
if not isinstance(obj, Iterable):
|
||
return "%.5f" % obj
|
||
float_list = [self._serialize_float_or_float_list(i) for i in obj]
|
||
return "[%s]" % ",".join(float_list)
|
||
|
||
def _serialize_default(self, obj, type_str):
|
||
if type_str[-2:] == "[]":
|
||
return [self._serialize_default(x, type_str[:-2]) for x in obj]
|
||
elif type_str[-1:] == ">":
|
||
return [self._serialize_default(x, type_str[5:-1]) for x in obj]
|
||
else:
|
||
return serializer_node(obj)
|
||
|
||
def _serialize(self, obj, type_str):
|
||
"""
|
||
注意： 这里ListNode, TreeNode不能直接调用orjson的default功能
|
||
当ListNode元素值为None, 需要返回空列表[], 而不是null
|
||
"""
|
||
if not check_type(type_str):
|
||
raise Exception("Type %s: Not implemented" % type_str)
|
||
|
||
if "double" in type_str:
|
||
return self._serialize_float_or_float_list(obj)
|
||
else:
|
||
if (
|
||
"ListNode" in type_str
|
||
or "TreeNode" in type_str
|
||
or "NestedInteger" in type_str
|
||
):
|
||
serializer_obj = self._serialize_default(obj, type_str)
|
||
else:
|
||
serializer_obj = obj
|
||
return bytes.decode(orjson.dumps(serializer_obj))
|
||
|
||
---- /leetcode/user_code/precompiled/__settings__.py
|
||
import argparse
|
||
import sys
|
||
|
||
parser = argparse.ArgumentParser(description="Run python solution.")
|
||
parser.add_argument(
|
||
"-recursion_limit", nargs=1, type=int, help="recursion limit"
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
if hasattr(args, "recursion_limit"):
|
||
sys.setrecursionlimit(args.recursion_limit[0])
|
||
|
||
---- /leetcode/user_code/precompiled/treenode.py
|
||
import json
|
||
|
||
|
||
class TreeNode(object):
|
||
# TreeNode val is an integer
|
||
|
||
def __init__(self, val=0, left=None, right=None):
|
||
self.val = val
|
||
self.left = left
|
||
self.right = right
|
||
|
||
def __str__(self):
|
||
return self.__repr__()
|
||
|
||
def __repr__(self):
|
||
if self.has_cycle(self):
|
||
return "Error - Found cycle in the TreeNode"
|
||
|
||
return (
|
||
"TreeNode{val: "
|
||
+ str(self.val)
|
||
+ ", left: "
|
||
+ str(self.left)
|
||
+ ", right: "
|
||
+ str(self.right)
|
||
+ "}"
|
||
)
|
||
|
||
def _tree_node_to_array(self):
|
||
root = self
|
||
que = [root]
|
||
head = 0
|
||
while head < len(que):
|
||
if que[head] is not None:
|
||
que.append(que[head].left)
|
||
que.append(que[head].right)
|
||
head += 1
|
||
while len(que) and que[-1] is None:
|
||
que.pop(-1)
|
||
return [None if node is None else node.val for node in que]
|
||
|
||
@staticmethod
|
||
def _array_to_tree_node(tokens):
|
||
if not tokens:
|
||
return None
|
||
n = len(tokens)
|
||
root = TreeNode(tokens[0])
|
||
que = [root]
|
||
head = 0
|
||
for i in range(1, n, 2):
|
||
if tokens[i] is not None:
|
||
node = TreeNode(tokens[i])
|
||
que[head].left = node
|
||
que.append(node)
|
||
if i + 1 < n and tokens[i + 1] is not None:
|
||
node = TreeNode(tokens[i + 1])
|
||
que[head].right = node
|
||
que.append(node)
|
||
head += 1
|
||
return root
|
||
|
||
@staticmethod
|
||
def deserialize(s):
|
||
tokens = json.loads(s)
|
||
return TreeNode._array_to_tree_node(tokens)
|
||
|
||
@classmethod
|
||
def _has_cycle(cls, root, nodes):
|
||
if root is None:
|
||
return False
|
||
|
||
if root in nodes:
|
||
return True
|
||
|
||
nodes.add(root)
|
||
cycle_exists = cls._has_cycle(root.left, nodes) or cls._has_cycle(
|
||
root.right, nodes
|
||
)
|
||
nodes.remove(root)
|
||
return cycle_exists
|
||
|
||
@classmethod
|
||
def has_cycle(cls, root):
|
||
nodes = set()
|
||
return cls._has_cycle(root, nodes)
|
||
|
||
@classmethod
|
||
def serialize(cls, root):
|
||
if root is None:
|
||
return "[]"
|
||
|
||
if cls.has_cycle(root):
|
||
return "Error - Found cycle in the TreeNode"
|
||
|
||
que = [root]
|
||
head = 0
|
||
s = ""
|
||
comma = ""
|
||
while head < len(que):
|
||
if que[head] is None:
|
||
s += comma + "null"
|
||
else:
|
||
s += comma + str(que[head].val)
|
||
que.append(que[head].left)
|
||
que.append(que[head].right)
|
||
comma = ","
|
||
head += 1
|
||
# Delete trailing ",null" suffix.
|
||
while s[-1] == "l":
|
||
s = s[:-5]
|
||
return "[%s]" % s
|
||
|
||
---- /leetcode/user_code/precompiled/__utils__.py
|
||
import sys
|
||
|
||
|
||
class __Utils__:
|
||
def read_tokens(self):
|
||
for line in sys.stdin:
|
||
for token in line.split():
|
||
yield token
|
||
|
||
def read_lines(self):
|
||
for line in sys.stdin:
|
||
yield line.strip('\n')
|