Newer
Older
#!/usr/bin/env python3
# Copyright 2020 Canonical Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
sys.path.append("lib")
from ops.charm import CharmBase
from ops.framework import StoredState
from ops.main import main
from ops.model import (
ActiveStatus,
BlockedStatus,
MaintenanceStatus,
WaitingStatus,
ModelError,
)
import subprocess
import logging
import asyncio
import random
def __init__(self, *args):
super().__init__(*args)
# An example of setting charm state
# that's persistent across events
self.state.set_default(is_started=False)
if not self.state.is_started:
self.state.is_started = True
# Register all of the events we want to observe
for event in (
# Charm events
self.on.config_changed,
self.on.install,
self.on.upgrade_charm,
self.on.register_action,
self.on.attach_ue_action,
self.on.unregister_action,
self.on.detach_ue_action,
def on_config_changed(self, event):
"""Handle changes in configuration"""
unit = self.model.unit
def on_install(self, event):
"""Called when the charm is being installed"""
unit = self.model.unit
# Install your software and its dependencies
unit.status = ActiveStatus()
def on_upgrade_charm(self, event):
"""Upgrade the charm."""
unit = self.model.unit
# Mark the unit as under Maintenance.
unit.status = MaintenanceStatus("Upgrading charm")
self.on_install(event)
# When maintenance is done, return to an Active state
unit.status = ActiveStatus()
def on_register_action(self, event):
"""Register to AGW (EPC)."""
try:
mme_addr = event.params["mme-addr"]
gtp_bind_addr = event.params["gtp-bind-addr"]
s1c_bind_addr = event.params["s1c-bind-addr"]
log_file = "/tmp/{}.log".format(random.randint(1000, 100000))
"/home/ubuntu/srsLTE/build/srsenb/src/srsenb",
"--enb.name=dummyENB01",
"--enb.mcc=901",
"--enb.mnc=70",
"--enb.mme_addr={}".format(mme_addr),
"--enb.gtp_bind_addr={}".format(gtp_bind_addr),
"--enb.s1c_bind_addr={}".format(s1c_bind_addr),
"--enb_files.rr_config=/configzmq/rr.conf",
"--enb_files.sib_config=/configzmq/sib.conf",
"--enb_files.drb_config=/configzmq/drb.conf",
"/configzmq/enb.conf",
"--rf.device_name=zmq",
"--rf.device_args='fail_on_disconnect=true,tx_port=tcp://*:2000,rx_port=tcp://localhost:2001,id=enb,base_srate=23.04e6'",
logger.debug("Register action: executing")
process = self._run_daemon(command, log_file)
logger.debug("Register action: executed")
event.set_results(
{"status": "ok", "pid": process.pid, "log-file": log_file}
)
except subprocess.CalledProcessError as e:
event.fail("Command error: {}".format(e.output))
except asyncio.TimeoutError as e:
event.fail("Timeout error")
except Exception as e:
event.fail(e)
def on_unregister_action(self, event):
"""Unregister action"""
command = "sudo killall -s KILL srsenb"
output = subprocess.check_output(command, shell=True)
{
"status": "ok",
"message": "Unregistered successfully",
"output": output,
}
)
except subprocess.CalledProcessError as e:
event.fail("Command error: {}".format(e.output))
except Exception as e:
event.fail(e)
def on_attach_ue_action(self, event):
"""Attach User Emulator to EnodeB."""
try:
usim_imsi = event.params["usim-imsi"]
usim_k = event.params["usim-k"]
usim_opc = event.params["usim-opc"]
log_file = "/tmp/{}.log".format(random.randint(1000, 100000))
"sudo",
"/home/ubuntu/srsLTE/build/srsue/src/srsue",
"--usim.imsi={}".format(usim_imsi),
"--usim.k={}".format(usim_k),
"--usim.algo=milenage",
"--usim.opc={}".format(usim_opc),
"--nas.apn=oai.ipv4",
"--rf.device_name=zmq",
"--rf.device_args='tx_port=tcp://*:2001,rx_port=tcp://localhost:2000,id=ue,base_srate=23.04e6'",
logger.debug("Attach UE action: executing")
process = self._run_daemon(command, log_file)
logger.debug("Attach UE action: executed")
event.set_results(
{"status": "ok", "pid": process.pid, "log-file": log_file}
except subprocess.CalledProcessError as ex:
event.fail(ex)
def on_detach_ue_action(self, event):
"""Detach UE action"""
try:
command = "sudo killall -s KILL srsue"
output = subprocess.check_output(command, shell=True)
event.set_results(
{"status": "ok", "message": "Detached successfully", "output": output}
)
except subprocess.CalledProcessError as e:
event.fail("Command error: {}".format(e.output))
except Exception as e:
event.fail(e)
"""Remove default gw"""
try:
command = "sudo route del default"
output = subprocess.check_output(command, shell=True)
event.set_results(
{"status": "ok", "message": "Default route removed!", "output": output}
)
except subprocess.CalledProcessError as e:
event.fail("Command error: {}".format(e.output))
except Exception as e:
event.fail(e)
def _run_daemon(self, cmd, stdout_file):
with open(stdout_file, "wb") as f:
return subprocess.Popen(cmd, shell=True, stdout=f)