krosh/network-diagrams-tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix vlan id bug

Browse Source 
add new examples
add magick for png output
This commit is contained in:
krosh
2026-03-26 01:36:34 +03:00
parent c6b97b7a13
commit 3ab25dfd3f
7 changed files with 93 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TODO
+9 -3
View File
@@ -1,3 +1,9 @@
- в данный момент маска никуда не прходит
- обработка виртуальных интерфейсов в yaml
- добавление bridge и vlan сущностей
- CI для тестов
- CI для автоматической генерации topology
- CI для автоматической генерации документации (если хватит сил)
- Флаг для генерации "чистых" yaml файлов
- Добавление к examples yaml файлов (?)
- тогда надо и диаграммы, а тогда и сами могут прогнать, лишние файлы
- Актуализировать flake
- проверить реальность yaml
- валидация сейчас раскинута повсюду
examples/2hostsswitch/table.csv
+10 -12
View File
@@ -1,12 +1,10 @@
Name,Role,Adapter,Interface,Master Interface,Network,VLAN,Network IP,Mask,Device IP,Default Gateway
PC1,Host,Adapter1,eth1,vlan7,A,,,,,
PC1,Host,,vlan7,,,7,10.10.10.0,/24,10.10.10.7,0.0.0.0
PC2,Host,Adapter1,eth1,vlan9,B,,,,,
PC2,Host,,vlan9,,,9,10.10.10.0,/24,10.10.10.9,0.0.0.0
com,Switch,Adapter1,eth1,vlan7,A,7,,,,
com,Switch,Adapter2,eth2,vlan9,B,9,,,,
com,Switch,,vlan7,bridge,,,,,,
com,Switch,,vlan9,bridge,,,,,,
com,Switch,,bridge,,,,,,,
## REMAKE
Name,Role,Adapter,Interface,Parent Interface,Slave Interfaces,Network,VLAN,Network IP,Mask,Device IP,Default Gateway
PC1,Host,Adapter1,eth1,,,A,,,,,
PC1,Host,,vlan7,eth1,,,7,10.10.10.0,/24,10.10.10.7,0.0.0.0
PC2,Host,Adapter1,eth1,,,B,,,,,
PC2,Host,,vlan9,eth1,,,9,10.10.10.0,/24,10.10.10.9,0.0.0.0
com,Switch,Adapter1,eth1,,,A,7,,,,
com,Switch,Adapter2,eth2,,,B,9,,,,
com,Switch,,vlan7,eth1,,,,,,,
com,Switch,,vlan9,eth2,,,,,,,
com,Switch,,bridge,,"vlan7,vlan9",,,,,,wa
1 Name Role Adapter Interface Master Interface Parent Interface Slave Interfaces Network VLAN Network IP Mask Device IP Default Gateway
2 PC1 Host Adapter1 eth1 vlan7 A
3 PC1 Host vlan7 eth1 7 10.10.10.0 /24 10.10.10.7 0.0.0.0
4 PC2 Host Adapter1 eth1 vlan9 B
5 PC2 Host vlan9 eth1 9 10.10.10.0 /24 10.10.10.9 0.0.0.0
6 com Switch Adapter1 eth1 vlan7 A 7
7 com Switch Adapter2 eth2 vlan9 B 9
8 com Switch vlan7 bridge eth1
9 com Switch vlan9 bridge eth2
10 com Switch bridge vlan7,vlan9 wa
## REMAKE
src/netdiag/domain/models.py
+39 -30
View File
@@ -1,6 +1,7 @@
from typing import Any, Dict, List, Optional
from typing import Dict, List, Optional
import logging
# ===========================
# TODO rewrite to dataclasses with validation
class Interface:
@@ -13,13 +14,13 @@ class Interface:
network: Optional[str]
subnet_mask: Optional[str]
default_gateway: Optional[str]
vlan: Optional[str]
# ---
device: "Device" # set by Device.add_interface() when the interface is added to a device
def __init__(
self,
name: str,
itype: Optional[str] = None,
adapter: Optional[str] = None,
slave_interfaces: Optional[List[str]] = None,
parent_interface: Optional[str] = None,
@@ -27,11 +28,10 @@ class Interface:
network: Optional[str] = None,
subnet_mask: Optional[str] = None,
default_gateway: Optional[str] = None,
vlan: Optional[str] = None,
):
if not isinstance(name, str):
raise ValueError("Interface 'name' must be a string")
if itype is not None and not isinstance(itype, str):
raise ValueError("Interface 'itype' must be a string or None")
if ip_address is not None and not isinstance(ip_address, str):
raise ValueError("Interface 'ip_address' must be a string or None")
if network is not None and not isinstance(network, str):
@@ -48,33 +48,43 @@ class Interface:
)
if parent_interface is not None and not isinstance(parent_interface, str):
raise ValueError("Interface 'parent_interface' must be a string or None")
if vlan is not None and not isinstance(vlan, str):
raise ValueError("Interface 'vlan' must be a string or None")
itype = None if itype == "" else itype
# edit this
adapter = None if adapter == "" else adapter
slave_interfaces = None if slave_interfaces == [] else slave_interfaces
parent_interface = None if parent_interface == "" else parent_interface
vlan = None if vlan == "" else vlan
if itype is None and adapter is None:
itype = "virtual"
if adapter is not None:
if itype is not None and itype != "physical":
raise ValueError(
"Interface with an adapter must have 'itype' set to 'physical'"
logging.debug(
f"Creating interface '{name}' with adapter='{adapter}', slave_interfaces='{slave_interfaces}', parent_interface='{parent_interface}'"
)
# looks like shit
if (
(adapter is not None)
+ (slave_interfaces is not None)
+ (parent_interface is not None)
) != 1:
raise ValueError(
"Interface must have exactly one of 'adapter', 'slave_interfaces', or 'parent_interface' defined"
)
if slave_interfaces is not None:
itype = "bridge"
network = None
elif parent_interface is not None:
if vlan is None:
raise ValueError(
"Interface with 'parent_interface' must have 'vlan' defined"
)
itype = "vlan"
network = None
else:
itype = "physical"
if itype not in (None, "physical", "virtual", "bridge", "vlan"):
raise ValueError(
"Interface 'itype' must be one of 'physical', 'virtual', 'bridge', 'vlan', or None"
)
# print(
# f"Creating interface '{name}' with itype='{itype}', adapter='{adapter}', slave_interfaces='{slave_interfaces}', parent_interface='{parent_interface}', ip_address='{ip_address}', network='{network}', subnet_mask='{subnet_mask}', default_gateway='{default_gateway}'"
# )
if itype == "bridge" and not slave_interfaces:
raise ValueError("Bridge interfaces must have 'slave_interfaces' defined")
if itype == "vlan" and not parent_interface:
raise ValueError("VLAN interfaces must have 'parent_interface' defined")
# sort all this values
self.name = name
self.itype = itype
self.ip_address = ip_address
@@ -84,9 +94,10 @@ class Interface:
self.adapter = adapter
self.slave_interfaces = slave_interfaces
self.parent_interface = parent_interface
self.vlan = vlan
def __repr__(self) -> str:
return f"Interface(name={self.name}, itype={self.itype}, ip_address={self.ip_address}, network={self.network}, default_gateway={self.default_gateway}, subnet_mask={self.subnet_mask}, adapter={self.adapter}, slave_interfaces={self.slave_interfaces})"
return f"Interface(name={self.name}, itype={self.itype}, ip_address={self.ip_address}, network={self.network}, default_gateway={self.default_gateway}, subnet_mask={self.subnet_mask}, adapter={self.adapter}, slave_interfaces={self.slave_interfaces}, vlan={self.vlan})"
# ===========================
@@ -113,7 +124,6 @@ class Device:
def rm_interface(self, interface: Interface):
if interface.name in self.interfaces:
del self.interfaces[interface.name]
# interface.device = None # clear the device attribute of the interface
del interface
else:
raise ValueError(
@@ -142,8 +152,7 @@ class Switch(Device):
class Network:
name: str
interfaces: List[Interface]
vlan: Optional[str] # need a proper parse
network_ip: Optional[str]
network_ip: Optional[str] # а что он делает?
def __init__(
self,
src/netdiag/output/d2.py
+31 -32
View File
@@ -10,42 +10,15 @@ from py_d2.connection import Direction
# https://d2lang.com/tour/themes/
THEME_NUMBER = 200
"""
PC2
PC2
PC3
PC3.adapter1 : {
shape: parallelogram
}
PC3.adapter2 : {
shape: parallelogram
}
netw_A : {
shape: cloud
}
netw_B : {
shape: cloud
}
PC1.adapter1 -- netw_A
PC2.adapter1 -- netw_A
PC3.adapter1 -- netw_A
PC1.adapter2 -- netw_B
PC3.adapter2 -- netw_B
"""
def _check_d2_installed() -> bool:
return shutil.which("d2") is not None
def _check_magick_installed() -> bool:
return shutil.which("magick") is not None
def generate_d2_diagram(topology: Topology, output_path: Path) -> None:
shapes = []
connections = []
@@ -80,7 +53,33 @@ def generate_d2_diagram(topology: Topology, output_path: Path) -> None:
with open(output_path, "w", encoding="utf-8") as f:
f.write(str(diagram))
_generate_picture(output_path, output_path.with_suffix(".png"))
_generate_picture(
output_path, output_path.with_suffix(".png")
) # output_path not used
_run_magick_command(
output_path.with_suffix(".svg"), output_path.with_suffix(".png")
)
def _run_magick_command(input_path: Path, output_path: Path) -> None:
if not _check_magick_installed():
raise EnvironmentError(
"ImageMagick is not installed or 'magick' command is not found in PATH. Please install ImageMagick to use this feature."
)
res = subprocess.run(
["magick", "convert", str(input_path), str(output_path)],
check=True,
capture_output=True,
)
if res.returncode != 0:
raise RuntimeError(
f"Image conversion failed (code {res.returncode})\n"
f"Stdout: {res.stdout.decode()}\n"
f"Stderr: {res.stderr.decode()}"
)
def _generate_picture(diagram: Path, output_path: Path) -> None:
src/netdiag/output/file_convert.py
+1 -2
View File
@@ -60,7 +60,6 @@ def make_yaml(topology: Topology, output_path: Path) -> None:
"name": interface.name,
"members": interface.slave_interfaces,
"ip": ip,
"network": interface.network if interface.network else None,
"gateway": (
interface.default_gateway
if interface.default_gateway
@@ -75,12 +74,12 @@ def make_yaml(topology: Topology, output_path: Path) -> None:
"name": interface.name,
"parent": interface.parent_interface,
"ip": ip,
"network": interface.network if interface.network else None,
"gateway": (
interface.default_gateway
if interface.default_gateway
else None
),
"id": interface.vlan,
}
)
continue
src/netdiag/output/graphviz.py
+4
View File
@@ -3,6 +3,10 @@ from pathlib import Path
import graphviz
import shutil
# TODO:
# - find ithers implementations of graphviz
# - actualize code
def _check_graphviz_installed() -> bool:
return shutil.which("dot") is not None
src/netdiag/parse/convert_raw.py
+1 -7
View File
@@ -15,7 +15,6 @@ name_matching = {
"DEVICE_TYPE": "Role",
"DEVICE_NAME": "Name",
"ADAPTER": "Adapter",
"INTERFACE TYPE": "Interface Type",
"MASTER": "Master Interface",
"SLAVES": "Slave Interfaces",
"PARENT": "Parent Interface",
@@ -83,13 +82,13 @@ def add_interfaces(devices: list[Device], raw_devices: list[RawDevices]) -> None
interface = Interface(
name=_get_from_field(raw_device.fields, "INTERFACE_NAME"),
itype=_get_from_field(raw_device.fields, "INTERFACE TYPE"),
adapter=_get_from_field(raw_device.fields, "ADAPTER"),
slave_interfaces=(
_get_from_field(raw_device.fields, "SLAVES").split(",")
if _get_from_field(raw_device.fields, "SLAVES")
else None
),
vlan=_get_from_field(raw_device.fields, "VLAN"),
parent_interface=_get_from_field(raw_device.fields, "PARENT"),
ip_address=_get_from_field(raw_device.fields, "IP_ADDRESS"),
network=_get_from_field(raw_device.fields, "NETWORK_NAME"),
@@ -109,21 +108,16 @@ def parse_networks(raw_devices: list[RawDevices]) -> list[Network]:
if not network_name:
continue
vlan = (raw_device.fields.get(name_matching["VLAN"], "").strip(),)
network_ip = (raw_device.fields.get(name_matching["NETWORK_IP"], "").strip(),)
if network_name in [n.name for n in networks]: # update if already exists
network = next(n for n in networks if n.name == network_name)
if not network.vlan and vlan:
network.vlan = vlan
if not network.network_ip and network_ip:
network.network_ip = network_ip
continue
network = Network(
name=network_name,
vlan=raw_device.fields.get(name_matching["VLAN"], "").strip(),
network_ip=raw_device.fields.get(name_matching["NETWORK_IP"], "").strip(),
)
networks.append(network)