Skip to content

Commit 38fa90f

Browse files
PeterC-DLSPeterC-DLS
authored
Minor server side fixes and clean ups (#233)
Reduce logging of validation errors from unpacking messages by only reporting errors when every model in union fails validation. Avoid mutating input on model validation. Fix typing glitch in simple demo script. Refactor message tests from plot server tests. Correct and add baton logging. Log warnings for unparseable client messages and catch exceptions in server handling messages
1 parent 6512db7 commit 38fa90f

6 files changed

Lines changed: 809 additions & 759 deletions

File tree

‎server/davidia/models/messages.py‎

Lines changed: 11 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -99,13 +99,6 @@ def validate_glyph_type(cls, v: GlyphType | str):
9999
v = GlyphType[v]
100100
return v
101101

102-
@model_validator(mode="before")
103-
@classmethod
104-
def check_glyph_type(cls, values: dict):
105-
if "glyphType" not in values and values.get("glyph_type") is None:
106-
values["glyph_type"] = GlyphType.Circle
107-
return values
108-
109102

110103
class LineData(DvDNpModel):
111104
"""Class for representing a line."""
@@ -135,24 +128,20 @@ def equal_axes(cls, v, values):
135128

136129
@model_validator(mode="before")
137130
@classmethod
138-
def are_indices_default(cls, values: Any):
139-
if not isinstance(values, dict):
140-
return values
141-
142-
for k in ("x", "y"):
143-
if k in values:
144-
v = values[k]
145-
if v is None:
146-
values.pop(k)
147-
else:
131+
def are_coords_ndarrays(cls, values: Any):
132+
if isinstance(values, dict):
133+
for k in ("x", "y"):
134+
v = values.get(k)
135+
if v is not None:
148136
values[k] = _asanyarray(v)
149-
150-
if not values.get("default_indices"):
151-
values["default_indices"] = (
152-
"y" not in values or "x" not in values or values["x"].size == 0
153-
)
154137
return values
155138

139+
@model_validator(mode="after")
140+
def are_indices_default(self): # type: (Self) -> Self
141+
if not self.default_indices:
142+
self.default_indices = self.x is None or self.x.size == 0
143+
return self
144+
156145

157146
class ImageData(DvDNpModel):
158147
"""Class for representing an image."""

‎server/davidia/server/fastapi_utils.py‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -148,14 +148,15 @@ async def root_request(payload: MyModel) -> OtherModel:
148148

149149
def _instantiate_obj(model_class, obj):
150150
if isinstance(model_class, UnionType):
151+
excs = {}
151152
for m in get_args(model_class):
152153
try:
153154
return m.model_validate(obj)
154-
except ValidationError:
155-
logger.warning(
156-
"Could not validate as %s: %s", m, obj, exc_info=True
157-
)
155+
except ValidationError as e:
156+
excs[m] = e
158157
logger.error("No valid models for", obj)
158+
for m, e in excs.items():
159+
logger.error("%s: %s", m, e)
159160
return None
160161

161162
if isinstance(obj, BaseModel):

‎server/davidia/server/plotserver.py‎

Lines changed: 133 additions & 101 deletions
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,112 @@ async def send_next_message(self):
9292
self.queue.task_done()
9393
await self.websocket.send_bytes(msg)
9494
except QueueEmpty:
95-
logger.debug("Queue for websocket %s is empty", self.websocket)
95+
logger.debug("Queue for client %s is empty", self.uuid)
96+
97+
98+
def combine_line_messages(
99+
ml_data_msg: MultiLineMessage, new_points_msg: MultiLineMessage
100+
) -> tuple[MultiLineMessage, MultiLineMessage]:
101+
"""
102+
Adds indices to data message and appends points to current multi-line
103+
data message
104+
105+
Parameters
106+
----------
107+
ml_data_msg : MultiLineMessage
108+
current data lines
109+
new_points_msg : MultiLineMessage
110+
new points to append to current data lines.
111+
"""
112+
if not new_points_msg.append:
113+
raise ValueError(f"New data is not marked as append: {new_points_msg}")
114+
115+
current_lines = ml_data_msg.ml_data
116+
add_colour_to_lines(new_points_msg.ml_data)
117+
new_points = new_points_msg.ml_data
118+
default_indices = current_lines[0].default_indices
119+
current_lines_len = len(current_lines)
120+
new_points_len = len(new_points)
121+
122+
def _append(a: DvDNDArray | None, b: DvDNDArray | None):
123+
if a is None:
124+
return b
125+
if b is None:
126+
return a
127+
return np.append(a, b)
128+
129+
if not default_indices:
130+
combined_lines = [
131+
LineData(
132+
line_params=c.line_params,
133+
x=_append(c.x, p.x),
134+
y=np.append(c.y, p.y),
135+
default_indices=False,
136+
)
137+
for c, p in zip(current_lines, new_points)
138+
]
139+
140+
if current_lines_len > new_points_len:
141+
combined_lines += current_lines[new_points_len:]
142+
143+
elif new_points_len > current_lines_len:
144+
combined_lines += new_points[current_lines_len:]
145+
146+
else:
147+
indexed_lines = []
148+
combined_lines = []
149+
for c, p in zip(current_lines, new_points):
150+
c_y_size = c.y.size
151+
total_y_size = c_y_size + p.y.size
152+
indexed_lines.append(
153+
LineData(
154+
line_params=p.line_params,
155+
x=np.arange(
156+
c_y_size,
157+
total_y_size,
158+
dtype=np.min_scalar_type(total_y_size),
159+
),
160+
y=p.y,
161+
default_indices=True,
162+
)
163+
)
164+
combined_lines.append(
165+
LineData(
166+
line_params=c.line_params,
167+
x=_append(
168+
c.x,
169+
np.arange(
170+
c_y_size,
171+
total_y_size,
172+
dtype=np.min_scalar_type(total_y_size),
173+
),
174+
),
175+
y=np.append(c.y, p.y),
176+
default_indices=True,
177+
)
178+
)
179+
if current_lines_len > new_points_len:
180+
combined_lines += current_lines[new_points_len:]
181+
182+
elif new_points_len > current_lines_len:
183+
extra_indexed_lines = [
184+
LineData(
185+
line_params=p.line_params,
186+
x=np.arange(p.y.size, dtype=np.min_scalar_type(p.y.size)),
187+
y=p.y,
188+
default_indices=True,
189+
)
190+
for p in new_points[current_lines_len:]
191+
]
192+
combined_lines += extra_indexed_lines
193+
indexed_lines += extra_indexed_lines
194+
195+
new_points_msg.ml_data = indexed_lines
196+
197+
return (
198+
MultiLineMessage(ml_data=combined_lines, plot_config=ml_data_msg.plot_config),
199+
new_points_msg,
200+
)
96201

97202

98203
class PlotState:
@@ -249,7 +354,7 @@ async def send_baton_approval_request(self, message: BatonRequestMessage) -> Non
249354
logger.warning("Ignoring baton request as client does not have baton")
250355
elif requester in self.uuids:
251356
processed_msg = BatonRequestMessage(requester=requester)
252-
logger.debug("Baton approved for %s", requester)
357+
logger.debug("Baton requested for %s", requester)
253358
msg = ws_pack(processed_msg)
254359
if msg is not None:
255360
for c in self.clients_with_uuid(self.baton):
@@ -271,6 +376,7 @@ async def take_baton(self, message: BatonDonateMessage) -> bool:
271376
uuid = message.receiver
272377
if uuid in self.uuids:
273378
self.baton = uuid
379+
logger.debug("Baton approved for %s", uuid)
274380
await self.update_baton()
275381
return True
276382

@@ -488,103 +594,12 @@ def combine_line_messages(
488594
new_points_msg : MultiLineMessage
489595
new points to append to current data lines.
490596
"""
491-
if not new_points_msg.append:
492-
raise ValueError(f"New data is not marked as append: {new_points_msg}")
493-
494597
ml_data_msg = self.plot_states[plot_id].current_data
495598
if not isinstance(ml_data_msg, MultiLineMessage):
496599
raise ValueError(
497600
f"Wrong type of message given: MultiLineMessage expected: {type(ml_data_msg)}"
498601
)
499-
500-
current_lines = ml_data_msg.ml_data
501-
add_colour_to_lines(new_points_msg.ml_data)
502-
new_points = new_points_msg.ml_data
503-
default_indices = current_lines[0].default_indices
504-
current_lines_len = len(current_lines)
505-
new_points_len = len(new_points)
506-
507-
def _append(a: DvDNDArray | None, b: DvDNDArray | None):
508-
if a is None:
509-
return b
510-
if b is None:
511-
return a
512-
return np.append(a, b)
513-
514-
if not default_indices:
515-
combined_lines = [
516-
LineData(
517-
line_params=c.line_params,
518-
x=_append(c.x, p.x),
519-
y=np.append(c.y, p.y),
520-
default_indices=False,
521-
)
522-
for c, p in zip(current_lines, new_points)
523-
]
524-
525-
if current_lines_len > new_points_len:
526-
combined_lines += current_lines[new_points_len:]
527-
528-
elif new_points_len > current_lines_len:
529-
combined_lines += new_points[current_lines_len:]
530-
531-
else:
532-
indexed_lines = []
533-
combined_lines = []
534-
for c, p in zip(current_lines, new_points):
535-
c_y_size = c.y.size
536-
total_y_size = c_y_size + p.y.size
537-
indexed_lines.append(
538-
LineData(
539-
line_params=p.line_params,
540-
x=np.arange(
541-
c_y_size,
542-
total_y_size,
543-
dtype=np.min_scalar_type(total_y_size),
544-
),
545-
y=p.y,
546-
default_indices=True,
547-
)
548-
)
549-
combined_lines.append(
550-
LineData(
551-
line_params=c.line_params,
552-
x=_append(
553-
c.x,
554-
np.arange(
555-
c_y_size,
556-
total_y_size,
557-
dtype=np.min_scalar_type(total_y_size),
558-
),
559-
),
560-
y=np.append(c.y, p.y),
561-
default_indices=True,
562-
)
563-
)
564-
if current_lines_len > new_points_len:
565-
combined_lines += current_lines[new_points_len:]
566-
567-
elif new_points_len > current_lines_len:
568-
extra_indexed_lines = [
569-
LineData(
570-
line_params=p.line_params,
571-
x=np.arange(p.y.size, dtype=np.min_scalar_type(p.y.size)),
572-
y=p.y,
573-
default_indices=True,
574-
)
575-
for p in new_points[current_lines_len:]
576-
]
577-
combined_lines += extra_indexed_lines
578-
indexed_lines += extra_indexed_lines
579-
580-
new_points_msg.ml_data = indexed_lines
581-
582-
return (
583-
MultiLineMessage(
584-
ml_data=combined_lines, plot_config=ml_data_msg.plot_config
585-
),
586-
new_points_msg,
587-
)
602+
return combine_line_messages(ml_data_msg, new_points_msg)
588603

589604
async def update_plot_states_with_message(
590605
self,
@@ -783,13 +798,13 @@ async def handle_client(server: PlotServer, plot_id: str, socket: WebSocket, uui
783798
try:
784799
while True:
785800
update_all = False
786-
message = await socket.receive()
787-
if message["type"] == "websocket.disconnect":
801+
raw_message = await socket.receive()
802+
if raw_message["type"] == "websocket.disconnect":
788803
logger.debug("Websocket disconnected: %s:%s", client.name, client.uuid)
789804
update_all = await server.remove_client(plot_id, client)
790805
break
791806

792-
message = ws_unpack(message["bytes"])
807+
message = ws_unpack(raw_message["bytes"])
793808
try:
794809
received_message = as_model(message)
795810
except ValidationError:
@@ -827,10 +842,18 @@ async def handle_client(server: PlotServer, plot_id: str, socket: WebSocket, uui
827842
update_all = await server.take_baton(received_message)
828843
else:
829844
logger.warning("Baton approval received from non-baton holder")
845+
case None:
846+
logger.warning(
847+
"Unparseable message from %s for %s: msg=%s\nraw=%s",
848+
uuid,
849+
plot_id,
850+
message,
851+
raw_message,
852+
)
853+
update_all = False
830854
case _:
831855
omit = None
832856

833-
is_valid = True
834857
logger.debug(
835858
"Got from %s from client %s: %s",
836859
plot_id,
@@ -847,9 +870,18 @@ async def handle_client(server: PlotServer, plot_id: str, socket: WebSocket, uui
847870
)
848871

849872
if is_valid:
850-
await server.prepare_client(
851-
plot_id, received_message, omit_client=omit
852-
)
873+
try:
874+
assert isinstance(received_message, ClientMessage)
875+
await server.prepare_client(
876+
plot_id, received_message, omit_client=omit
877+
)
878+
except Exception:
879+
logger.debug(
880+
"Failed with message type: %s",
881+
type(received_message),
882+
exc_info=True,
883+
)
884+
853885
update_all = True
854886

855887
if update_all:

0 commit comments

Comments
 (0)