Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Upgrade to capellambse v0.6.x #131

Merged
merged 4 commits into from
Sep 18, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,15 @@ repos:
rev: v1.11.1
hooks:
- id: mypy
args: [--follow-imports=silent]
additional_dependencies:
- capellambse==0.6.6
- mkdocs
- mkdocs-gen-files
- mkdocs-literate-nav
- pydantic==2.7.3
- pytest
- types-setuptools
- repo: https://github.com/Lucas-C/pre-commit-hooks
rev: v1.5.5
hooks:
Expand Down
63 changes: 32 additions & 31 deletions capellambse_context_diagrams/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,10 @@
import typing as t
from importlib import metadata

import capellambse.model as m
from capellambse.diagram import COLORS, CSSdef, capstyle
from capellambse.model import common
from capellambse.model.crosslayer import fa, information
from capellambse.model.layers import ctx, la, oa, pa
from capellambse.model.modeltypes import DiagramType
from capellambse.metamodel import fa, information, la, oa, pa, sa
from capellambse.model import DiagramType

from . import _elkjs, context, styling

Expand All @@ -38,9 +37,7 @@
del metadata

DefaultRenderParams = dict[str, t.Any]
SupportedClass = tuple[
type[common.GenericElement], DiagramType, DefaultRenderParams
]
SupportedClass = tuple[type[m.ModelElement], DiagramType, DefaultRenderParams]
logger = logging.getLogger(__name__)

ATTR_NAME = "context_diagram"
Expand Down Expand Up @@ -78,10 +75,14 @@ def register_classes() -> None:
{"display_parent_relation": True},
),
(oa.OperationalCapability, DiagramType.OCB, {}),
(ctx.Mission, DiagramType.MCB, {}),
(ctx.Capability, DiagramType.MCB, {"display_symbols_as_boxes": False}),
(sa.Mission, DiagramType.MCB, {}),
(
ctx.SystemComponent,
sa.Capability,
DiagramType.MCB,
{"display_symbols_as_boxes": False},
),
(
sa.SystemComponent,
DiagramType.SAB,
{
"display_symbols_as_boxes": True,
Expand All @@ -90,7 +91,7 @@ def register_classes() -> None:
},
),
(
ctx.SystemFunction,
sa.SystemFunction,
DiagramType.SAB,
{
"display_symbols_as_boxes": True,
Expand Down Expand Up @@ -132,10 +133,10 @@ def register_classes() -> None:
),
]
patch_styles(supported_classes)
class_: type[common.GenericElement]
class_: type[m.ModelElement]
for class_, dgcls, default_render_params in supported_classes:
accessor = context.ContextAccessor(dgcls.value, default_render_params)
common.set_accessor(class_, ATTR_NAME, accessor)
m.set_accessor(class_, ATTR_NAME, accessor)


def patch_styles(classes: cabc.Iterable[SupportedClass]) -> None:
Expand All @@ -157,14 +158,16 @@ def patch_styles(classes: cabc.Iterable[SupportedClass]) -> None:
capstyle.STYLES["Operational Capabilities Blank"][
"Box.OperationalCapability"
] = cap
circle_style = {"fill": COLORS["_CAP_xAB_Function_Border_Green"]}
circle_style: dict[str, CSSdef] = {
"fill": COLORS["_CAP_xAB_Function_Border_Green"],
}
for _, dt, _ in classes:
capstyle.STYLES[dt.value]["Circle.FunctionalExchange"] = circle_style


def register_interface_context() -> None:
"""Add the `context_diagram` property to interface model objects."""
common.set_accessor(
m.set_accessor(
oa.CommunicationMean,
ATTR_NAME,
context.InterfaceContextAccessor(
Expand All @@ -175,13 +178,13 @@ def register_interface_context() -> None:
{"include_interface": True},
),
)
common.set_accessor(
m.set_accessor(
fa.ComponentExchange,
ATTR_NAME,
context.InterfaceContextAccessor(
{
ctx.SystemComponentPkg: DiagramType.SAB.value,
ctx.SystemComponent: DiagramType.SAB.value,
sa.SystemComponentPkg: DiagramType.SAB.value,
sa.SystemComponent: DiagramType.SAB.value,
la.LogicalComponentPkg: DiagramType.LAB.value,
la.LogicalComponent: DiagramType.LAB.value,
pa.PhysicalComponentPkg: DiagramType.PAB.value,
Expand All @@ -200,17 +203,15 @@ def register_functional_context() -> None:
The functional context diagrams will be available soon.
"""
attr_name = f"functional_{ATTR_NAME}"
supported_classes: list[
tuple[type[common.GenericElement], DiagramType]
] = [
supported_classes: list[tuple[type[m.ModelElement], DiagramType]] = [
(oa.Entity, DiagramType.OAB),
(ctx.SystemComponent, DiagramType.SAB),
(sa.SystemComponent, DiagramType.SAB),
(la.LogicalComponent, DiagramType.LAB),
(pa.PhysicalComponent, DiagramType.PAB),
]
class_: type[common.GenericElement]
class_: type[m.ModelElement]
for class_, dgcls in supported_classes:
common.set_accessor(
m.set_accessor(
class_,
attr_name,
context.FunctionalContextAccessor(dgcls.value),
Expand All @@ -219,7 +220,7 @@ def register_functional_context() -> None:

def register_tree_view() -> None:
"""Add the ``tree_view`` attribute to ``Class``es."""
common.set_accessor(
m.set_accessor(
information.Class,
"tree_view",
context.ClassTreeAccessor(DiagramType.CDB.value),
Expand All @@ -235,16 +236,16 @@ def register_realization_view() -> None:
supported_classes: list[SupportedClass] = [
(oa.Entity, DiagramType.OAB, {}),
(oa.OperationalActivity, DiagramType.OAIB, {}),
(ctx.SystemComponent, DiagramType.SAB, {}),
(ctx.SystemFunction, DiagramType.SDFB, {}),
(sa.SystemComponent, DiagramType.SAB, {}),
(sa.SystemFunction, DiagramType.SDFB, {}),
(la.LogicalComponent, DiagramType.LAB, {}),
(la.LogicalFunction, DiagramType.LDFB, {}),
(pa.PhysicalComponent, DiagramType.PAB, {}),
(pa.PhysicalFunction, DiagramType.PDFB, {}),
]
styles: dict[str, dict[str, capstyle.CSSdef]] = {}
for class_, dgcls, _ in supported_classes:
common.set_accessor(
m.set_accessor(
class_,
"realization_view",
context.RealizationViewContextAccessor("RealizationView Diagram"),
Expand All @@ -265,9 +266,9 @@ def register_realization_view() -> None:
def register_data_flow_view() -> None:
supported_classes: list[SupportedClass] = [
(oa.OperationalCapability, DiagramType.OAIB, {}), # portless
(ctx.Capability, DiagramType.SDFB, {}), # default
(sa.Capability, DiagramType.SDFB, {}), # default
]
class_: type[common.GenericElement]
class_: type[m.ModelElement]
for class_, dgcls, default_render_params in supported_classes:
accessor = context.DataFlowAccessor(dgcls.value, default_render_params)
common.set_accessor(class_, "data_flow_view", accessor)
m.set_accessor(class_, "data_flow_view", accessor)
5 changes: 5 additions & 0 deletions capellambse_context_diagrams/_elkjs.py
Original file line number Diff line number Diff line change
Expand Up @@ -206,6 +206,7 @@ class ELKOutputNode(ELKOutputDiagramElement):
children: cabc.MutableSequence[ELKOutputChild] = pydantic.Field(
default_factory=list
)
context: list[str] = pydantic.Field(default_factory=list)


class ELKOutputJunction(ELKOutputElement):
Expand All @@ -217,6 +218,7 @@ class ELKOutputJunction(ELKOutputElement):
)

position: ELKPoint
context: list[str] = pydantic.Field(default_factory=list)


class ELKOutputPort(ELKOutputDiagramElement):
Expand All @@ -226,13 +228,15 @@ class ELKOutputPort(ELKOutputDiagramElement):
children: cabc.MutableSequence[ELKOutputLabel] = pydantic.Field(
default_factory=list
)
context: list[str] = pydantic.Field(default_factory=list)


class ELKOutputLabel(ELKOutputDiagramElement):
"""Label that comes out of ELK."""

type: t.Literal["label"]
text: str
context: list[str] = pydantic.Field(default_factory=list)


class ELKOutputEdge(ELKOutputElement):
Expand All @@ -246,6 +250,7 @@ class ELKOutputEdge(ELKOutputElement):
children: cabc.MutableSequence[
t.Union[ELKOutputLabel, ELKOutputJunction]
] = pydantic.Field(default_factory=list)
context: list[str] = pydantic.Field(default_factory=list)


ELKOutputChild = t.Union[
Expand Down
11 changes: 5 additions & 6 deletions capellambse_context_diagrams/collectors/dataflow_view.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,16 +10,15 @@
import typing as t
from itertools import chain

from capellambse.model import modeltypes
from capellambse.model.crosslayer import fa
from capellambse.model.layers import oa
import capellambse.model as m
from capellambse.metamodel import fa, oa

from .. import _elkjs, context
from . import default, generic, makers, portless

COLLECTOR_PARAMS: dict[modeltypes.DiagramType, dict[str, t.Any]] = {
modeltypes.DiagramType.OAIB: {"attribute": "involved_activities"},
modeltypes.DiagramType.SDFB: {
COLLECTOR_PARAMS: dict[m.DiagramType, dict[str, t.Any]] = {
m.DiagramType.OAIB: {"attribute": "involved_activities"},
m.DiagramType.SDFB: {
"attribute": "involved_functions",
"filter_attrs": ("source.owner", "target.owner"),
"port_collector": default.port_collector,
Expand Down
50 changes: 24 additions & 26 deletions capellambse_context_diagrams/collectors/default.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,12 +10,10 @@
import typing as t
from itertools import chain

import capellambse.model as m
from capellambse import helpers
from capellambse.model import common
from capellambse.model.crosslayer import cs, fa
from capellambse.model.layers import ctx as sa
from capellambse.model.layers import la
from capellambse.model.modeltypes import DiagramType as DT
from capellambse.metamodel import cs, fa, la, sa
from capellambse.model import DiagramType as DT

from .. import _elkjs
from . import exchanges, generic, makers
Expand All @@ -29,8 +27,8 @@
]

Filter: t.TypeAlias = cabc.Callable[
[cabc.Iterable[common.GenericElement]],
cabc.Iterable[common.GenericElement],
[cabc.Iterable[m.ModelElement]],
cabc.Iterable[m.ModelElement],
]


Expand Down Expand Up @@ -134,7 +132,7 @@ def _process_port_spread(
port_spread[owner] += inc

def _process_exchanges(self) -> tuple[
list[common.GenericElement],
list[m.ModelElement],
list[generic.ExchangeData],
]:
inc, out = port_collector(self.diagram.target, self.diagram.type)
Expand Down Expand Up @@ -285,13 +283,13 @@ def collector(


def port_collector(
target: common.GenericElement | common.ElementList, diagram_type: DT
) -> tuple[list[common.GenericElement], list[common.GenericElement]]:
target: m.ModelElement | m.ElementList, diagram_type: DT
) -> tuple[list[m.ModelElement], list[m.ModelElement]]:
"""Savely collect ports from `target`."""

def __collect(target):
incoming_ports: list[common.GenericElement] = []
outgoing_ports: list[common.GenericElement] = []
incoming_ports: list[m.ModelElement] = []
outgoing_ports: list[m.ModelElement] = []
for attr in generic.DIAGRAM_TYPE_TO_CONNECTOR_NAMES[diagram_type]:
try:
ports = getattr(target, attr)
Expand All @@ -315,9 +313,9 @@ def __collect(target):
return incoming_ports, outgoing_ports

if isinstance(target, cabc.Iterable):
assert not isinstance(target, common.GenericElement)
incoming_ports: list[common.GenericElement] = []
outgoing_ports: list[common.GenericElement] = []
assert not isinstance(target, m.ModelElement)
incoming_ports: list[m.ModelElement] = []
outgoing_ports: list[m.ModelElement] = []
for obj in target:
inc, out = __collect(obj)
incoming_ports.extend(inc)
Expand All @@ -328,37 +326,37 @@ def __collect(target):


def _extract_edges(
obj: common.GenericElement,
obj: m.ModelElement,
attribute: str,
filter: Filter,
) -> cabc.Iterable[common.GenericElement]:
) -> cabc.Iterable[m.ModelElement]:
return filter(getattr(obj, attribute, []))


def port_exchange_collector(
ports: t.Iterable[common.GenericElement],
ports: t.Iterable[m.ModelElement],
filter: Filter = lambda i: i,
) -> dict[str, common.ElementList[fa.AbstractExchange]]:
) -> dict[str, list[fa.AbstractExchange]]:
"""Collect exchanges from `ports` savely."""
edges: dict[str, common.ElementList[fa.AbstractExchange]] = {}
edges: dict[str, list[fa.AbstractExchange]] = {}

for port in ports:
if exs := _extract_edges(port, "exchanges", filter):
edges[port.uuid] = exs
edges[port.uuid] = t.cast(list[fa.AbstractExchange], exs)
continue

if links := _extract_edges(port, "links", filter):
edges[port.uuid] = links
edges[port.uuid] = t.cast(list[fa.AbstractExchange], links)

return edges


class ContextInfo(t.NamedTuple):
"""ContextInfo data."""

element: common.GenericElement
element: m.ModelElement
"""An element of context."""
ports: list[common.GenericElement]
ports: list[m.ModelElement]
"""The context element's relevant ports.

This list only contains ports that at least one of the exchanges
Expand All @@ -370,7 +368,7 @@ class ContextInfo(t.NamedTuple):

def port_context_collector(
exchange_datas: t.Iterable[generic.ExchangeData],
local_ports: t.Container[common.GenericElement],
local_ports: t.Container[m.ModelElement],
) -> t.Iterator[ContextInfo]:
"""Collect the context objects.

Expand Down Expand Up @@ -492,7 +490,7 @@ def derive_from_functions(
)


DERIVATORS: dict[type[common.GenericElement], DerivatorFunction] = {
DERIVATORS: dict[type[m.ModelElement], DerivatorFunction] = {
la.LogicalComponent: derive_from_functions,
sa.SystemComponent: derive_from_functions,
}
Expand Down
Loading
Loading