telecommunications-expert

name	telecommunications-expert
version	1.0.0
description	Expert-level telecommunications systems, network management, billing, 5G, SDN, and telecom infrastructure
category	domains
tags	["telecom","networking","5g","billing","oss","bss"]
allowed-tools	["Read","Write","Edit"]

Telecommunications Expert

Expert guidance for telecommunications systems, network management, billing systems, 5G networks, SDN/NFV, and telecom infrastructure management.

Core Concepts

Telecommunications Systems

• Operations Support Systems (OSS)
• Business Support Systems (BSS)
• Network Management Systems (NMS)
• Service Assurance
• Inventory Management
• Provisioning systems
• Customer care platforms

Network Technologies

• 5G/4G/LTE networks
• Fiber optic networks
• Software-Defined Networking (SDN)
• Network Functions Virtualization (NFV)
• Edge computing
• IoT connectivity
• Satellite communications

Standards and Protocols

• 3GPP standards
• TM Forum Frameworx
• ETSI specifications
• ITU-T recommendations
• SIP (Session Initiation Protocol)
• Diameter protocol
• SNMP for network management

Network Management System

from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional, Dict
from enum import Enum
import numpy as np

class NetworkElementType(Enum):
    BASE_STATION = "base_station"
    ROUTER = "router"
    SWITCH = "switch"
    FIBER_NODE = "fiber_node"
    GATEWAY = "gateway"
    FIREWALL = "firewall"

class AlarmSeverity(Enum):
    CRITICAL = "critical"
    MAJOR = "major"
    MINOR = "minor"
    WARNING = "warning"
    CLEARED = "cleared"

@dataclass
class NetworkElement:
    """Network infrastructure element"""
    element_id: str
    element_type: NetworkElementType
    name: str
    location: dict
    ip_address: str
    status: str  # 'active', 'inactive', 'maintenance'
    vendor: str
    model: str
    software_version: str
    capacity: dict
    utilization: dict

@dataclass
class NetworkAlarm:
    
    alarm_id: 
    element_id: 
    severity: AlarmSeverity
    alarm_type: 
    description: 
    timestamp: datetime
    acknowledged: 
    cleared: 
    clear_timestamp: [datetime]


 :
    
    element_id: 
    metric_name: 
    value: 
    unit: 
    timestamp: datetime
    threshold_warning: 
    threshold_critical: 

 :
    

     ():
        .network_elements = {}
        .alarms = []
        .performance_data = []

     () -> :
        
        element = .network_elements.get(element_id)
          element:
             {: }

        
        metrics = ._collect_snmp_metrics(element)

        
        violations = []
         metric  metrics:
             metric.value >= metric.threshold_critical:
                violations.append({
                    : metric.metric_name,
                    : metric.value,
                    : metric.threshold_critical,
                    : 
                })
                ._raise_alarm(element_id, AlarmSeverity.CRITICAL,
                                )

             metric.value >= metric.threshold_warning:
                violations.append({
                    : metric.metric_name,
                    : metric.value,
                    : metric.threshold_warning,
                    : 
                })

         {
            : element_id,
            : element.status,
            : [
                {
                    : m.metric_name,
                    : m.value,
                    : m.unit
                }
                 m  metrics
            ],
            : violations,
            : ._calculate_health_score(element, metrics)
        }

     () -> [PerformanceMetric]:
        
        metrics = []
        timestamp = datetime.now()

        
        cpu_util = ._get_cpu_utilization(element)
        metrics.append(PerformanceMetric(
            element_id=element.element_id,
            metric_name=,
            value=cpu_util,
            unit=,
            timestamp=timestamp,
            threshold_warning=,
            threshold_critical=
        ))

        
        mem_util = ._get_memory_utilization(element)
        metrics.append(PerformanceMetric(
            element_id=element.element_id,
            metric_name=,
            value=mem_util,
            unit=,
            timestamp=timestamp,
            threshold_warning=,
            threshold_critical=
        ))

        
         interface  [, ]:
            traffic = ._get_interface_traffic(element, interface)
            metrics.append(PerformanceMetric(
                element_id=element.element_id,
                metric_name=,
                value=traffic,
                unit=,
                timestamp=timestamp,
                threshold_warning=,
                threshold_critical=
            ))

         metrics

     () -> :
        
         element.status != :
             

        score = 

         metric  metrics:
             metric.value >= metric.threshold_critical:
                score -= 
             metric.value >= metric.threshold_warning:
                score -= 

         (, score)

     ():
        
        alarm = NetworkAlarm(
            alarm_id=._generate_alarm_id(),
            element_id=element_id,
            severity=severity,
            alarm_type=,
            description=description,
            timestamp=datetime.now(),
            acknowledged=,
            cleared=,
            clear_timestamp=
        )

        .alarms.append(alarm)

        
         severity == AlarmSeverity.CRITICAL:
            ._send_alarm_notification(alarm)

     () -> :
        
        
        region_elements = [
            e  e  .network_elements.values()
             e.location.get() == region
        ]

          region_elements:
             {: }

        
        total_capacity = 
        total_used = 

         element  region_elements:
            capacity = element.capacity.get(, )
            utilization = element.utilization.get(, )

            total_capacity += capacity
            total_used += capacity * (utilization / )

        utilization_percent = (total_used / total_capacity * )  total_capacity >   

        
        growth_rate =   
        months_until_full = ._predict_capacity_exhaustion(
            total_capacity,
            total_used,
            growth_rate
        )

         {
            : region,
            : total_capacity,
            : total_used,
            : total_capacity - total_used,
            : utilization_percent,
            : months_until_full,
            : months_until_full < 
        }

     () -> :
        
         current_usage >= total_capacity:
             

        available = total_capacity - current_usage
        monthly_growth_rate = growth_rate / 

        
        target_usage = total_capacity * 
        usage_needed = target_usage - current_usage

         usage_needed <= :
             

        months = np.log( + (usage_needed / current_usage)) / np.log( + monthly_growth_rate)

         months

     () -> :
        
        
         np.random.uniform(, )  

     () -> :
        
         np.random.uniform(, )  

     () -> :
        
         np.random.uniform(, )  

     ():
        
        
        

     () -> :
         uuid
         

Billing System

from decimal import Decimal

@dataclass
class Subscriber:
    """Telecom subscriber"""
    subscriber_id: str
    account_number: str
    name: str
    phone_number: str
    email: str
    address: dict
    plan_id: str
    status: str  # 'active', 'suspended', 'terminated'
    activation_date: datetime

@dataclass
class ServicePlan:
    """Service plan/package"""
    plan_id: str
    name: str
    description: str
    monthly_fee: Decimal
    data_allowance_gb: float
    voice_minutes: int
    sms_count: int
    overage_rates: dict

@dataclass
class UsageRecord:
    """Usage record for billing"""
    record_id: str
    subscriber_id: str
    usage_type: str  # 'voice', 'sms', 'data'
    timestamp: datetime
    quantity: float
    unit: str
    destination: Optional[str]
    charged: bool

class BillingSystem:
    """Telecom billing and charging system"""

    def __init__(self):
        self.subscribers = {}
        .service_plans = {}
        .usage_records = []
        .invoices = []

     () -> :
        
        subscriber = .subscribers.get(usage.subscriber_id)
          subscriber:
             {: }

         subscriber.status != :
             {: }

        plan = .service_plans.get(subscriber.plan_id)
          plan:
             {: }

        
        current_usage = ._get_current_month_usage(usage.subscriber_id, usage.usage_type)

        charge = Decimal()

         usage.usage_type == :
             current_usage > plan.data_allowance_gb:
                
                overage_gb = usage.quantity
                charge = Decimal((overage_gb)) * plan.overage_rates[]

         usage.usage_type == :
             current_usage > plan.voice_minutes:
                
                overage_minutes = usage.quantity
                charge = Decimal((overage_minutes)) * plan.overage_rates[]

         usage.usage_type == :
             current_usage > plan.sms_count:
                
                overage_sms = usage.quantity
                charge = Decimal((overage_sms)) * plan.overage_rates[]

        usage.charged = 
        .usage_records.append(usage)

         {
            : usage.subscriber_id,
            : usage.usage_type,
            : usage.quantity,
            : (charge),
            : charge == 
        }

     () -> :
        
        subscriber = .subscribers.get(subscriber_id)
          subscriber:
             {: }

        plan = .service_plans.get(subscriber.plan_id)
        start_date, end_date = billing_period

        
        monthly_fee = plan.monthly_fee

        
        period_usage = [
            u  u  .usage_records
             u.subscriber_id == subscriber_id 
            start_date <= u.timestamp <= end_date
        ]

        usage_charges = ._calculate_usage_charges(period_usage, plan)

        
        subtotal = monthly_fee + usage_charges[]
        tax_rate = Decimal()  
        taxes = subtotal * tax_rate

        total = subtotal + taxes

        invoice = {
            : ._generate_invoice_id(),
            : subscriber_id,
            : subscriber.account_number,
            : {
                : start_date.isoformat(),
                : end_date.isoformat()
            },
            : {
                : (monthly_fee),
                : (usage_charges[]),
                : (usage_charges[]),
                : (usage_charges[]),
                : (usage_charges[])
            },
            : (subtotal),
            : (taxes),
            : (total),
            : (end_date + timedelta(days=)).isoformat()
        }

        .invoices.append(invoice)

         invoice

     () -> :
        
        current_month_start = datetime.now().replace(day=, hour=, minute=, second=)

        usage = [
            u  u  .usage_records
             u.subscriber_id == subscriber_id 
            u.usage_type == usage_type 
            u.timestamp >= current_month_start
        ]

        total = (u.quantity  u  usage)
         total

     () -> :
        
        charges = {
            : Decimal(),
            : Decimal(),
            : Decimal(),
            : Decimal(),
            : Decimal()
        }

        
        data_usage = (u.quantity  u  usage_records  u.usage_type == )
        voice_usage = (u.quantity  u  usage_records  u.usage_type == )
        sms_usage = (u.quantity  u  usage_records  u.usage_type == )

        
         data_usage > plan.data_allowance_gb:
            overage = data_usage - plan.data_allowance_gb
            charges[] = Decimal((overage)) * plan.overage_rates[]

         voice_usage > plan.voice_minutes:
            overage = voice_usage - plan.voice_minutes
            charges[] = Decimal((overage)) * plan.overage_rates[]

         sms_usage > plan.sms_count:
            overage = sms_usage - plan.sms_count
            charges[] = Decimal((overage)) * plan.overage_rates[]

        charges[] = ([charges[], charges[], charges[], charges[]])

         charges

     () -> :
         uuid
         

5G Network Management

class FiveGNetworkManagement:
    """5G network management and optimization"""

    def __init__(self):
        self.base_stations = {}
        self.network_slices = {}

    def configure_network_slice(self, slice_config: dict) -> dict:
        """Configure 5G network slice"""
        slice_id = self._generate_slice_id()

        network_slice = {
            'slice_id': slice_id,
            'name': slice_config['name'],
            'slice_type': slice_config['slice_type'],  # 'eMBB', 'URLLC', 'mMTC'
            'resources': {
                'bandwidth_mhz': slice_config['bandwidth'],
                'latency_ms': slice_config['max_latency'],
                'reliability': slice_config['reliability']
            },
            'qos_profile': slice_config['qos_profile'],
            'status': 'active'
        }

        self.network_slices[slice_id] = network_slice

        # Allocate resources
        self._allocate_slice_resources(network_slice)

        return {
            'slice_id': slice_id,
            'status': 'configured',
            'resources_allocated': True
        }

    def () -> :
        
        
        

        num_users = (user_positions)
        num_antennas =   

        
        beam_directions = []
         position  user_positions:
            angle = ._calculate_beam_angle(position)
            beam_directions.append(angle)

        
        

         {
            : base_station_id,
            : num_users,
            : num_antennas,
            : beam_directions,
            :   
        }

     () -> :
        
        
        source_rsrp = ._measure_rsrp(ue_id, source_cell)
        target_rsrp = ._measure_rsrp(ue_id, target_cell)

        
        handover_threshold =   
         target_rsrp > source_rsrp + handover_threshold:
            
            result = ._execute_handover(ue_id, source_cell, target_cell)

             {
                : ue_id,
                : ,
                : source_cell,
                : target_cell,
                : source_rsrp,
                : target_rsrp
            }
        :
             {
                : ue_id,
                : ,
                : source_rsrp,
                : target_rsrp
            }

     ():
        
        
        

     () -> :
        
        
           

     () -> :
        
        
         np.random.uniform(-, -)  

     () -> :
        
        
         

     () -> :
         uuid
         

Best Practices

Network Management

• Implement proactive monitoring
• Use predictive analytics for fault detection
• Automate routine tasks
• Maintain network documentation
• Implement configuration management
• Use centralized logging
• Monitor key performance indicators (KPIs)

Billing Systems

• Ensure real-time charging
• Implement usage mediation
• Support multiple rating models
• Provide transparent billing
• Enable self-service portal
• Automate invoice generation
• Implement payment processing

5G Networks

• Implement network slicing
• Optimize for low latency
• Use edge computing
• Enable dynamic resource allocation
• Support massive IoT connectivity
• Implement security measures
• Monitor QoS metrics

Service Assurance

• Track service level agreements (SLAs)
• Implement automated testing
• Monitor customer experience
• Provide real-time diagnostics
• Enable root cause analysis
• Track mean time to repair (MTTR)
• Implement service quality metrics

Anti-Patterns

❌ Reactive network management only ❌ Manual provisioning processes ❌ No capacity planning ❌ Inaccurate billing ❌ Poor alarm management (alarm storms) ❌ No network redundancy ❌ Ignoring customer experience metrics ❌ Manual configuration changes ❌ No disaster recovery plan

Resources

• TM Forum: https://www.tmforum.org/
• 3GPP: https://www.3gpp.org/
• ETSI: https://www.etsi.org/
• ITU-T: https://www.itu.int/
• GSMA: https://www.gsma.com/
• ONF (Open Networking Foundation): https://opennetworking.org/
• O-RAN Alliance: https://www.o-ran.org/