joppe/BondGraph
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Bond graph ports and drawing added

Browse Source
This commit is contained in:
Joppe Blondel
2026-07-22 12:24:58 +02:00
parent 0c578af85d
commit 09248421d0
12 changed files with 614 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
BEdit/src/bedit/core/model.py
View File

@@ -6,6 +6,7 @@ from typing import Any
from uuid import uuid4
from bedit.core.port_types import PortTypeRegistry
from bedit.core.power_domains import POWER_CAUSALITIES, POWER_DOMAINS
def _dimensions(data: Any, subject: str) -> tuple[int, int]:
@@ -36,6 +37,9 @@ class Port:
rows: int = 1
columns: int = 1
description: str = ""
orientation: str = "input"
domain: str = "power"
causality: str = "indifferent"
def to_dict(self) -> dict[str, Any]:
return {
@@ -50,10 +54,13 @@ class Port:
"unit": self.unit,
"dimensions": {"rows": self.rows, "columns": self.columns},
"description": self.description,
"orientation": self.orientation,
"domain": self.domain,
"causality": self.causality,
}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "Port":
def from_dict(cls, data: dict[str, Any], default_orientation: str = "input") -> "Port":
position = data.get("position", {})
rows, columns = _dimensions(data.get("dimensions", {}), "Port")
return cls(
@@ -70,6 +77,9 @@ class Port:
rows=rows,
columns=columns,
description=str(data.get("description", "")),
orientation=str(data.get("orientation", default_orientation)),
domain=str(data.get("domain", "power")),
causality=str(data.get("causality", "indifferent")),
)
@@ -417,8 +427,8 @@ class Component:
x=float(position.get("x", 0.0)),
y=float(position.get("y", 0.0)),
rotation=float(data.get("rotation", 0.0)),
inputs=[Port.from_dict(item) for item in interface.get("inputs", [])],
outputs=[Port.from_dict(item) for item in interface.get("outputs", [])],
inputs=[Port.from_dict(item, "input") for item in interface.get("inputs", [])],
outputs=[Port.from_dict(item, "output") for item in interface.get("outputs", [])],
parameters=[Parameter.from_dict(item) for item in parameters],
icon=Icon.from_dict(data.get("icon")),
properties=dict(data.get("properties", {})),
@@ -528,6 +538,15 @@ class GraphDocument:
raise ValueError(f"Component {owner.name} contains duplicate port IDs")
for port in (*owner.inputs, *owner.outputs):
PortTypeRegistry.get(port.type)
if port.orientation not in {"input", "output", "indifferent"}:
raise ValueError(f"Port {port.name!r} has an invalid orientation")
if port.orientation == "indifferent" and port.type != "power":
raise ValueError("Only power ports may have indifferent orientation")
if port.type == "power":
if port.domain not in {domain.id for domain in POWER_DOMAINS}:
raise ValueError(f"Power port {port.name!r} has an unknown domain")
if port.causality not in POWER_CAUSALITIES:
raise ValueError(f"Power port {port.name!r} has invalid causality")
if not port.value_type.strip():
raise ValueError(f"Port {port.name!r} has no value type")
if port.rows < 1 or port.columns < 1:
@@ -549,9 +568,17 @@ class GraphDocument:
source_port = next(p for p in owner.inputs if p.id == connection.source.interface)
else:
source = owner.graph.blocks.get(connection.source.block or "")
if source is None or connection.source.port not in {p.id for p in source.outputs}:
source_ports = (
[]
if source is None
else [
*source.outputs,
*(p for p in source.inputs if p.orientation == "indifferent"),
]
)
if source is None or connection.source.port not in {p.id for p in source_ports}:
raise ValueError(f"Connection {connection.id} uses an unknown block output")
source_port = next(p for p in source.outputs if p.id == connection.source.port)
source_port = next(p for p in source_ports if p.id == connection.source.port)
if connection.target.junction is not None:
junction = owner.graph.junctions.get(connection.target.junction)
if junction is None:
@@ -565,15 +592,27 @@ class GraphDocument:
allows_multiple_connections=False,
)
elif connection.target.interface is not None:
if connection.target.interface not in output_ids:
indifferent_ids = {
port.id for port in owner.inputs if port.orientation == "indifferent"
}
if connection.target.interface not in output_ids | indifferent_ids:
raise ValueError(f"Connection {connection.id} uses an unknown interface output")
target_port = next(p for p in owner.outputs if p.id == connection.target.interface)
target_port = next(
p
for p in (*owner.outputs, *owner.inputs)
if p.id == connection.target.interface
)
else:
target = owner.graph.blocks.get(connection.target.block or "")
if target is None or connection.target.port not in {p.id for p in target.inputs}:
raise ValueError(f"Connection {connection.id} uses an unknown block input")
target_port = next(p for p in target.inputs if p.id == connection.target.port)
if not PortTypeRegistry.compatible(source_port.type, target_port.type):
if not PortTypeRegistry.compatible(
source_port.type,
target_port.type,
source_port.domain,
target_port.domain,
):
raise ValueError(f"Connection {connection.id} joins incompatible port types")
source_key = (
"source-junction"
@@ -670,27 +709,11 @@ def clone_component(source: Component) -> Component:
x=current.x,
y=current.y,
inputs=[
Port(
port.id,
port.name,
port.x,
port.y,
deepcopy(port.properties),
port.type,
port.allows_multiple_connections,
)
Port.from_dict(port.to_dict(), "input")
for port in current.inputs
],
outputs=[
Port(
port.id,
port.name,
port.x,
port.y,
deepcopy(port.properties),
port.type,
port.allows_multiple_connections,
)
Port.from_dict(port.to_dict(), "output")
for port in current.outputs
],
parameters=deepcopy(current.parameters),

13
BEdit/src/bedit/core/port_types.py
View File

@@ -14,6 +14,7 @@ class PortType:
class PortTypeRegistry:
_types = {
"signal": PortType("signal", "Signal", "A scalar signal connection"),
"power": PortType("power", "Power", "A two-variable power connection"),
}
@classmethod
@@ -28,5 +29,13 @@ class PortTypeRegistry:
raise ValueError(f"Unknown port type: {type_id}") from error
@classmethod
def compatible(cls, first: str, second: str) -> bool:
return cls.get(first).accepts(cls.get(second))
def compatible(
cls,
first: str,
second: str,
first_domain: str = "",
second_domain: str = "",
) -> bool:
if not cls.get(first).accepts(cls.get(second)):
return False
return first != "power" or first_domain == second_domain

31
BEdit/src/bedit/core/power_domains.py Normal file
View File

@@ -0,0 +1,31 @@
"""Editable power-port domain definitions used by the port editor."""
from dataclasses import dataclass
@dataclass(frozen=True)
class PowerDomain:
id: str
display_name: str
effort: str
flow: str
POWER_DOMAINS: tuple[PowerDomain, ...] = (
PowerDomain("power", "Power", "p.e", "p.f"),
)
def power_domain(domain_id: str) -> PowerDomain:
return next((domain for domain in POWER_DOMAINS if domain.id == domain_id), POWER_DOMAINS[0])
POWER_CAUSALITIES = (
"fixed flow out",
"fixed effort out",
"preferred flow out",
"preferred effort out",
"likes flow out",
"likes effort out",
"indifferent",
)

30
BEdit/src/bedit/gui/controllers/document.py
View File

@@ -285,7 +285,12 @@ class DocumentController(QObject):
target_port = self._port_for_endpoint(target, "target")
if source_port is None or target_port is None:
raise ValueError("A connection endpoint no longer exists")
if not PortTypeRegistry.compatible(source_port.type, target_port.type):
if not PortTypeRegistry.compatible(
source_port.type,
target_port.type,
source_port.domain,
target_port.domain,
):
raise ValueError(f"Cannot connect {source_port.type!r} to {target_port.type!r}")
if not self.endpoint_accepts_connection(source, "source"):
raise ValueError(
@@ -573,11 +578,24 @@ class DocumentController(QObject):
)
if endpoint.interface is not None:
ports = owner.inputs if role == "source" else owner.outputs
ports = [
*ports,
*(
port
for port in owner.inputs
if port.orientation == "indifferent" and port not in ports
),
]
else:
component = owner.graph.blocks.get(endpoint.block or "")
if component is None:
return None
ports = component.outputs if role == "source" else component.inputs
if role == "source":
ports = [
*ports,
*(port for port in component.inputs if port.orientation == "indifferent"),
]
return next(
(port for port in ports if port.id == (endpoint.interface or endpoint.port)), None
)
@@ -664,6 +682,9 @@ class DocumentController(QObject):
raise ValueError("Only text-defined components can be edited here")
input_ids = [port.id for port in inputs]
output_ids = [port.id for port in outputs]
source_ids = output_ids + [
port.id for port in inputs if port.orientation == "indifferent"
]
if len(set(input_ids)) != len(input_ids) or len(set(output_ids)) != len(output_ids):
raise ValueError("Input and output IDs must be unique")
if any(not port.name.strip() for port in (*inputs, *outputs)):
@@ -686,7 +707,7 @@ class DocumentController(QObject):
)
if (
connection.source.block == component.id
and connection.source.port not in output_ids
and connection.source.port not in source_ids
):
raise ValueError(
f"Output {connection.source.port!r} is still connected in the containing graph"
@@ -842,6 +863,9 @@ class DocumentController(QObject):
return
input_ids = {port.id for port in inputs}
output_ids = {port.id for port in outputs}
source_ids = output_ids | {
port.id for port in inputs if port.orientation == "indifferent"
}
parent = self.document.find_parent(component_id)
if parent is not None:
for connection in parent.graph.connections.values():
@@ -852,7 +876,7 @@ class DocumentController(QObject):
raise ValueError("An input cannot be removed or reoriented while connected")
if (
connection.source.block == component_id
and connection.source.port not in output_ids
and connection.source.port not in source_ids
):
raise ValueError("An output cannot be removed or reoriented while connected")
for connection in component.graph.connections.values():

68
BEdit/src/bedit/gui/dialogs/port_options.py
View File

@@ -6,6 +6,7 @@ from PySide6.QtWidgets import QDialog, QDialogButtonBox, QListWidgetItem, QMessa
from bedit.core.model import Component, Port
from bedit.core.physical_types import QUANTITIES, UNITS
from bedit.core.power_domains import POWER_CAUSALITIES, POWER_DOMAINS, power_domain
from bedit.core.port_types import PortTypeRegistry
from bedit.gui.generated.ui_port_options_dialog import Ui_PortOptionsDialog
@@ -13,6 +14,13 @@ from bedit.gui.generated.ui_port_options_dialog import Ui_PortOptionsDialog
PORT_ROLE = Qt.ItemDataRole.UserRole
def _oriented_copy(port: Port, fallback: str) -> Port:
copied = deepcopy(port)
if copied.orientation != "indifferent":
copied.orientation = fallback
return copied
class PortOptionsDialog(QDialog):
"""Unified editor for a component's typed, oriented ports."""
@@ -25,8 +33,8 @@ class PortOptionsDialog(QDialog):
self.setWindowTitle(f"Port Options — {component.name}")
self.read_only = read_only
self.ports: list[tuple[Port, str]] = [
*((deepcopy(port), "input") for port in component.inputs),
*((deepcopy(port), "output") for port in component.outputs),
*((_oriented_copy(port, "input"), port.orientation or "input") for port in component.inputs),
*((_oriented_copy(port, "output"), "output") for port in component.outputs),
]
self._loading = False
self.ui.typeCombo.clear()
@@ -36,6 +44,10 @@ class PortOptionsDialog(QDialog):
self.ui.unitCombo.addItems(UNITS)
self.ui.orientationCombo.setItemData(0, "input")
self.ui.orientationCombo.setItemData(1, "output")
self.ui.orientationCombo.setItemData(2, "indifferent")
for domain in POWER_DOMAINS:
self.ui.domainCombo.addItem(domain.display_name, domain.id)
self.ui.causalityCombo.addItems(POWER_CAUSALITIES)
self.ui.portList.currentRowChanged.connect(self._load_current)
self.ui.addPortButton.clicked.connect(self.add_port)
self.ui.removePortButton.clicked.connect(self.remove_port)
@@ -49,6 +61,9 @@ class PortOptionsDialog(QDialog):
self.ui.rowsSpin.valueChanged.connect(self._store_current)
self.ui.columnsSpin.valueChanged.connect(self._store_current)
self.ui.descriptionEdit.textChanged.connect(self._store_current)
self.ui.domainCombo.currentIndexChanged.connect(self._power_domain_changed)
self.ui.causalityCombo.currentTextChanged.connect(self._store_current)
self.ui.powerDescriptionEdit.textChanged.connect(self._store_current)
self.ui.portSplitter.setSizes([250, 370])
if read_only:
self.ui.addPortButton.setEnabled(False)
@@ -64,7 +79,11 @@ class PortOptionsDialog(QDialog):
@property
def inputs(self) -> list[Port]:
return [port for port, orientation in self.ports if orientation == "input"]
return [
port
for port, orientation in self.ports
if orientation in {"input", "indifferent"}
]
@property
def outputs(self) -> list[Port]:
@@ -96,6 +115,9 @@ class PortOptionsDialog(QDialog):
self.ui.rowsSpin.setValue(port.rows)
self.ui.columnsSpin.setValue(port.columns)
self.ui.descriptionEdit.setPlainText(port.description)
self.ui.domainCombo.setCurrentIndex(self.ui.domainCombo.findData(port.domain))
self.ui.causalityCombo.setCurrentText(port.causality)
self.ui.powerDescriptionEdit.setPlainText(port.description)
self._show_type_options(port.type)
else:
self.ui.nameEdit.clear()
@@ -113,15 +135,27 @@ class PortOptionsDialog(QDialog):
self.ui.typeOptionsStack.setEnabled(enabled and not self.read_only)
def _port_type_changed(self) -> None:
self._show_type_options(self.ui.typeCombo.currentData())
port_type = self.ui.typeCombo.currentData()
indifferent_item = self.ui.orientationCombo.model().item(2)
indifferent_item.setEnabled(port_type == "power")
if port_type != "power" and self.ui.orientationCombo.currentData() == "indifferent":
self.ui.orientationCombo.setCurrentIndex(0)
self._show_type_options(port_type)
self._store_current()
def _show_type_options(self, port_type: str) -> None:
self.ui.typeOptionsStack.setCurrentWidget(
self.ui.signalOptionsPage
if port_type == "signal"
else self.ui.unsupportedTypePage
)
self.ui.orientationCombo.model().item(2).setEnabled(port_type == "power")
page = {
"signal": self.ui.signalOptionsPage,
"power": self.ui.powerOptionsPage,
}.get(port_type, self.ui.unsupportedTypePage)
self.ui.typeOptionsStack.setCurrentWidget(page)
def _power_domain_changed(self) -> None:
domain = power_domain(self.ui.domainCombo.currentData())
self.ui.effortValueLabel.setText(domain.effort)
self.ui.flowValueLabel.setText(domain.flow)
self._store_current()
def _store_current(self) -> None:
row = self.ui.portList.currentRow()
@@ -136,8 +170,16 @@ class PortOptionsDialog(QDialog):
port.unit = self.ui.unitCombo.currentText().strip()
port.rows = self.ui.rowsSpin.value()
port.columns = self.ui.columnsSpin.value()
port.description = self.ui.descriptionEdit.toPlainText()
self.ports[row] = (port, self.ui.orientationCombo.currentData())
port.domain = self.ui.domainCombo.currentData() or "power"
port.causality = self.ui.causalityCombo.currentText()
port.description = (
self.ui.powerDescriptionEdit.toPlainText()
if port.type == "power"
else self.ui.descriptionEdit.toPlainText()
)
orientation = self.ui.orientationCombo.currentData()
port.orientation = orientation
self.ports[row] = (port, orientation)
self.ui.portList.item(row).setText(
f"{port.name} [{self.ports[row][1]}, {port.type}]"
)
@@ -146,8 +188,8 @@ class PortOptionsDialog(QDialog):
def set_ports(self, inputs: list[Port], outputs: list[Port]) -> None:
self._loading = True
self.ports = [
*((deepcopy(port), "input") for port in inputs),
*((deepcopy(port), "output") for port in outputs),
*((_oriented_copy(port, "input"), port.orientation or "input") for port in inputs),
*((_oriented_copy(port, "output"), "output") for port in outputs),
]
self._loading = False
self._rebuild_list(0 if self.ports else -1)

64
BEdit/src/bedit/gui/generated/ui_port_options_dialog.py
View File

@@ -92,6 +92,7 @@ class Ui_PortOptionsDialog(object):
self.orientationCombo = QComboBox(self.portDetailsPanel)
self.orientationCombo.addItem("")
self.orientationCombo.addItem("")
self.orientationCombo.addItem("")
self.orientationCombo.setObjectName(u"orientationCombo")
self.portDetailsForm.setWidget(2, QFormLayout.ItemRole.FieldRole, self.orientationCombo)
@@ -185,6 +186,61 @@ class Ui_PortOptionsDialog(object):
self.signalOptionsForm.setWidget(4, QFormLayout.ItemRole.FieldRole, self.descriptionEdit)
self.typeOptionsStack.addWidget(self.signalOptionsPage)
self.powerOptionsPage = QWidget()
self.powerOptionsPage.setObjectName(u"powerOptionsPage")
self.powerOptionsForm = QFormLayout(self.powerOptionsPage)
self.powerOptionsForm.setObjectName(u"powerOptionsForm")
self.domainLabel = QLabel(self.powerOptionsPage)
self.domainLabel.setObjectName(u"domainLabel")
self.powerOptionsForm.setWidget(0, QFormLayout.ItemRole.LabelRole, self.domainLabel)
self.domainCombo = QComboBox(self.powerOptionsPage)
self.domainCombo.setObjectName(u"domainCombo")
self.powerOptionsForm.setWidget(0, QFormLayout.ItemRole.FieldRole, self.domainCombo)
self.effortLabel = QLabel(self.powerOptionsPage)
self.effortLabel.setObjectName(u"effortLabel")
self.powerOptionsForm.setWidget(1, QFormLayout.ItemRole.LabelRole, self.effortLabel)
self.effortValueLabel = QLabel(self.powerOptionsPage)
self.effortValueLabel.setObjectName(u"effortValueLabel")
self.powerOptionsForm.setWidget(1, QFormLayout.ItemRole.FieldRole, self.effortValueLabel)
self.flowLabel = QLabel(self.powerOptionsPage)
self.flowLabel.setObjectName(u"flowLabel")
self.powerOptionsForm.setWidget(2, QFormLayout.ItemRole.LabelRole, self.flowLabel)
self.flowValueLabel = QLabel(self.powerOptionsPage)
self.flowValueLabel.setObjectName(u"flowValueLabel")
self.powerOptionsForm.setWidget(2, QFormLayout.ItemRole.FieldRole, self.flowValueLabel)
self.causalityLabel = QLabel(self.powerOptionsPage)
self.causalityLabel.setObjectName(u"causalityLabel")
self.powerOptionsForm.setWidget(3, QFormLayout.ItemRole.LabelRole, self.causalityLabel)
self.causalityCombo = QComboBox(self.powerOptionsPage)
self.causalityCombo.setObjectName(u"causalityCombo")
self.powerOptionsForm.setWidget(3, QFormLayout.ItemRole.FieldRole, self.causalityCombo)
self.powerDescriptionLabel = QLabel(self.powerOptionsPage)
self.powerDescriptionLabel.setObjectName(u"powerDescriptionLabel")
self.powerOptionsForm.setWidget(4, QFormLayout.ItemRole.LabelRole, self.powerDescriptionLabel)
self.powerDescriptionEdit = QPlainTextEdit(self.powerOptionsPage)
self.powerDescriptionEdit.setObjectName(u"powerDescriptionEdit")
self.powerOptionsForm.setWidget(4, QFormLayout.ItemRole.FieldRole, self.powerDescriptionEdit)
self.typeOptionsStack.addWidget(self.powerOptionsPage)
self.unsupportedTypePage = QWidget()
self.unsupportedTypePage.setObjectName(u"unsupportedTypePage")
self.unsupportedTypeLayout = QVBoxLayout(self.unsupportedTypePage)
@@ -228,6 +284,7 @@ class Ui_PortOptionsDialog(object):
self.orientationLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"Orientation:", None))
self.orientationCombo.setItemText(0, QCoreApplication.translate("PortOptionsDialog", u"Input", None))
self.orientationCombo.setItemText(1, QCoreApplication.translate("PortOptionsDialog", u"Output", None))
self.orientationCombo.setItemText(2, QCoreApplication.translate("PortOptionsDialog", u"Indifferent", None))
self.multipleConnectionsCheckBox.setText(QCoreApplication.translate("PortOptionsDialog", u"Allow multiple connections", None))
self.positionHintLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"New ports start at (0, 0) in the icon editor.", None))
@@ -241,6 +298,13 @@ class Ui_PortOptionsDialog(object):
self.unitLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"Unit:", None))
self.rowsColumnsLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"Matrix size:", None))
self.descriptionLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"Description:", None))
self.domainLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"Domain:", None))
self.effortLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"Effort:", None))
self.effortValueLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"p.e", None))
self.flowLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"Flow:", None))
self.flowValueLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"p.f", None))
self.causalityLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"Causality:", None))
self.powerDescriptionLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"Description:", None))
self.unsupportedTypeLabel.setText(QCoreApplication.translate("PortOptionsDialog", u"This port type does not have an options editor yet.", None))
# retranslateUi

10
BEdit/src/bedit/gui/graphics/connection_styles.py
View File

@@ -1,8 +1,12 @@
from dataclasses import dataclass
from typing import Literal
from PySide6.QtCore import Qt
ArrowStyle = Literal["open", "half", "filled"]
@dataclass(frozen=True)
class ConnectionStyle:
color: str = "#285f9e"
@@ -13,11 +17,17 @@ class ConnectionStyle:
arrow_at_source: bool = False
arrow_at_target: bool = True
arrow_size: float = 10.0
arrow_style: ArrowStyle = "filled"
def __post_init__(self) -> None:
if self.arrow_style not in {"open", "half", "filled"}:
raise ValueError(f"Unknown connection arrow style: {self.arrow_style}")
# This is the intentional code-level styling point for every port/connection type.
CONNECTION_STYLES: dict[str, ConnectionStyle] = {
"signal": ConnectionStyle(),
"power": ConnectionStyle(width=3.0, arrow_style="half", color="#000000"),
}

10
BEdit/src/bedit/gui/graphics/icon_editor.py
View File

@@ -1,6 +1,6 @@
from copy import deepcopy
from PySide6.QtCore import QPointF, QRectF, QSizeF, Qt
from PySide6.QtCore import QPointF, QRectF, QSizeF, Qt, QTimer
from PySide6.QtGui import QColor, QPainter, QPainterPath, QPen, QPolygonF
from PySide6.QtWidgets import (
QColorDialog,
@@ -487,7 +487,13 @@ class IconEditorDialog(QDialog):
self.scene.addItem(ShapeItem(element))
self._add_ports(self.inputs, "input", 0.0)
self._add_ports(self.outputs, "output", self.icon.width)
self.view.center_icon()
self._initial_fit_pending = True
def showEvent(self, event) -> None: # noqa: N802 (Qt API name)
super().showEvent(event)
if self._initial_fit_pending:
self._initial_fit_pending = False
QTimer.singleShot(0, self.view.center_icon)
def _add_ports(self, ports: list[Port], direction: str, default_x: float) -> None:
spacing = self.icon.height / (len(ports) + 1)

67
BEdit/src/bedit/gui/graphics/workspace.py
View File

@@ -485,23 +485,39 @@ class ConnectionGraphicsItem(QGraphicsPathItem):
handle.setVisible(self.isSelected())
@staticmethod
def _arrow(end: QPointF, direction: QPointF, size: float) -> QPolygonF:
def _arrow_points(
end: QPointF, direction: QPointF, size: float
) -> tuple[QPointF, QPointF, QPointF]:
length = max(0.001, (direction.x() ** 2 + direction.y() ** 2) ** 0.5)
unit = QPointF(direction.x() / length, direction.y() / length)
normal = QPointF(-unit.y(), unit.x())
base = end - unit * size
return QPolygonF([end, base + normal * size * 0.45, base - normal * size * 0.45])
return end, base + normal * size * 0.45, base - normal * size * 0.45
def _draw_arrow(
self, painter: QPainter, end: QPointF, direction: QPointF
) -> None:
tip, left, right = self._arrow_points(
end, direction, self.style.arrow_size
)
color = self.pen().color()
if self.style.arrow_style == "filled":
painter.setPen(Qt.PenStyle.NoPen)
painter.setBrush(color)
painter.drawPolygon(QPolygonF([tip, left, right]))
return
painter.setPen(QPen(color, self.pen().widthF(), Qt.PenStyle.SolidLine))
painter.setBrush(Qt.BrushStyle.NoBrush)
painter.drawLine(tip, left)
if self.style.arrow_style == "open":
painter.drawLine(tip, right)
def paint(self, painter: QPainter, option, widget=None) -> None:
super().paint(painter, option, widget)
painter.setPen(Qt.PenStyle.NoPen)
painter.setBrush(self.pen().color())
if self.style.arrow_at_target and not self.target_is_junction:
painter.drawPolygon(self._arrow(self.end, self.end_direction, self.style.arrow_size))
self._draw_arrow(painter, self.end, self.end_direction)
if self.style.arrow_at_source and not self.source_is_junction:
painter.drawPolygon(
self._arrow(self.start, -self.start_direction, self.style.arrow_size)
)
self._draw_arrow(painter, self.start, -self.start_direction)
class WaypointHandle(QGraphicsEllipseItem):
@@ -1207,9 +1223,22 @@ class GraphScene(QGraphicsScene):
def add_pairs(
source_item: ComponentGraphicsItem, target_item: ComponentGraphicsItem
) -> None:
for output in source_item.component.outputs:
source_ports = [
*source_item.component.outputs,
*(
port
for port in source_item.component.inputs
if port.orientation == "indifferent"
),
]
for output in source_ports:
for input_port in target_item.component.inputs:
if not PortTypeRegistry.compatible(output.type, input_port.type):
if not PortTypeRegistry.compatible(
output.type,
input_port.type,
output.domain,
input_port.domain,
):
continue
source = Endpoint(block=source_item.component_id, port=output.id)
target = Endpoint(block=target_item.component_id, port=input_port.id)
@@ -1240,7 +1269,9 @@ class GraphScene(QGraphicsScene):
if terminal.direction == "input":
for port in component.component.inputs:
target = Endpoint(block=component.component_id, port=port.id)
if not PortTypeRegistry.compatible(terminal.port.type, port.type):
if not PortTypeRegistry.compatible(
terminal.port.type, port.type, terminal.port.domain, port.domain
):
continue
if not self.controller.endpoint_accepts_connection(interface, "source"):
continue
@@ -1254,9 +1285,19 @@ class GraphScene(QGraphicsScene):
)
)
else:
for port in component.component.outputs:
ports = [
*component.component.outputs,
*(
port
for port in component.component.inputs
if port.orientation == "indifferent"
),
]
for port in ports:
source = Endpoint(block=component.component_id, port=port.id)
if not PortTypeRegistry.compatible(port.type, terminal.port.type):
if not PortTypeRegistry.compatible(
port.type, terminal.port.type, port.domain, terminal.port.domain
):
continue
if not self.controller.endpoint_accepts_connection(source, "source"):
continue