Module: PWN::SDR::Decoder::RDS

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

Overview

RDS Decoder Module for FM Radio Signals.

Two entry points:

.sample  — non-interactive structured Hash (agents / cron / tools)
.decode  — interactive TTY spinner (human REPL via GQRX.init_freq)

Both share the same GQRX RDS protocol path (U RDS, p RDS_PI / PS_NAME / RADIOTEXT). .sample is the canonical mid-layer API that Extrospection and any other automation should call.

Constant Summary collapse

DEFAULT_SETTLE_SECS = 
8.0
DEFAULT_INTERVAL = 
0.75
CALLSIGN_RX = 
/\A[A-Z]{1,2}[A-Z0-9]{2,4}\z/
CALLSIGN_RT_RX = 
/\A([A-Z]{1,2}[A-Z0-9]{2,4})\b/

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com



187
188
189
190
191
 
# File 'lib/pwn/sdr/decoder/rds.rb', line 187

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

.decode(opts = {}) ⇒ Object

Supported Method Parameters

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

Interactive TTY UX: enables RDS, spins a live status line until the operator presses ENTER, then disables RDS. Does not return a Hash — callers that need structured data should use .sample instead.



107
108
109
110
111
112
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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/pwn/sdr/decoder/rds.rb', line 107

public_class_method def self.decode(opts = {})
  freq_obj = opts[:freq_obj]
  raise ArgumentError, 'freq_obj: required' unless freq_obj.is_a?(Hash)

  gqrx_sock = freq_obj[:gqrx_sock]
  raise ArgumentError, 'freq_obj[:gqrx_sock] required' unless gqrx_sock

  # Pretty-print the tuned frequency context without leaking the socket.
  display = freq_obj.dup
  display.delete(:gqrx_sock)
  skip_freq_char = "\n"
  puts JSON.pretty_generate(display)
  puts "\n*** FM Radio RDS Decoder ***"
  puts 'Press [ENTER] to continue to next frequency...'

  unless enable_rds!(sock: gqrx_sock)
    puts 'ERROR: RDS not supported by this radio backend'
    return nil
  end

   = TTY::Spinner.new(
    '[:spinner] :status',
    format: :arrow_pulse,
    clear: true,
    hide_cursor: true
  )

   = 12
  max_title_length = [TTY::Screen.width - , 50].max

  initial_title = 'INFO: Decoding FM radio RDS data...'
  initial_title = initial_title[0...max_title_length] if initial_title.length > max_title_length
  .update(status: initial_title)
  .auto_spin

  last_resp = {}

  loop do
    snap = poll_once(sock: gqrx_sock)
    rds_resp = {
      rds_pi: snap[:pi],
      rds_ps_name: snap[:ps],
      rds_radiotext: snap[:rt]
    }

    if rds_resp[:rds_pi] != '0000' && !rds_resp[:rds_pi].empty? && rds_resp != last_resp
      rds_pi = rds_resp[:rds_pi].upcase.rjust(4, '0')[0, 4]
      rds_ps = "#{rds_resp[:rds_ps_name]}        "[0, 8]
      rds_rt = rds_resp[:rds_radiotext].rstrip

      prefix = "Program ID: #{rds_pi} | Station Name: #{rds_ps} | Radio Txt: "
      available_for_term = [max_title_length - prefix.length, 10].max
      rt_display = rds_rt
      rt_display = "#{rt_display[0...available_for_term]}..." if rt_display.length > available_for_term

      .update(status: "#{prefix}#{rt_display}")
      last_resp = rds_resp.dup
    end

    if $stdin.wait_readable(0)
      begin
        char = $stdin.read_nonblock(1)
        break if char == skip_freq_char
      rescue IO::WaitReadable, EOFError
        # No-op
      end
    end

    sleep 0.01
  end
rescue StandardError => e
  .error('Decoding failed') if defined?() && 
  raise e
ensure
  disable_rds!(sock: gqrx_sock) if defined?(gqrx_sock) && gqrx_sock
  .stop if defined?() && 
end

.help ⇒ Object

Display Usage for this Module



195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
 
# File 'lib/pwn/sdr/decoder/rds.rb', line 195

public_class_method def self.help
  puts "USAGE:
    # Non-interactive (agents / automation) — returns a Hash:
    rds = #{self}.sample(
      gqrx_sock: sock,          # or freq_obj: from init_freq
      settle_secs: 8,           # default 8, max 30
      leave_enabled: false
    )
    # => { pi:, ps_name:, radiotext:, station:, samples:, settle_secs: }

    # Interactive TTY spinner (human REPL via GQRX.init_freq decoder: :rds):
    #{self}.decode(
      freq_obj: 'required - freq_obj returned from PWN::SDR::GQRX.init_freq method'
    )

    #{self}.authors
  "
end

.sample(opts = {}) ⇒ Object

Supported Method Parameters

rds_hash = PWN::SDR::Decoder::RDS.sample( gqrx_sock: 'required unless freq_obj - TCPSocket from GQRX.connect', freq_obj: 'required unless gqrx_sock - Hash from GQRX.init_freq', settle_secs: 'optional - seconds to sample (default 8, max 30)', interval: 'optional - poll interval seconds (default 0.75)', leave_enabled: 'optional - leave RDS decoder ON after sample (default false)' )

Returns

{ pi:, ps_name:, radiotext:, station:, samples: Integer, settle_secs: Float, error: String? # present when RDS backend is unavailable }



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
91
92
93
94
95
96
 
# File 'lib/pwn/sdr/decoder/rds.rb', line 40

public_class_method def self.sample(opts = {})
  sock = resolve_sock(opts)
  raise ArgumentError, 'gqrx_sock: or freq_obj: with :gqrx_sock required' unless sock

  settle   = (opts[:settle_secs] || DEFAULT_SETTLE_SECS).to_f.clamp(0.5, 30.0)
  interval = [(opts[:interval] || DEFAULT_INTERVAL).to_f, 0.1].max
  leave_on = opts[:leave_enabled] ? true : false
  samples  = []

  unless enable_rds!(sock: sock)
    return {
      pi: nil,
      ps_name: nil,
      radiotext: nil,
      station: nil,
      samples: 0,
      settle_secs: settle,
      error: 'RDS not supported by this radio backend'
    }
  end

  deadline = Time.now + settle
  while Time.now < deadline
    snap = poll_once(sock: sock)
    samples << snap unless snap[:pi].empty? && snap[:ps].empty? && snap[:rt].empty?

    # Early exit once we have a non-zero PI and a non-trivial RT —
    # give one more interval for RadioText to finish filling.
    pi = snap[:pi]
    rt = snap[:rt]
    if pi =~ /\A[0-9A-F]{4}\z/ && pi != '0000' && rt.length >= 8
      sleep interval
      snap2 = poll_once(sock: sock)
      samples << snap2
      break if snap2[:rt].length >= rt.length
    end

    sleep interval
  end

  disable_rds!(sock: sock) unless leave_on

  aggregate(samples: samples, settle_secs: settle)
rescue ArgumentError
  raise
rescue StandardError => e
  disable_rds!(sock: sock) if sock && !leave_on
  {
    pi: nil,
    ps_name: nil,
    radiotext: nil,
    station: nil,
    samples: samples&.length.to_i,
    settle_secs: settle,
    error: "#{e.class}: #{e.message}"
  }
end