"""ODF 光纤配线单元管理系统 - 业务逻辑层。 使用内存假数据存储，模拟 ODF 配线单元的管理功能。 提供机架、配线单元、端口、跳接连接的 CRUD 操作。 """ import uuid from datetime import datetime from typing import Optional from app.models import Rack, Unit, Port, Connection, PortType, PortStatus class OdfService: """ODF 光纤配线单元管理服务。 使用内存字典存储数据，提供对机架、配线单元、端口和跳接连接的管理能力。 此类为单例服务，所有数据在服务运行期间保存在内存中。 """ def __init__(self) -> None: """初始化内存数据存储和初始假数据。""" self._racks: dict[str, Rack] = {} self._units: dict[str, Unit] = {} self._ports: dict[str, Port] = {} self._connections: dict[str, Connection] = {} self._init_demo_data() def _init_demo_data(self) -> None: """初始化演示用假数据，方便测试和展示。""" # 创建 2 个机架 rack1 = Rack( rack_id="rack-001", rack_name="ODF-001", location="A机房-1列-01排", ) rack2 = Rack( rack_id="rack-002", rack_name="ODF-002", location="A机房-1列-02排", ) # 为机架1创建2个单元（每个单元12口） unit1 = self._create_unit_internal("rack-001", 1, "上排单元A", "上排", 12, PortType.LC) unit2 = self._create_unit_internal("rack-001", 2, "下排单元B", "下排", 12, PortType.SC) # 为机架2创建1个单元（24口） unit3 = self._create_unit_internal("rack-002", 1, "上排单元C", "上排", 24, PortType.LC) rack1.units = [unit1, unit2] rack2.units = [unit3] self._racks[rack1.rack_id] = rack1 self._racks[rack2.rack_id] = rack2 # 创建一条示例跳接连接：单元1的端口1 -> 单元2的端口1 unit1_ports = [p for p in self._ports.values() if p.unit_id == unit1.unit_id] unit2_ports = [p for p in self._ports.values() if p.unit_id == unit2.unit_id] if unit1_ports and unit2_ports: pa = unit1_ports[0] pb = unit2_ports[0] pa.status = PortStatus.IN_USE pb.status = PortStatus.IN_USE conn = Connection( connection_id=f"conn-{uuid.uuid4().hex[:8]}", port_a_id=pa.port_id, port_b_id=pb.port_id, fiber_length=3.5, create_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), remark="示例跳接", ) self._connections[conn.connection_id] = conn def _create_unit_internal( self, rack_id: str, unit_number: int, unit_name: str, position: str, port_count: int, port_type: PortType, ) -> Unit: """内部方法：创建一个配线单元及其端口。 Args: rack_id: 所属机架 ID。 unit_number: 单元编号。 unit_name: 单元名称。 position: 位置描述。 port_count: 端口数量。 port_type: 端口类型。 Returns: 创建的 Unit 实例。 """ unit_id = f"unit-{uuid.uuid4().hex[:8]}" unit = Unit( unit_id=unit_id, rack_id=rack_id, unit_number=unit_number, unit_name=unit_name, position=position, ports=[], ) ports: list[Port] = [] for i in range(1, port_count + 1): port_id = f"port-{uuid.uuid4().hex[:8]}" port = Port( port_id=port_id, unit_id=unit_id, port_number=i, port_type=port_type, status=PortStatus.FREE, label=f"{rack_id}-{unit_number}-{i:02d}", ) self._ports[port_id] = port ports.append(port) unit.ports = ports self._units[unit_id] = unit return unit # ==================== 机架管理 ==================== def list_racks(self) -> list[dict]: """获取所有机架的简要列表。 Returns: 包含机架信息和单元数量的字典列表。 """ result = [] for rack in self._racks.values(): result.append({ "rack_id": rack.rack_id, "rack_name": rack.rack_name, "location": rack.location, "unit_count": len(rack.units), }) return result def get_rack(self, rack_id: str) -> Optional[Rack]: """根据 ID 获取机架详情（包含单元和端口）。 Args: rack_id: 机架 ID。 Returns: 机架对象，未找到时返回 None。 """ return self._racks.get(rack_id) def create_rack(self, rack_name: str, location: str) -> Rack: """创建一个新的空机架。 Args: rack_name: 机架名称/编号。 location: 物理位置。 Returns: 新创建的机架对象。 """ rack_id = f"rack-{uuid.uuid4().hex[:8]}" rack = Rack( rack_id=rack_id, rack_name=rack_name, location=location, units=[], ) self._racks[rack_id] = rack return rack def delete_rack(self, rack_id: str) -> bool: """删除指定机架（同时删除其下的单元和端口）。 Args: rack_id: 机架 ID。 Returns: 删除成功返回 True，机架不存在返回 False。 """ if rack_id not in self._racks: return False rack = self._racks[rack_id] for unit in rack.units: for port in unit.ports: self._ports.pop(port.port_id, None) self._units.pop(unit.unit_id, None) del self._racks[rack_id] return True # ==================== 配线单元管理 ==================== def create_unit( self, rack_id: str, unit_number: int, unit_name: str, position: str, port_count: int, port_type: PortType, ) -> Optional[Unit]: """在指定机架下创建配线单元。 Args: rack_id: 所属机架 ID。 unit_number: 单元编号。 unit_name: 单元名称。 position: 位置描述。 port_count: 端口数量。 port_type: 端口类型。 Returns: 创建的 Unit 对象，机架不存在时返回 None。 """ if rack_id not in self._racks: return None unit = self._create_unit_internal(rack_id, unit_number, unit_name, position, port_count, port_type) self._racks[rack_id].units.append(unit) return unit def delete_unit(self, unit_id: str) -> bool: """删除指定配线单元（同时删除其端口）。 Args: unit_id: 单元 ID。 Returns: 删除成功返回 True，单元不存在返回 False。 """ if unit_id not in self._units: return False unit = self._units[unit_id] for port in unit.ports: self._ports.pop(port.port_id, None) rack = self._racks.get(unit.rack_id) if rack: rack.units = [u for u in rack.units if u.unit_id != unit_id] del self._units[unit_id] return True # ==================== 端口管理 ==================== def get_free_ports(self, rack_id: Optional[str] = None) -> list[dict]: """查询空闲端口列表。 Args: rack_id: 可选的机架 ID，指定时只查询该机架下的空闲端口。 Returns: 空闲端口信息列表，包含所属单元和机架信息。 """ result = [] for port in self._ports.values(): if port.status != PortStatus.FREE: continue unit = self._units.get(port.unit_id) if not unit: continue rack = self._racks.get(unit.rack_id) if not rack: continue if rack_id and rack.rack_id != rack_id: continue result.append({ "port_id": port.port_id, "port_number": port.port_number, "port_type": port.port_type, "label": port.label, "unit_id": unit.unit_id, "unit_number": unit.unit_number, "rack_id": rack.rack_id, "rack_name": rack.rack_name, }) return result def get_port_detail(self, port_id: str) -> Optional[dict]: """获取端口的详细信息（含所属单元和机架）。 Args: port_id: 端口 ID。 Returns: 端口详细信息字典，端口不存在时返回 None。 """ port = self._ports.get(port_id) if not port: return None unit = self._units.get(port.unit_id) if not unit: return None rack = self._racks.get(unit.rack_id) if not rack: return None return { "port_id": port.port_id, "port_number": port.port_number, "port_type": port.port_type, "status": port.status, "label": port.label, "unit_id": unit.unit_id, "unit_number": unit.unit_number, "rack_id": rack.rack_id, "rack_name": rack.rack_name, } # ==================== 跳接连接管理 ==================== def create_connection( self, port_a_id: str, port_b_id: str, fiber_length: float, remark: str, ) -> Optional[Connection]: """创建端口之间的跳接连接。 两个端口必须都存在且状态为空闲（FREE）。 Args: port_a_id: A 端端口 ID。 port_b_id: B 端端口 ID。 fiber_length: 光纤长度（米）。 remark: 备注。 Returns: 创建的 Connection 对象，端口不存在或端口繁忙时返回 None。 """ port_a = self._ports.get(port_a_id) port_b = self._ports.get(port_b_id) if not port_a or not port_b: return None if port_a.status != PortStatus.FREE or port_b.status != PortStatus.FREE: return None if port_a_id == port_b_id: return None connection_id = f"conn-{uuid.uuid4().hex[:8]}" conn = Connection( connection_id=connection_id, port_a_id=port_a_id, port_b_id=port_b_id, fiber_length=fiber_length, create_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), remark=remark, ) port_a.status = PortStatus.IN_USE port_b.status = PortStatus.IN_USE self._connections[connection_id] = conn return conn def list_connections(self) -> list[Connection]: """获取所有跳接连接列表。 Returns: 连接对象列表。 """ return list(self._connections.values()) def delete_connection(self, connection_id: str) -> bool: """删除跳接连接（释放两个端口状态为空闲）。 Args: connection_id: 连接 ID。 Returns: 删除成功返回 True，连接不存在返回 False。 """ conn = self._connections.get(connection_id) if not conn: return False port_a = self._ports.get(conn.port_a_id) port_b = self._ports.get(conn.port_b_id) if port_a: port_a.status = PortStatus.FREE if port_b: port_b.status = PortStatus.FREE del self._connections[connection_id] return True def get_port_path(self, port_id: str) -> Optional[dict]: """查询指定端口的连接路径信息。 Args: port_id: 端口 ID。 Returns: 包含端口信息和关联连接的字典，端口不存在时返回 None。 """ port = self._ports.get(port_id) if not port: return None unit = self._units.get(port.unit_id) rack = self._racks.get(unit.rack_id) if unit else None connected_conn = None for conn in self._connections.values(): if conn.port_a_id == port_id or conn.port_b_id == port_id: connected_conn = conn break other_port = None if connected_conn: other_port_id = ( connected_conn.port_b_id if connected_conn.port_a_id == port_id else connected_conn.port_a_id ) other_port = self.get_port_detail(other_port_id) return { "port": { "port_id": port.port_id, "port_number": port.port_number, "port_type": port.port_type, "status": port.status, "label": port.label, "unit_id": unit.unit_id if unit else "", "unit_number": unit.unit_number if unit else 0, "rack_id": rack.rack_id if rack else "", "rack_name": rack.rack_name if rack else "", }, "connection": { "connection_id": connected_conn.connection_id if connected_conn else None, "fiber_length": connected_conn.fiber_length if connected_conn else 0, "create_time": connected_conn.create_time if connected_conn else "", "remark": connected_conn.remark if connected_conn else "", } if connected_conn else None, "connected_port": other_port, } # 全局单例服务实例 service = OdfService()