forked from KD8EYF/DMRlink
-
Notifications
You must be signed in to change notification settings - Fork 0
/
log.py
executable file
·90 lines (73 loc) · 4.21 KB
/
log.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
#!/usr/bin/env python
#
# This work is licensed under the Creative Commons Attribution-ShareAlike
# 3.0 Unported License.To view a copy of this license, visit
# http://creativecommons.org/licenses/by-sa/3.0/ or send a letter to
# Creative Commons, 444 Castro Street, Suite 900, Mountain View,
# California, 94041, USA.
# This is a sample application that snoops voice traffic to log calls
from __future__ import print_function
from twisted.internet import reactor
from binascii import b2a_hex as h
import time
from dmrlink import IPSC, NETWORK, networks, get_info, int_id, subscriber_ids, peer_ids, talkgroup_ids, logger
__author__ = 'Cortney T. Buffington, N0MJS'
__copyright__ = 'Copyright (c) 2013, 2014 Cortney T. Buffington, N0MJS and the K0USY Group'
__credits__ = 'Adam Fast, KC0YLK, Dave K, and he who wishes not to be named'
__license__ = 'Creative Commons Attribution-ShareAlike 3.0 Unported'
__version__ = '1.0'
__maintainer__ = 'Cort Buffington, N0MJS'
__email__ = 'n0mjs@me.com'
__status__ = 'Production'
class logIPSC(IPSC):
def __init__(self, *args, **kwargs):
IPSC.__init__(self, *args, **kwargs)
self.ACTIVE_CALLS = []
#************************************************
# CALLBACK FUNCTIONS FOR USER PACKET TYPES
#************************************************
def group_voice(self, _network, _src_sub, _dst_sub, _ts, _end, _peerid, _data):
# _log = logger.debug
if (_ts not in self.ACTIVE_CALLS) or _end:
_time = time.strftime('%m/%d/%y %H:%M:%S')
_dst_sub = get_info(int_id(_dst_sub), talkgroup_ids)
_peerid = get_info(int_id(_peerid), peer_ids)
_src_sub = get_info(int_id(_src_sub), subscriber_ids)
if not _end: self.ACTIVE_CALLS.append(_ts)
if _end: self.ACTIVE_CALLS.remove(_ts)
if _ts: _ts = 2
else: _ts = 1
if _end: _end = 'END'
else: _end = 'START'
print('{} ({}) Call {} Group Voice: \n\tIPSC Source:\t{}\n\tSubscriber:\t{}\n\tDestination:\t{}\n\tTimeslot\t{}' .format(_time, _network, _end, _peerid, _src_sub, _dst_sub, _ts))
def private_voice(self, _network, _src_sub, _dst_sub, _ts, _end, _peerid, _data):
# _log = logger.debug
if (_ts not in self.ACTIVE_CALLS) or _end:
_time = time.strftime('%m/%d/%y %H:%M:%S')
_dst_sub = get_info(int_id(_dst_sub), subscriber_ids)
_peerid = get_info(int_id(_peerid), peer_ids)
_src_sub = get_info(int_id(_src_sub), subscriber_ids)
if not _end: self.ACTIVE_CALLS.append(_ts)
if _end: self.ACTIVE_CALLS.remove(_ts)
if _ts: _ts = 2
else: _ts = 1
if _end: _end = 'END'
else: _end = 'START'
print('{} ({}) Call {} Private Voice: \n\tIPSC Source:\t{}\n\tSubscriber:\t{}\n\tDestination:\t{}\n\tTimeslot\t{}' .format(_time, _network, _end, _peerid, _src_sub, _dst_sub, _ts))
def group_data(self, _network, _src_sub, _dst_sub, _ts, _end, _peerid, _data):
_dst_sub = get_info(int_id(_dst_sub), talkgroup_ids)
_peerid = get_info(int_id(_peerid), peer_ids)
_src_sub = get_info(int_id(_src_sub), subscriber_ids)
print('({}) Group Data Packet Received From: {}' .format(_network, _src_sub))
def private_data(self, _network, _src_sub, _dst_sub, _ts, _end, _peerid, _data):
_dst_sub = get_info(int_id(_dst_sub), subscriber_ids)
_peerid = get_info(int_id(_peerid), peer_ids)
_src_sub = get_info(int_id(_src_sub), subscriber_ids)
print('({}) Private Data Packet Received From: {} To: {}' .format(_network, _src_sub, _dst_sub))
if __name__ == '__main__':
logger.info('DMRlink \'log.py\' (c) 2013, 2014 N0MJS & the K0USY Group - SYSTEM STARTING...')
for ipsc_network in NETWORK:
if NETWORK[ipsc_network]['LOCAL']['ENABLED']:
networks[ipsc_network] = logIPSC(ipsc_network)
reactor.listenUDP(NETWORK[ipsc_network]['LOCAL']['PORT'], networks[ipsc_network], interface=NETWORK[ipsc_network]['LOCAL']['IP'])
reactor.run()