src/target/trx_toolkit/data_msg.py

   1 # -*- coding: utf-8 -*-
   2
   3 # TRX Toolkit
   4 # DATA interface message definitions and helpers
   5 #
   6 # (C) 2018-2019 by Vadim Yanitskiy <axilirator@gmail.com>
   7 #
   8 # All Rights Reserved
   9 #
  10 # This program is free software; you can redistribute it and/or modify
  11 # it under the terms of the GNU General Public License as published by
  12 # the Free Software Foundation; either version 2 of the License, or
  13 # (at your option) any later version.
  14 #
  15 # This program is distributed in the hope that it will be useful,
  16 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  17 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  18 # GNU General Public License for more details.
  19
  20 import random
  21 import struct
  22 import abc
  23
  24 from typing import List
  25 from enum import Enum
  26 from gsm_shared import *
  27
  28 class Modulation(Enum):
  29         """ Modulation types defined in 3GPP TS 45.002 """
  30         ModGMSK         = (0b0000, 1 * GMSK_BURST_LEN)
  31         Mod8PSK         = (0b0100, 3 * GMSK_BURST_LEN)
  32         ModGMSK_AB      = (0b0110, 1 * GMSK_BURST_LEN)
  33         # ModRFU        = (0b0111, 0) # Reserved for Future Use
  34         Mod16QAM        = (0b1000, 4 * GMSK_BURST_LEN)
  35         Mod32QAM        = (0b1010, 5 * GMSK_BURST_LEN)
  36         ModAQPSK        = (0b1100, 2 * GMSK_BURST_LEN)
  37
  38         def __init__(self, coding, bl):
  39                 # Coding in TRXD header
  40                 self.coding = coding
  41                 # Burst length
  42                 self.bl = bl
  43
  44         @classmethod
  45         def pick(self, coding):
  46                 for mod in list(self):
  47                         if mod.coding == coding:
  48                                 return mod
  49                 return None
  50
  51         @classmethod
  52         def pick_by_bl(self, bl):
  53                 for mod in list(self):
  54                         if mod.bl == bl:
  55                                 return mod
  56                 return None
  57
  58 class Msg(abc.ABC):
  59         ''' TRXD (DATA) message coding API (common part). '''
  60
  61         # NOTE: up to 16 versions can be encoded
  62         CHDR_VERSION_MAX = 0b1111
  63         KNOWN_VERSIONS = (0, 1)
  64
  65         def __init__(self, fn = None, tn = None, burst = None, ver = 0):
  66                 self.burst = burst
  67                 self.ver = ver
  68                 self.fn = fn
  69                 self.tn = tn
  70
  71         @property
  72         def CHDR_LEN(self):
  73                 ''' The common header length. '''
  74                 return 1 + 4 # (VER + TN) + FN
  75
  76         @abc.abstractmethod
  77         def gen_hdr(self):
  78                 ''' Generate message specific header. '''
  79
  80         @abc.abstractmethod
  81         def parse_hdr(self, hdr):
  82                 ''' Parse message specific header. '''
  83
  84         @abc.abstractmethod
  85         def gen_burst(self):
  86                 ''' Generate message specific burst. '''
  87
  88         @abc.abstractmethod
  89         def parse_burst(self, burst):
  90                 ''' Parse message specific burst. '''
  91
  92         @abc.abstractmethod
  93         def rand_burst(self):
  94                 ''' Generate a random message specific burst. '''
  95
  96         def rand_fn(self):
  97                 ''' Generate a random frame number. '''
  98                 return random.randint(0, GSM_HYPERFRAME)
  99
 100         def rand_tn(self):
 101                 ''' Generate a random timeslot number. '''
 102                 return random.randint(0, 7)
 103
 104         def rand_hdr(self):
 105                 ''' Randomize the message header. '''
 106                 self.fn = self.rand_fn()
 107                 self.tn = self.rand_tn()
 108
 109         def desc_hdr(self):
 110                 ''' Generate human-readable header description. '''
 111
 112                 result = ""
 113
 114                 if self.ver > 0:
 115                         result += ("ver=%u " % self.ver)
 116
 117                 if self.fn is not None:
 118                         result += ("fn=%u " % self.fn)
 119
 120                 if self.tn is not None:
 121                         result += ("tn=%u " % self.tn)
 122
 123                 if self.burst is not None and len(self.burst) > 0:
 124                         result += ("bl=%u " % len(self.burst))
 125
 126                 return result
 127
 128         @staticmethod
 129         def usbit2sbit(bits: List[int]) -> List[int]:
 130                 ''' Convert unsigned soft-bits {254..0} to soft-bits {-127..127}. '''
 131                 return [-127 if (b == 0xff) else 127 - b for b in bits]
 132
 133         @staticmethod
 134         def sbit2usbit(bits: List[int]) -> List[int]:
 135                 ''' Convert soft-bits {-127..127} to unsigned soft-bits {254..0}. '''
 136                 return [127 - b for b in bits]
 137
 138         @staticmethod
 139         def sbit2ubit(bits: List[int]) -> List[int]:
 140                 ''' Convert soft-bits {-127..127} to bits {1..0}. '''
 141                 return [int(b < 0) for b in bits]
 142
 143         @staticmethod
 144         def ubit2sbit(bits: List[int]) -> List[int]:
 145                 ''' Convert bits {1..0} to soft-bits {-127..127}. '''
 146                 return [-127 if b else 127 for b in bits]
 147
 148         def validate(self):
 149                 ''' Validate the message fields (throws ValueError). '''
 150
 151                 if not self.ver in self.KNOWN_VERSIONS:
 152                         raise ValueError("Unknown TRXD header version %d" % self.ver)
 153
 154                 if self.fn is None:
 155                         raise ValueError("TDMA frame-number is not set")
 156
 157                 if self.fn < 0 or self.fn > GSM_HYPERFRAME:
 158                         raise ValueError("TDMA frame-number %d is out of range" % self.fn)
 159
 160                 if self.tn is None:
 161                         raise ValueError("TDMA time-slot is not set")
 162
 163                 if self.tn < 0 or self.tn > 7:
 164                         raise ValueError("TDMA time-slot %d is out of range" % self.tn)
 165
 166         def gen_msg(self, legacy = False):
 167                 ''' Generate a TRX DATA message. '''
 168
 169                 # Validate all the fields
 170                 self.validate()
 171
 172                 # Allocate an empty byte-array
 173                 buf = bytearray()
 174
 175                 # Put version (4 bits) and TDMA TN (3 bits)
 176                 buf.append((self.ver << 4) | (self.tn & 0x07))
 177
 178                 # Put TDMA FN (4 octets, BE)
 179                 buf += struct.pack(">L", self.fn)
 180
 181                 # Generate message specific header part
 182                 hdr = self.gen_hdr()
 183                 buf += hdr
 184
 185                 # Generate burst
 186                 if self.burst is not None:
 187                         buf += self.gen_burst()
 188
 189                 # This is a rudiment from (legacy) OpenBTS transceiver,
 190                 # some L1 implementations still expect two dummy bytes.
 191                 if legacy and self.ver == 0x00:
 192                         buf += bytearray(2)
 193
 194                 return buf
 195
 196         def parse_msg(self, msg):
 197                 ''' Parse a TRX DATA message. '''
 198
 199                 # Make sure we have at least common header
 200                 if len(msg) < self.CHDR_LEN:
 201                         raise ValueError("Message is to short: missing common header")
 202
 203                 # Parse the header version first
 204                 self.ver = (msg[0] >> 4)
 205                 if not self.ver in self.KNOWN_VERSIONS:
 206                         raise ValueError("Unknown TRXD header version %d" % self.ver)
 207
 208                 # Parse TDMA TN and FN
 209                 self.tn = (msg[0] & 0x07)
 210                 self.fn = struct.unpack(">L", msg[1:5])[0]
 211
 212                 # Make sure we have the whole header,
 213                 # including the version specific fields
 214                 if len(msg) < self.HDR_LEN:
 215                         raise ValueError("Message is to short: missing version specific header")
 216
 217                 # Specific message part
 218                 self.parse_hdr(msg)
 219
 220                 # Copy burst, skipping header
 221                 msg_burst = msg[self.HDR_LEN:]
 222                 if len(msg_burst) > 0:
 223                         self.parse_burst(msg_burst)
 224                 else:
 225                         self.burst = None
 226
 227 class TxMsg(Msg):
 228         ''' Tx (L1 -> TRX) message coding API. '''
 229
 230         # Constants
 231         PWR_MIN = 0x00
 232         PWR_MAX = 0xff
 233
 234         # Specific message fields
 235         pwr = None
 236
 237         @property
 238         def HDR_LEN(self):
 239                 ''' Calculate header length depending on its version. '''
 240
 241                 # Common header length
 242                 length = self.CHDR_LEN
 243
 244                 # Message specific header length
 245                 if self.ver in (0x00, 0x01):
 246                         length += 1 # PWR
 247                 else:
 248                         raise IndexError("Unhandled version %u" % self.ver)
 249
 250                 return length
 251
 252         def validate(self):
 253                 ''' Validate the message fields (throws ValueError). '''
 254
 255                 # Validate common fields
 256                 Msg.validate(self)
 257
 258                 if self.pwr is None:
 259                         raise ValueError("Tx Attenuation level is not set")
 260
 261                 if self.pwr < self.PWR_MIN or self.pwr > self.PWR_MAX:
 262                         raise ValueError("Tx Attenuation %d is out of range" % self.pwr)
 263
 264                 # FIXME: properly handle IDLE / NOPE indications
 265                 if self.burst is None:
 266                         raise ValueError("Tx burst bits are not set")
 267
 268                 # FIXME: properly handle IDLE / NOPE indications
 269                 if len(self.burst) not in (GMSK_BURST_LEN, EDGE_BURST_LEN):
 270                         raise ValueError("Tx burst has odd length %u" % len(self.burst))
 271
 272         def rand_pwr(self, min = None, max = None):
 273                 ''' Generate a random power level. '''
 274
 275                 if min is None:
 276                         min = self.PWR_MIN
 277
 278                 if max is None:
 279                         max = self.PWR_MAX
 280
 281                 return random.randint(min, max)
 282
 283         def rand_hdr(self):
 284                 ''' Randomize message specific header. '''
 285
 286                 Msg.rand_hdr(self)
 287                 self.pwr = self.rand_pwr()
 288
 289         def desc_hdr(self):
 290                 ''' Generate human-readable header description. '''
 291
 292                 # Describe the common part
 293                 result = Msg.desc_hdr(self)
 294
 295                 if self.pwr is not None:
 296                         result += ("pwr=%u " % self.pwr)
 297
 298                 # Strip useless whitespace and return
 299                 return result.strip()
 300
 301         def gen_hdr(self):
 302                 ''' Generate message specific header part. '''
 303
 304                 # Allocate an empty byte-array
 305                 buf = bytearray()
 306
 307                 # Put power
 308                 buf.append(self.pwr)
 309
 310                 return buf
 311
 312         def parse_hdr(self, hdr):
 313                 ''' Parse message specific header part. '''
 314
 315                 # Parse power level
 316                 self.pwr = hdr[5]
 317
 318         def gen_burst(self):
 319                 ''' Generate message specific burst. '''
 320
 321                 # Copy burst 'as is'
 322                 return bytearray(self.burst)
 323
 324         def parse_burst(self, burst):
 325                 ''' Parse message specific burst. '''
 326
 327                 length = len(burst)
 328
 329                 # Distinguish between GSM and EDGE
 330                 if length >= EDGE_BURST_LEN:
 331                         self.burst = list(burst[:EDGE_BURST_LEN])
 332                 else:
 333                         self.burst = list(burst[:GMSK_BURST_LEN])
 334
 335         def rand_burst(self, length = GMSK_BURST_LEN):
 336                 ''' Generate a random message specific burst. '''
 337                 self.burst = [random.randint(0, 1) for _ in range(length)]
 338
 339         def trans(self, ver = None):
 340                 ''' Transform this message into RxMsg. '''
 341
 342                 # Allocate a new message
 343                 msg = RxMsg(fn = self.fn, tn = self.tn,
 344                         ver = self.ver if ver is None else ver)
 345
 346                 # Convert burst bits
 347                 if self.burst is not None:
 348                         msg.burst = self.ubit2sbit(self.burst)
 349                 else:
 350                         msg.nope_ind = True
 351
 352                 return msg
 353
 354 class RxMsg(Msg):
 355         ''' Rx (TRX -> L1) message coding API. '''
 356
 357         # rxlev2dbm(0..63) gives us [-110..-47], plus -10 dbm for noise
 358         RSSI_MIN = -120
 359         RSSI_MAX = -47
 360
 361         # Min and max values of int16_t
 362         TOA256_MIN = -32768
 363         TOA256_MAX = 32767
 364
 365         # TSC (Training Sequence Code) range
 366         TSC_RANGE = range(0, 8)
 367
 368         # C/I range (in centiBels)
 369         CI_MIN = -1280
 370         CI_MAX = 1280
 371
 372         # IDLE frame / nope detection indicator
 373         NOPE_IND = (1 << 7)
 374
 375         # Specific message fields
 376         rssi = None
 377         toa256 = None
 378
 379         # Version 0x01 specific (default values)
 380         mod_type = Modulation.ModGMSK
 381         nope_ind = False
 382
 383         tsc_set = None
 384         tsc = None
 385         ci = None
 386
 387         @property
 388         def HDR_LEN(self):
 389                 ''' Calculate header length depending on its version. '''
 390
 391                 # Common header length
 392                 length = self.CHDR_LEN
 393
 394                 # Message specific header length
 395                 if self.ver == 0x00:
 396                         # RSSI + ToA
 397                         length += 1 + 2
 398                 elif self.ver == 0x01:
 399                         # RSSI + ToA + TS + C/I
 400                         length += 1 + 2 + 1 + 2
 401                 else:
 402                         raise IndexError("Unhandled version %u" % self.ver)
 403
 404                 return length
 405
 406         def _validate_burst_v0(self):
 407                 # Burst is mandatory
 408                 if self.burst is None:
 409                         raise ValueError("Rx burst bits are not set")
 410
 411                 # ... and can be either of GSM (GMSK) or EDGE (8-PSK)
 412                 if len(self.burst) not in (GMSK_BURST_LEN, EDGE_BURST_LEN):
 413                         raise ValueError("Rx burst has odd length %u" % len(self.burst))
 414
 415         def _validate_burst_v1(self):
 416                 # Burst is omitted in case of an IDLE / NOPE indication
 417                 if self.nope_ind and self.burst is None:
 418                         return
 419
 420                 if self.nope_ind and self.burst is not None:
 421                         raise ValueError("NOPE.ind comes with burst?!?")
 422                 if self.burst is None:
 423                         raise ValueError("Rx burst bits are not set")
 424
 425                 # Burst length depends on modulation type
 426                 if len(self.burst) != self.mod_type.bl:
 427                         raise ValueError("Rx burst has odd length %u" % len(self.burst))
 428
 429         def validate_burst(self):
 430                 ''' Validate the burst (throws ValueError). '''
 431
 432                 if self.ver == 0x00:
 433                         self._validate_burst_v0()
 434                 elif self.ver >= 0x01:
 435                         self._validate_burst_v1()
 436
 437         def validate(self):
 438                 ''' Validate the message header fields (throws ValueError). '''
 439
 440                 # Validate common fields
 441                 Msg.validate(self)
 442
 443                 if self.rssi is None:
 444                         raise ValueError("RSSI is not set")
 445
 446                 if self.rssi < self.RSSI_MIN or self.rssi > self.RSSI_MAX:
 447                         raise ValueError("RSSI %d is out of range" % self.rssi)
 448
 449                 if self.toa256 is None:
 450                         raise ValueError("ToA256 is not set")
 451
 452                 if self.toa256 < self.TOA256_MIN or self.toa256 > self.TOA256_MAX:
 453                         raise ValueError("ToA256 %d is out of range" % self.toa256)
 454
 455                 # Version specific parameters (omited for NOPE.ind)
 456                 if self.ver >= 0x01 and not self.nope_ind:
 457                         if type(self.mod_type) is not Modulation:
 458                                 raise ValueError("Unknown Rx modulation type")
 459
 460                         if self.tsc_set is None:
 461                                 raise ValueError("TSC set is not set")
 462
 463                         if self.mod_type is Modulation.ModGMSK:
 464                                 if self.tsc_set not in range(0, 4):
 465                                         raise ValueError("TSC set %d is out of range" % self.tsc_set)
 466                         else:
 467                                 if self.tsc_set not in range(0, 2):
 468                                         raise ValueError("TSC set %d is out of range" % self.tsc_set)
 469
 470                         if self.tsc is None:
 471                                 raise ValueError("TSC is not set")
 472
 473                         if self.tsc not in self.TSC_RANGE:
 474                                 raise ValueError("TSC %d is out of range" % self.tsc)
 475
 476                 # Version specific parameters (also present in NOPE.ind)
 477                 if self.ver >= 0x01:
 478                         if self.ci is None:
 479                                 raise ValueError("C/I is not set")
 480
 481                         if self.ci < self.CI_MIN or self.ci > self.CI_MAX:
 482                                 raise ValueError("C/I %d is out of range" % self.ci)
 483
 484                 self.validate_burst()
 485
 486         def rand_rssi(self, min = None, max = None):
 487                 ''' Generate a random RSSI value. '''
 488
 489                 if min is None:
 490                         min = self.RSSI_MIN
 491
 492                 if max is None:
 493                         max = self.RSSI_MAX
 494
 495                 return random.randint(min, max)
 496
 497         def rand_toa256(self, min = None, max = None):
 498                 ''' Generate a random ToA (Time of Arrival) value. '''
 499
 500                 if min is None:
 501                         min = self.TOA256_MIN
 502
 503                 if max is None:
 504                         max = self.TOA256_MAX
 505
 506                 return random.randint(min, max)
 507
 508         def rand_hdr(self):
 509                 ''' Randomize message specific header. '''
 510
 511                 Msg.rand_hdr(self)
 512                 self.rssi = self.rand_rssi()
 513                 self.toa256 = self.rand_toa256()
 514
 515                 if self.ver >= 0x01:
 516                         self.mod_type = random.choice(list(Modulation))
 517                         if self.mod_type is Modulation.ModGMSK:
 518                                 self.tsc_set = random.randint(0, 3)
 519                         else:
 520                                 self.tsc_set = random.randint(0, 1)
 521                         self.tsc = random.choice(self.TSC_RANGE)
 522
 523                         # C/I: Carrier-to-Interference ratio
 524                         self.ci = random.randint(self.CI_MIN, self.CI_MAX)
 525
 526         def desc_hdr(self):
 527                 ''' Generate human-readable header description. '''
 528
 529                 # Describe the common part
 530                 result = Msg.desc_hdr(self)
 531
 532                 if self.rssi is not None:
 533                         result += ("rssi=%d " % self.rssi)
 534
 535                 if self.toa256 is not None:
 536                         result += ("toa256=%d " % self.toa256)
 537
 538                 if self.ver >= 0x01:
 539                         if not self.nope_ind:
 540                                 if self.mod_type is not None:
 541                                         result += ("%s " % self.mod_type)
 542                                 if self.tsc_set is not None:
 543                                         result += ("set=%u " % self.tsc_set)
 544                                 if self.tsc is not None:
 545                                         result += ("tsc=%u " % self.tsc)
 546                                 if self.ci is not None:
 547                                         result += ("C/I=%d cB " % self.ci)
 548                         else:
 549                                 result += "(IDLE / NOPE IND) "
 550
 551                 # Strip useless whitespace and return
 552                 return result.strip()
 553
 554         def gen_mts(self):
 555                 ''' Encode Modulation and Training Sequence info. '''
 556
 557                 # IDLE / nope indication has no MTS info
 558                 if self.nope_ind:
 559                         return self.NOPE_IND
 560
 561                 # TSC: . . . . . X X X
 562                 mts = self.tsc & 0b111
 563
 564                 # MTS: . X X X X . . .
 565                 mts |= self.mod_type.coding << 3
 566                 mts |= self.tsc_set << 3
 567
 568                 return mts
 569
 570         def parse_mts(self, mts):
 571                 ''' Parse Modulation and Training Sequence info. '''
 572
 573                 # IDLE / nope indication has no MTS info
 574                 self.nope_ind = (mts & self.NOPE_IND) > 0
 575                 if self.nope_ind:
 576                         self.mod_type = None
 577                         self.tsc_set = None
 578                         self.tsc = None
 579                         return
 580
 581                 # TSC: . . . . . X X X
 582                 self.tsc = mts & 0b111
 583
 584                 # MTS: . X X X X . . .
 585                 mts = (mts >> 3) & 0b1111
 586                 if (mts & 0b1100) > 0:
 587                         # Mask: . . . . M M M S
 588                         self.mod_type = Modulation.pick(mts & 0b1110)
 589                         self.tsc_set = mts & 0b1
 590                 else:
 591                         # GMSK: . . . . 0 0 S S
 592                         self.mod_type = Modulation.ModGMSK
 593                         self.tsc_set = mts & 0b11
 594
 595         def gen_hdr(self):
 596                 ''' Generate message specific header part. '''
 597
 598                 # Allocate an empty byte-array
 599                 buf = bytearray()
 600
 601                 # Put RSSI
 602                 buf.append(-self.rssi)
 603
 604                 # Encode ToA (Time of Arrival)
 605                 # Big endian, 2 bytes (int32_t)
 606                 buf += struct.pack(">h", self.toa256)
 607
 608                 if self.ver >= 0x01:
 609                         # Modulation and Training Sequence info
 610                         mts = self.gen_mts()
 611                         buf.append(mts)
 612
 613                         # C/I: Carrier-to-Interference ratio (in centiBels)
 614                         buf += struct.pack(">h", self.ci)
 615
 616                 return buf
 617
 618         def parse_hdr(self, hdr):
 619                 ''' Parse message specific header part. '''
 620
 621                 # Parse RSSI
 622                 self.rssi = -(hdr[5])
 623
 624                 # Parse ToA (Time of Arrival)
 625                 self.toa256 = struct.unpack(">h", hdr[6:8])[0]
 626
 627                 if self.ver >= 0x01:
 628                         # Modulation and Training Sequence info
 629                         self.parse_mts(hdr[8])
 630
 631                         # C/I: Carrier-to-Interference ratio (in centiBels)
 632                         self.ci = struct.unpack(">h", hdr[9:11])[0]
 633
 634         def gen_burst(self):
 635                 ''' Generate message specific burst. '''
 636
 637                 # Convert soft-bits to unsigned soft-bits
 638                 burst_usbits = self.sbit2usbit(self.burst)
 639
 640                 # Encode to bytes
 641                 return bytearray(burst_usbits)
 642
 643         def _parse_burst_v0(self, burst):
 644                 ''' Parse message specific burst for header version 0. '''
 645
 646                 bl = len(burst)
 647
 648                 # We need to guess modulation by the length of burst
 649                 self.mod_type = Modulation.pick_by_bl(bl)
 650                 if self.mod_type is None:
 651                         # Some old transceivers append two dummy bytes
 652                         self.mod_type = Modulation.pick_by_bl(bl - 2)
 653
 654                 if self.mod_type is None:
 655                         raise ValueError("Odd burst length %u" % bl)
 656
 657                 return burst[:self.mod_type.bl]
 658
 659         def parse_burst(self, burst):
 660                 ''' Parse message specific burst. '''
 661
 662                 burst = list(burst)
 663
 664                 if self.ver == 0x00:
 665                         burst = self._parse_burst_v0(burst)
 666
 667                 # Convert unsigned soft-bits to soft-bits
 668                 self.burst = self.usbit2sbit(burst)
 669
 670         def rand_burst(self, length = None):
 671                 ''' Generate a random message specific burst. '''
 672
 673                 if length is None:
 674                         length = self.mod_type.bl
 675
 676                 self.burst = [random.randint(-127, 127) for _ in range(length)]
 677
 678         def trans(self, ver = None):
 679                 ''' Transform this message to TxMsg. '''
 680
 681                 # Allocate a new message
 682                 msg = TxMsg(fn = self.fn, tn = self.tn,
 683                         ver = self.ver if ver is None else ver)
 684
 685                 # Convert burst bits
 686                 if self.burst is not None:
 687                         msg.burst = self.sbit2ubit(self.burst)
 688
 689                 return msg