Module: PWN::SDR::Decoder::POCSAG

Defined in:
lib/pwn/sdr/decoder/pocsag.rb

Overview

Pure-Ruby POCSAG (CCIR Radiopaging Code No. 1) decoder.

GQRX supplies NBFM-discriminator audio on its 48 kHz UDP tap; for a 2-FSK pager channel that is already an NRZ baseband whose sign encodes the bit. This module NRZ-slices at 512/1200/2400 baud, locks onto the 32-bit Frame Sync Codeword (0x7CD215D8), then walks each 8-frame batch of BCH(31,21)+parity codewords, extracting address (RIC/capcode + function bits) and message codewords (numeric BCD or 7-bit ASCII). No multimon-ng, no sox.

Defined Under Namespace

Classes: Demod

Constant Summary collapse

FSC = 
0x7CD215D8
IDLE_CW = 
0x7A89C197
BAUDS = 
[1200, 512, 2400].freeze
BCD_TABLE = 
['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '*', 'U', ' ', '-', ')', '('].freeze
FUNCTION_DESC = 
{
  0 => 'Numeric (Tone/A)',
  1 => 'Tone only (B)',
  2 => 'Tone only (C)',
  3 => 'Alphanumeric (D)'
}.freeze

Class Method Summary collapse

Class Method Details

.alpha_decode(opts = {}) ⇒ Object

Supported Method Parameters

str = PWN::SDR::Decoder::POCSAG.alpha_decode(words: [Integer, ...])



164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
 
# File 'lib/pwn/sdr/decoder/pocsag.rb', line 164

public_class_method def self.alpha_decode(opts = {})
  words = opts[:words] || []
  bitstream = []
  words.each do |w|
    19.downto(0) { |b| bitstream << ((w >> b) & 1) }
  end
  out = +''
  bitstream.each_slice(7) do |ch|
    break if ch.length < 7

    # 7-bit ASCII, LSB first within each character
    code = ch.each_with_index.sum { |b, i| b << i }
    next if code.zero? || code == 0x03 || code == 0x17

    out << (code.between?(0x20, 0x7E) ? code.chr : '.')
  end
  out.strip
end

.assemble(opts = {}) ⇒ Object

Supported Method Parameters

h = PWN::SDR::Decoder::POCSAG.assemble(pending: ric:,func:,msg_words:, baud: 1200)



113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
 
# File 'lib/pwn/sdr/decoder/pocsag.rb', line 113

public_class_method def self.assemble(opts = {})
  pending = opts[:pending]
  baud    = opts[:baud]
  return {} unless pending

  words = pending[:msg_words] || []
  func  = pending[:func]
  type, text =
    if words.empty?
      ['Tone', nil]
    elsif func == 3
      ['Alpha', alpha_decode(words: words)]
    else
      ['Numeric', numeric_decode(words: words)]
    end
  out = {
    protocol: 'POCSAG',
    baud: baud,
    address: pending[:ric],
    capcode: pending[:ric].to_s.rjust(7, '0'),
    function: func,
    function_desc: FUNCTION_DESC[func] || 'Unknown',
    type: type,
    message: text
  }.compact
  summary = ["POCSAG#{baud}", "RIC=#{out[:capcode]}", "F#{func}(#{out[:function_desc]})"]
  summary << "#{type}: #{text}" if text
  out[:summary] = summary.join(' ')
  out
end

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com



216
217
218
 
# File 'lib/pwn/sdr/decoder/pocsag.rb', line 216

public_class_method def self.authors
  "AUTHOR(S):\n  0day Inc. <support@0dayinc.com>\n"
end

.decode(opts = {}) ⇒ Object

Supported Method Parameters

PWN::SDR::Decoder::POCSAG.decode( freq_obj: 'required - freq_obj returned from PWN::SDR::GQRX.init_freq' )



188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
 
# File 'lib/pwn/sdr/decoder/pocsag.rb', line 188

public_class_method def self.decode(opts = {})
  freq_obj = opts[:freq_obj]
  # Prefer true-air I/Q (FM-demod → existing audio demod) when the
  # operator asks for a source/file or sets freq_obj[:iq_source].
  # Otherwise keep the GQRX 48 kHz UDP audio path (run_native).
  want_iq = opts[:source] || opts[:file] || freq_obj[:iq_source] || freq_obj[:iq_file]
  if want_iq
    PWN::SDR::Decoder::Base.run_iq(
      freq_obj: freq_obj,
      protocol: 'POCSAG',
      demod: Demod.new,
      sample_rate: (opts[:sample_rate] || freq_obj[:iq_rate] || 240_000).to_i,
      source: opts[:source],
      file: opts[:file],
      fm_demod: true,
      note: 'POCSAG true-air: FM-demod I/Q then native bit recovery; falls back to detector without SDR hardware.'
    )
  else
    PWN::SDR::Decoder::Base.run_native(
      freq_obj: freq_obj,
      protocol: 'POCSAG',
      demod: Demod.new
    )
  end
end

.decode_bits(opts = {}) ⇒ Object

Supported Method Parameters

carry = PWN::SDR::Decoder::POCSAG.decode_bits(bits: [...], baud: 1200) { |msg| ... } Returns the trailing (unconsumed) bits so the caller can prepend them to the next chunk for streaming continuity.



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/pwn/sdr/decoder/pocsag.rb', line 70

public_class_method def self.decode_bits(opts = {})
  bits = opts[:bits] || []
  baud = opts[:baud]
  i = 0
  pending = nil
  flush = proc do
    yield assemble(pending: pending, baud: baud) if pending && block_given?
    pending = nil
  end
  loop do
    idx = PWN::SDR::Decoder::DSP.find_sync(bits: bits, pattern: FSC, width: 32, max_err: 2, from: i)
    break unless idx

    i = idx + 32
    # One batch = 8 frames × 2 codewords × 32 bits = 512 bits
    8.times do |frame|
      2.times do
        break if i + 32 > bits.length

        cw = PWN::SDR::Decoder::DSP.bits_to_int(bits: bits[i, 32])
        i += 32
        next if [IDLE_CW, FSC].include?(cw)

        if cw.nobits?(0x80000000)
          flush.call
          addr18 = (cw >> 13) & 0x3FFFF
          func   = (cw >> 11) & 0x3
          ric    = (addr18 << 3) | frame
          pending = { ric: ric, func: func, msg_words: [] }
        elsif pending
          pending[:msg_words] << ((cw >> 11) & 0xFFFFF)
        end
      end
    end
  end
  flush.call
  tail_from = [bits.length - 576, 0].max
  bits[tail_from..] || []
end

.help ⇒ Object

Display Usage for this Module



222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
 
# File 'lib/pwn/sdr/decoder/pocsag.rb', line 222

public_class_method def self.help
  puts "USAGE (true-air I/Q + GQRX-audio native paths, no external binaries):
    #{self}.decode(
      freq_obj: 'required - freq_obj returned from PWN::SDR::GQRX.init_freq'
    )

    #{self}.decode_bits(bits: [0,1,...], baud: 1200) { |msg| ... }
    #{self}.assemble(pending: {ric:, func:, msg_words:[]}, baud: 1200)
    #{self}.numeric_decode(words: [Integer, ...])
    #{self}.alpha_decode(words: [Integer, ...])

    NOTE: Set GQRX to Narrow FM. Baud (512/1200/2400) and NRZ
          polarity are auto-detected from the FSC 0x7CD215D8.

    #{self}.authors
  "
end

.numeric_decode(opts = {}) ⇒ Object

Supported Method Parameters

str = PWN::SDR::Decoder::POCSAG.numeric_decode(words: [Integer, ...])



147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/pwn/sdr/decoder/pocsag.rb', line 147

public_class_method def self.numeric_decode(opts = {})
  words = opts[:words] || []
  out = +''
  words.each do |w|
    5.times do |d|
      nib = (w >> (16 - (d * 4))) & 0xF
      # POCSAG BCD nibbles are bit-reversed within each 4-bit group
      rev = ((nib & 1) << 3) | ((nib & 2) << 1) | ((nib & 4) >> 1) | ((nib & 8) >> 3)
      out << BCD_TABLE[rev]
    end
  end
  out.gsub(/ +$/, '')
end