ShaYmez
/
dmr_utils
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

UPDATE FOR JSON DATABASE INSTEAD OF CSV

This commit is contained in:
Cort Buffington 2019-03-01 08:32:57 -06:00
parent 605698ca09
commit 22cc9d1920
2 changed files with 95 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

126
dmr_utils/utils.py
View File

@ -1,7 +1,7 @@
#!/usr/bin/env python
#
###############################################################################
# Copyright (C) 2016 Cortney T. Buffington, N0MJS <n0mjs@me.com>
# Copyright (C) 2016-2019 Cortney T. Buffington, N0MJS <n0mjs@me.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@ -20,16 +20,15 @@
from __future__ import print_function
from json import load as jload
from os.path import isfile, getmtime
from time import time
from urllib import URLopener
from csv import reader as csv_reader
from csv import DictReader as csv_dict_reader
from binascii import b2a_hex as ahex
# Does anybody read this stuff? There's a PEP somewhere that says I should do this.
__author__ = 'Cortney T. Buffington, N0MJS'
__copyright__ = 'Copyright (c) 2016-2017 Cortney T. Buffington, N0MJS and the K0USY Group'
__copyright__ = 'Copyright (c) 2016-2019 Cortney T. Buffington, N0MJS and the K0USY Group'
__credits__ = 'Colin Durbridge, G4EML, Steve Zingman, N4IRS; Mike Zingman'
__license__ = 'GNU GPLv3'
__maintainer__ = 'Cort Buffington, N0MJS'
@ -97,45 +96,60 @@ def try_download(_path, _file, _url, _stale,):
url.close()
return result
# LEGACY VERSION - MAKES A SIMPLE {INTEGER ID: 'CALLSIGN'} DICTIONARY
# SHORT VERSION - MAKES A SIMPLE {INTEGER ID: 'CALLSIGN'} DICTIONARY
def mk_id_dict(_path, _file):
dict = {}
_dict = {}
try:
with open(_path+_file, 'rU') as _handle:
ids = csv_reader(_handle, dialect='excel', delimiter=',')
for row in ids:
try:
dict[int(row[0])] = (row[1])
except:
pass
records = jload(_handle)['results']
_handle.close
return dict
for record in records:
_dict[int(record['id'])] = record['callsign'].encode('ascii','ignore')
return _dict
except IOError:
return dict
return _dict
# NEW VERSION - MAKES A FULL DICTIONARY OF INFORMATION BASED ON TYPE OF ALIAS FILE
# BASED ON DOWNLOADS FROM DMR-MARC, TGID IS STILL A "SIMPLE" DICTIONARY
# LONG VERSION - MAKES A FULL DICTIONARY OF INFORMATION BASED ON TYPE OF ALIAS FILE
# BASED ON DOWNLOADS FROM RADIOID.NET
def mk_full_id_dict(_path, _file, _type):
dict = {}
if _type == 'subscriber':
fields = SUB_FIELDS
elif _type == 'peer':
fields = PEER_FIELDS
elif _type == 'tgid':
fields = TGID_FIELDS
_dict = {}
try:
with open(_path+_file, 'rU') as _handle:
ids = csv_dict_reader(_handle, fieldnames=fields, restkey='OTHER', dialect='excel', delimiter=',')
for row in ids:
for item in row:
try:
dict[int(row['ID'])] = row
except:
pass
records = jload(_handle)['results']
_handle.close
return dict
if _type == 'peer':
for record in records:
_dict[int(record['id'])] = {
'CALLSIGN': record['callsign'],
'CITY': record['city'],
'STATE': record['state'],
'COUNTRY': record['country'],
'FREQ': record['frequency'],
'CC': record['color_code'],
'OFFSET': record['offset'],
'LINKED': record['ts_linked'],
'TRUSTEE': record['trustee'],
'NETWORK': record['ipsc_network']
}
elif _type == 'subscriber':
for record in records:
_dict[int(record['id'])] = {
'CALLSIGN': record['callsign'],
'NAME': (record['fname'] + ' ' + record['surname']),
'CITY': record['city'],
'STATE': record['state'],
'COUNTRY': record['country']
}
elif _type == 'tgid':
for record in records:
_dict[int(record['tgid'])] = {
'NAME': record['name'],
'ID': record['id']
}
return _dict
except IOError:
return dict
return _dict
# THESE ARE THE SAME THING FOR LEGACY PURPOSES
def get_alias(_id, _dict, *args):
@ -169,8 +183,56 @@ def get_info(_id, _dict, *args):
else:
return _dict[_id]
return _id
if __name__ == '__main__':
'''
repeater file: ('callsign', 'city', 'color_code', 'country', 'frequency', 'ipsc_network', 'locator', 'offset', 'state', 'trustee', 'ts_linked')
user file: ('callsign', 'city', 'country', 'fname', 'radio_id', 'remarks', 'state', 'surname')
'''
# Try updating peer aliases file
result = try_download('/tmp/', 'peers.json', 'https://www.radioid.net/api/dmr/repeater/?country=united%20states', 0)
print(result)
# Try updating subscriber aliases file
result = try_download('/tmp/', 'subscribers.json', 'https://www.radioid.net/api/dmr/user/?country=united%20states', 0)
print(result)
# Make Dictionaries
peer_ids = mk_id_dict('/tmp/', 'peers.json')
if peer_ids:
print('ID ALIAS MAPPER: peer_ids dictionary is available')
subscriber_ids = mk_id_dict('/tmp/', 'subscribers.json')
if subscriber_ids:
print('ID ALIAS MAPPER: subscriber_ids dictionary is available')
full_peer_ids = mk_full_id_dict('/tmp/', 'peers.json', 'peer')
if peer_ids:
print('ID ALIAS MAPPER: full_peer_ids dictionary is available')
full_subscriber_ids = mk_full_id_dict('/tmp/', 'subscribers.json', 'subscriber')
if subscriber_ids:
print('ID ALIAS MAPPER: full_subscriber_ids dictionary is available')
print(get_alias('\x2f\x9b\xe5', subscriber_ids))
print(get_info(3120101, subscriber_ids))
print(get_alias(3120101, full_subscriber_ids))
print(get_info(3120101, full_subscriber_ids))
print(get_alias(31201010, subscriber_ids))
print(get_info(31201010, subscriber_ids))
print(get_alias(312000, peer_ids))
print(get_info(312000, peer_ids))
print(get_alias(312000, full_peer_ids))
print(get_info(312000, full_peer_ids))
print(repr(hex_str_2(65535)))
print(repr(hex_str_3(65535)))
print(ahex(hex_str_4(13120101)))
print(int_id('\x00\xc8\x32\x65'))

2
setup.py
View File

@ -7,7 +7,7 @@ def readme():
return file.read()
setup(name='dmr_utils',
version='0.1.19',
version='0.1.21',
description='ETSI DMR (Digital Mobile Radio) Tier II Utilities',
long_description='Modules to disassemble and assemble DMR packets, including generating and decoding various FEC routines',
classifiers=[