Module: PWN::SDR::Decoder::RTL433

Defined in:

lib/pwn/sdr/decoder/rtl433.rb

Overview

Protocol frames via .decode; energy observations only via .detect.

Defined Under Namespace

Classes: AcuriteIQ, DetectorIQ

Constant Summary

SUPPORTED_MODES =

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).

%i[acurite_609txc native].freeze

DemodIQ =

AcuriteIQ

Class Method Summary

• .authors ⇒ Object

  Author(s)

  0day Inc.

• .decode(opts = {}) ⇒ Object

  Never silently downgrade decode to energy detection.

• .detect(opts = {}) ⇒ Object
• .help ⇒ Object
• .parse_frame(opts = {}) ⇒ Object
• .parse_line(opts = {}) ⇒ Object

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com

# File 'lib/pwn/sdr/decoder/rtl433.rb', line 333

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

.decode(opts = {}) ⇒ Object

Never silently downgrade decode to energy detection.

Raises:

• (ArgumentError)

# File 'lib/pwn/sdr/decoder/rtl433.rb', line 171

public_class_method def self.decode(opts = {})
  mode = opts.fetch(:mode, :acurite_609txc).to_s.to_sym
  raise ArgumentError, "unsupported RTL433 mode #{mode}; supported: #{SUPPORTED_MODES.join(', ')}" unless SUPPORTED_MODES.include?(mode)

  return native_replay(opts) if mode == :native

  freq_obj = opts[:freq_obj] || {}
  rate = opts[:sample_rate] || freq_obj[:iq_rate] || 250_000
  demod = AcuriteIQ.new(rate: rate, threshold: opts.fetch(:threshold, 0.5))
  PWN::SDR::Decoder::Base.run_iq(opts.merge(freq_obj: freq_obj, protocol: 'RTL433',
                                            sample_rate: rate, demod: demod, fallback: :raise))
end

.detect(opts = {}) ⇒ Object

# File 'lib/pwn/sdr/decoder/rtl433.rb', line 282

public_class_method def self.detect(opts = {})
  freq_obj = opts[:freq_obj]

  rate  = (opts[:sample_rate] || freq_obj[:iq_rate] || 250_000).to_i
  proto = 'ISM-433'
  demod = DetectorIQ.new(
    rate: rate, protocol: proto, modulation: 'OOK/ASK/FSK',
    extra: { threshold: 10.0 }
  )
  PWN::SDR::Decoder::Base.run_iq(
    **opts,
    freq_obj: freq_obj,
    protocol: proto,
    sample_rate: rate,
    source: opts[:source],
    file: opts[:file],
    demod: demod,
    threshold: 10.0,
    note: 'True-air I/Q path characterizes OOK/FSK bursts; offline rtl_433 JSON via .parse_line.',
    describe: proc { |b|
      { modulation: 'OOK/ASK/FSK', classification: (if b[:duration_ms] < 20
                                                      'keyfob/OOK-short'
                                                    else
                                                      (b[:duration_ms] < 120 ? 'sensor/OOK-packet' : 'FSK-continuous')
                                                    end) }
    }
  )
end

.help ⇒ Object

# File 'lib/pwn/sdr/decoder/rtl433.rb', line 337

public_class_method def self.help
  puts "USAGE:
    # Default :acurite_609txc is the Ruby OOK/PPM implementation.
    # mode: :native runs installed rtl_433 with its enabled device catalogue (OOK/FSK).
    # Native mode is finite FILE replay only: file required, no auto RF/config loading.
    # Native options: format: :cu8/:cs16/:cf32, protocols: [positive native IDs],
    # executable: 'rtl_433', duration: 30 seconds (deadline), stop: callable.
    # Missing executable raises IOError with install/path guidance; already-stopped replay never spawns.
    # Output is per-device native JSON, device_protocol preserves native numeric ID.
    # Individual sensors/integrity depend on installed version; mic absent means not-reported.
    # sample_rate: 250000 default, >=20000; threshold: 0.5 normalized envelope amplitude.
    # Parses ID/status/battery/signed temperature/humidity and checks additive checksum.
    # The Ruby mode checksum is weak, not authentication; use native mode for FSK/other sensors.
    # decode never falls back to a detector; unsupported mode raises ArgumentError.
    # #{self}.detect uses the same source/lifecycle options for energy-only observations.
    # Validate a complete frame and return decoded fields or nil.
    #{self}.parse_frame(bits: 'required - Array of 40 binary digits in transmitted order')
    # 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',
      sample_rate: 'optional - sample rate value consumed by #decode',
      source: 'optional - source value consumed by #decode',
      file: 'optional - filesystem path'
    )

    # Run parse line and return its result
    #{self}.parse_line(
      line: 'optional - line value consumed by #parse_line'
    )

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

.parse_frame(opts = {}) ⇒ Object

# File 'lib/pwn/sdr/decoder/rtl433.rb', line 142

public_class_method def self.parse_frame(opts = {})
  bits = opts[:bits]
  return nil unless bits.is_a?(Array) && bits.length == 40 && bits.all? { |b| [0, 1].include?(b) }

  bytes = bits.each_slice(8).map { |byte| byte.join.to_i(2) }
  checksum = bytes.first(4).sum
  return nil if checksum.zero? || (checksum & 0xff) != bytes[4] || bytes[3] > 100

  temperature = ((bytes[1] & 15) << 8) | bytes[2]
  temperature -= 4096 if temperature >= 2048
  { protocol: 'RTL433', mode: :acurite_609txc, source: 'iq', decoded: true,
    model: 'Acurite-609TXC', id: bytes[0], status: bytes[1] >> 4,
    battery_ok: bytes[1].nobits?(0x80) ? 1 : 0, temperature_C: temperature / 10.0,
    humidity: bytes[3], integrity: 'additive-checksum', payload_hex: bytes.pack('C*').unpack1('H*'),
    summary: "Acurite-609TXC id=#{bytes[0]} temperature=#{temperature / 10.0}C humidity=#{bytes[3]}%" }
end

.parse_line(opts = {}) ⇒ Object

# File 'lib/pwn/sdr/decoder/rtl433.rb', line 311

public_class_method def self.parse_line(opts = {})
  line = opts[:line].to_s
  h = begin
    JSON.parse(line, symbolize_names: true)
  rescue StandardError
    { unparsed: line }
  end
  out = { protocol: 'RTL433' }.merge(h)
  bits = []
  bits << out[:model].to_s if out[:model]
  bits << "id=#{out[:id]}" if out[:id]
  bits << "ch=#{out[:channel]}" if out[:channel]
  bits << "code=#{out[:code]}" if out[:code]
  bits << "cmd=#{out[:cmd] || out[:button]}" if out[:cmd] || out[:button]
  bits << "temp=#{out[:temperature_C]}C" if out[:temperature_C]
  bits << "rssi=#{out[:rssi]}" if out[:rssi]
  out[:summary] = bits.empty? ? line[0, 120] : bits.join(' ')
  out
end