Skip to content

Commit

Permalink
Limit text tables by screen size
Browse files Browse the repository at this point in the history
  • Loading branch information
raulikak committed May 10, 2024
1 parent 34bd317 commit 0122f3c
Show file tree
Hide file tree
Showing 5 changed files with 60 additions and 13 deletions.
6 changes: 5 additions & 1 deletion tcsfw/builder_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import json
import logging
import pathlib
import shutil
import sys
from typing import Any, Callable, Dict, List, Optional, Self, Tuple, Union

Expand Down Expand Up @@ -1099,8 +1100,11 @@ def run(self):
print(json.dumps(resp, indent=4))
return
if args.test_get:
wid, hei = shutil.get_terminal_size()
for res in args.test_get:
print(api.api_get(APIRequest.parse(res), pretty=True))
api_req = APIRequest.parse(res)
api_req.set_param("screen", f"{wid}x{hei}")
print(api.api_get(api_req, pretty=True))
return

out_form = args.output
Expand Down
7 changes: 6 additions & 1 deletion tcsfw/client_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,11 @@ def parse(cls, url_str: str) -> 'APIRequest':
req.get_visual = "visual" in qs # Assumes we have VisualAPI
return req

def set_param(self, key: str, value: str) -> 'APIRequest':
"""Set parameter"""
self.parameters[key] = value
return self

def change_path(self, path: str) -> 'APIRequest':
"""Change the path"""
r = APIRequest(path)
Expand Down Expand Up @@ -122,7 +127,7 @@ def api_get(self, request: APIRequest, pretty=False) -> str:
context = RequestContext(request, self)
path = request.path
if path.startswith("table/"):
text = TableView.get_print(self.registry.system, path[6:])
text = TableView.get_print(self.registry.system, path[6:], parameters=request.parameters)
return text
if path == "all":
request.get_connections = False
Expand Down
4 changes: 4 additions & 0 deletions tcsfw/entity.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,10 @@ def is_relevant(self) -> bool:
"""Is this entity relevant, i.e. not placeholder or external?"""
return True

def is_admin(self) -> bool:
"""Is an admin entity?"""
return False

def is_host(self) -> bool:
"""Is a host?"""
return False
Expand Down
6 changes: 6 additions & 0 deletions tcsfw/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,9 @@ def is_original(self) -> bool:
system = self.source.get_system()
return self in system.originals

def is_admin(self) -> bool:
return self.target.is_admin()

def is_relevant(self, ignore_ends=False) -> bool:
"""Is this connection relevant, i.e. not placeholder or external?"""
if self.status in {Status.EXPECTED, Status.UNEXPECTED}:
Expand Down Expand Up @@ -160,6 +163,9 @@ def is_multicast(self) -> bool:
def is_relevant(self) -> bool:
return self.status in {Status.EXPECTED, Status.UNEXPECTED}

def is_admin(self) -> bool:
return self.host_type == HostType.ADMINISTRATIVE

def get_connections(self, relevant_only=True) -> List[Connection]:
"""Get relevant conneciions"""
cs = []
Expand Down
50 changes: 39 additions & 11 deletions tcsfw/text_tables.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,10 @@
"""Text tables"""

from io import StringIO
from typing import Any, List, TextIO, Tuple
from typing import Any, Dict, List, TextIO, Tuple

from tcsfw.basics import Status
from tcsfw.entity import Entity
from tcsfw.model import Host, IoTSystem, NetworkNode, Service


Expand All @@ -17,15 +19,32 @@ def __init__(self, columns: List[Tuple[str, int]], screen_size: Tuple[int, int])
ratio = screen_size[0] / min_wid
self.columns = [(c[0], int(c[1] * ratio)) for c in columns]
self.relevant_only = True
self.include_admin = False
self.include_external = False

def _include(self, entity: Entity) -> bool:
if self.relevant_only and not entity.is_relevant():
return False
if not self.include_external and entity.status == Status.EXTERNAL:
return False
if not self.include_admin and entity.is_admin():
return False
return True

def print(self, stream: TextIO):
"""Print!"""
raise NotImplementedError()

def print_rows(self, rows: List[List[Any]], stream: TextIO) -> str:
"""Print the rows"""
show_rows = len(rows)
drop_rows = show_rows - self.screen_size[1]
if drop_rows > 0:
drop_rows += 1
show_rows -= drop_rows

screen_wid = self.screen_size[0]
for row in rows:
for row in rows[:show_rows]:
line = []
x, target_x = 0, 0
assert len(row) == len(self.columns), f"Row and columns mismatch: {len(row)} != {len(self.columns)}"
Expand All @@ -42,14 +61,14 @@ def print_rows(self, rows: List[List[Any]], stream: TextIO) -> str:
# space between columns
target_x += 1
x += 1

line_s = "".join(line)
stream.write(f"{line_s}\n")

if drop_rows > 0:
stream.write(f"[...{drop_rows} rows omitted]\n")

class HostTable(BaseTable):
"""Host table"""
def __init__(self, root: IoTSystem, screen_size=(80, 50)):
def __init__(self, root: IoTSystem, screen_size: Tuple[int, int]):
super().__init__([
("Host", 10),
("Service", 20),
Expand All @@ -67,12 +86,12 @@ def _components(node: NetworkNode):

for h in self.root.get_children():
if isinstance(h, Host):
if self.relevant_only and not h.is_relevant():
if not self._include(h):
continue
rows.append([h.long_name(), '', '', h.status_string()])
_components(h)
for s in h.get_children():
if self.relevant_only and not s.is_relevant():
if not self._include(s):
continue
if isinstance(s, Service):
rows.append(['', s.long_name(), '', s.status_string()])
Expand All @@ -83,7 +102,7 @@ def _components(node: NetworkNode):

class ConnectionTable(BaseTable):
"""Host table"""
def __init__(self, root: IoTSystem, screen_size=(80, 50)):
def __init__(self, root: IoTSystem, screen_size: Tuple[int, int]):
super().__init__([
("Source", 20),
("Target", 20),
Expand All @@ -94,7 +113,9 @@ def __init__(self, root: IoTSystem, screen_size=(80, 50)):

def print(self, stream: TextIO):
rows = [[h[0] for h in self.columns]]
for c in self.root.get_connections(self.relevant_only):
for c in self.root.get_connections(relevant_only=False):
if not self._include(c):
continue
s, t = c.source, c.target
proto = ""
if isinstance(t, Service) and t.protocol is not None:
Expand All @@ -110,9 +131,16 @@ def __init__(self, tables: List[BaseTable]) -> None:
self.tables = tables

@classmethod
def get_print(cls, model: IoTSystem, _name: str) -> str:
def get_print(cls, model: IoTSystem, _name: str, parameters: Dict[str, str]) -> str:
"""Get printout by name"""
view = TableView([HostTable(model), ConnectionTable(model)])
screen = parameters.get("screen")
if screen:
screen_size = tuple([int(x) for x in screen.split("x")])
assert len(screen_size) == 2
else:
screen_size = (80, 50)
screen_size = screen_size[0], int(screen_size[1] / 2)
view = TableView([HostTable(model, screen_size), ConnectionTable(model, screen_size)])
buf = StringIO()
view.print(buf)
return buf.getvalue()
Expand Down

0 comments on commit 0122f3c

Please sign in to comment.