1#!/usr/bin/env python3

2# Copyright 2015-2020 Canonical Ltd.

3# All rights reserved.

4#

5# Written by:

6# Jonathan Cave <jonathan.cave@canonical.com>

7# Po-Hsu Lin <po-hsu.lin@canonical.com>

8

9 import argparse

10 import logging

11 import os

12 import subprocess

13 import sys

14 import time

15

16 import dbus

17

18

19 TEST_IP = "8.8.8.8"

20 GSM_CON_ID = "GSMCONN"

21

22

23 DBUS_PROPERTIES = "org.freedesktop.DBus.Properties"

24 DBUS_OBJECTMANAGER = "org.freedesktop.DBus.ObjectManager"

25

26 DBUS_MM1_SERVICE = "org.freedesktop.ModemManager1"

27 DBUS_MM1_PATH = "/org/freedesktop/ModemManager1"

28 DBUS_MM1_IF = "org.freedesktop.ModemManager1"

29 DBUS_MM1_IF_MODEM = "org.freedesktop.ModemManager1.Modem"

30 DBUS_MM1_IF_MODEM_SIMPLE = "org.freedesktop.ModemManager1.Modem.Simple"

31 DBUS_MM1_IF_MODEM_3GPP = "org.freedesktop.ModemManager1.Modem.Modem3gpp"

32 DBUS_MM1_IF_MODEM_CDMA = "org.freedesktop.ModemManager1.Modem.ModemCdma"

33 DBUS_MM1_IF_SIM = "org.freedesktop.ModemManager1.Sim"

34

35 MMModemCapability = {

36 "MM_MODEM_CAPABILITY_NONE": 0,

37 "MM_MODEM_CAPABILITY_POTS": 1 << 0,

38 "MM_MODEM_CAPABILITY_CDMA_EVDO": 1 << 1,

39 "MM_MODEM_CAPABILITY_GSM_UMTS": 1 << 2,

40 "MM_MODEM_CAPABILITY_LTE": 1 << 3,

41 "MM_MODEM_CAPABILITY_LTE_ADVANCED": 1 << 4,

42 "MM_MODEM_CAPABILITY_IRIDIUM": 1 << 5,

43 "MM_MODEM_CAPABILITY_ANY": 0xFFFFFFFF,

44}

45

46

47 def print_head(txt):

48 print("##", txt, flush=True)

49

50

51 def print_cmd(cmd):

52 if isinstance(cmd, list):

53 print("+", *cmd, sep=" ", flush=True)

54 else:

55 print("+", cmd, flush=True)

56

57

58 class MMDbus:

59 def __init__(self):

60 self._bus = dbus.SystemBus()

61 self._modems = []

62 try:

63 manager_proxy = self._bus.get_object(

64 DBUS_MM1_SERVICE, DBUS_MM1_PATH

65 )

66 om = dbus.Interface(manager_proxy, DBUS_OBJECTMANAGER)

67 self._modems = om.GetManagedObjects()

68 except dbus.exceptions.DBusException as excp:

69 if (

70 excp.get_dbus_name()

71 == "org.freedesktop.DBus.Error.ServiceUnknown"

72 ):

73 logging.error(excp.get_dbus_message())

74 logging.error(

75 "Note: wwan_tests.py requires ModemManager >=1.0"

76 )

77 else:

78 logging.error(excp.get_dbus_message())

79 return

80

81 def _modem_by_id(self, mm_id):

82 for m in self._modems:

83 if mm_id == (int(os.path.basename(m))):

84 return m

85

86 def _modem_props_iface(self, mm_id):

87 m = self._modem_by_id(mm_id)

88 if m is not None:

89 proxy = self._bus.get_object(DBUS_MM1_SERVICE, m)

90 return dbus.Interface(proxy, dbus_interface=DBUS_PROPERTIES)

91

92 def get_modem_ids(self):

93 modem_ids = []

94 for m in self._modems:

95 modem_ids.append((int(os.path.basename(m))))

96 return modem_ids

97

98 def equipment_id_to_mm_id(self, equipment_id):

99 for mm_id in self.get_modem_ids():

100 if equipment_id == self.get_equipment_id(mm_id):

101 return mm_id

102

103 def get_rat_support(self, mm_id):

104 pi = self._modem_props_iface(mm_id)

105 return pi.Get(DBUS_MM1_IF_MODEM, "CurrentCapabilities")

106

107 def get_equipment_id(self, mm_id):

108 pi = self._modem_props_iface(mm_id)

109 return pi.Get(DBUS_MM1_IF_MODEM, "EquipmentIdentifier")

110

111 def get_manufacturer(self, mm_id):

112 pi = self._modem_props_iface(mm_id)

113 return pi.Get(DBUS_MM1_IF_MODEM, "Manufacturer")

114

115 def get_model_name(self, mm_id):

116 pi = self._modem_props_iface(mm_id)

117 return pi.Get(DBUS_MM1_IF_MODEM, "Model")

118

119 def get_firmware_revision(self, mm_id):

120 pi = self._modem_props_iface(mm_id)

121 return pi.Get(DBUS_MM1_IF_MODEM, "Revision").replace("\r\n", " ")

122

123 def get_hardware_revision(self, mm_id):

124 pi = self._modem_props_iface(mm_id)

125 return pi.Get(DBUS_MM1_IF_MODEM, "HardwareRevision")

126

127 def sim_present(self, mm_id):

128 pi = self._modem_props_iface(mm_id)

129 if pi.Get(DBUS_MM1_IF_MODEM, "Sim") != "/":

130 return True

131 return False

132

133 def _get_sim_pi(self, mm_id):

134 pi = self._modem_props_iface(mm_id)

135 sim_path = pi.Get(DBUS_MM1_IF_MODEM, "Sim")

136 if sim_path != "/":

137 sim_proxy = self._bus.get_object(DBUS_MM1_SERVICE, sim_path)

138 return dbus.Interface(sim_proxy, dbus_interface=DBUS_PROPERTIES)

139

140 def get_sim_operatorname(self, mm_id):

141 sim_pi = self._get_sim_pi(mm_id)

142 if sim_pi is None:

143 return "No card"

144 return sim_pi.Get(DBUS_MM1_IF_SIM, "OperatorName")

145

146 def get_sim_operatoridentifier(self, mm_id):

147 sim_pi = self._get_sim_pi(mm_id)

148 if sim_pi is None:

149 return "No card"

150 return sim_pi.Get(DBUS_MM1_IF_SIM, "OperatorIdentifier")

151

152 def get_sim_imsi(self, mm_id):

153 sim_pi = self._get_sim_pi(mm_id)

154 if sim_pi is None:

155 return "No card"

156 return sim_pi.Get(DBUS_MM1_IF_SIM, "Imsi")

157

158 def get_sim_simidentifier(self, mm_id):

159 sim_pi = self._get_sim_pi(mm_id)

160 if sim_pi is None:

161 return "No card"

162 return sim_pi.Get(DBUS_MM1_IF_SIM, "SimIdentifier")

163

164

165 def _value_from_table(item, item_id, key):

166 if item == "modem":

167 flag = "-m"

168 if item == "sim":

169 flag = "-i"

170 proc = subprocess.Popen(

171 ["mmcli", flag, str(item_id)], stdout=subprocess.PIPE

172 )

173 while True:

174 line = proc.stdout.readline().decode(sys.stdout.encoding)

175 if line == "":

176 break

177 if key in line:

178 chars = " |'\n\t"

179 for c in chars:

180 if c in line:

181 line = line.replace(c, "")

182 value = line.split(":", 1)[1]

183 return value

184

185

186 class MMCLI:

187 def __init__(self):

188 self._modem_ids = []

189 try:

190 proc = subprocess.Popen(["mmcli", "-L"], stdout=subprocess.PIPE)

191 while True:

192 line = proc.stdout.readline().decode(sys.stdout.encoding)

193 if line == "":

194 break

195 if "/org/freedesktop/ModemManager1/Modem" in line:

196 path = line.strip().split()[0]

197 self._modem_ids.append(int(os.path.basename(path)))

198 except OSError:

199 logging.error("mmcli not found")

200

201 def _get_sim_id(self, mm_id):

202 sim_value = _value_from_table("modem", mm_id, "SIM")

203 if sim_value == "none":

204 return None

205 else:

206 return int(os.path.basename(sim_value))

207

208 def get_modem_ids(self):

209 return self._modem_ids

210

211 def equipment_id_to_mm_id(self, equipment_id):

212 for mm_id in self.get_modem_ids():

213 if equipment_id == self.get_equipment_id(mm_id):

214 return mm_id

215

216 def get_equipment_id(self, mm_id):

217 return _value_from_table("modem", mm_id, "equipment id")

218

219 def get_manufacturer(self, mm_id):

220 return _value_from_table("modem", mm_id, "manufacturer")

221

222 def get_model_name(self, mm_id):

223 return _value_from_table("modem", mm_id, "model")

224

225 def get_primary_port(self, mm_id):

226 return _value_from_table("modem", mm_id, "primary port")

227

228 def get_firmware_revision(self, mm_id):

229 return _value_from_table("modem", mm_id, "firmware revision")

230

231 def get_hardware_revision(self, mm_id):

232 return _value_from_table("modem", mm_id, "h/w revision")

233

234 def sim_present(self, mm_id):

235 if self._get_sim_id(mm_id) is None:

236 return False

237 return True

238

239 def get_sim_operatorname(self, mm_id):

240 sim_id = self._get_sim_id(mm_id)

241 if sim_id is None:

242 return "No card"

243 return _value_from_table("sim", sim_id, "operator name")

244

245 def get_sim_imsi(self, mm_id):

246 sim_id = self._get_sim_id(mm_id)

247 if sim_id is None:

248 return "No card"

249 return _value_from_table("sim", sim_id, "imsi")

250

251 def get_sim_operatoridentifier(self, mm_id):

252 sim_id = self._get_sim_id(mm_id)

253 if sim_id is None:

254 return "No card"

255 return _value_from_table("sim", sim_id, "operator id")

256

257 def get_sim_simidentifier(self, mm_id):

258 sim_id = self._get_sim_id(mm_id)

259 if sim_id is None:

260 return "No card"

261 return _value_from_table("sim", sim_id, " id")

262

263

264 def _create_3gpp_connection(wwan_if, apn):

265 print_head("Creating 3GPP Connection")

266 cmd = [

267 "nmcli",

268 "c",

269 "add",

270 "con-name",

271 GSM_CON_ID,

272 "type",

273 "gsm",

274 "ifname",

275 wwan_if,

276 "apn",

277 apn,

278 ]

279 print_cmd(cmd)

280 subprocess.check_call(cmd)

281 print()

282

283

284 def _wwan_radio_on():

285 print_head("Turn radio on")

286 cmd = ["nmcli", "r", "wwan", "on"]

287 print_cmd(cmd)

288 subprocess.check_call(cmd)

289 print()

290

291

292 def _wwan_radio_off():

293 print_head("Turn radio off")

294 cmd = ["nmcli", "r", "wwan", "off"]

295 print_cmd(cmd)

296 subprocess.check_call(cmd)

297 print()

298

299

300 def _wwan_radio_status():

301 print_head("Get radio status")

302 cmd = ["nmcli", "r", "wwan"]

303 ret_sp = subprocess.run(

304 cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=True

305 )

306 print(ret_sp.stdout.decode("utf-8"))

307 return ret_sp.stdout.decode("utf-8").strip()

308

309

310 def _destroy_3gpp_connection():

311 print_head("Destroy 3GPP Connection")

312 cmd = ["nmcli", "c", "delete", GSM_CON_ID]

313 print_cmd(cmd)

314 subprocess.check_call(cmd)

315 print()

316

317

318 def _ping_test(if_name):

319 print_head("Perfrom ping test")

320 ret_code = 1

321 cmd = ["ip", "route", "add", TEST_IP, "dev", if_name]

322 print_cmd(cmd)

323 route = subprocess.call(cmd)

324 if route == 0:

325 cmd = ["ping", "-c", "4", "-I", if_name, TEST_IP]

326 print_cmd(cmd)

327 ret_code = subprocess.check_call(cmd)

328 cmd = ["ip", "route", "del", TEST_IP, "dev", if_name]

329 print_cmd(cmd)

330 subprocess.call(cmd)

331 print()

332 return ret_code

333

334

335 class ThreeGppConnection:

336

337 def invoked(self):

338 parser = argparse.ArgumentParser()

339 parser.add_argument(

340 "hw_id", type=str, help="The hardware ID of the modem"

341 )

342 parser.add_argument(

343 "wwan_net_if",

344 type=str,

345 help="The network interface used when connected",

346 )

347 parser.add_argument(

348 "apn", type=str, help="The APN for data connection"

349 )

350 parser.add_argument(

351 "wwan_setup_time",

352 type=int,

353 default=30,

354 help="delay before ping test",

355 )

356 args = parser.parse_args(sys.argv[2:])

357

358 mm = MMCLI()

359 mm_id = mm.equipment_id_to_mm_id(args.hw_id)

360 wwan_control_if = mm.get_primary_port(mm_id)

361

362 ret_code = 1

363 try:

364 _create_3gpp_connection(wwan_control_if, args.apn)

365 _wwan_radio_on()

366 time.sleep(args.wwan_setup_time)

367 ret_code = _ping_test(args.wwan_net_if)

368 except subprocess.SubprocessError:

369 pass

370 _destroy_3gpp_connection()

371 _wwan_radio_off()

372 sys.exit(ret_code)

373

374

375 class WWANTestCtx:

376

377 def __init__(self, hardware_id, use_mmcli=True, need_enable=False):

378 self.mm_obj = MMCLI() if use_mmcli else MMDbus()

379 self.need_enable = need_enable

380 self.original_status = ""

381 self.modem_idx = self.mm_obj.equipment_id_to_mm_id(hardware_id)

382

383 def __enter__(self):

384 self.original_status = _wwan_radio_status()

385 if self.need_enable and self.original_status == "disabled":

386 _wwan_radio_on()

387 return self

388

389 def __exit__(self, type, value, traceback):

390 if self.need_enable and self.original_status == "disabled":

391 _wwan_radio_off()

392

393

394 class ThreeGppScanTest:

395

396 def register_argument(self):

397

398 parser = argparse.ArgumentParser()

399 parser.add_argument(

400 "hw_id", type=str, help="The hardware ID of the modem"

401 )

402 parser.add_argument(

403 "--timeout",

404 type=int,

405 default=300,

406 help="timeout for 3gpp-scan",

407 )

408 return parser.parse_args(sys.argv[2:])

409

410 def invoked(self):

411

412 args = self.register_argument()

413 ret_code = 1

414 try:

415 with WWANTestCtx(args.hw_id, True, True) as ctx:

416 cmd = [

417 "mmcli",

418 "-m",

419 str(ctx.modem_idx),

420 "--3gpp-scan",

421 "--timeout",

422 str(args.timeout),

423 ]

424 print_head("Scanning 3GPP available network")

425 print("+ {}".format(" ".join(cmd)))

426 ret_sp = subprocess.run(

427 cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT

428 )

429 print(ret_sp.stdout.decode("utf-8"))

430 ret_code = ret_sp.returncode

431 except subprocess.CalledProcessError:

432 pass

433

434 sys.exit(ret_code)

435

436

437 class CountModems:

438

439 def invoked(self):

440 parser = argparse.ArgumentParser()

441 parser.add_argument(

442 "--use-cli",

443 action="store_true",

444 help="Use mmcli for all calls rather than dbus",

445 )

446 args = parser.parse_args(sys.argv[2:])

447 if args.use_cli:

448 mm = MMCLI()

449 else:

450 mm = MMDbus()

451 print(len(mm.get_modem_ids()))

452

453

454 class Resources:

455

456 def invoked(self):

457 parser = argparse.ArgumentParser()

458 parser.add_argument(

459 "--use-cli",

460 action="store_true",

461 help="Use mmcli for all calls rather than dbus",

462 )

463 args = parser.parse_args(sys.argv[2:])

464 if args.use_cli:

465 mm = MMCLI()

466 else:

467 mm = MMDbus()

468 for m in mm.get_modem_ids():

469 print("mm_id: {}".format(m))

470 print("hw_id: {}".format(mm.get_equipment_id(m)))

471 print("manufacturer: {}".format(mm.get_manufacturer(m)))

472 print("model: {}".format(mm.get_model_name(m)))

473 print("firmware_revision: {}".format(mm.get_firmware_revision(m)))

474 print("hardware_revision: {}".format(mm.get_hardware_revision(m)))

475 print()

476

477

478 class SimPresent:

479

480 def invoked(self):

481 parser = argparse.ArgumentParser()

482 parser.add_argument(

483 "hw_id",

484 type=str,

485 help="The hardware ID of the modem whose attached"

486 "SIM we want to query",

487 )

488 parser.add_argument(

489 "--use-cli",

490 action="store_true",

491 help="Use mmcli for all calls rather than dbus",

492 )

493 args = parser.parse_args(sys.argv[2:])

494 if args.use_cli:

495 mm = MMCLI()

496 else:

497 mm = MMDbus()

498 mm_id = mm.equipment_id_to_mm_id(args.hw_id)

499 if not mm.sim_present(mm_id):

500 sys.exit(1)

501

502

503 class SimInfo:

504

505 def invoked(self):

506 parser = argparse.ArgumentParser()

507 parser.add_argument(

508 "hw_id",

509 type=str,

510 help="The hardware ID of the modem whose attached"

511 "SIM we want to query",

512 )

513 parser.add_argument(

514 "--use-cli",

515 action="store_true",

516 help="Use mmcli for all calls rather than dbus",

517 )

518 args = parser.parse_args(sys.argv[2:])

519 if args.use_cli:

520 mm = MMCLI()

521 else:

522 mm = MMDbus()

523 mm_id = mm.equipment_id_to_mm_id(args.hw_id)

524 print("Operator: {}".format(mm.get_sim_operatorname(mm_id)))

525 print("IMSI: {}".format(mm.get_sim_imsi(mm_id)))

526 print("MCC/MNC: {}".format(mm.get_sim_operatoridentifier(mm_id)))

527 print("ICCID: {}".format(mm.get_sim_simidentifier(mm_id)))

528

529

530 class WWANTests:

531

532 def main(self):

533 sub_commands = {

534 "count": CountModems,

535 "resources": Resources,

536 "3gpp-connection": ThreeGppConnection,

537 "3gpp-scan": ThreeGppScanTest,

538 "sim-present": SimPresent,

539 "sim-info": SimInfo,

540 }

541 parser = argparse.ArgumentParser()

542 parser.add_argument("subcommand", type=str, choices=sub_commands)

543 args = parser.parse_args(sys.argv[1:2])

544 sub_commands[args.subcommand]().invoked()

545

546

547 if __name__ == "__main__":

548 WWANTests().main()