Commit 6022f5b8 authored by eckhart's avatar eckhart

- server.py: typing info added

parent 3b58429f
......@@ -1495,7 +1495,7 @@ class Option(UnaryParser):
>>> number = Option(TKN('-')) + RegExp(r'\d+') + Option(RegExp(r'\.\d+'))
>>> Grammar(number)('3.14159').content
'3.14159'
>>> Grammar(number)('3.14159').structure
>>> Grammar(number)('3.14159').as_sxpr()
'(:Series (:RegExp "3") (:RegExp ".14159"))'
>>> Grammar(number)('-1').content
'-1'
......
......@@ -49,10 +49,23 @@ from typing import Callable, Optional, Union, Dict, List, Sequence, cast
from DHParser.toolkit import get_config_value, re
__all__ = ('RPC_Table',
'RPC_Type',
'JSON_Type',
'SERVER_ERROR',
'SERVER_OFFLINE',
'SERVER_STARTING',
'SERVER_ONLINE',
'SERVER_TERMINATE',
'Server')
RPC_Table = Dict[str, Callable]
RPC_Type = Union[RPC_Table, List[Callable], Callable]
JSON_Type = Union[Dict, Sequence, str, int, None]
RX_IS_JSON = re.compile(r'\s*(?:{|\[|"|\d|true|false|null)')
SERVER_ERROR = "COMPILER-SERVER-ERROR"
SERVER_OFFLINE = 0
......@@ -60,11 +73,7 @@ SERVER_STARTING = 1
SERVER_ONLINE = 2
SERVER_TERMINATE = 3
RX_MAYBE_JSON = re.compile(r'\s*(?:{|\[|"|\d|true|false|null)')
response = b'''HTTP/1.1 200 OK
response_test = b'''HTTP/1.1 200 OK
Date: Sun, 18 Oct 2009 08:56:53 GMT
Server: Apache/2.2.14 (Win32)
Last-Modified: Sat, 20 Nov 2004 07:16:26 GMT
......@@ -170,10 +179,17 @@ class Server:
self.stage.value = SERVER_ONLINE
self.server_messages.put(SERVER_ONLINE)
await self.server.serve_forever()
# self.server.wait_until_closed()
# self.server.wait_until_closed()
def run_server(self, address: str = '127.0.0.1', port: int = 8888):
self.stage.value = SERVER_STARTING
# loop = asyncio.get_event_loop()
# try:
# loop.run_until_complete(self.serve(address, port))
# finally:
# print(type(self.server))
# # self.server.cancel()
# loop.close()
asyncio.run(self.serve(address, port))
def wait_until_server_online(self):
......
......@@ -71,6 +71,10 @@ ZOMBIE_TAG = "__ZOMBIE__"
#######################################################################
RX_IS_SXPR = re.compile(r'\s*\(')
RX_IS_XML = re.compile(r'\s*<')
def flatten_sxpr(sxpr: str, threshold: int = -1) -> str:
"""
Returns S-expression ``sxpr`` as a one-liner without unnecessary
......@@ -87,9 +91,11 @@ def flatten_sxpr(sxpr: str, threshold: int = -1) -> str:
>>> flatten_sxpr('(a\\n (b\\n c\\n )\\n)\\n')
'(a (b c))'
"""
assert RX_IS_SXPR.match(sxpr)
if threshold == 0:
return sxpr
flat = re.sub(r'\s(?=\))', '', re.sub(r'\s+', ' ', sxpr)).strip()
if threshold >= 0 and len(flat) > threshold:
if len(flat) > threshold >= 0:
return sxpr.strip()
return flat
......@@ -101,9 +107,9 @@ def flatten_xml(xml: str) -> str:
A more precise alternative to `flatten_xml` is to use Node.as_xml()
ans passing a set containing the top level tag to parameter `inline_tags`.
"""
# works only with regex
# return re.sub(r'\s+(?=<\w)', '', re.sub(r'(?<=</\w+>)\s+', '', xml))
assert RX_IS_XML.match(xml)
def tag_only(m):
return m.groupdict()['closing_tag']
return re.sub(r'\s+(?=<[\w:])', '', re.sub(r'(?P<closing_tag></:?\w+>)\s+', tag_only, xml))
......@@ -385,15 +391,6 @@ class Node: # (collections.abc.Sized): Base class omitted for cython-compatibil
# else str(self._result)
@property
def structure(self) -> str:
"""
Return structure (and content) as S-expression on a single line
without any line breaks.
"""
return flatten_sxpr(self.as_sxpr())
@property
def pos(self) -> int:
"""Returns the position of the Node's content in the source text."""
......@@ -628,7 +625,7 @@ class Node: # (collections.abc.Sized): Base class omitted for cython-compatibil
def as_sxpr(self, src: str = None,
indentation: int = 2,
compact: bool = False,
flatten_threshold: int = 0) -> str:
flatten_threshold: int = 92) -> str:
"""
Returns content as S-expression, i.e. in lisp-like form. If this
method is callad on a RootNode-object,
......@@ -675,7 +672,7 @@ class Node: # (collections.abc.Sized): Base class omitted for cython-compatibil
else '"%s"' % strg.replace('"', r'\"')
sxpr = self._tree_repr(' ' * indentation, opening, closing, pretty, density=density)
return flatten_sxpr(sxpr, flatten_threshold)
return sxpr if compact else flatten_sxpr(sxpr, flatten_threshold)
def as_xml(self, src: str = None,
......@@ -1048,7 +1045,7 @@ def parse_sxpr(sxpr: Union[str, StringView]) -> Node:
generate test data.
Example:
>>> parse_sxpr("(a (b c))").as_sxpr()
>>> parse_sxpr("(a (b c))").as_sxpr(flatten_threshold=0)
'(a\\n (b\\n "c"\\n )\\n)'
"""
......
......@@ -456,3 +456,5 @@ try:
except AttributeError:
# somebody has already taken care of this !?
pass
......@@ -47,7 +47,7 @@ class TestServer:
reader, writer = await asyncio.open_connection('127.0.0.1', 8888)
writer.write(src.encode())
data = await reader.read(500)
print(f'Received: {data.decode()!r}')
# print(f'Received: {data.decode()!r}')
writer.close()
asyncio.run(compile('Test', ''))
......
......@@ -302,20 +302,20 @@ class TestSerialization:
def test_sexpr(self):
tree = parse_sxpr('(A (B "C") (D "E"))')
s = tree.as_sxpr()
s = tree.as_sxpr(flatten_threshold=0)
assert s == '(A\n (B\n "C"\n )\n (D\n "E"\n )\n)', s
tree = parse_sxpr('(A (B (C "D") (E "F")) (G "H"))')
s = tree.as_sxpr()
s = tree.as_sxpr(flatten_threshold=0)
assert s == '(A\n (B\n (C\n "D"\n )\n (E\n "F"\n )' \
'\n )\n (G\n "H"\n )\n)', s
tree = parse_sxpr('(A (B (C "D\nX") (E "F")) (G " H \n Y "))')
s = tree.as_sxpr()
s = tree.as_sxpr(flatten_threshold=0)
assert s == '(A\n (B\n (C\n "D"\n "X"\n )' \
'\n (E\n "F"\n )\n )\n (G\n " H "\n " Y "\n )\n)', s
def test_compact_representation(self):
tree = parse_sxpr('(A (B (C "D") (E "F")) (G "H"))')
compact = tree.as_sxpr(compact=True)
compact = tree.as_sxpr(compact=True, )
assert compact == 'A\n B\n C "D"\n E "F"\n G "H"', compact
tree = parse_sxpr('(A (B (C "D\nX") (E "F")) (G " H \n Y "))')
compact = tree.as_sxpr(compact=True)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment