Coverage for pyTooling/Graph/GraphML.py: 88%

1# ==================================================================================================================== #

2# _____ _ _ ____ _ #

3# _ __ _ |_ _|__ ___ | (_)_ __ __ _ / ___|_ __ __ _ _ __ | |__ #

4# | '_ \| | | || |/ _ \ / _ \| | | '_ \ / _` || | _| '__/ _` | '_ \| '_ \ #

5# | |_) | |_| || | (_) | (_) | | | | | | (_| || |_| | | | (_| | |_) | | | | #

6# | .__/ \__, ||_|\___/ \___/|_|_|_| |_|\__, (_)____|_| \__,_| .__/|_| |_| #

7# |_| |___/ |___/ |_| #

8# ==================================================================================================================== #

9# Authors: #

10# Patrick Lehmann #

11# #

12# License: #

13# ==================================================================================================================== #

15# #

16# Licensed under the Apache License, Version 2.0 (the "License"); #

17# you may not use this file except in compliance with the License. #

18# You may obtain a copy of the License at #

19# #

20# http://www.apache.org/licenses/LICENSE-2.0 #

21# #

22# Unless required by applicable law or agreed to in writing, software #

23# distributed under the License is distributed on an "AS IS" BASIS, #

24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #

25# See the License for the specific language governing permissions and #

26# limitations under the License. #

27# #

28# SPDX-License-Identifier: Apache-2.0 #

29# ==================================================================================================================== #

30#

31"""

32A data model to write out GraphML XML files.

34.. seealso::

36 * http://graphml.graphdrawing.org/primer/graphml-primer.html

37"""

38from enum import Enum, auto

39from pathlib import Path

40from typing import Any, List, Dict, Union, Optional as Nullable

42from pyTooling.Decorators import export, readonly

43from pyTooling.MetaClasses import ExtendedType

44from pyTooling.Graph import Graph as pyToolingGraph, Subgraph as pyToolingSubgraph

45from pyTooling.Tree import Node as pyToolingNode

48@export

49class AttributeContext(Enum):

50 """

51 Enumeration of all attribute contexts.

53 An attribute context describes to what kind of GraphML node an attribute can be applied.

54 """

55 GraphML = auto()

56 Graph = auto()

57 Node = auto()

58 Edge = auto()

59 Port = auto()

61 def __str__(self) -> str:

62 return f"{self.name.lower()}"

65@export

66class AttributeTypes(Enum):

67 """

68 Enumeration of all attribute types.

70 An attribute type describes what datatype can be applied to an attribute.

71 """

72 Boolean = auto()

73 Int = auto()

74 Long = auto()

75 Float = auto()

76 Double = auto()

77 String = auto()

79 def __str__(self) -> str:

80 return f"{self.name.lower()}"

83@export

84class EdgeDefault(Enum):

85 """An enumeration describing the default edge direction."""

86 Undirected = auto()

87 Directed = auto()

89 def __str__(self) -> str:

90 return f"{self.name.lower()}"

93@export

94class ParsingOrder(Enum):

95 """An enumeration describing the parsing order of the graph's representation."""

96 NodesFirst = auto() #: First, all nodes are given, then followed by all edges.

97 AdjacencyList = auto()

98 Free = auto()

100 def __str__(self) -> str:

101 return f"{self.name.lower()}"

102

103

104@export

105class IDStyle(Enum):

106 """An enumeration describing the style of identifiers (IDs)."""

107 Canonical = auto()

108 Free = auto()

109

110 def __str__(self) -> str:

111 return f"{self.name.lower()}"

112

113

114@export

115class Base(metaclass=ExtendedType, slots=True):

116 """

117 Base-class for all GraphML data model classes.

118 """

119 @readonly

120 def HasClosingTag(self) -> bool:

121 return True

122

123 def Tag(self, indent: int = 0) -> str:

124 raise NotImplementedError()

125

126 def OpeningTag(self, indent: int = 0) -> str:

127 raise NotImplementedError()

128

129 def ClosingTag(self, indent: int = 0) -> str:

130 raise NotImplementedError()

131

132 def ToStringLines(self, indent: int = 0) -> List[str]:

133 raise NotImplementedError()

134

135

136@export

137class BaseWithID(Base):

138 _id: str

139

140 def __init__(self, identifier: str) -> None:

141 super().__init__()

142 self._id = identifier

143

144 @readonly

145 def ID(self) -> str:

146 return self._id

147

148

149@export

150class BaseWithData(BaseWithID):

151 _data: List['Data']

152

153 def __init__(self, identifier: str) -> None:

154 super().__init__(identifier)

155

156 self._data = []

157

158 @readonly

159 def Data(self) -> List['Data']:

160 return self._data

161

162 def AddData(self, data: Data) -> Data:

163 self._data.append(data)

164 return data

165

166

167@export

168class Key(BaseWithID):

169 _context: AttributeContext

170 _attributeName: str

171 _attributeType: AttributeTypes

172

173 def __init__(self, identifier: str, context: AttributeContext, name: str, type: AttributeTypes) -> None:

174 super().__init__(identifier)

175

176 self._context = context

177 self._attributeName = name

178 self._attributeType = type

179

180 @readonly

181 def Context(self) -> AttributeContext:

182 return self._context

183

184 @readonly

185 def AttributeName(self) -> str:

186 return self._attributeName

187

188 @readonly

189 def AttributeType(self) -> AttributeTypes:

190 return self._attributeType

191

192 @readonly

193 def HasClosingTag(self) -> bool:

194 return False

195

196 def Tag(self, indent: int = 2) -> str:

197 return f"""{' '*indent}<key id="{self._id}" for="{self._context}" attr.name="{self._attributeName}" attr.type="{self._attributeType}" />\n"""

198

199 def ToStringLines(self, indent: int = 2) -> List[str]:

200 return [self.Tag(indent)]

201

202

203@export

204class Data(Base):

205 _key: Key

206 _data: Any

207

208 def __init__(self, key: Key, data: Any) -> None:

209 super().__init__()

210

211 self._key = key

212 self._data = data

213

214 @readonly

215 def Key(self) -> Key:

216 return self._key

217

218 @readonly

219 def Data(self) -> Any:

220 return self._data

221

222 @readonly

223 def HasClosingTag(self) -> bool:

224 return False

225

226 def Tag(self, indent: int = 2) -> str:

227 data = str(self._data)

228 data = data.replace("&", "&")

229 data = data.replace("<", "<")

230 data = data.replace(">", ">")

231 data = data.replace("\n", "\\n")

232 return f"""{' '*indent}<data key="{self._key._id}">{data}</data>\n"""

233

234 def ToStringLines(self, indent: int = 2) -> List[str]:

235 return [self.Tag(indent)]

236

237

238@export

239class Node(BaseWithData):

240 def __init__(self, identifier: str) -> None:

241 super().__init__(identifier)

242

243 @readonly

244 def HasClosingTag(self) -> bool:

245 return len(self._data) > 0

246

247 def Tag(self, indent: int = 2) -> str:

248 return f"""{' '*indent}<node id="{self._id}" />\n"""

249

250 def OpeningTag(self, indent: int = 2) -> str:

251 return f"""{' '*indent}<node id="{self._id}">\n"""

252

253 def ClosingTag(self, indent: int = 2) -> str:

254 return f"""{' ' * indent}</node>\n"""

255

256 def ToStringLines(self, indent: int = 2) -> List[str]:

257 if not self.HasClosingTag:

258 return [self.Tag(indent)]

259

260 lines = [self.OpeningTag(indent)]

261 for data in self._data:

262 lines.extend(data.ToStringLines(indent + 1))

263 lines.append(self.ClosingTag(indent))

264

265 return lines

266

267

268@export

269class Edge(BaseWithData):

270 _source: Node

271 _target: Node

272

273 def __init__(self, identifier: str, source: Node, target: Node) -> None:

274 super().__init__(identifier)

275

276 self._source = source

277 self._target = target

278

279 @readonly

280 def Source(self) -> Node:

281 return self._source

282

283 @readonly

284 def Target(self) -> Node:

285 return self._target

286

287 @readonly

288 def HasClosingTag(self) -> bool:

289 return len(self._data) > 0

290

291 def Tag(self, indent: int = 2) -> str:

292 return f"""{' ' * indent}<edge id="{self._id}" source="{self._source._id}" target="{self._target._id}" />\n"""

293

294 def OpeningTag(self, indent: int = 2) -> str:

295 return f"""{' '*indent}<edge id="{self._id}" source="{self._source._id}" target="{self._target._id}">\n"""

296

297 def ClosingTag(self, indent: int = 2) -> str:

298 return f"""{' ' * indent}</edge>\n"""

299

300 def ToStringLines(self, indent: int = 2) -> List[str]:

301 if not self.HasClosingTag:

302 return [self.Tag(indent)]

303

304 lines = [self.OpeningTag(indent)]

305 for data in self._data:

306 lines.extend(data.ToStringLines(indent + 1))

307 lines.append(self.ClosingTag(indent))

308

309 return lines

310

311

312@export

313class BaseGraph(BaseWithData, mixin=True):

314 _subgraphs: Dict[str, 'Subgraph']

315 _nodes: Dict[str, Node]

316 _edges: Dict[str, Edge]

317 _edgeDefault: EdgeDefault

318 _parseOrder: ParsingOrder

319 _nodeIDStyle: IDStyle

320 _edgeIDStyle: IDStyle

321

322 def __init__(self, identifier: Nullable[str] = None) -> None:

323 super().__init__(identifier)

324

325 self._subgraphs = {}

326 self._nodes = {}

327 self._edges = {}

328 self._edgeDefault = EdgeDefault.Directed

329 self._parseOrder = ParsingOrder.NodesFirst

330 self._nodeIDStyle = IDStyle.Free

331 self._edgeIDStyle = IDStyle.Free

332

333 @readonly

334 def Subgraphs(self) -> Dict[str, 'Subgraph']:

335 return self._subgraphs

336

337 @readonly

338 def Nodes(self) -> Dict[str, Node]:

339 return self._nodes

340

341 @readonly

342 def Edges(self) -> Dict[str, Edge]:

343 return self._edges

344

345 def AddSubgraph(self, subgraph: 'Subgraph') -> 'Subgraph':

346 self._subgraphs[subgraph._subgraphID] = subgraph

347 self._nodes[subgraph._id] = subgraph

348 return subgraph

349

350 def GetSubgraph(self, subgraphName: str) -> 'Subgraph':

351 return self._subgraphs[subgraphName]

352

353 def AddNode(self, node: Node) -> Node:

354 self._nodes[node._id] = node

355 return node

356

357 def GetNode(self, nodeName: str) -> Node:

358 return self._nodes[nodeName]

359

360 def AddEdge(self, edge: Edge) -> Edge:

361 self._edges[edge._id] = edge

362 return edge

363

364 def GetEdge(self, edgeName: str) -> Edge:

365 return self._edges[edgeName]

366

367 def OpeningTag(self, indent: int = 1) -> str:

368 return f"""\

369{' '*indent}<graph id="{self._id}"

370{' '*indent} edgedefault="{self._edgeDefault!s}"

371{' '*indent} parse.nodes="{len(self._nodes)}"

372{' '*indent} parse.edges="{len(self._edges)}"

373{' '*indent} parse.order="{self._parseOrder!s}"

374{' '*indent} parse.nodeids="{self._nodeIDStyle!s}"

375{' '*indent} parse.edgeids="{self._edgeIDStyle!s}">

376"""

377

378 def ClosingTag(self, indent: int = 1) -> str:

379 return f"{' '*indent}</graph>\n"

380

381 def ToStringLines(self, indent: int = 1) -> List[str]:

382 lines = [self.OpeningTag(indent)]

383 for node in self._nodes.values():

384 lines.extend(node.ToStringLines(indent + 1))

385 for edge in self._edges.values():

386 lines.extend(edge.ToStringLines(indent + 1))

387 # for data in self._data:

388 # lines.extend(data.ToStringLines(indent + 1))

389 lines.append(self.ClosingTag(indent))

390

391 return lines

392

393

394@export

395class Graph(BaseGraph):

396 _document: 'GraphMLDocument'

397 _ids: Dict[str, Union[Node, Edge, 'Subgraph']]

398

399 def __init__(self, document: 'GraphMLDocument', identifier: str) -> None:

400 super().__init__(identifier)

401 self._document = document

402 self._ids = {}

403

404 def GetByID(self, identifier: str) -> Union[Node, Edge, 'Subgraph']:

405 return self._ids[identifier]

406

407 def AddSubgraph(self, subgraph: 'Subgraph') -> 'Subgraph':

408 result = super().AddSubgraph(subgraph)

409 self._ids[subgraph._subgraphID] = subgraph

410 subgraph._root = self

411 return result

412

413 def AddNode(self, node: Node) -> Node:

414 result = super().AddNode(node)

415 self._ids[node._id] = node

416 return result

417

418 def AddEdge(self, edge: Edge) -> Edge:

419 result = super().AddEdge(edge)

420 self._ids[edge._id] = edge

421 return result

422

423

424@export

425class Subgraph(Node, BaseGraph):

426 _subgraphID: str

427 _root: Nullable[Graph]

428

429 def __init__(self, nodeIdentifier: str, graphIdentifier: str) -> None:

430 super().__init__(nodeIdentifier)

431 BaseGraph.__init__(self, nodeIdentifier)

432

433 self._subgraphID = graphIdentifier

434 self._root = None

435

436 @readonly

437 def RootGraph(self) -> Graph:

438 return self._root

439

440 @readonly

441 def SubgraphID(self) -> str:

442 return self._subgraphID

443

444 @readonly

445 def HasClosingTag(self) -> bool:

446 return True

447

448 def AddNode(self, node: Node) -> Node:

449 result = super().AddNode(node)

450 self._root._ids[node._id] = node

451 return result

452

453 def AddEdge(self, edge: Edge) -> Edge:

454 result = super().AddEdge(edge)

455 self._root._ids[edge._id] = edge

456 return result

457

458 def Tag(self, indent: int = 2) -> str:

459 raise NotImplementedError()

460

461 def OpeningTag(self, indent: int = 1) -> str:

462 return f"""\

463{' ' * indent}<graph id="{self._subgraphID}"

464{' ' * indent} edgedefault="{self._edgeDefault!s}"

465{' ' * indent} parse.nodes="{len(self._nodes)}"

466{' ' * indent} parse.edges="{len(self._edges)}"

467{' ' * indent} parse.order="{self._parseOrder!s}"

468{' ' * indent} parse.nodeids="{self._nodeIDStyle!s}"

469{' ' * indent} parse.edgeids="{self._edgeIDStyle!s}">

470"""

471

472 def ClosingTag(self, indent: int = 2) -> str:

473 return BaseGraph.ClosingTag(self, indent)

474

475 def ToStringLines(self, indent: int = 2) -> List[str]:

476 lines = [super().OpeningTag(indent)]

477 for data in self._data: 477 ↛ 478line 477 didn't jump to line 478 because the loop on line 477 never started

478 lines.extend(data.ToStringLines(indent + 1))

479 # lines.extend(Graph.ToStringLines(self, indent + 1))

480 lines.append(self.OpeningTag(indent + 1))

481 for node in self._nodes.values():

482 lines.extend(node.ToStringLines(indent + 2))

483 for edge in self._edges.values():

484 lines.extend(edge.ToStringLines(indent + 2))

485 # for data in self._data:

486 # lines.extend(data.ToStringLines(indent + 1))

487 lines.append(self.ClosingTag(indent + 1))

488 lines.append(super().ClosingTag(indent))

489

490 return lines

491

492

493@export

494class GraphMLDocument(Base):

495 xmlNS = {

496 None: "http://graphml.graphdrawing.org/xmlns",

497 "xsi": "http://www.w3.org/2001/XMLSchema-instance"

498 }

499 xsi = {

500 "schemaLocation": "http://graphml.graphdrawing.org/xmlns/1.0/graphml.xsd"

501 }

502

503 _graph: Graph

504 _keys: Dict[str, Key]

505

506 def __init__(self, identifier: str = "G") -> None:

507 super().__init__()

508

509 self._graph = Graph(self, identifier)

510 self._keys = {}

511

512 @readonly

513 def Graph(self) -> BaseGraph:

514 return self._graph

515

516 @readonly

517 def Keys(self) -> Dict[str, Key]:

518 return self._keys

519

520 def AddKey(self, key: Key) -> Key:

521 self._keys[key._id] = key

522 return key

523

524 def GetKey(self, keyName: str) -> Key:

525 return self._keys[keyName]

526

527 def HasKey(self, keyName: str) -> bool:

528 return keyName in self._keys

529

530 def FromGraph(self, graph: pyToolingGraph) -> None:

531 document = self

532 self._graph._id = graph._name

533

534 nodeValue = self.AddKey(Key("nodeValue", AttributeContext.Node, "value", AttributeTypes.String))

535 edgeValue = self.AddKey(Key("edgeValue", AttributeContext.Edge, "value", AttributeTypes.String))

536

537 def translateGraph(rootGraph: Graph, pyTGraph: pyToolingGraph):

538 for vertex in pyTGraph.IterateVertices():

539 newNode = Node(vertex._id)

540 newNode.AddData(Data(nodeValue, vertex._value))

541 for key, value in vertex._dict.items(): 541 ↛ 542line 541 didn't jump to line 542 because the loop on line 541 never started

542 if document.HasKey(str(key)):

543 nodeKey = document.GetKey(f"node{key!s}")

544 else:

545 nodeKey = document.AddKey(Key(f"node{key!s}", AttributeContext.Node, str(key), AttributeTypes.String))

546 newNode.AddData(Data(nodeKey, value))

547

548 rootGraph.AddNode(newNode)

549

550 for edge in pyTGraph.IterateEdges():

551 source = rootGraph.GetByID(edge._source._id)

552 target = rootGraph.GetByID(edge._destination._id)

553

554 newEdge = Edge(edge._id, source, target)

555 newEdge.AddData(Data(edgeValue, edge._value))

556 for key, value in edge._dict.items(): 556 ↛ 557line 556 didn't jump to line 557 because the loop on line 556 never started

557 if self.HasKey(str(key)):

558 edgeKey = self.GetBy(f"edge{key!s}")

559 else:

560 edgeKey = self.AddKey(Key(f"edge{key!s}", AttributeContext.Edge, str(key), AttributeTypes.String))

561 newEdge.AddData(Data(edgeKey, value))

562

563 rootGraph.AddEdge(newEdge)

564

565 for link in pyTGraph.IterateLinks():

566 source = rootGraph.GetByID(link._source._id)

567 target = rootGraph.GetByID(link._destination._id)

568

569 newEdge = Edge(link._id, source, target)

570 newEdge.AddData(Data(edgeValue, link._value))

571 for key, value in link._dict.items(): 571 ↛ 572line 571 didn't jump to line 572 because the loop on line 571 never started

572 if self.HasKey(str(key)):

573 edgeKey = self.GetKey(f"link{key!s}")

574 else:

575 edgeKey = self.AddKey(Key(f"link{key!s}", AttributeContext.Edge, str(key), AttributeTypes.String))

576 newEdge.AddData(Data(edgeKey, value))

577

578 rootGraph.AddEdge(newEdge)

579

580 def translateSubgraph(nodeGraph: Subgraph, pyTSubgraph: pyToolingSubgraph):

581 rootGraph = nodeGraph.RootGraph

582

583 for vertex in pyTSubgraph.IterateVertices():

584 newNode = Node(vertex._id)

585 newNode.AddData(Data(nodeValue, vertex._value))

586 for key, value in vertex._dict.items(): 586 ↛ 587line 586 didn't jump to line 587 because the loop on line 586 never started

587 if self.HasKey(str(key)):

588 nodeKey = self.GetKey(f"node{key!s}")

589 else:

590 nodeKey = self.AddKey(Key(f"node{key!s}", AttributeContext.Node, str(key), AttributeTypes.String))

591 newNode.AddData(Data(nodeKey, value))

592

593 nodeGraph.AddNode(newNode)

594

595 for edge in pyTSubgraph.IterateEdges():

596 source = nodeGraph.GetNode(edge._source._id)

597 target = nodeGraph.GetNode(edge._destination._id)

598

599 newEdge = Edge(edge._id, source, target)

600 newEdge.AddData(Data(edgeValue, edge._value))

601 for key, value in edge._dict.items(): 601 ↛ 602line 601 didn't jump to line 602 because the loop on line 601 never started

602 if self.HasKey(str(key)):

603 edgeKey = self.GetKey(f"edge{key!s}")

604 else:

605 edgeKey = self.AddKey(Key(f"edge{key!s}", AttributeContext.Edge, str(key), AttributeTypes.String))

606 newEdge.AddData(Data(edgeKey, value))

607

608 nodeGraph.AddEdge(newEdge)

609

610 for subgraph in graph.Subgraphs:

611 nodeGraph = Subgraph(subgraph.Name, "sg" + subgraph.Name)

612 self._graph.AddSubgraph(nodeGraph)

613 translateSubgraph(nodeGraph, subgraph)

614

615 translateGraph(self._graph, graph)

616

617 def FromTree(self, tree: pyToolingNode) -> None:

618 self._graph._id = tree._id

619

620 nodeValue = self.AddKey(Key("nodeValue", AttributeContext.Node, "value", AttributeTypes.String))

621

622 rootNode = self._graph.AddNode(Node(tree._id))

623 rootNode.AddData(Data(nodeValue, tree._value))

624

625 for i, node in enumerate(tree.GetDescendants()):

626 newNode = self._graph.AddNode(Node(node._id))

627 newNode.AddData(Data(nodeValue, node._value))

628

629 newEdge = self._graph.AddEdge(Edge(f"e{i}", newNode, self._graph.GetNode(node._parent._id)))

630

631 def OpeningTag(self, indent: int = 0) -> str:

632 return f"""\

633{' '*indent}<graphml xmlns="{self.xmlNS[None]}"

634{' '*indent} xmlns:xsi="{self.xmlNS["xsi"]}"

635{' '*indent} xsi:schemaLocation="{self.xsi["schemaLocation"]}">

636"""

637

638 def ClosingTag(self, indent: int = 0) -> str:

639 return f"{' '*indent}</graphml>\n"

640

641 def ToStringLines(self, indent: int = 0) -> List[str]:

642 lines = [self.OpeningTag(indent)]

643 for key in self._keys.values():

644 lines.extend(key.ToStringLines(indent + 1))

645 lines.extend(self._graph.ToStringLines(indent + 1))

646 lines.append(self.ClosingTag(indent))

647

648 return lines

649

650 def WriteToFile(self, file: Path) -> None:

651 with file.open("w", encoding="utf-8") as f:

652 f.write(f"""<?xml version="1.0" encoding="utf-8"?>""")

653 f.writelines(self.ToStringLines())

Coverage for pyTooling / Graph / GraphML.py: 88%

411 statements