517 lines
17 KiB
Python
517 lines
17 KiB
Python
|
#!/usr/bin/env python
|
||
|
|
|
||
|
|
from builtins import bytes
|
||
|
|
from functools import total_ordering
|
||
|
|
import sys, math
|
||
|
|
from math import ceil, log
|
||
|
|
from collections import defaultdict, OrderedDict
|
||
|
|
import attr
|
||
|
|
|
||
|
|
# must match <linux/nl80211.h> enum nl80211_reg_rule_flags
|
||
|
|
|
||
|
|
flag_definitions = {
|
||
|
|
'NO-OFDM': 1<<0,
|
||
|
|
'NO-CCK': 1<<1,
|
||
|
|
'NO-INDOOR': 1<<2,
|
||
|
|
'NO-OUTDOOR': 1<<3,
|
||
|
|
'DFS': 1<<4,
|
||
|
|
'PTP-ONLY': 1<<5,
|
||
|
|
'PTMP-ONLY': 1<<6,
|
||
|
|
'NO-IR': 1<<7,
|
||
|
|
# hole at bit 8
|
||
|
|
# hole at bit 9. FIXME: Where is NO-HT40 defined?
|
||
|
|
'NO-HT40': 1<<10,
|
||
|
|
'AUTO-BW': 1<<11,
|
||
|
|
}
|
||
|
|
|
||
|
|
dfs_regions = {
|
||
|
|
'DFS-FCC': 1,
|
||
|
|
'DFS-ETSI': 2,
|
||
|
|
'DFS-JP': 3,
|
||
|
|
}
|
||
|
|
|
||
|
|
@total_ordering
|
||
|
|
|
||
|
|
@attr.s(frozen=True, cmp=False)
|
||
|
|
class WmmRule(object):
|
||
|
|
vo_c = attr.ib()
|
||
|
|
vi_c = attr.ib()
|
||
|
|
be_c = attr.ib()
|
||
|
|
bk_c = attr.ib()
|
||
|
|
vo_ap = attr.ib()
|
||
|
|
vi_ap = attr.ib()
|
||
|
|
be_ap = attr.ib()
|
||
|
|
bk_ap = attr.ib()
|
||
|
|
|
||
|
|
def _as_tuple(self):
|
||
|
|
return (self.vo_c, self.vi_c, self.be_c, self.bk_c,
|
||
|
|
self.vo_ap, self.vi_ap, self.be_ap, self.bk_ap)
|
||
|
|
|
||
|
|
def __eq__(self, other):
|
||
|
|
if other is None:
|
||
|
|
return False
|
||
|
|
return (self._as_tuple() == other._as_tuple())
|
||
|
|
|
||
|
|
def __ne__(self, other):
|
||
|
|
return not (self == other)
|
||
|
|
|
||
|
|
def __lt__(self, other):
|
||
|
|
if other is None:
|
||
|
|
return False
|
||
|
|
return (self._as_tuple() < other._as_tuple())
|
||
|
|
|
||
|
|
def __hash__(self):
|
||
|
|
return hash(self._as_tuple())
|
||
|
|
|
||
|
|
class FreqBand(object):
|
||
|
|
def __init__(self, start, end, bw, comments=None):
|
||
|
|
self.start = start
|
||
|
|
self.end = end
|
||
|
|
self.maxbw = bw
|
||
|
|
self.comments = comments or []
|
||
|
|
|
||
|
|
def _as_tuple(self):
|
||
|
|
return (self.start, self.end, self.maxbw)
|
||
|
|
|
||
|
|
def __eq__(self, other):
|
||
|
|
return (self._as_tuple() == other._as_tuple())
|
||
|
|
|
||
|
|
def __ne__(self, other):
|
||
|
|
return not (self == other)
|
||
|
|
|
||
|
|
def __lt__(self, other):
|
||
|
|
return (self._as_tuple() < other._as_tuple())
|
||
|
|
|
||
|
|
def __hash__(self):
|
||
|
|
return hash(self._as_tuple())
|
||
|
|
|
||
|
|
def __str__(self):
|
||
|
|
return '<FreqBand %.3f - %.3f @ %.3f>' % (
|
||
|
|
self.start, self.end, self.maxbw)
|
||
|
|
|
||
|
|
@total_ordering
|
||
|
|
class PowerRestriction(object):
|
||
|
|
def __init__(self, max_ant_gain, max_eirp, comments = None):
|
||
|
|
self.max_ant_gain = max_ant_gain
|
||
|
|
self.max_eirp = max_eirp
|
||
|
|
self.comments = comments or []
|
||
|
|
|
||
|
|
def _as_tuple(self):
|
||
|
|
return (self.max_ant_gain, self.max_eirp)
|
||
|
|
|
||
|
|
def __eq__(self, other):
|
||
|
|
return (self._as_tuple() == other._as_tuple())
|
||
|
|
|
||
|
|
def __ne__(self, other):
|
||
|
|
return not (self == other)
|
||
|
|
|
||
|
|
def __lt__(self, other):
|
||
|
|
return (self._as_tuple() < other._as_tuple())
|
||
|
|
|
||
|
|
def __hash__(self):
|
||
|
|
return hash(self._as_tuple())
|
||
|
|
|
||
|
|
def __str__(self):
|
||
|
|
return '<PowerRestriction ...>'
|
||
|
|
|
||
|
|
class DFSRegionError(Exception):
|
||
|
|
def __init__(self, dfs_region):
|
||
|
|
self.dfs_region = dfs_region
|
||
|
|
|
||
|
|
class FlagError(Exception):
|
||
|
|
def __init__(self, flag):
|
||
|
|
self.flag = flag
|
||
|
|
|
||
|
|
@total_ordering
|
||
|
|
class Permission(object):
|
||
|
|
def __init__(self, freqband, power, flags, wmmrule):
|
||
|
|
assert isinstance(freqband, FreqBand)
|
||
|
|
assert isinstance(power, PowerRestriction)
|
||
|
|
assert isinstance(wmmrule, WmmRule) or wmmrule is None
|
||
|
|
self.freqband = freqband
|
||
|
|
self.power = power
|
||
|
|
self.wmmrule = wmmrule
|
||
|
|
self.flags = 0
|
||
|
|
for flag in flags:
|
||
|
|
if not flag in flag_definitions:
|
||
|
|
raise FlagError(flag)
|
||
|
|
self.flags |= flag_definitions[flag]
|
||
|
|
self.textflags = flags
|
||
|
|
|
||
|
|
def _as_tuple(self):
|
||
|
|
return (self.freqband, self.power, self.flags, self.wmmrule)
|
||
|
|
|
||
|
|
def __eq__(self, other):
|
||
|
|
return (self._as_tuple() == other._as_tuple())
|
||
|
|
|
||
|
|
def __ne__(self, other):
|
||
|
|
return not (self == other)
|
||
|
|
|
||
|
|
def __lt__(self, other):
|
||
|
|
return (self._as_tuple() < other._as_tuple())
|
||
|
|
|
||
|
|
def __hash__(self):
|
||
|
|
return hash(self._as_tuple())
|
||
|
|
|
||
|
|
def __str__(self):
|
||
|
|
return str(self.freqband) + str(self.power) + str(self.wmmrule)
|
||
|
|
|
||
|
|
class Country(object):
|
||
|
|
def __init__(self, dfs_region, permissions=None, comments=None):
|
||
|
|
self._permissions = permissions or []
|
||
|
|
self.comments = comments or []
|
||
|
|
self.dfs_region = 0
|
||
|
|
|
||
|
|
if dfs_region:
|
||
|
|
if not dfs_region in dfs_regions:
|
||
|
|
raise DFSRegionError(dfs_region)
|
||
|
|
self.dfs_region = dfs_regions[dfs_region]
|
||
|
|
|
||
|
|
def add(self, perm):
|
||
|
|
assert isinstance(perm, Permission)
|
||
|
|
self._permissions.append(perm)
|
||
|
|
self._permissions.sort()
|
||
|
|
|
||
|
|
def __contains__(self, perm):
|
||
|
|
assert isinstance(perm, Permission)
|
||
|
|
return perm in self._permissions
|
||
|
|
|
||
|
|
def __str__(self):
|
||
|
|
r = ['(%s, %s)' % (str(b), str(p)) for b, p in self._permissions]
|
||
|
|
return '<Country (%s)>' % (', '.join(r))
|
||
|
|
|
||
|
|
def _get_permissions_tuple(self):
|
||
|
|
return tuple(self._permissions)
|
||
|
|
permissions = property(_get_permissions_tuple)
|
||
|
|
|
||
|
|
class SyntaxError(Exception):
|
||
|
|
pass
|
||
|
|
|
||
|
|
class DBParser(object):
|
||
|
|
def __init__(self, warn=None):
|
||
|
|
self._warn_callout = warn or sys.stderr.write
|
||
|
|
|
||
|
|
def _syntax_error(self, txt=None):
|
||
|
|
txt = txt and ' (%s)' % txt or ''
|
||
|
|
raise SyntaxError("Syntax error in line %d%s" % (self._lineno, txt))
|
||
|
|
|
||
|
|
def _warn(self, txt):
|
||
|
|
self._warn_callout("Warning (line %d): %s\n" % (self._lineno, txt))
|
||
|
|
|
||
|
|
def _parse_band_def(self, bname, banddef, dupwarn=True):
|
||
|
|
try:
|
||
|
|
freqs, bw = banddef.split('@')
|
||
|
|
bw = float(bw)
|
||
|
|
except ValueError:
|
||
|
|
bw = 20.0
|
||
|
|
|
||
|
|
try:
|
||
|
|
start, end = freqs.split('-')
|
||
|
|
start = float(start)
|
||
|
|
end = float(end)
|
||
|
|
# The kernel will reject these, so might as well reject this
|
||
|
|
# upon building it.
|
||
|
|
if start <= 0:
|
||
|
|
self._syntax_error("Invalid start freq (%d)" % start)
|
||
|
|
if end <= 0:
|
||
|
|
self._syntax_error("Invalid end freq (%d)" % end)
|
||
|
|
if start > end:
|
||
|
|
self._syntax_error("Inverted freq range (%d - %d)" % (start, end))
|
||
|
|
if start == end:
|
||
|
|
self._syntax_error("Start and end freqs are equal (%d)" % start)
|
||
|
|
except ValueError:
|
||
|
|
self._syntax_error("band must have frequency range")
|
||
|
|
|
||
|
|
b = FreqBand(start, end, bw, comments=self._comments)
|
||
|
|
self._comments = []
|
||
|
|
self._banddup[bname] = bname
|
||
|
|
if b in self._bandrev:
|
||
|
|
if dupwarn:
|
||
|
|
self._warn('Duplicate band definition ("%s" and "%s")' % (
|
||
|
|
bname, self._bandrev[b]))
|
||
|
|
self._banddup[bname] = self._bandrev[b]
|
||
|
|
self._bands[bname] = b
|
||
|
|
self._bandrev[b] = bname
|
||
|
|
self._bandline[bname] = self._lineno
|
||
|
|
|
||
|
|
def _parse_band(self, line):
|
||
|
|
try:
|
||
|
|
bname, line = line.split(':', 1)
|
||
|
|
if not bname:
|
||
|
|
self._syntax_error("'band' keyword must be followed by name")
|
||
|
|
except ValueError:
|
||
|
|
self._syntax_error("band name must be followed by colon")
|
||
|
|
|
||
|
|
if bname in flag_definitions:
|
||
|
|
self._syntax_error("Invalid band name")
|
||
|
|
|
||
|
|
self._parse_band_def(bname, line)
|
||
|
|
|
||
|
|
def _parse_power(self, line):
|
||
|
|
try:
|
||
|
|
pname, line = line.split(':', 1)
|
||
|
|
if not pname:
|
||
|
|
self._syntax_error("'power' keyword must be followed by name")
|
||
|
|
except ValueError:
|
||
|
|
self._syntax_error("power name must be followed by colon")
|
||
|
|
|
||
|
|
if pname in flag_definitions:
|
||
|
|
self._syntax_error("Invalid power name")
|
||
|
|
|
||
|
|
self._parse_power_def(pname, line)
|
||
|
|
|
||
|
|
def _parse_power_def(self, pname, line, dupwarn=True):
|
||
|
|
try:
|
||
|
|
max_eirp = line
|
||
|
|
if max_eirp == 'N/A':
|
||
|
|
max_eirp = '0'
|
||
|
|
max_ant_gain = float(0)
|
||
|
|
def conv_pwr(pwr):
|
||
|
|
if pwr.endswith('mW'):
|
||
|
|
pwr = float(pwr[:-2])
|
||
|
|
return 10.0 * math.log10(pwr)
|
||
|
|
else:
|
||
|
|
return float(pwr)
|
||
|
|
max_eirp = conv_pwr(max_eirp)
|
||
|
|
except ValueError:
|
||
|
|
self._syntax_error("invalid power data")
|
||
|
|
|
||
|
|
p = PowerRestriction(max_ant_gain, max_eirp,
|
||
|
|
comments=self._comments)
|
||
|
|
self._comments = []
|
||
|
|
self._powerdup[pname] = pname
|
||
|
|
if p in self._powerrev:
|
||
|
|
if dupwarn:
|
||
|
|
self._warn('Duplicate power definition ("%s" and "%s")' % (
|
||
|
|
pname, self._powerrev[p]))
|
||
|
|
self._powerdup[pname] = self._powerrev[p]
|
||
|
|
self._power[pname] = p
|
||
|
|
self._powerrev[p] = pname
|
||
|
|
self._powerline[pname] = self._lineno
|
||
|
|
|
||
|
|
def _parse_wmmrule(self, line):
|
||
|
|
regions = line[:-1].strip()
|
||
|
|
if not regions:
|
||
|
|
self._syntax_error("'wmmrule' keyword must be followed by region")
|
||
|
|
|
||
|
|
regions = regions.split(',')
|
||
|
|
|
||
|
|
self._current_regions = {}
|
||
|
|
for region in regions:
|
||
|
|
if region in self._wmm_rules:
|
||
|
|
self._warn("region %s was added already to wmm rules" % region)
|
||
|
|
self._current_regions[region] = 1
|
||
|
|
self._comments = []
|
||
|
|
|
||
|
|
def _validate_input(self, cw_min, cw_max, aifsn, cot):
|
||
|
|
if cw_min < 1:
|
||
|
|
self._syntax_error("Invalid cw_min value (%d)" % cw_min)
|
||
|
|
if cw_max < 1:
|
||
|
|
self._syntax_error("Invalid cw_max value (%d)" % cw_max)
|
||
|
|
if cw_min > cw_max:
|
||
|
|
self._syntax_error("Inverted contention window (%d - %d)" %
|
||
|
|
(cw_min, cw_max))
|
||
|
|
if not (bin(cw_min + 1).count('1') == 1 and cw_min < 2**15):
|
||
|
|
self._syntax_error("Invalid cw_min value should be power of 2 - 1 (%d)"
|
||
|
|
% cw_min)
|
||
|
|
if not (bin(cw_max + 1).count('1') == 1 and cw_max < 2**15):
|
||
|
|
self._syntax_error("Invalid cw_max value should be power of 2 - 1 (%d)"
|
||
|
|
% cw_max)
|
||
|
|
if aifsn < 1:
|
||
|
|
self._syntax_error("Invalid aifsn value (%d)" % aifsn)
|
||
|
|
if cot < 0:
|
||
|
|
self._syntax_error("Invalid cot value (%d)" % cot)
|
||
|
|
|
||
|
|
|
||
|
|
def _validate_size(self, var, bytcnt):
|
||
|
|
return bytcnt < ceil(len(bin(var)[2:]) / 8.0)
|
||
|
|
|
||
|
|
def _parse_wmmrule_item(self, line):
|
||
|
|
bytcnt = (2.0, 2.0, 1.0, 2.0)
|
||
|
|
try:
|
||
|
|
ac, cval = line.split(':')
|
||
|
|
if not ac:
|
||
|
|
self._syntax_error("wmm item must have ac prefix")
|
||
|
|
except ValueError:
|
||
|
|
self._syntax_error("access category must be followed by colon")
|
||
|
|
p = tuple([int(v.split('=', 1)[1]) for v in cval.split(',')])
|
||
|
|
self._validate_input(*p)
|
||
|
|
for v, b in zip(p, bytcnt):
|
||
|
|
if self._validate_size(v, b):
|
||
|
|
self._syntax_error("unexpected input size expect %d got %d"
|
||
|
|
% (b, v))
|
||
|
|
|
||
|
|
for r in self._current_regions:
|
||
|
|
self._wmm_rules[r][ac] = p
|
||
|
|
|
||
|
|
def _parse_country(self, line):
|
||
|
|
try:
|
||
|
|
cname, cvals= line.split(':', 1)
|
||
|
|
dfs_region = cvals.strip()
|
||
|
|
if not cname:
|
||
|
|
self._syntax_error("'country' keyword must be followed by name")
|
||
|
|
except ValueError:
|
||
|
|
self._syntax_error("country name must be followed by colon")
|
||
|
|
|
||
|
|
cnames = cname.split(',')
|
||
|
|
|
||
|
|
self._current_countries = {}
|
||
|
|
for cname in cnames:
|
||
|
|
if len(cname) != 2:
|
||
|
|
self._warn("country '%s' not alpha2" % cname)
|
||
|
|
cname = bytes(cname, 'ascii')
|
||
|
|
if not cname in self._countries:
|
||
|
|
self._countries[cname] = Country(dfs_region, comments=self._comments)
|
||
|
|
self._current_countries[cname] = self._countries[cname]
|
||
|
|
self._comments = []
|
||
|
|
|
||
|
|
def _parse_country_item(self, line):
|
||
|
|
if line[0] == '(':
|
||
|
|
try:
|
||
|
|
band, line = line[1:].split('),', 1)
|
||
|
|
bname = 'UNNAMED %d' % self._lineno
|
||
|
|
self._parse_band_def(bname, band, dupwarn=False)
|
||
|
|
except:
|
||
|
|
self._syntax_error("Badly parenthesised band definition")
|
||
|
|
else:
|
||
|
|
try:
|
||
|
|
bname, line = line.split(',', 1)
|
||
|
|
if not bname:
|
||
|
|
self._syntax_error("country definition must have band")
|
||
|
|
if not line:
|
||
|
|
self._syntax_error("country definition must have power")
|
||
|
|
except ValueError:
|
||
|
|
self._syntax_error("country definition must have band and power")
|
||
|
|
|
||
|
|
if line[0] == '(':
|
||
|
|
items = line.split('),', 1)
|
||
|
|
if len(items) == 1:
|
||
|
|
pname = items[0]
|
||
|
|
line = ''
|
||
|
|
if not pname[-1] == ')':
|
||
|
|
self._syntax_error("Badly parenthesised power definition")
|
||
|
|
pname = pname[:-1]
|
||
|
|
flags = []
|
||
|
|
else:
|
||
|
|
pname = items[0]
|
||
|
|
flags = items[1].split(',')
|
||
|
|
power = pname[1:]
|
||
|
|
pname = 'UNNAMED %d' % self._lineno
|
||
|
|
self._parse_power_def(pname, power, dupwarn=False)
|
||
|
|
else:
|
||
|
|
line = line.split(',')
|
||
|
|
pname = line[0]
|
||
|
|
flags = line[1:]
|
||
|
|
w = None
|
||
|
|
if flags and 'wmmrule' in flags[-1]:
|
||
|
|
try:
|
||
|
|
region = flags.pop().split('=', 1)[1]
|
||
|
|
if region not in self._wmm_rules.keys():
|
||
|
|
self._syntax_error("No wmm rule for %s" % region)
|
||
|
|
except IndexError:
|
||
|
|
self._syntax_error("flags is empty list or no region was found")
|
||
|
|
w = WmmRule(*self._wmm_rules[region].values())
|
||
|
|
|
||
|
|
if not bname in self._bands:
|
||
|
|
self._syntax_error("band does not exist")
|
||
|
|
if not pname in self._power:
|
||
|
|
self._syntax_error("power does not exist")
|
||
|
|
self._bands_used[bname] = True
|
||
|
|
self._power_used[pname] = True
|
||
|
|
# de-duplicate so binary database is more compact
|
||
|
|
bname = self._banddup[bname]
|
||
|
|
pname = self._powerdup[pname]
|
||
|
|
b = self._bands[bname]
|
||
|
|
p = self._power[pname]
|
||
|
|
try:
|
||
|
|
perm = Permission(b, p, flags, w)
|
||
|
|
except FlagError as e:
|
||
|
|
self._syntax_error("Invalid flag '%s'" % e.flag)
|
||
|
|
for cname, c in self._current_countries.items():
|
||
|
|
if perm in c:
|
||
|
|
self._warn('Rule "%s, %s" added to "%s" twice' % (
|
||
|
|
bname, pname, cname))
|
||
|
|
else:
|
||
|
|
c.add(perm)
|
||
|
|
|
||
|
|
def parse(self, f):
|
||
|
|
self._current_countries = None
|
||
|
|
self._current_regions = None
|
||
|
|
self._bands = {}
|
||
|
|
self._power = {}
|
||
|
|
self._countries = {}
|
||
|
|
self._bands_used = {}
|
||
|
|
self._power_used = {}
|
||
|
|
self._bandrev = {}
|
||
|
|
self._powerrev = {}
|
||
|
|
self._banddup = {}
|
||
|
|
self._powerdup = {}
|
||
|
|
self._bandline = {}
|
||
|
|
self._powerline = {}
|
||
|
|
self._wmm_rules = defaultdict(lambda: OrderedDict())
|
||
|
|
|
||
|
|
self._comments = []
|
||
|
|
|
||
|
|
self._lineno = 0
|
||
|
|
for line in f:
|
||
|
|
self._lineno += 1
|
||
|
|
line = line.strip()
|
||
|
|
if line[0:1] == '#':
|
||
|
|
self._comments.append(line[1:].strip())
|
||
|
|
line = line.replace(' ', '').replace('\t', '')
|
||
|
|
if not line:
|
||
|
|
self._current_regions = None
|
||
|
|
self._comments = []
|
||
|
|
line = line.split('#')[0]
|
||
|
|
if not line:
|
||
|
|
continue
|
||
|
|
if line[0:4] == 'band':
|
||
|
|
self._parse_band(line[4:])
|
||
|
|
self._current_countries = None
|
||
|
|
self._current_regions = None
|
||
|
|
self._comments = []
|
||
|
|
elif line[0:5] == 'power':
|
||
|
|
self._parse_power(line[5:])
|
||
|
|
self._current_countries = None
|
||
|
|
self._current_regions = None
|
||
|
|
self._comments = []
|
||
|
|
elif line[0:7] == 'country':
|
||
|
|
self._parse_country(line[7:])
|
||
|
|
self._comments = []
|
||
|
|
self._current_regions = None
|
||
|
|
elif self._current_countries is not None:
|
||
|
|
self._current_regions = None
|
||
|
|
self._parse_country_item(line)
|
||
|
|
self._comments = []
|
||
|
|
elif line[0:7] == 'wmmrule':
|
||
|
|
self._parse_wmmrule(line[7:])
|
||
|
|
self._current_countries = None
|
||
|
|
self._comments = []
|
||
|
|
elif self._current_regions is not None:
|
||
|
|
self._parse_wmmrule_item(line)
|
||
|
|
self._current_countries = None
|
||
|
|
self._comments = []
|
||
|
|
else:
|
||
|
|
self._syntax_error("Expected band, power or country definition")
|
||
|
|
|
||
|
|
countries = self._countries
|
||
|
|
bands = {}
|
||
|
|
for k, v in self._bands.items():
|
||
|
|
if k in self._bands_used:
|
||
|
|
bands[self._banddup[k]] = v
|
||
|
|
continue
|
||
|
|
# we de-duplicated, but don't warn again about the dupes
|
||
|
|
if self._banddup[k] == k:
|
||
|
|
self._lineno = self._bandline[k]
|
||
|
|
self._warn('Unused band definition "%s"' % k)
|
||
|
|
power = {}
|
||
|
|
for k, v in self._power.items():
|
||
|
|
if k in self._power_used:
|
||
|
|
power[self._powerdup[k]] = v
|
||
|
|
continue
|
||
|
|
# we de-duplicated, but don't warn again about the dupes
|
||
|
|
if self._powerdup[k] == k:
|
||
|
|
self._lineno = self._powerline[k]
|
||
|
|
self._warn('Unused power definition "%s"' % k)
|
||
|
|
return countries
|