Module: PWN::SDR::Decoder::Morse

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

Overview

Pure-Ruby CW / Morse decoder for the amateur CW sub-bands.

GQRX's CW/USB demodulator produces a ~600–800 Hz sidetone in the 48 kHz UDP audio stream. This module envelope-detects the tone with a state-preserving single-pole low-pass, adaptively thresholds it into on/off runs, classifies each run as dit / dah / char-gap / word-gap by timing, and looks the resulting .- sequences up in DSP::MORSE_TABLE. No multimon-ng, no sox.

Defined Under Namespace

Classes: Demod

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. [email protected]



160
161
162
 
# File 'lib/pwn/sdr/decoder/morse.rb', line 160

public_class_method def self.authors
  "AUTHOR(S):\n  0day Inc. <[email protected]>\n"
end

.decode(opts = {}) ⇒ Object 



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
 
# File 'lib/pwn/sdr/decoder/morse.rb', line 120

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(
      **opts,
      fallback: :raise,
      freq_obj: freq_obj,
      protocol: 'MORSE-CW',
      demod: Demod.new(rate: (opts[:sample_rate] || freq_obj[:iq_rate] || 48_000).to_i),
      sample_rate: (opts[:sample_rate] || freq_obj[:iq_rate] || 48_000).to_i,
      source: opts[:source],
      file: opts[:file],
      fm_demod: true,
      note: 'MORSE-CW true-air: FM-demod I/Q then native bit recovery; missing I/Q raises (use .detect for energy only).'
    )
  else
    PWN::SDR::Decoder::Base.run_native(
      **opts,
      freq_obj: freq_obj,
      protocol: 'MORSE-CW',
      demod: Demod.new(rate: (opts[:rate] || 48_000).to_i)
    )
  end
end

.decode_string(opts = {}) ⇒ Object

Supported Method Parameters

h = PWN::SDR::Decoder::Morse.decode_string(pattern: '.- -...')



152
153
154
155
156
 
# File 'lib/pwn/sdr/decoder/morse.rb', line 152

public_class_method def self.decode_string(opts = {})
  pattern = opts[:pattern].to_s
  txt = pattern.strip.split(/\s+/).map { |s| PWN::SDR::Decoder::DSP::MORSE_TABLE[s] || '_' }.join
  { protocol: 'MORSE-CW', text: txt, summary: "CW #{txt}" }
end

.detect(opts = {}) ⇒ Object

Realtime options forwarded to Base: on_frame (Hash callback), output (writable IO), interactive (default true), duration (seconds), stop (callable), queue_size (bounded chunks), log_file (path or false). Energy detection only; does not identify or decode Morse payloads.

Supported Method Parameters

Morse.detect(freq_obj: Hash, threshold: 8.0, on_frame: Proc)



112
113
114
115
116
117
118
 
# File 'lib/pwn/sdr/decoder/morse.rb', line 112

public_class_method def self.detect(opts = {})
  Base.run_detector(opts.merge(
                      protocol: 'MORSE',
                      note: 'Energy detection only; no protocol payload decoding.',
                      describe: proc { |_burst| { event: 'detection', capability: 'energy-detection', decoded: false } }
                    ))
end

.help ⇒ Object

Display Usage for this Module



166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
 
# File 'lib/pwn/sdr/decoder/morse.rb', line 166

public_class_method def self.help
  puts "USAGE:
    # Detect energy only (not protocol payloads); accepts Base runner controls.
    #{self}.detect(freq_obj: {}, threshold: 8.0, on_frame: nil)
    # Run decode and return its result
    #{self}.decode(
      freq_obj: 'required - freq_obj returned from PWN::SDR::GQRX.init_freq',
      on_frame: 'optional - callback receiving each emitted Hash',
      output: 'optional - writable IO (default stdout)',
      interactive: 'optional - false disables ENTER input',
      duration: 'optional - finite seconds to run',
      stop: 'optional - callable returning true to stop',
      queue_size: 'optional - bounded pending chunks (default 8)',
      log_file: 'optional - JSONL path or false to disable logging',
      source: 'optional - source value consumed by #decode',
      file: 'optional - filesystem path',
      sample_rate: 'optional - sample rate value consumed by #decode'
    )

    # Run decode string and return its result
    #{self}.decode_string(
      pattern: 'optional - pattern value consumed by #decode_string'
    )

    # Print the AUTHOR(S) string for this module.
    #{self}.authors
  "
  constants.sort
end