Coverage for src/distopf/plot.py: 6%
460 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-13 17:34 -0800
1"""
2plot.py - Plotly visualization functions for LinDistModel results
3"""
5from typing import Optional
6import numpy as np
7import pandas as pd
8import plotly.express as px
9import plotly.graph_objects as go
11from distopf.matrix_models.base import LinDistBase
12from distopf.importer import Case
14def _choose_t(df, t=None):
15 df = df.copy()
16 if t is None:
17 t = 0
18 if "t" in df.columns:
19 if t is None and len(df) > 0:
20 t = min(df.t)
21 df = df.loc[df.t == t, :]
22 df = df.drop("t", axis=1)
23 return df
26def plot_voltages(v: pd.DataFrame, t=None) -> go.Figure:
27 """
28 Parameters
29 ----------
30 v : pd.DataFrame, Dataframe containing solved voltages for each bus.
31 Typically generated by the LinDistModel.get_voltages() method.
33 Returns
34 -------
35 fig : Plotly figure object
36 Plotly figure object containing the voltage magnitudes for each bus.
37 """
38 v = v.copy()
39 if "id" not in v.columns:
40 v["id"] = v.index
41 if "name" not in v.columns:
42 v["name"] = v["id"]
43 # if t is None and "t" in v.columns:
44 # t = min(v.t)
45 # if "t" in v.columns:
46 # v = v.loc[v.t == t, :]
47 # v = v.drop("t", axis=1)
48 v = _choose_t(v, t=t)
49 v = v.melt(
50 ignore_index=False, id_vars=["id", "name"], var_name="phase", value_name="v"
51 )
52 v["v"] = v["v"].astype(float)
53 fig = px.scatter(v, x=v.name, y="v", facet_col="phase", color="phase")
54 fig.for_each_annotation(lambda a: a.update(text=a.text.split("=")[-1].upper()))
55 fig.for_each_xaxis(lambda a: a.update(title="Bus Name"))
56 return fig
59def compare_voltages(v1: pd.DataFrame, v2: pd.DataFrame, t=None) -> go.Figure:
60 """
61 Visually compare voltages by plotting two different results.
62 Parameters
63 ----------
64 v1 : pd.DataFrame
65 v2 : pd.DataFrame
67 Returns
68 -------
69 fig : Plotly figure object
70 """
71 v1 = v1.copy()
72 v2 = v2.copy()
73 if "id" not in v1.columns:
74 v1["id"] = v1.index
75 if "name" not in v1.columns:
76 v1["name"] = v1["id"]
77 if "id" not in v2.columns:
78 v2["id"] = v2.index
79 if "name" not in v2.columns:
80 v2["name"] = v2["id"]
82 if t is None and "t" in v1.columns:
83 t1 = min(v1.t)
84 if t is None and "t" in v2.columns:
85 t2 = min(v2.t)
86 assert t1 == t2
87 if "t" in v1.columns:
88 v1 = pd.DataFrame(v1.loc[v1.t == t1, :])
89 v1 = v1.drop("t", axis=1)
90 if "t" in v2.columns:
91 v2 = pd.DataFrame(v2.loc[v2.t == t2, :])
92 v2 = v2.drop("t", axis=1)
93 v1 = v1.melt(
94 ignore_index=True, var_name="phase", id_vars=["id", "name"], value_name="v1"
95 )
96 v2 = v2.melt(
97 ignore_index=True, var_name="phase", id_vars=["id", "name"], value_name="v2"
98 )
99 v = pd.merge(v1, v2, on=["id", "name", "phase"])
100 v = v.melt(
101 ignore_index=True,
102 var_name="value",
103 id_vars=["id", "name", "phase"],
104 value_name="v",
105 ).sort_values(by=["id", "phase"])
106 # v1["v1"] = v1["v1"].astype(float)
107 # v2["v2"] = v2["v2"].astype(float)
108 # v = pd.merge(v1, v2, on=["name", "phase"])
109 fig = px.line(v, x="name", facet_col="phase", y="v", color="value", markers=True)
110 fig.for_each_annotation(lambda a: a.update(text=a.text.split("=")[-1].upper()))
111 fig.for_each_xaxis(lambda a: a.update(title="Bus Name"))
112 return fig
115def voltage_differences(v1: pd.DataFrame, v2: pd.DataFrame, t=None) -> go.Figure:
116 """
117 Visually compare voltages from two different results by plotting the difference v1-v2.
118 Parameters
119 ----------
120 v1 : pd.DataFrame
121 v2 : pd.DataFrame
123 Returns
124 -------
125 fig : Plotly figure object
126 """
127 v1 = v1.copy()
128 v2 = v2.copy()
129 if "id" not in v1.columns:
130 v1["id"] = v1.index
131 if "name" not in v1.columns:
132 v1["name"] = v1["id"]
133 if "id" not in v2.columns:
134 v2["id"] = v2.index
135 if "name" not in v2.columns:
136 v2["name"] = v2["id"]
137 if "t" in v1.columns:
138 if t is None:
139 t1 = min(v1.t)
140 v1 = pd.DataFrame(v1.loc[v1.t == t1, :])
141 v1 = v1.drop("t", axis=1)
142 if "t" in v2.columns:
143 if t is None:
144 t2 = min(v2.t)
145 v2 = pd.DataFrame(v2.loc[v2.t == t2, :])
146 v2 = v2.drop("t", axis=1)
147 v1 = v1.melt(
148 ignore_index=True, var_name="phase", id_vars=["id", "name"], value_name="v1"
149 )
150 v2 = v2.melt(
151 ignore_index=True, var_name="phase", id_vars=["id", "name"], value_name="v2"
152 )
153 v = pd.merge(v1, v2, on=["id", "name", "phase"])
154 v["diff"] = v["v1"] - v["v2"]
155 # fig = px.line(v, x="id", y="diff", facet_col="phase")
156 fig = px.bar(
157 v,
158 x="name",
159 y="diff",
160 color="phase",
161 barmode="group",
162 )
163 fig.for_each_annotation(lambda a: a.update(text=a.text.split("=")[-1].upper()))
164 fig.for_each_xaxis(lambda a: a.update(title="Bus Name"))
165 return fig
168def plot_power_flows(s: pd.DataFrame) -> go.Figure:
169 """
170 Plot the active and reactive power flowing into each bus on each phase.
171 Parameters
172 ----------
173 s : pd.DataFrame
175 Returns
176 -------
177 fig : Plotly figure object
178 """
179 s = s.copy()
180 s = s.melt(
181 ignore_index=True,
182 id_vars=["fb", "tb", "from_name", "to_name"],
183 var_name="phase",
184 value_name="s",
185 )
186 s["p"] = s.s.apply(lambda x: x.real)
187 s["q"] = s.s.apply(lambda x: x.imag)
188 del s["s"]
189 s = s.melt(
190 ignore_index=True,
191 id_vars=["fb", "tb", "from_name", "to_name", "phase"],
192 var_name="part",
193 value_name="power",
194 )
195 fig = px.bar(
196 s,
197 x="to_name",
198 y="power",
199 facet_col="phase",
200 facet_row="part",
201 color="phase",
202 labels={"to_name": "To-Bus Name"},
203 )
204 fig.for_each_annotation(lambda a: a.update(text=a.text.split("=")[-1].upper()))
205 fig.update_layout(
206 yaxis4_title="Active Power (p.u.)",
207 yaxis_title="Reactive Power (p.u.)",
208 )
209 return fig
212def plot_gens(p_gens: pd.DataFrame, q_gens: pd.DataFrame, t=None) -> go.Figure:
213 """
214 Plot the active and reactive power flowing into each bus on each phase.
215 Parameters
216 ----------
217 p_gens : pd.DataFrame
218 q_gens : pd.DataFrame
219 Returns
220 -------
221 fig : Plotly figure object
222 """
223 p_gens = p_gens.copy()
224 q_gens = q_gens.copy()
225 if t is None:
226 t = 0
227 if "t" in p_gens.columns:
228 t1 = t
229 if t is None and len(p_gens) > 0:
230 t1 = min(p_gens.t)
231 p_gens = pd.DataFrame(p_gens.loc[p_gens.t == t1, :])
232 p_gens = p_gens.drop("t", axis=1)
233 if "t" in q_gens.columns:
234 t2 = t
235 if t is None and len(q_gens) > 0:
236 t2 = min(q_gens.t)
237 q_gens = pd.DataFrame(q_gens.loc[q_gens.t == t2, :])
238 q_gens = q_gens.drop("t", axis=1)
240 p_gens = p_gens.melt(
241 ignore_index=True,
242 id_vars=["id", "name"],
243 var_name="phase",
244 value_name="P",
245 )
246 q_gens = q_gens.melt(
247 ignore_index=True,
248 id_vars=["id", "name"],
249 var_name="phase",
250 value_name="Q",
251 )
252 gens = pd.merge(p_gens, q_gens, how="outer", on=["id", "name", "phase"])
253 gens.id = p_gens.id
254 gens.name = p_gens.name
255 gens.phase = p_gens.phase
256 gens["P"] = p_gens["P"]
257 gens["Q"] = q_gens["Q"]
258 gens = gens.melt(
259 ignore_index=False,
260 id_vars=["id", "name", "phase"],
261 var_name="part",
262 value_name="power",
263 )
264 fig = px.bar(
265 gens,
266 x="name",
267 y="power",
268 facet_col="phase",
269 facet_row="part",
270 color="phase",
271 labels={"name": "Bus Name"},
272 )
273 fig.for_each_annotation(lambda a: a.update(text=a.text.split("=")[-1].upper()))
274 fig.update_layout(
275 yaxis4_title="Active Power (p.u.)",
276 yaxis_title="Reactive Power (p.u.)",
277 )
278 return fig
281def plot_pq(p: pd.DataFrame, q: pd.DataFrame, t=None) -> go.Figure:
282 """
283 Plot the active and reactive power.
284 Parameters
285 ----------
286 p : pd.DataFrame
287 q : pd.DataFrame
288 Returns
289 -------
290 fig : Plotly figure object
291 """
292 p = p.copy()
293 q = q.copy()
294 if t is None:
295 t = 0
296 if "t" in p.columns:
297 t1 = t
298 if t is None and len(p) > 0:
299 t1 = min(p.t)
300 p = pd.DataFrame(p.loc[p.t == t1, :])
301 p = p.drop("t", axis=1)
302 if "t" in q.columns:
303 t2 = t
304 if t is None and len(q) > 0:
305 t2 = min(q.t)
306 q = pd.DataFrame(q.loc[q.t == t2, :])
307 q = q.drop("t", axis=1)
309 p = p.melt(
310 ignore_index=True,
311 id_vars=["id", "name"],
312 var_name="phase",
313 value_name="P",
314 )
315 q = q.melt(
316 ignore_index=True,
317 id_vars=["id", "name"],
318 var_name="phase",
319 value_name="Q",
320 )
321 pq = pd.merge(p, q, how="outer", on=["id", "name", "phase"])
322 pq.id = p.id
323 pq.name = p.name
324 pq.phase = p.phase
325 pq["P"] = p["P"]
326 pq["Q"] = q["Q"]
327 pq = pq.melt(
328 ignore_index=False,
329 id_vars=["id", "name", "phase"],
330 var_name="part",
331 value_name="power",
332 )
333 fig = px.bar(
334 pq,
335 x="name",
336 y="power",
337 facet_col="phase",
338 facet_row="part",
339 color="phase",
340 labels={"name": "Bus Name"},
341 )
342 fig.for_each_annotation(lambda a: a.update(text=a.text.split("=")[-1].upper()))
343 fig.update_layout(
344 yaxis4_title="Active Power (p.u.)",
345 yaxis_title="Reactive Power (p.u.)",
346 )
347 return fig
350def compare_flows(s1: pd.DataFrame, s2: pd.DataFrame) -> go.Figure:
351 """
352 Similar to plot_power_flows but plots two results side by side.
353 Parameters
354 ----------
355 s1 : pd.DataFrame
356 s2 : pd.DataFrame
358 Returns
359 -------
360 fig : Plotly figure object
361 """
362 s1 = s1.copy()
363 s2 = s2.copy()
364 if "from_name" not in s1.columns:
365 s1["from_name"] = s1.fb
366 if "to_name" not in s1.columns:
367 s1["to_name"] = s1.tb
368 if "from_name" not in s2.columns:
369 s2["from_name"] = s2.fb
370 if "to_name" not in s2.columns:
371 s2["to_name"] = s2.tb
373 s1 = s1.melt(
374 ignore_index=True,
375 id_vars=["fb", "tb", "from_name", "to_name"],
376 var_name="phase",
377 value_name="s",
378 )
379 s1["p"] = s1.s.apply(np.real)
380 s1["q"] = s1.s.apply(np.imag)
381 del s1["s"]
382 s1 = s1.melt(
383 ignore_index=True,
384 id_vars=["fb", "tb", "from_name", "to_name", "phase"],
385 var_name="part",
386 value_name="s1",
387 )
388 s2 = s2.melt(
389 ignore_index=True,
390 id_vars=["fb", "tb", "from_name", "to_name"],
391 var_name="phase",
392 value_name="s",
393 )
394 s2["p"] = s2.s.apply(np.real)
395 s2["q"] = s2.s.apply(np.imag)
396 del s2["s"]
397 s2 = s2.melt(
398 ignore_index=True,
399 id_vars=["fb", "tb", "from_name", "to_name", "phase"],
400 var_name="part",
401 value_name="s2",
402 )
403 s = pd.merge(
404 s1, s2, on=["fb", "tb", "from_name", "to_name", "phase", "part"], how="outer"
405 )
406 fig = px.bar(
407 s,
408 x="to_name",
409 y=["s1", "s2"],
410 facet_col="phase",
411 facet_row="part",
412 barmode="group",
413 labels={"to_name": "To-Bus Name"},
414 )
415 fig.for_each_annotation(lambda a: a.update(text=a.text.split("=")[-1].upper()))
416 fig.update_layout(
417 yaxis4_title="Active Power (p.u.)",
418 yaxis_title="Reactive Power (p.u.)",
419 )
420 return fig
423def plot_ders(ders: pd.DataFrame) -> go.Figure:
424 """
425 Plot the generated power for each DER.
426 Parameters
427 ----------
428 ders : pd.DataFrame
430 Returns
431 -------
432 fig : Plotly figure object
433 """
434 ders = ders.copy()
435 dec_var = ders.melt(
436 ignore_index=False,
437 var_name="phase",
438 value_name="Generated Power (p.u.)",
439 id_vars=["id", "name"],
440 )
441 fig = px.bar(
442 dec_var,
443 x=dec_var.name,
444 y="Generated Power (p.u.)",
445 color="phase",
446 barmode="group",
447 labels={"name": "DER Bus Name"},
448 )
449 fig.for_each_annotation(lambda a: a.update(text=a.text.split("=")[-1]).upper())
450 return fig
453def plot_polar(p: pd.DataFrame, q: pd.DataFrame, t=None) -> go.Figure:
454 p = p.copy()
455 q = q.copy()
456 if t is None:
457 t = 0
458 if "t" in p.columns:
459 t1 = t
460 if t is None and len(p) > 0:
461 t1 = min(p.t)
462 p = pd.DataFrame(p.loc[p.t == t1, :])
463 p = p.drop("t", axis=1)
464 if "t" in q.columns:
465 t2 = t
466 if t is None and len(q) > 0:
467 t2 = min(q.t)
468 q = pd.DataFrame(q.loc[q.t == t2, :])
469 q = q.drop("t", axis=1)
471 p = p.melt(
472 ignore_index=True,
473 id_vars=["id", "name"],
474 var_name="phase",
475 value_name="value",
476 )
477 q = q.melt(
478 ignore_index=True,
479 id_vars=["id", "name"],
480 var_name="phase",
481 value_name="value",
482 )
483 pq = pd.merge(p, q, how="outer", on=["id", "name", "phase"])
484 pq.id = p.id
485 pq.name = p.name
486 pq.phase = p.phase
487 pq["p"] = p.value
488 pq["q"] = q.value
489 pq["s"] = pq.p + 1j * pq.q
490 pq["r"] = np.abs(pq.s)
491 pq["th"] = np.angle(pq.s, deg=True)
492 fig = px.scatter_polar(
493 pq,
494 r="r",
495 theta="th",
496 color="phase",
497 # range_theta=[-90, 90],
498 start_angle=0,
499 direction="counterclockwise",
500 hover_data="name",
501 )
502 return fig
505def plot_batteries(p: pd.DataFrame, soc: pd.DataFrame) -> go.Figure:
506 df = p.copy()
507 p_last = df.loc[df.t == df.t.max(), :].copy()
508 p_last.loc[:, "t"] = df.t.max() + 1
509 df = pd.concat([df, p_last]).reset_index(drop=True)
510 soc = soc.copy()
511 soc.t = soc.t + 1
512 soc.value = soc.value * 100
513 df["value"] = df.a + df.b + df.c
514 df["variable"] = "power"
515 df.drop(["a", "b", "c"], axis=1)
516 soc["variable"] = "soc"
517 df = pd.concat([df, soc])
518 fig = px.line(
519 df,
520 x="t",
521 y="value",
522 facet_row="variable",
523 color="name",
524 labels={"name": "Bus Name"},
525 )
526 for trace in fig.data:
527 if trace.yaxis == "y2": # Top row uses y2 axis
528 trace.update(line_shape="hv")
529 fig.for_each_annotation(lambda a: a.update(text=""))
530 fig.update_layout(
531 yaxis2_title="Discharging Power (p.u.)",
532 yaxis_title="SOC (%)",
533 yaxis=dict(matches=None), # Make bottom y-axis independent
534 yaxis2=dict(matches=None), # Make top y-axis independent
535 )
537 return fig
540def plot_network(
541 model: LinDistBase | Case,
542 v: Optional[pd.DataFrame] = None,
543 s: Optional[pd.DataFrame] = None,
544 p_flow: Optional[pd.DataFrame] = None,
545 q_flow: Optional[pd.DataFrame] = None,
546 p_gen: Optional[pd.DataFrame] = None,
547 q_gen: Optional[pd.DataFrame] = None,
548 v_min: float = 0.95,
549 v_max: float = 1.05,
550 show_phases: str = "abc",
551 show_reactive_power: bool = False,
552 t: Optional[int] = None,
553) -> go.Figure:
554 """
555 Plot the distribution network showing voltage and power results.
556 Parameters
557 ----------
558 model : LinDistBase
559 v : pd.DataFrame, (default=None) Dataframe containing voltage magnitudes for each bus.
560 s : pd.DataFrame, (default=None) Dataframe containing power flows for each branch. Either s or p_flow and q_flow should be provided.
561 p_flow : pd.DataFrame, (default=None) Dataframe containing power flows for each branch.
562 q_flow : pd.DataFrame, (default=None) Dataframe containing power flows for each branch.
563 p_gen : pd.DataFrame, (default=None) Dataframe containing actve power generated by each generator.
564 q_gen : pd.DataFrame, (default=None) Dataframe containing reactive power generated by each generator.
565 v_min : (default=0.95) Used for scaling node colors.
566 v_max : (default=1.05) Used for scaling node colors.
567 show_phases : (default="abc") valid options: "a", "b", "c", or "abc"
568 show_reactive_power : (default=False) If True, show reactive power flows instead of active power flows.
570 Returns
571 -------
572 fig: plotly.graph_objects.Figure
573 """
574 _v = None
575 _s = None
576 if s is not None:
577 _s = _choose_t(s.copy(), t)
578 if v is not None:
579 _v = _choose_t(v.copy(), t)
580 if s is None and p_flow is not None and q_flow is not None:
581 from_bus_map = {
582 int(tb): int(fb)
583 for fb, tb in model.branch_data.loc[:, ["fb", "tb"]].to_numpy()
584 }
585 _s = p_flow.copy()
586 _s = _s.drop(["a", "b", "c"], axis=1)
587 _s["a"] = p_flow.a + 1j * q_flow.a
588 _s["b"] = p_flow.b + 1j * q_flow.b
589 _s["c"] = p_flow.c + 1j * q_flow.c
590 _s["tb"] = _s["id"]
591 _s["fb"] = _s["tb"].map(from_bus_map)
592 _s = _choose_t(_s, t)
593 if p_gen is not None:
594 p_gen = _choose_t(p_gen, t)
595 if q_gen is not None:
596 q_gen = _choose_t(q_gen, t)
598 # validate phases
599 if show_phases.lower() not in ["a", "b", "c", "abc"]:
600 raise ValueError("Invalid phase. Must be 'a', 'b', 'c', or 'abc'.")
601 show_phases = show_phases.lower()
602 phase_list = sorted([ph.lower() for ph in show_phases])
603 bus_data = model.bus_data.copy()
604 branch_data = model.branch_data.copy()
605 gen_data = model.gen_data.copy()
606 cap_data = model.cap_data.copy()
608 node_size = 10
609 edge_scale = 10
610 edge_min = 1
612 bus_data = _process_bus_data(bus_data, _v, phase_list)
613 branch_data = _process_branch_data(
614 branch_data, bus_data, _s, phase_list, edge_scale, edge_min
615 )
616 gen_data = _process_gen_data(gen_data, p_gen, q_gen)
618 node_trace = _make_node_trace(bus_data, node_size, v_max, v_min)
619 cap_trace, gen_trace, substation_trace = _make_asset_markers(
620 bus_data, cap_data, gen_data, node_size
621 )
622 edge_traces = _make_edge_traces(branch_data, show_phases, show_reactive_power)
623 reverse_flow_markers_trace = _make_reverse_flow_marker_trace(
624 branch_data, node_size, show_reactive_power
625 )
627 node_trace.text = _make_hover_text(branch_data, bus_data, cap_data, gen_data)
628 title = _make_title(show_phases, show_reactive_power)
630 fig = go.Figure(
631 data=[
632 *edge_traces,
633 substation_trace,
634 reverse_flow_markers_trace,
635 cap_trace,
636 gen_trace,
637 node_trace,
638 ]
639 )
641 fig.update_layout(
642 title=title,
643 plot_bgcolor="White",
644 paper_bgcolor="White",
645 title_font_color="Black",
646 legend_font_color="Black",
647 legend_bgcolor="White",
648 margin=dict(t=50, b=10, l=10, r=10),
649 xaxis=dict(visible=False),
650 yaxis=dict(visible=False),
651 legend=dict(x=0.8, y=0.9),
652 font=dict(family="Droid Sans Mono", color="Black"),
653 # annotations=arrow_list[0:1],
654 )
656 return fig
659def _process_bus_data(bus_data, _v, phase_list):
660 bus_data["y"] = bus_data.latitude - bus_data.latitude.mean()
661 bus_data["x"] = bus_data.longitude - bus_data.longitude.mean()
662 bus_data["color"] = "white"
663 if bus_data.x.abs().max() > 0:
664 bus_data.x = bus_data.x / bus_data.x.abs().max()
665 if bus_data.y.abs().max() > 0:
666 bus_data.y = bus_data.y / bus_data.y.abs().max()
667 if _v is not None:
668 _v.index = _v.id - 1
669 bus_data.v_a = _v.a
670 bus_data.v_b = _v.b
671 bus_data.v_c = _v.c
672 bus_data.color = _v[phase_list].mean(axis=1)
673 return bus_data
676def _process_branch_data(branch_data, bus_data, _s, phase_list, edge_scale, edge_min):
677 branch_data = branch_data.loc[branch_data.status != "OPEN", :]
678 branch_data.index = branch_data.tb.to_numpy() - 1
679 branch_data["s_a"] = complex(np.nan, np.nan)
680 branch_data["s_b"] = complex(np.nan, np.nan)
681 branch_data["s_c"] = complex(np.nan, np.nan)
682 branch_data["p_abs"] = 1
683 branch_data["q_abs"] = 1
684 branch_data["p_norm"] = edge_min
685 branch_data["q_norm"] = edge_min
686 branch_data["p_direction"] = 1
687 branch_data["q_direction"] = 1
688 branch_data["x0"] = bus_data.loc[branch_data.fb.to_numpy() - 1].x.to_numpy()
689 branch_data["x1"] = bus_data.loc[branch_data.tb.to_numpy() - 1].x.to_numpy()
690 branch_data["y0"] = bus_data.loc[branch_data.fb.to_numpy() - 1].y.to_numpy()
691 branch_data["y1"] = bus_data.loc[branch_data.tb.to_numpy() - 1].y.to_numpy()
692 branch_data["x_mid"] = 1 / 2 * (branch_data.x0 + branch_data.x1)
693 branch_data["y_mid"] = 1 / 2 * (branch_data.y0 + branch_data.y1)
694 if _s is not None:
695 _s.index = _s.tb.to_numpy() - 1
696 branch_data["s_a"] = _s.a
697 branch_data["s_b"] = _s.b
698 branch_data["s_c"] = _s.c
699 branch_data["p_abs"] = np.abs(np.real(_s.loc[:, phase_list].sum(axis=1)))
700 branch_data["q_abs"] = np.abs(np.imag(_s.loc[:, phase_list].sum(axis=1)))
701 branch_data["p_norm"] = (
702 branch_data["p_abs"].to_numpy() / branch_data["p_abs"].max() * edge_scale
703 + edge_min
704 )
705 branch_data["q_norm"] = (
706 branch_data["q_abs"].to_numpy() / branch_data["q_abs"].max() * edge_scale
707 + edge_min
708 )
709 branch_data["p_direction"] = np.sign(
710 np.real(_s.loc[:, phase_list].sum(axis=1)) + 1e-6
711 )
712 branch_data["q_direction"] = np.sign(
713 np.imag(_s.loc[:, phase_list].sum(axis=1)) + 1e-6
714 )
715 return branch_data
718def _process_gen_data(gen_data, p_gen, q_gen):
719 gen_data.index = gen_data.id.to_numpy() - 1
720 if p_gen is not None:
721 p_gen.index = p_gen.id.to_numpy() - 1
722 gen_data.pa = p_gen.a
723 gen_data.pb = p_gen.b
724 gen_data.pc = p_gen.c
725 if q_gen is not None:
726 q_gen.index = q_gen.id.to_numpy() - 1
727 gen_data.qa = q_gen.a
728 gen_data.qb = q_gen.b
729 gen_data.qc = q_gen.c
730 return gen_data
733def _make_reverse_flow_marker_trace(branch_data, node_size, show_reactive_power):
734 if show_reactive_power:
735 reverse_list_x = branch_data.loc[
736 branch_data.q_direction < 0, "x_mid"
737 ].to_numpy()
738 reverse_list_y = branch_data.loc[
739 branch_data.q_direction < 0, "y_mid"
740 ].to_numpy()
741 else:
742 reverse_list_x = branch_data.loc[
743 branch_data.p_direction < 0, "x_mid"
744 ].to_numpy()
745 reverse_list_y = branch_data.loc[
746 branch_data.p_direction < 0, "y_mid"
747 ].to_numpy()
748 reverse_flow_markers_trace = go.Scatter()
749 if not show_reactive_power:
750 reverse_flow_markers_trace = go.Scatter(
751 x=reverse_list_x,
752 y=reverse_list_y,
753 mode="markers",
754 marker=dict(
755 symbol="star-triangle-up",
756 size=node_size * 1,
757 color="orange",
758 line_width=0.5,
759 line_color="black",
760 ),
761 showlegend=True,
762 name="Reverse Power Flow",
763 hoverinfo="none",
764 )
765 return reverse_flow_markers_trace
768def _make_edge_traces(branch_data, show_phases, show_reactive_power):
769 edge_traces = []
770 for _, edge in branch_data.iterrows():
771 x0, x1 = edge.x0, edge.x1
772 y0, y1 = edge.y0, edge.y1
773 dash = "solid"
774 linewidth = edge.p_norm
775 direction = edge.p_direction
776 if show_reactive_power:
777 linewidth = edge.q_norm
778 direction = edge.q_direction
779 if (
780 show_phases.lower() != "abc"
781 and show_phases.lower() not in edge.phases.lower()
782 ):
783 dash = "dot"
785 color = "darkblue"
786 if direction < 0:
787 color = "maroon"
788 edge_trace = go.Scatter(
789 x=[x0, x1],
790 y=[y0, y1],
791 mode="lines",
792 line=dict(color=color, width=linewidth, dash=dash),
793 showlegend=False,
794 hoverinfo="none",
795 )
796 edge_traces.append(edge_trace)
797 return edge_traces
800# def _make_hover_text(branch_data, bus_data, cap_data, gen_data):
801# text = [f"Bus: '{name}' A || B || C" for name in bus_data["name"]]
802# for i, bus_row in enumerate(bus_data.itertuples()):
803# # if _v is not None:
804# va = bus_row.v_a
805# vb = bus_row.v_b
806# vc = bus_row.v_c
807# text[i] = text[i] + f"<br> |V|: {va:.3f} {vb:.3f} {vc:.3f}"
809# pla = bus_row.pl_a
810# plb = bus_row.pl_b
811# plc = bus_row.pl_c
812# qla = bus_row.ql_a
813# qlb = bus_row.ql_b
814# qlc = bus_row.ql_c
815# text[i] += f"<br> P-Load: {pla:.3f} {plb:.3f} {plc:.3f}"
816# text[i] += f"<br> Q-Load: {qla:.3f} {qlb:.3f} {qlc:.3f}"
818# if cap_data is not None:
819# if bus_row.id in cap_data.id.to_numpy():
820# q_cap = cap_data.loc[
821# cap_data.id == bus_row.id, ["qa", "qb", "qc"]
822# ].to_numpy()[0]
823# text[i] += (
824# f"<br> Q-Cap: {q_cap[0]:.3f} {q_cap[1]:.3f} {q_cap[2]:.3f}"
825# )
827# if bus_row.id in gen_data.id.to_numpy():
828# p_gen = gen_data.loc[
829# gen_data.id == bus_row.id, ["pa", "pb", "pc"]
830# ].to_numpy()[0]
831# q_gen = gen_data.loc[
832# gen_data.id == bus_row.id, ["qa", "qb", "qc"]
833# ].to_numpy()[0]
834# text[i] += f"<br> P-Gen: {p_gen[0]:.3f} {p_gen[1]:.3f} {p_gen[2]:.3f}"
835# text[i] += f"<br> Q-Gen: {q_gen[0]:.3f} {q_gen[1]:.3f} {q_gen[2]:.3f}"
836# edge = branch_data.loc[branch_data.tb == bus_row.id, :]
837# if len(edge) == 0:
838# continue
839# to_name = bus_row.name
840# fb = edge.fb.to_numpy()[0]
841# from_name = bus_data.loc[bus_data.id == fb, "name"].to_numpy()[0]
842# sa, sb, sc = (
843# edge.s_a.to_numpy()[0],
844# edge.s_b.to_numpy()[0],
845# edge.s_c.to_numpy()[0],
846# )
847# new_text = (
848# f"<br>Branch {from_name}→{to_name}"
849# f"<br> P flow: {np.real(sa):.3f} {np.real(sb):.3f} {np.real(sc):.3f}"
850# f"<br> Q flow: {np.imag(sa):.3f} {np.imag(sb):.3f} {np.imag(sc):.3f}"
851# )
852# text[i] += new_text
853# return text
856def _make_hover_text(branch_data, bus_data, cap_data, gen_data):
857 text = []
858 for i, bus_row in enumerate(bus_data.itertuples()):
859 bus_phases = bus_row.phases.lower() # e.g. "abc", "ac", "ab", etc.
861 # Helper function to format value or dash with better alignment
862 def format_phase_value(value, phase):
863 if phase in bus_phases:
864 return f"{value:>7.3f}" # Increased width for better alignment
865 else:
866 return " --- " # Centered dashes with proper spacing
868 # Start with bus header
869 hover_text = ""
870 hover_text = f"<b> Bus: {bus_row.name:<5} (p.u.)</b><br>"
871 # Create a formatted table-like structure using spaces and HTML
872 hover_text += "<b> A B C</b><br>" # Adjusted spacing
873 # hover_text += "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━<br>"
875 # Voltage data
876 va, vb, vc = bus_row.v_a, bus_row.v_b, bus_row.v_c
877 va_str = format_phase_value(va, "a")
878 vb_str = format_phase_value(vb, "b")
879 vc_str = format_phase_value(vc, "c")
880 hover_text += f"⚡ V Mag {va_str} {vb_str} {vc_str}<br>"
882 # Load data
883 pla, plb, plc = bus_row.pl_a, bus_row.pl_b, bus_row.pl_c
884 qla, qlb, qlc = bus_row.ql_a, bus_row.ql_b, bus_row.ql_c
885 pla_str = format_phase_value(pla, "a")
886 plb_str = format_phase_value(plb, "b")
887 plc_str = format_phase_value(plc, "c")
888 qla_str = format_phase_value(qla, "a")
889 qlb_str = format_phase_value(qlb, "b")
890 qlc_str = format_phase_value(qlc, "c")
891 hover_text += f"💡 P Load {pla_str} {plb_str} {plc_str}<br>"
892 hover_text += f"💡 Q Load {qla_str} {qlb_str} {qlc_str}<br>"
894 # Capacitor data (if present)
895 if cap_data is not None and bus_row.id in cap_data.id.to_numpy():
896 q_cap = cap_data.loc[
897 cap_data.id == bus_row.id, ["qa", "qb", "qc"]
898 ].to_numpy()[0]
899 qcap_a_str = format_phase_value(q_cap[0], "a")
900 qcap_b_str = format_phase_value(q_cap[1], "b")
901 qcap_c_str = format_phase_value(q_cap[2], "c")
902 hover_text += f"🔃 Q Cap {qcap_a_str} {qcap_b_str} {qcap_c_str}<br>"
904 # Generator data (if present)
905 if bus_row.id in gen_data.id.to_numpy():
906 p_gen = gen_data.loc[
907 gen_data.id == bus_row.id, ["pa", "pb", "pc"]
908 ].to_numpy()[0]
909 q_gen = gen_data.loc[
910 gen_data.id == bus_row.id, ["qa", "qb", "qc"]
911 ].to_numpy()[0]
912 pgen_a_str = format_phase_value(p_gen[0], "a")
913 pgen_b_str = format_phase_value(p_gen[1], "b")
914 pgen_c_str = format_phase_value(p_gen[2], "c")
915 qgen_a_str = format_phase_value(q_gen[0], "a")
916 qgen_b_str = format_phase_value(q_gen[1], "b")
917 qgen_c_str = format_phase_value(q_gen[2], "c")
918 hover_text += f"⚙️ P Gen {pgen_a_str} {pgen_b_str} {pgen_c_str}<br>"
919 hover_text += f"⚙️ Q Gen {qgen_a_str} {qgen_b_str} {qgen_c_str}<br>"
921 # Branch flow data (if present)
922 edge = branch_data.loc[branch_data.tb == bus_row.id, :]
923 if len(edge) > 0:
924 to_name = bus_row.name
925 fb = edge.fb.to_numpy()[0]
926 from_name = bus_data.loc[bus_data.id == fb, "name"].to_numpy()[0]
927 sa, sb, sc = (
928 edge.s_a.to_numpy()[0],
929 edge.s_b.to_numpy()[0],
930 edge.s_c.to_numpy()[0],
931 )
933 pflow_a_str = format_phase_value(np.real(sa), "a")
934 pflow_b_str = format_phase_value(np.real(sb), "b")
935 pflow_c_str = format_phase_value(np.real(sc), "c")
936 qflow_a_str = format_phase_value(np.imag(sa), "a")
937 qflow_b_str = format_phase_value(np.imag(sb), "b")
938 qflow_c_str = format_phase_value(np.imag(sc), "c")
940 hover_text += "<br>"
941 hover_text += f"<b> Branch: {from_name} → {to_name}</b><br>"
942 hover_text += f"➡️ P Flow {pflow_a_str} {pflow_b_str} {pflow_c_str}<br>"
943 hover_text += f"➡️ Q Flow {qflow_a_str} {qflow_b_str} {qflow_c_str}<br>"
944 text.append(hover_text)
945 return text
948def _make_title(show_phases, show_reactive_power):
949 title = "<b>Network Plot (P.U.)</b>"
950 # if _v is not None:
951 title = title + "<br>Node color: "
952 if show_phases == "abc":
953 title = title + "Average voltage magnitude"
954 else:
955 title = title + f"Phase-{show_phases.upper()} voltage magnitude"
956 # if _s is not None:
957 title = title + "<br>Line width:"
958 if show_phases == "abc":
959 title = title + " Total"
960 else:
961 title = title + f" Phase-{show_phases.upper()}"
962 if show_reactive_power:
963 title = title + " <i>reactive</i> power flow"
964 else:
965 title = title + " <i>active</i> power flow"
966 title = title + " (reverse flow in red)."
967 return title
970def _make_asset_markers(bus_data, cap_data, gen_data, node_size):
971 # Add substation marker
972 substation_buses = bus_data.loc[bus_data.bus_type == "SWING", :]
973 substation_trace = go.Scatter(
974 x=substation_buses["x"],
975 y=substation_buses["y"],
976 mode="markers",
977 marker=dict(
978 symbol="square",
979 size=node_size * 2,
980 color="black",
981 line_width=1,
982 line_color="black",
983 ),
984 showlegend=True,
985 name="Substation",
986 hoverinfo="none",
987 )
988 # Add generator markers
989 gen_buses = bus_data.loc[bus_data.id.isin(gen_data.id), :]
990 gen_trace = go.Scatter(
991 x=gen_buses["x"],
992 y=gen_buses["y"],
993 mode="markers",
994 marker=dict(
995 symbol="square",
996 size=node_size * 2,
997 color="white",
998 line_width=1,
999 line_color="black",
1000 ),
1001 showlegend=True,
1002 name="Generators",
1003 hoverinfo="none",
1004 )
1005 # Add capacitor markers
1006 cap_buses = bus_data.loc[bus_data.id.isin(cap_data.id), :]
1007 cap_trace = go.Scatter(
1008 x=cap_buses["x"],
1009 y=cap_buses["y"],
1010 mode="markers",
1011 marker=dict(
1012 symbol="star-diamond",
1013 size=node_size * 2,
1014 color="white",
1015 line_width=1,
1016 line_color="black",
1017 ),
1018 showlegend=True,
1019 name="Capacitors",
1020 hoverinfo="none",
1021 )
1022 return cap_trace, gen_trace, substation_trace
1025def _make_node_trace(bus_data, node_size, v_max, v_min):
1026 node_trace = go.Scatter(
1027 x=bus_data["x"],
1028 y=bus_data["y"],
1029 mode="markers",
1030 marker=dict(
1031 showscale=True,
1032 cmin=v_min,
1033 cmax=v_max,
1034 # colorscale options
1035 # 'Greys' | 'YlGnBu' | 'Greens' | 'YlOrRd' | 'Bluered' | 'RdBu' |
1036 # 'Reds' | 'Blues' | 'Picnic' | 'Rainbow' | 'Portland' | 'Jet' |
1037 # 'Hot' | 'Blackbody' | 'Earth' | 'Electric' | 'Viridis' | 'Aggrnyl'
1038 colorscale="Viridis",
1039 reversescale=False,
1040 size=node_size,
1041 color=bus_data.color,
1042 line_width=node_size / 5,
1043 line_color="black",
1044 ),
1045 showlegend=False,
1046 text=bus_data["name"],
1047 hovertemplate="%{text}",
1048 hoverlabel=dict(font=dict(family="Monospace")),
1049 textposition="top center",
1050 )
1051 return node_trace