CW4ISR2

[JG1VPP]

問題意識

複数信号の追跡

複数信号の分離、誤表示の抑制、リアルタイム性の改善が課題である。現状の方式は、その時点で強い周波数 bin を中心に処理するため、近接した信号同士で強い方に引き寄せられやすく、中心周波数が 1〜2 bin 変動しただけで別信号として扱われやすい。Track を導入することで、各信号を継続的に追跡する対象として保持できるようになり、近接した 2 局を別々に維持しやすくなり、フェージングによるわずかな周波数変動でも追尾が切れにくくなり、強い局の近傍にある弱い局も残りやすくなる。これは、強い音をその都度拾う方式から、個々の信号を継続して追う方式へ移行することを意味する。

復調の安定性

1 bin を直接読む方式では、信号が bin 境界に乗った場合や微小な周波数ずれが生じた場合にレベル系列が不安定になりやすい。近傍 bin を合成して包絡を構成することで、ON/OFF 判定がぶれにくくなり、短点と長点の長さ推定が安定し、隣接信号やスペクトル漏れの影響も受けにくくなる。その結果、点と線の取り違えや文字間、語間の誤判定を減らすことができる。

確定前表示の抑制

表示の確定タイミングにも改善余地がある。従来の Trim ベースの扱いは末尾の未完成要素を隠すには有効であるが、文字が確定したかどうかを明示的に管理していない。PendingCode と CommitText を分離することで、ON が終わっただけでは表示せず、文字間 gap が十分に観測された時点で文字を確定し、語間 gap が観測された時点で空白も確定できる。これにより、未確定の文字が先に表示される問題を大きく減らすことができる。

リアルタイム性の改善

リアルタイム性についても、未確定の内容を先に出すことで見かけ上の速さを得るのではなく、確定した内容のみを即時に表示する方向へ改める必要がある。この変更により、表示がわずかに遅れる場面はあり得るが、表示された文字の信頼性は高まり、UI の揺れも抑えられる。結果として、速さだけを優先した表示ではなく、確定ベースのリアルタイム表示に近づく。

速度変動への追従

Dit を track ごとに保持して更新することで、送信者ごとの打鍵速度の違い、やや詰まったり伸びたりする癖、運用中の微小な速度変化に追従しやすくなる。これにより、固定しきい値による判定よりも、短点と長点、要素間と文字間、語間の分類が安定する。

混信とノイズへの耐性

ヒステリシス付きの ON/OFF 判定や短すぎるパルスの除去を導入することで、一瞬のノイズで点が出る誤り、短い谷によって文字が不自然に切れる誤り、隣接局の裾で ON 扱いになる誤りを抑制できる。特に 7MHz 帯のような混雑した CW 環境では、この改善の効果が大きいと考えられる。

全体としての変化

総じて、この提案により、アプリはスペクトル上の強い箇所を毎回読み直すデコーダから、複数の CW 信号をそれぞれ追跡し、確定した文字だけを出力するデコーダへ近づくことになる。改善が期待できる点は、複数局の分離性能、表示の安定性、誤表示の抑制、確定タイミングの明確化、混雑帯での粘り強さである。これは精度の改善にとどまらず、実運用における使いやすさの向上にも直結する。受信者の体感としては、読める文字数が増えること以上に、表示結果を信用しやすくなる変化として現れるはずである。

解決方法

Track の導入

既存の History ベースの継続判定を見直し、内部状態として Track を導入する。これにより、周波数 bin の一致ではなく、継続的に追跡している信号そのものを単位として扱えるようにする。各 Track は中心周波数、強度、雑音床、最終観測時刻、未確定の符号列、確定済みの文字列、推定中の Dit 長などを保持し、フレームごとに更新する構造とする。

ピーク検出とトラック割り当て

各更新では、スペクトル全体から局所ピークを抽出し、そのピークを既存の Track に割り当てる。割り当てには周波数差と強度差を用い、近傍かつ継続性の高いものを優先して対応付ける。適切な Track が存在しない場合には新規に Track を生成し、一定時間観測されなくなった Track は破棄する。この変更により、一時的な周波数揺れや強弱変動があっても、同一信号を継続して追跡しやすくする。

近傍 bin の合成

各 Track に対する復調入力は単一 bin から直接取得するのではなく、中心周波数の近傍数 bin を重み付きで合成した包絡を用いる。これにより、信号が bin 境界に位置する場合や微小な周波数変動がある場合でも、より安定したレベル系列を得られるようにする。結果として、ON/OFF 判定、短点と長点の推定、文字間と語間の判定を安定化させる。

Track 単位の復調状態管理

従来の stateless な再判定に加えて、Track ごとに状態を持ちながら復調を進める方式を導入する。各 Track は現在の ON/OFF 状態、継続フレーム数、推定 Dit 長、文字間 gap、語間 gap を保持し、入力系列を逐次処理しながら更新する。これにより、各信号の時間的連続性を利用した判定が可能になり、単発の揺れや局所的な乱れに強い構造へ移行する。

Pending と Commit の分離

未確定の符号列と確定済みの文字列を明確に分離する。ON 区間が終わった時点では短点または長点を PendingCode に追加するだけにとどめ、文字間 gap が十分に観測された時点で初めて CommitCode と CommitText に反映する。語間 gap が観測された場合には空白も同時に確定する。この方式により、確定前の文字が UI に出ることを防ぎ、確定した瞬間のみ表示される挙動を実現する。

適応的な速度推定

Dit は固定値ではなく Track ごとに保持し、観測された ON 区間の長さに応じて徐々に更新する。短点と判断された場合はその長さを、長点と判断された場合は 3 分の 1 に正規化した長さを用いて Dit を更新し、同時に文字間 gap と語間 gap も再計算する。この方式により、送信者ごとの癖や運用中の速度変化に自然に追従できるようにする。

ノイズ耐性の強化

ON/OFF 判定にはヒステリシスを導入し、立ち上がりと立ち下がりで異なる閾値を用いる。これにより、境界付近の揺れによるチャタリングを抑制する。また、短すぎる ON パルスや短すぎる OFF ギャップはノイズとして扱うことで、一瞬の雑音による誤検出や不自然な文字分断を減らす。この変更により、混信環境やノイズの多い受信条件においても、より安定した出力を目指す。

既存インターフェースとの整合

内部では Track を中心とした状態管理へ移行する一方で、外部への出力は引き続き Message とする。Track の状態から Snapshot を生成することで、既存の Program フックや表示処理との互換性を維持する。これにより、内部実装を大きく改善しつつ、周辺コードへの影響を最小限に抑える構成とする。

[JG1VPP]

実装

core/track.go

package core

type Track struct {
	ID       int
	Center   float64
	Power    float64
	Noise    float64
	LastSeen int
	Miss     int

	State     bool
	OnFrames  int
	OffFrames int

	Dit     float64
	GapChar float64
	GapWord float64

	PendingCode string
	CommitCode  string
	CommitText  string

	Env []float64
}

func NewTrack(id int, center, power float64, now int) *Track {
	return &Track{
		ID:       id,
		Center:   center,
		Power:    power,
		LastSeen: now,
		Dit:      3.0,
		GapChar:  9.0,
		GapWord:  21.0,
	}
}

func (tr *Track) Snapshot(now int) Message {
	return Message{
		Freq: int(tr.Center + 0.5),
		Time: now,
		Miss: tr.Miss,
		Code: tr.CommitCode,
		Trim: tr.CommitCode,
		Text: tr.CommitText,
	}
}

core/decoder.go

package core

import "math"

type Decoder struct {
	Scatter int
	MaxMiss int
	Squelch float64
	Monitor Monitor
	Scanner Scanner

	Tracks []*Track
	NextID int

	Program func(Message) Message
}

type Peak struct {
	Center float64
	Power  float64
	Bin    int
}

func DefaultDecoder(rate int) Decoder {
	return Decoder{
		Scatter: 1,
		MaxMiss: 2,
		Monitor: DefaultMonitor(rate),
		Scanner: DefaultScanner(),
	}
}

func (d *Decoder) Read(signal []float64) (result []Message) {
	data := d.Monitor.Next(signal)
	now := d.Monitor.Time

	peaks := detectPeaks(data, d.Squelch)
	d.assignTracks(peaks, now)

	for _, tr := range d.Tracks {
		if tr.Miss > d.MaxMiss {
			continue
		}

		src := integrateAround(data, tr.Center)
		d.Scanner.UpdateTrack(tr, src)

		msg := tr.Snapshot(now)
		if d.Program != nil {
			msg = d.Program(msg)
		}
		if msg.Miss <= d.MaxMiss {
			result = append(result, msg)
		}
	}

	d.pruneTracks()
	return
}

func detectPeaks(data []Message, squelch float64) []Peak {
	pows := make([]float64, len(data))
	for i, msg := range data {
		pows[i] = pow64(msg.Data)
	}

	var peaks []Peak
	for i := 1; i < len(pows)-1; i++ {
		if pows[i] <= squelch {
			continue
		}
		if pows[i] < pows[i-1] || pows[i] < pows[i+1] {
			continue
		}

		center := float64(i)

		l := pows[i-1]
		c := pows[i]
		r := pows[i+1]
		den := l - 2*c + r
		if den != 0 {
			delta := 0.5 * (l - r) / den
			if math.Abs(delta) <= 1.0 {
				center += delta
			}
		}

		peaks = append(peaks, Peak{
			Center: center,
			Power:  c,
			Bin:    i,
		})
	}

	return peaks
}

func (d *Decoder) assignTracks(peaks []Peak, now int) {
	used := map[int]bool{}

	for _, pk := range peaks {
		best := -1
		bestCost := math.MaxFloat64

		for i, tr := range d.Tracks {
			if used[i] {
				continue
			}

			df := math.Abs(pk.Center - tr.Center)
			dp := math.Abs(pk.Power - tr.Power)
			cost := df + 0.2*dp

			if df <= float64(d.Scatter+2) && cost < bestCost {
				best = i
				bestCost = cost
			}
		}

		if best >= 0 {
			tr := d.Tracks[best]
			tr.Center = 0.8*tr.Center + 0.2*pk.Center
			tr.Power = 0.7*tr.Power + 0.3*pk.Power
			tr.LastSeen = now
			tr.Miss = 0
			used[best] = true
			continue
		}

		d.Tracks = append(d.Tracks, NewTrack(d.NextID, pk.Center, pk.Power, now))
		d.NextID++
	}

	for _, tr := range d.Tracks {
		if tr.LastSeen != now {
			tr.Miss++
		}
	}
}

func (d *Decoder) pruneTracks() {
	var keep []*Track
	for _, tr := range d.Tracks {
		if tr.Miss <= d.MaxMiss {
			keep = append(keep, tr)
		}
	}
	d.Tracks = keep
}

func integrateAround(data []Message, center float64) Message {
	c := int(center + 0.5)
	if c < 0 {
		c = 0
	}
	if c >= len(data) {
		c = len(data) - 1
	}

	out := Message{
		Freq: c,
		Time: data[c].Time,
		Data: make([]float64, len(data[c].Data)),
	}

	weights := []float64{0.2, 0.6, 1.0, 0.6, 0.2}
	for k := -2; k <= 2; k++ {
		f := c + k
		if f < 0 || f >= len(data) {
			continue
		}
		w := weights[k+2]
		for i, v := range data[f].Data {
			out.Data[i] += w * v
		}
	}

	return out
}

core/scanner.go

package core

func DefaultScanner() Scanner {
	return Scanner{
		Gain: 2.5,
		Iter: 16,
	}
}

type Scanner struct {
	Gain float64
	Iter int
}

func (s *Scanner) Scan(source Message) Message {
	level := &Classes{X: source.AGC(s.Gain)}
	level.Train(2, s.Iter)

	segments := level.Segments(0)

	var spans []float64
	for _, seg := range segments {
		if seg.Class {
			spans = append(spans, seg.Width)
		}
	}

	speed := &Classes{X: spans}
	if len(spans) > 0 {
		speed.Train(2, s.Iter)
		source.Code = speed.Code(segments)
		source.Trim = speed.Trim(segments)
	} else {
		source.Code = ""
		source.Trim = ""
	}

	source.Body = segments
	source.Text = CodeToText(source.Trim)
	return source
}

func (s *Scanner) UpdateTrack(tr *Track, source Message) {
	level := source.AGC(s.Gain)
	if len(level) == 0 {
		return
	}

	var avg float64
	for _, x := range level {
		avg += x
	}
	avg /= float64(len(level))

	if tr.Noise == 0 {
		tr.Noise = avg
	} else {
		tr.Noise = 0.9*tr.Noise + 0.1*avg
	}

	thOn := tr.Noise * 1.8
	thOff := tr.Noise * 1.3

	for _, x := range level {
		if !tr.State {
			if x >= thOn {
				flushOff(tr)
				tr.State = true
				tr.OnFrames = 1
				tr.OffFrames = 0
			} else {
				tr.OffFrames++
				commitIfNeeded(tr)
			}
		} else {
			if x <= thOff {
				flushOn(tr)
				tr.State = false
				tr.OffFrames = 1
				tr.OnFrames = 0
				commitIfNeeded(tr)
			} else {
				tr.OnFrames++
			}
		}
		tr.Env = append(tr.Env, x)
	}
}

func flushOff(tr *Track) {
	if tr.OffFrames <= 0 {
		return
	}
}

func flushOn(tr *Track) {
	if tr.OnFrames <= 0 {
		return
	}

	w := float64(tr.OnFrames)

	if w < 0.5*tr.Dit {
		return
	}

	if w < 2.0*tr.Dit {
		tr.PendingCode += "."
		tr.Dit = 0.9*tr.Dit + 0.1*w
	} else {
		tr.PendingCode += "_"
		tr.Dit = 0.9*tr.Dit + 0.1*(w/3.0)
	}

	tr.GapChar = 3.0 * tr.Dit
	tr.GapWord = 7.0 * tr.Dit
}

func commitIfNeeded(tr *Track) {
	if tr.PendingCode == "" {
		return
	}

	off := float64(tr.OffFrames)

	if off >= tr.GapWord {
		commitPending(tr)
		if tr.CommitText == "" || tr.CommitText[len(tr.CommitText)-1] != ' ' {
			tr.CommitText += " "
		}
		return
	}

	if off >= tr.GapChar {
		commitPending(tr)
	}
}

func commitPending(tr *Track) {
	if tr.PendingCode == "" {
		return
	}

	if tr.CommitCode != "" {
		tr.CommitCode += " "
	}
	tr.CommitCode += tr.PendingCode
	tr.CommitText += CodeToText(tr.PendingCode)
	tr.PendingCode = ""
}

core/message.go

package core

import "math"

type Message struct {
	Data []float64
	Body []Segment
	Code string
	Trim string
	Text string
	Freq int
	Time int
	Miss int
	Hide bool
}

func (m Message) AGC(gain float64) []float64 {
	seq := make([]float64, len(m.Data))
	max := max64(m.Data)
	if max == 0 {
		return seq
	}
	for n, v := range m.Data {
		if v == 0 {
			continue
		}
		seq[n] = v * math.Min(gain, max/v)
	}
	return seq
}

func (m Message) Merge(next Message) Message {
	return Message{
		Data: append(m.Data, next.Data...),
		Freq: m.Freq,
		Time: m.Time,
	}
}

core/history.go

package core

type History []Message

func (h History) Active(maxMiss int) (out []Message) {
	for _, msg := range h {
		if msg.Miss <= maxMiss && !msg.Hide {
			out = append(out, msg)
		}
	}
	return
}

func (h History) FindByFreq(freq int) (Message, bool) {
	for _, msg := range h {
		if msg.Freq == freq {
			return msg, true
		}
	}
	return Message{}, false
}

core/segment.go

package core

import "strings"

type Segment struct {
	Class bool
	Since int
	Until int
	Width float64
	Level float64
}

func (m *Classes) Segments(first int) (result []Segment) {
	if len(m.X) == 0 {
		return nil
	}

	since := 0
	final := len(m.X) - 1

	for until, value := range m.X {
		class := m.Class(value)
		if first != class || until == final {
			end := until
			if until == final {
				end = until + 1
			}
			if end > since {
				result = append(result, Segment{
					Class: first == 1,
					Since: since,
					Until: end,
					Width: float64(end - since),
					Level: med64(m.X[since:end]),
				})
			}
			since = until
			first = class
		}
	}

	if len(result) > 1 {
		return result[1:]
	}
	return nil
}

func (m *Classes) Code(segments []Segment) (code string) {
	for _, s := range segments {
		if s.Class {
			switch m.Class(s.Width) {
			case 0:
				code += "."
			case 1:
				code += "_"
			}
			continue
		}

		switch m.Extra(s.Width) {
		case 0:
		case 1:
			code += " "
		default:
			code += " ; "
		}
	}
	return
}

func (m *Classes) Trim(segments []Segment) string {
	return strings.TrimRight(m.Code(segments), "._")
}

view/decoder.go

package view

import "github.com/nextzlog/cw4i/core"

type Decoder struct {
	Initial func(rate int) (core.Decoder, error)

	core.Decoder
	History
}

func (d *Decoder) Update(rate int) {
	decoder, err := d.Initial(rate)
	if err != nil {
		return
	}
	d.Decoder = decoder
	d.History.Clear()
}

func (d *Decoder) Restart(rate int) {
	d.Update(rate)
}

func (d *Decoder) Resquelch(value float64) {
	d.Decoder.Squelch = value
	d.Decoder.Tracks = nil
	d.Decoder.NextID = 0
	d.History.Clear()
}

func (d *Decoder) Decode(signal []float64) {
	messages := d.Decoder.Read(signal)
	d.History.Add(messages...)
}

view/history.go

package view

import (
	"fmt"

	"fyne.io/fyne/v2"
	"fyne.io/fyne/v2/widget"

	"github.com/nextzlog/cw4i/core"
)

type History struct {
	Items []core.Message
	Added int
}

func (h *History) Clear() {
	h.Items = nil
	h.Added = 0
}

func (h *History) Add(messages ...core.Message) {
	if len(messages) == 0 {
		return
	}
	h.Items = append(h.Items, messages...)
	h.Added += len(messages)
}

func (h *History) Length() int {
	return len(h.Items)
}

func (h *History) CanvasObject() fyne.CanvasObject {
	list := widget.NewList(
		func() int {
			return h.Length()
		},
		func() fyne.CanvasObject {
			return widget.NewLabel("")
		},
		func(id widget.ListItemID, object fyne.CanvasObject) {
			label := object.(*widget.Label)
			msg := h.Items[id]
			label.SetText(formatMessage(msg))
		},
	)
	list.Refresh()
	return list
}

func (h *History) Refresh(list *widget.List) {
	if h.Added == 0 {
		return
	}
	list.Refresh()
	h.Added = 0
}

func formatMessage(msg core.Message) string {
	if msg.Text != "" {
		return fmt.Sprintf("%d %s", msg.Freq, msg.Text)
	}
	if msg.Trim != "" {
		return fmt.Sprintf("%d %s", msg.Freq, msg.Trim)
	}
	if msg.Code != "" {
		return fmt.Sprintf("%d %s", msg.Freq, msg.Code)
	}
	return fmt.Sprintf("%d", msg.Freq)
}

[jucky154]

import numpy as np
from pydub import AudioSegment

def generate_morse_wave(text, fs=44100, freq=600, wpm=20, amplitude=1.0):
    """モールス信号の波形生成（各音ごとにランダム位相）"""
    MORSE_CODE = {
        'J': '.---', 'A': '.-', '1': '.----', 'Z': '--..', 'L': '.-..', 'O': '---',
        'Y': '-.--', 'W': '.--', 'X': '-..-'
    }
    dot_duration = 1.2 / wpm
    signal = np.array([])

    for char in text.upper():
        if char not in MORSE_CODE: continue
        for symbol in MORSE_CODE[char]:
            duration = dot_duration if symbol == '.' else dot_duration * 3
            t = np.linspace(0, duration, int(fs * duration), endpoint=False)
            # 指定された条件：短音/長音ごとにランダムな初期位相
            phase = np.random.uniform(0, 2 * np.pi)
            wave = amplitude * np.sin(2 * np.pi * freq * t + phase)
            signal = np.concatenate([signal, wave, np.zeros(int(fs * dot_duration))])
        signal = np.concatenate([signal, np.zeros(int(fs * dot_duration * 2))])
    return signal

def apply_fading(signal, fs, speed=0.5, depth=0.4):
    """信号にフェージング（QSB）を付加"""
    t = np.linspace(0, len(signal)/fs, len(signal), endpoint=False)
    # 振幅を (1-depth) ~ 1.0 の間で変動させる
    fading_envelope = (1 - depth/2) + (depth/2) * np.sin(2 * np.pi * speed * t)
    return signal * fading_envelope

def add_white_noise(signal, snr_db):
    """指定されたS/N比(dB)に基づいてホワイトノイズを付加"""
    # 信号の平均電力を計算
    sig_avg_watts = np.mean(signal**2)
    sig_avg_db = 10 * np.log10(sig_avg_watts)
    
    # 指定SNRからノイズの電力を計算
    noise_avg_db = sig_avg_db - snr_db
    noise_avg_watts = 10 ** (noise_avg_db / 10)
    
    # ホワイトノイズ生成
    white_noise = np.random.normal(0, np.sqrt(noise_avg_watts), len(signal))
    return signal + white_noise

def add_impulse_noise(signal, fs, intensity=0.5, probability=0.0005):
    """雷のようなバリバリ音（空電ノイズ）を付加"""
    # 確率的にスパイク（非常に短い強ノイズ）を発生させる
    noise_mask = np.random.choice([0, 1], size=len(signal), p=[1-probability, probability])
    impulse = noise_mask * np.random.normal(0, intensity, len(signal))
    return signal + impulse

# --- メイン設定 ---
fs = 44100
SNR_DB = 3            # ホワイトノイズのS/N比 (dB)
QRM_AMPLITUDE = 0.3    # 混信局(JA1YWX)の相対強度 (0.0〜1.0)

# 1. 目的信号 (JA1ZLO / 600Hz) の生成
sig_target = generate_morse_wave("JA1ZLO", fs, freq=600, wpm=22)
sig_target = apply_fading(sig_target, fs, speed=0.4) # フェージング適用

# 2. 混信信号 (JA1YWX / 500Hz) の生成
sig_interfere = generate_morse_wave("JA1YWX", fs, freq=500, wpm=18, amplitude=QRM_AMPLITUDE)

# 長さを合わせる
max_len = max(len(sig_target), len(sig_interfere))
sig_target = np.pad(sig_target, (0, max_len - len(sig_target)))
sig_interfere = np.pad(sig_interfere, (0, max_len - len(sig_interfere)))

# 3. 信号の合成とノイズ付加
# まず信号同士を混ぜる
mixed_signal = sig_target + sig_interfere

# ホワイトノイズをS/N比に基づいて付加（基準はsig_target）
final_signal = add_white_noise(mixed_signal, SNR_DB)

# 空電ノイズ（バリバリ）を付加
final_signal = add_impulse_noise(final_signal, fs, intensity=0.6)

# 4. MP3保存
# クリッピング（音割れ）防止のため正規化
max_val = np.max(np.abs(final_signal))
if max_val > 0:
    final_signal = (final_signal / max_val) * 0.9

int_signal = (final_signal * 32767).astype(np.int16)
audio = AudioSegment(int_signal.tobytes(), frame_rate=fs, sample_width=2, channels=1)
audio.export("realistic_morse.mp3", format="mp3", bitrate="192k")

print(f"完了: SNR={SNR_DB}dB で保存しました。")

これまで一応は、ノイズのないテストデータでテストするみたいなことが内部で行われていたり、実際のコンテストの音声とかを試しに回していましたが、もう少し課題点を通ったのかをテストするために、下のような信号を用意しました
つまり、JA1ZLOとJA1YWXが500と600Hzに存在し、ZLOはフェージングし、S/N比が-5, 1, 3[dB]くらいで、空電ノイズも入っています。これらが実際にS/N比としてそんなものか怪しい気がしていますが、数字の上では、マイナスをとれるのはすごいので、この辺をパスさせたいですね…

realistic_morse_m5db.mp3
realistic_morse_3db.mp3
realistic_morse_1db.mp3

[jucky154]

ちなみに今回のコードで対策されているという納豆モールスも作りました

natto_morse_experiment.mp3
ちょっと納豆ぽくはない？（sin波で長さが変わるというより？　納豆ってもう少し癖がある気がするので）

import numpy as np
from pydub import AudioSegment

def generate_morse_wave(text, fs=44100, freq=600, wpm=20, amplitude=1.0, natto_mode=False):
    MORSE_CODE = {
        'J': '.---', 'A': '.-', '1': '.----', 'Z': '--..', 'L': '.-..', 'O': '---',
        'Y': '-.--', 'W': '.--', 'X': '-..-'
    }
    dot_duration = 1.2 / wpm
    signal = np.array([])
    
    # 納豆モード用の時間軸
    elapsed_time = 0

    for char in text.upper():
        if char not in MORSE_CODE: continue
        for symbol in MORSE_CODE[char]:
            # 比率の計算 (デフォルト 1:3)
            dash_ratio = 3.0
            if natto_mode:
                # 時間経過とともに 3.5倍 〜 5.5倍 の間で「粘り」が変動する
                dash_ratio = 4.5 + 1.0 * np.sin(2 * np.pi * 0.1 * elapsed_time)
            
            duration = dot_duration if symbol == '.' else dot_duration * dash_ratio
            
            t = np.linspace(0, duration, int(fs * duration), endpoint=False)
            phase = np.random.uniform(0, 2 * np.pi)
            wave = amplitude * np.sin(2 * np.pi * freq * t + phase)
            
            signal = np.concatenate([signal, wave, np.zeros(int(fs * dot_duration))])
            elapsed_time += (duration + dot_duration)
            
        signal = np.concatenate([signal, np.zeros(int(fs * dot_duration * 2))])
        elapsed_time += (dot_duration * 2)
        
    return signal

def add_white_noise(signal, snr_db):
    sig_avg_watts = np.mean(signal**2)
    sig_avg_db = 10 * np.log10(sig_avg_watts)
    noise_avg_db = sig_avg_db - snr_db
    noise_avg_watts = 10 ** (noise_avg_db / 10)
    white_noise = np.random.normal(0, np.sqrt(noise_avg_watts), len(signal))
    return signal + white_noise

# --- メイン設定 ---
FS = 44100
SNR_DB = 10         # S/N比
NATTO_ON = True     # ここをTrueにすると「納豆モールス」化

# 1. 信号生成
# 目的局 (JA1ZLO) - 納豆モード設定
sig_target = generate_morse_wave("JA1ZLO", FS, freq=600, wpm=20, natto_mode=NATTO_ON)

# 2. 混信局 (JA1YWX) - 標準的な叩き方
sig_interfere = generate_morse_wave("JA1YWX", FS, freq=500, wpm=18, amplitude=0.4, natto_mode=False)

# 長さ合わせ
max_len = max(len(sig_target), len(sig_interfere))
sig_target = np.pad(sig_target, (0, max_len - len(sig_target)))
sig_interfere = np.pad(sig_interfere, (0, max_len - len(sig_interfere)))

# 3. 合成とノイズ
combined = sig_target + sig_interfere
final_signal = add_white_noise(combined, SNR_DB)

# 4. MP3書き出し
max_val = np.max(np.abs(final_signal))
int_signal = (final_signal / max_val * 0.9 * 32767).astype(np.int16)
audio = AudioSegment(int_signal.tobytes(), frame_rate=FS, sample_width=2, channels=1)
audio.export("natto_morse_experiment.mp3", format="mp3")

print(f"生成完了: NattoMode={'ON' if NATTO_ON else 'OFF'}, SNR={SNR_DB}dB")

[JG1VPP]

問題意識

広帯域一括処理の限界

複数信号の分離、誤表示の抑制、リアルタイム性の改善が課題である。現状の方式は、広帯域の STFT 結果からその時点で強い周波数 bin を選び、その後に ON/OFF 判定と長短判定を進める構造である。この構造では、前段で周波数選択を誤ると、その誤差が後段まで伝播しやすい。特にパイルアップ環境では、強い局に引き寄せられたり、弱い局を見失ったりしやすく、広帯域を一括で hard に扱うこと自体が制約になっている。

bin 単位判定の不安定さ

1 bin をそのまま信号の代表として扱う方式では、同じ信号であっても時間ごとにピーク bin が揺れるため、安定した追跡が難しい。ある瞬間には 31 番目の bin が最大であっても、次の瞬間には 32 番目の bin が最大になることは自然に起こる。これは周波数ドリフト、窓関数による漏れ、雑音、隣接信号の影響によって生じるものであり、bin 番号そのものを信号 ID とみなす設計では不安定になりやすい。

hard decision の早すぎる確定

現状の方式では、周波数選択、ON/OFF 判定、短点と長点の分類が比較的早い段階で hard に決定される。このため、最初のしきい値判定やクラスタリングが崩れた時点で、後段の記号列全体が崩れやすい。低 SNR や混信環境では、まず観測そのものが不確かであるため、初期段階で 0/1 や dot/dash に落とし込む設計は不利である。

周波数判定と復号の分離不足

実運用では、符号だけでなく、どの周波数に信号が存在するかもリアルタイムに判定しなければならない。しかもその周波数は固定ではなく、わずかに揺れたり、近接した複数信号が同時に存在したりする。したがって、周波数判定を先に固定し、その結果だけを使って復号する方式ではなく、周波数と ON/OFF の両方を時間方向の連続性の中で扱う必要がある。

単一トーン前提からの脱却

単一の狭帯域信号を読むだけであれば、狭帯域抽出と包絡処理でかなり安定に読める場合がある。しかし、実際の用途は複数周波数の信号が同時に重なる環境であるため、単一チャネル専用のデコーダだけでは不十分である。必要なのは、単一チャネルで安定に動作する復号器を基礎部品とし、それを多周波数へ並列化し、候補信号ごとに独立に扱える構造である。

リアルタイム表示の信頼性

リアルタイム性についても、未確定の内容を先に表示して見かけ上の速さを得るのではなく、確定した内容だけを表示する方向へ改める必要がある。低品質な観測に基づく早すぎる表示は、結果的に訂正や揺れを増やし、運用上の信頼性を下げる。リアルタイム性は単なる速さではなく、確定した内容を遅れを抑えて出すこととして再定義すべきである。

全体としての見直し

総じて、現状の課題は track 管理だけにあるのではなく、その前段の観測モデルと判定順序にある。広帯域から強い bin を選び、早い段階で hard decision を行う構造から、狭帯域チャネル群から soft な観測を作り、それを時間方向に統合して周波数と ON/OFF を同時に推定する構造へ移行する必要がある。改善の本質は、強い箇所をその都度拾う方式から、複数の信号候補を連続値として追跡し、確率的に復号する方式へ変えることにある。

解決方法

狭帯域チャネル群への分解

広帯域信号をそのまま 1 回で判定するのではなく、まず受信帯域を複数の狭帯域チャネルへ分解する。各チャネルは単一周波数近傍の信号成分を表し、周波数候補ごとのエネルギー、局所雑音床、SNR 風スコアをフレームごとに算出する。この段階では 0/1 判定を行わず、観測値を連続量として保持する。

peak bin ではなく連続周波数の推定

各フレームで最大 bin をそのまま採用するのではなく、局所ピーク近傍の複数 bin から重心周波数を推定する。これにより、同一信号が隣接する bin 間を揺れる場合でも、離散的な bin 番号ではなく連続値の周波数として扱えるようにする。周波数は観測値であり、内部状態としては滑らかに更新される連続量として保持する。

ridge 候補の追跡

各フレームで得られた周波数候補を独立に処理するのではなく、時間周波数平面上で連続する ridge として追跡する。周波数が急に飛びすぎないこと、一定時間継続して存在すること、観測スコアが高いことを評価に用い、複数の候補経路を並行して維持する。これにより、近接信号や一時的なノイズでピーク bin が揺れても、同一信号として経路を維持しやすくする。

soft score による観測表現

各チャネルまたは各 ridge に対して、frame ごとの tone likelihood を計算する。ここでは hard な ON/OFF 判定を行わず、その frame が CW トーンを含むらしさを連続値として持つ。これにより、観測の不確かさを後段へ持ち越し、早すぎる確定を避ける構造とする。

時系列平滑化の導入

tone likelihood の列に対して、2 状態または 3 状態の時系列モデルを適用し、最も自然な ON/OFF 列を復元する。実装手段としては HMM と Viterbi を有力候補とするが、本質は soft な観測を時間方向に統合して hard decision を後ろへ送ることにある。これにより、局所的なノイズや短時間の観測崩れに対して頑健な ON/OFF 推定を行う。

記号長推定の分離

ON/OFF 列が得られたあとで、ON 区間長と OFF 区間長から短点、長点、文字間、語間を判定する。この判定では固定しきい値を用いるのではなく、推定ユニット長 T を逐次更新し、1T、3T、7T に対する尤度で決定する。これにより、送信者ごとの速度差や途中の速度変化に自然に追従できるようにする。

多候補同時復号

単一の最良周波数だけを選んで復号するのではなく、複数の ridge 候補を並列に維持し、それぞれについて復号を進める。これにより、パイルアップ時でも複数局を同時に読み進めることができる。重い全探索を避けるため、各フレームでは有力候補のみを beam 的に保持し、計算量を制御する構成とする。

確定表示への変換

復号結果は未確定状態と確定済み状態を分離して保持し、文字間または語間が十分に確定した時点でのみ UI に反映する。これにより、未確定の記号列を先に表示して揺らすのではなく、確定した内容だけをリアルタイムに出力する。リアルタイム性は維持しつつ、表示の信頼性を優先する構成とする。

段階的な導入

実装は一度に全面置換するのではなく、まず単一チャネルで soft score と時系列平滑化の有効性を確認し、その後に多チャネル化と ridge tracking を導入する。これにより、前段特徴、時系列復号、多信号追跡の各要素を切り分けて検証できる。最終的には、広帯域から強い bin を 1 本選ぶ方式ではなく、複数の周波数候補を連続的に追跡しながら soft に復号する構造へ移行する。

[JG1VPP]

実装

core/scanner.go

package core

import "math"

func DefaultScanner() Scanner {
	return Scanner{
		StayOn:     0.985,
		StayOff:    0.985,
		MinScore:   0.15,
		FloorAlpha: 0.995,
		ScoreAlpha: 0.80,
	}
}

type Scanner struct {
	StayOn     float64
	StayOff    float64
	MinScore   float64
	FloorAlpha float64
	ScoreAlpha float64
}

func (s *Scanner) Scan(source Message) Message {
	score := s.ToneScore(source.Data)
	flags := s.Viterbi(score)
	flags = Debounce(flags, 2)

	segments := BoolSegments(flags)
	unit := EstimateUnitRobust(segments)
	code := DecodeSegments(segments, unit)
	trim := TrimCode(code)

	return Message{
		Freq: source.Freq,
		Time: source.Time,
		Data: source.Data,
		Code: code,
		Trim: trim,
		Text: CodeToText(trim),
	}
}

func (s *Scanner) ToneScore(x []float64) []float64 {
	if len(x) == 0 {
		return nil
	}

	floor := SmoothMinLike(x, s.FloorAlpha)
	score := make([]float64, len(x))

	for i := range x {
		base := floor[i] + 1e-9
		ratio := x[i] / base
		score[i] = math.Log(ratio)

		if score[i] < -4 {
			score[i] = -4
		}
		if score[i] > 4 {
			score[i] = 4
		}
	}

	score = Smooth(score, s.ScoreAlpha)

	for i := range score {
		score[i] -= s.MinScore
	}

	return score
}

func Smooth(x []float64, alpha float64) []float64 {
	if len(x) == 0 {
		return nil
	}

	y := make([]float64, len(x))
	y[0] = x[0]

	for i := 1; i < len(x); i++ {
		y[i] = alpha*y[i-1] + (1-alpha)*x[i]
	}

	return y
}

func SmoothMinLike(x []float64, alpha float64) []float64 {
	if len(x) == 0 {
		return nil
	}

	y := make([]float64, len(x))
	y[0] = x[0]

	for i := 1; i < len(x); i++ {
		if x[i] < y[i-1] {
			y[i] = x[i]
		} else {
			y[i] = alpha*y[i-1] + (1-alpha)*x[i]
		}
	}

	return y
}

func (s *Scanner) Viterbi(score []float64) []bool {
	n := len(score)
	if n == 0 {
		return nil
	}

	dpOn := make([]float64, n)
	dpOff := make([]float64, n)
	prOn := make([]bool, n)
	prOff := make([]bool, n)

	p0 := sigmoid(score[0])
	dpOn[0] = logp(p0)
	dpOff[0] = logp(1 - p0)

	for t := 1; t < n; t++ {
		pOn := sigmoid(score[t])
		pOff := 1 - pOn

		onOn := dpOn[t-1] + logp(s.StayOn)
		offOn := dpOff[t-1] + logp(1-s.StayOff)
		if onOn >= offOn {
			dpOn[t] = onOn + logp(pOn)
			prOn[t] = true
		} else {
			dpOn[t] = offOn + logp(pOn)
			prOn[t] = false
		}

		offOff := dpOff[t-1] + logp(s.StayOff)
		onOff := dpOn[t-1] + logp(1-s.StayOn)
		if offOff >= onOff {
			dpOff[t] = offOff + logp(pOff)
			prOff[t] = false
		} else {
			dpOff[t] = onOff + logp(pOff)
			prOff[t] = true
		}
	}

	state := dpOn[n-1] > dpOff[n-1]
	out := make([]bool, n)
	out[n-1] = state

	for t := n - 1; t >= 1; t-- {
		if state {
			state = prOn[t]
		} else {
			state = prOff[t]
		}
		out[t-1] = state
	}

	return out
}

func Debounce(x []bool, minWidth int) []bool {
	if len(x) == 0 || minWidth <= 1 {
		return x
	}

	out := make([]bool, len(x))
	copy(out, x)

	segments := BoolSegments(out)
	for _, seg := range segments {
		if seg.Width >= minWidth {
			continue
		}

		fill := false
		if seg.Since > 0 && seg.Until < len(out) {
			if out[seg.Since-1] == out[seg.Until] {
				fill = out[seg.Since-1]
			} else {
				fill = false
			}
		} else if seg.Since > 0 {
			fill = out[seg.Since-1]
		} else if seg.Until < len(out) {
			fill = out[seg.Until]
		}

		for i := seg.Since; i < seg.Until; i++ {
			out[i] = fill
		}
	}

	return out
}

func sigmoid(x float64) float64 {
	if x > 40 {
		return 1
	}
	if x < -40 {
		return 0
	}
	return 1 / (1 + math.Exp(-x))
}

func logp(x float64) float64 {
	if x < 1e-12 {
		x = 1e-12
	}
	return math.Log(x)
}

core/segment.go

package core

import (
	"sort"
	"strings"
)

type Segment struct {
	On    bool
	Since int
	Until int
	Width int
}

func BoolSegments(x []bool) []Segment {
	if len(x) == 0 {
		return nil
	}

	var out []Segment
	since := 0
	prev := x[0]

	for i := 1; i < len(x); i++ {
		if x[i] == prev {
			continue
		}
		out = append(out, Segment{
			On:    prev,
			Since: since,
			Until: i,
			Width: i - since,
		})
		since = i
		prev = x[i]
	}

	out = append(out, Segment{
		On:    prev,
		Since: since,
		Until: len(x),
		Width: len(x) - since,
	})

	return out
}

func EstimateUnitRobust(segments []Segment) float64 {
	var on []float64

	for _, s := range segments {
		if s.On && s.Width > 0 {
			on = append(on, float64(s.Width))
		}
	}

	if len(on) == 0 {
		return 1
	}
	if len(on) == 1 {
		return on[0]
	}

	sort.Float64s(on)

	n := len(on)
	shorts := on[:maxInt(1, (n+1)/2)]

	unit := med64(shorts)
	if unit <= 0 {
		unit = med64(on)
	}
	if unit <= 0 {
		unit = 1
	}

	return unit
}

func DecodeSegments(segments []Segment, unit float64) string {
	if unit <= 0 {
		unit = 1
	}

	var code string

	for _, s := range segments {
		w := float64(s.Width)

		if s.On {
			switch {
			case w < 1.8*unit:
				code += "."
			default:
				code += "_"
			}
			continue
		}

		switch {
		case w < 1.5*unit:
		case w < 5.0*unit:
			code += " "
		default:
			code += " ; "
		}
	}

	return code
}

func TrimCode(code string) string {
	return strings.TrimRight(code, "._")
}

func maxInt(a, b int) int {
	if a > b {
		return a
	}
	return b
}

core/decoder.go

package core

import "math"

type Decoder struct {
	Scatter int
	MaxN    int
	MaxMiss int
	Squelch float64
	Monitor Monitor
	Scanner Scanner
	Program func(Message) Message

	Tracks []*Track
	NextID int
}

type Candidate struct {
	Bin    int
	Freq   float64
	Power  float64
	Series []float64
}

func DefaultDecoder(rate int) Decoder {
	return Decoder{
		Scatter: 2,
		MaxN:    8,
		MaxMiss: 3,
		Squelch: 0,
		Monitor: DefaultMonitor(rate),
		Scanner: DefaultScanner(),
	}
}

func (d *Decoder) Read(signal []float64) (result []Message) {
	data := d.Monitor.Next(signal)
	if len(data) == 0 {
		return nil
	}

	cands := d.Candidates(data)
	d.UpdateTracks(cands)

	for _, tr := range d.Tracks {
		if tr.Miss > d.MaxMiss {
			continue
		}

		msg := tr.Message(d.Scanner)
		if d.Program != nil {
			msg = d.Program(msg)
		}
		if msg.Text != "" || msg.Trim != "" || msg.Code != "" {
			result = append(result, msg)
		}
	}

	d.PruneTracks()
	return
}

func (d *Decoder) Candidates(data []Message) []Candidate {
	if len(data) == 0 {
		return nil
	}

	pows := make([]float64, len(data))
	for i, msg := range data {
		pows[i] = pow64(msg.Data)
	}

	var peaks []int
	for i := 1; i < len(pows)-1; i++ {
		if pows[i] <= d.Squelch {
			continue
		}
		if pows[i] >= pows[i-1] && pows[i] >= pows[i+1] {
			peaks = append(peaks, i)
		}
	}

	peaks = d.UniquePeaks(peaks, pows)

	var out []Candidate
	for _, i := range peaks {
		series := d.MergeBins(data, i)
		freq := d.EstimateFreq(data, pows, i)

		out = append(out, Candidate{
			Bin:    i,
			Freq:   freq,
			Power:  pows[i],
			Series: series,
		})
	}

	if len(out) > d.MaxN {
		out = topCandidates(out, d.MaxN)
	}

	return out
}

func (d *Decoder) UniquePeaks(peaks []int, pows []float64) []int {
	if len(peaks) == 0 {
		return nil
	}

	var out []int
	for _, i := range peaks {
		keep := true
		for j, k := range out {
			if absInt(i-k) <= d.Scatter {
				if pows[i] > pows[k] {
					out[j] = i
				}
				keep = false
				break
			}
		}
		if keep {
			out = append(out, i)
		}
	}
	return out
}

func (d *Decoder) MergeBins(data []Message, center int) []float64 {
	n := len(data[center].Data)
	out := make([]float64, n)

	for k := -d.Scatter; k <= d.Scatter; k++ {
		i := center + k
		if i < 0 || i >= len(data) {
			continue
		}

		w := 1.0 / (1.0 + math.Abs(float64(k)))
		for t, v := range data[i].Data {
			out[t] += w * v
		}
	}

	return out
}

func (d *Decoder) EstimateFreq(data []Message, pows []float64, center int) float64 {
	var num float64
	var den float64

	for k := -d.Scatter; k <= d.Scatter; k++ {
		i := center + k
		if i < 0 || i >= len(pows) {
			continue
		}

		w := math.Max(pows[i], 0)
		num += float64(data[i].Freq) * w
		den += w
	}

	if den == 0 {
		return float64(data[center].Freq)
	}
	return num / den
}

func (d *Decoder) UpdateTracks(cands []Candidate) {
	now := d.Monitor.Time
	used := make([]bool, len(d.Tracks))

	for _, cand := range cands {
		best := -1
		bestCost := math.MaxFloat64

		for i, tr := range d.Tracks {
			if used[i] {
				continue
			}

			df := math.Abs(tr.Freq - cand.Freq)
			if df > d.AssignRadiusHz(tr) {
				continue
			}

			cost := d.TrackCost(tr, cand)
			if cost < bestCost {
				best = i
				bestCost = cost
			}
		}

		if best >= 0 {
			d.Tracks[best].Update(cand.Freq, cand.Power, now, cand.Series)
			used[best] = true
			continue
		}

		d.Tracks = append(d.Tracks, NewTrack(d.NextID, cand.Freq, cand.Power, now, cand.Series))
		d.NextID++
		used = append(used, true)
	}

	for i, tr := range d.Tracks {
		if !used[i] {
			tr.Decay()
		}
	}
}

func (d *Decoder) TrackCost(tr *Track, cand Candidate) float64 {
	df := math.Abs(tr.Freq - cand.Freq)
	ds := math.Abs(tr.Score - cand.Power)

	ageBonus := 1.0 / (1.0 + 0.2*float64(tr.Age))
	missPenalty := 1.0 + 0.5*float64(tr.Miss)

	return (df + 0.05*ds) * ageBonus * missPenalty
}

func (d *Decoder) AssignRadiusHz(tr *Track) float64 {
	base := 30.0 + 10.0*float64(d.Scatter)

	if tr.Age > 10 {
		base += 10
	}
	if tr.Miss > 0 {
		base -= 5 * float64(tr.Miss)
	}
	if base < 15 {
		base = 15
	}
	return base
}

func (d *Decoder) PruneTracks() {
	var keep []*Track
	for _, tr := range d.Tracks {
		if tr.Miss > d.MaxMiss {
			continue
		}
		if tr.Score < d.Squelch*0.5 && tr.Age < 2 {
			continue
		}
		keep = append(keep, tr)
	}
	d.Tracks = keep
}

func topCandidates(in []Candidate, n int) []Candidate {
	if len(in) <= n {
		return in
	}

	out := make([]Candidate, 0, n)
	for _, cand := range in {
		if len(out) < n {
			out = append(out, cand)
			continue
		}

		minIdx := 0
		for i := 1; i < len(out); i++ {
			if out[i].Power < out[minIdx].Power {
				minIdx = i
			}
		}

		if cand.Power > out[minIdx].Power {
			out[minIdx] = cand
		}
	}

	return out
}

func absInt(x int) int {
	if x < 0 {
		return -x
	}
	return x
}

core/monitor.go

package core

import (
	"math/cmplx"

	"github.com/r9y9/gossp/stft"
)

func DefaultMonitor(rate int) Monitor {
	return Monitor{
		Bass:   100,
		Treble: 2000,
		Rate:   rate,
		STFT:   *stft.New(rate/100, 2048),
	}
}

type Monitor struct {
	Bass   int
	Treble int
	Rate   int
	Time   int
	STFT   stft.STFT
	prev   []float64
}

func (m *Monitor) Next(signal []float64) (result []Message) {
	bassBin := int(m.Bass * m.STFT.FrameLen / m.Rate)
	trebleBin := int(m.Treble * m.STFT.FrameLen / m.Rate)

	specs := m.STFT.STFT(append(m.prev, signal...))
	size := len(signal) / m.STFT.FrameShift
	if len(specs) < size {
		size = len(specs)
	}
	if size <= 0 || bassBin >= trebleBin {
		m.prev = signal
		return nil
	}

	result = make([]Message, trebleBin-bassBin)
	for bin := bassBin; bin < trebleBin; bin++ {
		freqHz := int(float64(bin*m.Rate)/float64(m.STFT.FrameLen) + 0.5)
		result[bin-bassBin] = Message{
			Data: make([]float64, size),
			Freq: freqHz,
			Time: m.Time,
		}
	}

	for t, spec := range specs[len(specs)-size:] {
		for bin := bassBin; bin < trebleBin; bin++ {
			result[bin-bassBin].Data[t] = cmplx.Abs(spec[bin])
		}
	}

	m.prev = signal
	m.Time += size
	return
}

core/track.go

package core

type Track struct {
	ID int

	Freq  float64
	Score float64

	Age  int
	Miss int
	Time int

	Window []float64
	Limit  int

	Code string
	Trim string
	Text string
}

func NewTrack(id int, freq float64, score float64, now int, data []float64) *Track {
	tr := &Track{
		ID:     id,
		Freq:   freq,
		Score:  score,
		Age:    1,
		Miss:   0,
		Time:   now,
		Limit:  4096,
		Window: nil,
		Code:   "",
		Trim:   "",
		Text:   "",
	}
	tr.Push(data)
	return tr
}

func (tr *Track) Update(freq float64, score float64, now int, data []float64) {
	tr.Freq = 0.8*tr.Freq + 0.2*freq
	tr.Score = 0.8*tr.Score + 0.2*score
	tr.Age++
	tr.Miss = 0
	tr.Time = now

	tr.Push(data)
}

func (tr *Track) Push(data []float64) {
	if len(data) == 0 {
		return
	}

	tr.Window = append(tr.Window, data...)

	if tr.Limit > 0 && len(tr.Window) > tr.Limit {
		drop := len(tr.Window) - tr.Limit
		tr.Window = tr.Window[drop:]

		tr.Code = ""
		tr.Trim = ""
		tr.Text = ""
	}
}

func (tr *Track) Decay() {
	tr.Miss++
	if tr.Score > 0 {
		tr.Score *= 0.95
	}
}

func (tr *Track) Message(scanner Scanner) Message {
	full := Message{
		Freq: int(tr.Freq + 0.5),
		Time: tr.Time,
		Data: tr.Window,
	}

	full = scanner.Scan(full)

	oldTrim := tr.Trim
	oldText := tr.Text

	tr.Code = full.Code
	tr.Trim = full.Trim
	tr.Text = full.Text

	msg := Message{
		Freq: full.Freq,
		Time: full.Time,
	}

	pTrim := commonPrefixLen(oldTrim, full.Trim)
	pText := commonPrefixLen(oldText, full.Text)

	if pTrim < len(full.Trim) {
		msg.Trim = full.Trim[pTrim:]
	}
	if pText < len(full.Text) {
		msg.Text = full.Text[pText:]
	}

	msg.Code = msg.Trim
	return msg
}

func commonPrefixLen(a, b string) int {
	n := len(a)
	if len(b) < n {
		n = len(b)
	}
	i := 0
	for i < n && a[i] == b[i] {
		i++
	}
	return i
}

core/message.go

package core

import "math"

type Message struct {
	Data []float64
	Code string
	Trim string
	Text string
	Freq int
	Time int
	Miss int
	Hide bool
}

func (m Message) AGC(gain float64) []float64 {
	if len(m.Data) == 0 {
		return nil
	}

	out := make([]float64, len(m.Data))
	maxv := max64(m.Data)
	if maxv <= 0 {
		return out
	}

	for i, v := range m.Data {
		if v <= 0 {
			continue
		}
		scale := gain
		if maxv/v < scale {
			scale = maxv / v
		}
		out[i] = v * scale
	}

	return out
}

func (m Message) Merge(next Message) Message {
	out := Message{
		Freq: m.Freq,
		Time: m.Time,
		Data: make([]float64, 0, len(m.Data)+len(next.Data)),
	}
	out.Data = append(out.Data, m.Data...)
	out.Data = append(out.Data, next.Data...)
	return out
}

func max64(x []float64) float64 {
	if len(x) == 0 {
		return 0
	}
	m := x[0]
	for _, v := range x[1:] {
		if v > m {
			m = v
		}
	}
	return m
}

func pow64(x []float64) float64 {
	if len(x) == 0 {
		return 0
	}
	var s float64
	for _, v := range x {
		s += v * v
	}
	return math.Sqrt(s / float64(len(x)))
}

func med64(x []float64) float64 {
	n := len(x)
	if n == 0 {
		return 0
	}

	cp := make([]float64, n)
	copy(cp, x)

	for i := 1; i < n; i++ {
		v := cp[i]
		j := i - 1
		for j >= 0 && cp[j] > v {
			cp[j+1] = cp[j]
			j--
		}
		cp[j+1] = v
	}

	m := n / 2
	if n%2 == 1 {
		return cp[m]
	}
	return 0.5 * (cp[m-1] + cp[m])
}

core/history.go

package core

type History []Message

func (h *History) Clear() {
	*h = nil
}

func (h *History) Add(messages ...Message) {
	if len(messages) == 0 {
		return
	}
	*h = append(*h, messages...)
}

func (h History) Length() int {
	return len(h)
}

func (h History) Active(maxMiss int) (out []Message) {
	for _, msg := range h {
		if msg.Miss <= maxMiss && !msg.Hide {
			out = append(out, msg)
		}
	}
	return
}

view/decoder.go

package view

import "github.com/nextzlog/cw4i/core"

type Decoder struct {
	Initial func(rate int) (core.Decoder, error)

	core.Decoder
	History
}

func (d *Decoder) Update(rate int) {
	decoder, err := d.Initial(rate)
	if err != nil {
		return
	}
	d.Decoder = decoder
	d.History.Clear()
}

func (d *Decoder) Restart(rate int) {
	d.Update(rate)
}

func (d *Decoder) Resquelch(value float64) {
	d.Decoder.Squelch = value
	d.Decoder.Tracks = nil
	d.Decoder.NextID = 0
	d.History.Clear()
}

func (d *Decoder) Decode(signal []float64) {
	messages := d.Decoder.Read(signal)
	d.History.Add(messages...)
}

view/history.go

package view

import (
	"fmt"

	"fyne.io/fyne/v2"
	"fyne.io/fyne/v2/widget"

	"github.com/nextzlog/cw4i/core"
)

type History struct {
	Items []core.Message
	Added int
}

func (h *History) Clear() {
	h.Items = nil
	h.Added = 0
}

func (h *History) Add(messages ...core.Message) {
	if len(messages) == 0 {
		return
	}
	h.Items = append(h.Items, messages...)
	h.Added += len(messages)
}

func (h *History) Length() int {
	return len(h.Items)
}

func (h *History) CanvasObject() fyne.CanvasObject {
	list := widget.NewList(
		func() int {
			return h.Length()
		},
		func() fyne.CanvasObject {
			return widget.NewLabel("")
		},
		func(id widget.ListItemID, object fyne.CanvasObject) {
			label := object.(*widget.Label)
			msg := h.Items[id]
			label.SetText(formatMessage(msg))
		},
	)
	list.Refresh()
	return list
}

func (h *History) Refresh(list *widget.List) {
	if h.Added == 0 {
		return
	}
	list.Refresh()
	h.Added = 0
}

func formatMessage(msg core.Message) string {
	switch {
	case msg.Text != "":
		return fmt.Sprintf("%d %s", msg.Freq, msg.Text)
	case msg.Trim != "":
		return fmt.Sprintf("%d %s", msg.Freq, msg.Trim)
	case msg.Code != "":
		return fmt.Sprintf("%d %s", msg.Freq, msg.Code)
	default:
		return fmt.Sprintf("%d", msg.Freq)
	}
}

[JG1VPP]

問題意識

現在は各周波数 bin を絶対的な強さで評価しているため、ホワイトノイズや広帯域雑音、隣接信号の裾も候補に入りやすい。特に混信時には、本当に読みたい CW トーンは周囲より狭帯域に突出して現れる一方、ノイズは周辺 bin にも同程度に広がることが多い。その違いを見ていないため、復調前の段階で不要な bin を多く通してしまい、後段での誤判定や誤表示につながっている。

解決方法

各 bin の評価を絶対パワーだけでなく、周辺 bin に対する相対的な突出度でも行う。具体的には、対象 bin のパワーを近傍 bin の中央値で割った局所 SNR を導入し、これが一定以上のものだけを候補とする。さらに、その bin が近傍内で局所ピークであることも条件に加える。これにより、広がったノイズや強い信号の裾を落としやすくなり、狭帯域に立った CW トーンを前段で選びやすくする。

実装

core/decoder.go

/*******************************************************************************
 * Audio Morse Decoder 'CW4ISR' (forked from Project ZyLO since 2023 July 15th)
 * Released under the MIT License (or GPL v3 until 2021 Oct 28th) (see LICENSE)
 * Univ. Tokyo Amateur Radio Club Development Task Force (https://nextzlog.dev)
*******************************************************************************/

package core

import "sort"
import "github.com/thoas/go-funk"

type Decoder struct {
	Scatter  int
	MaxMiss  int
	Squelch  float64
	Monitor  Monitor
	Scanner  Scanner
	History  []Message
	Program  func(Message) Message
	SNRWidth int
	MinSNR   float64
}

func DefaultDecoder(rate int) Decoder {
	return Decoder{
		Scatter:  1,
		MaxMiss:  2,
		Monitor:  DefaultMonitor(rate),
		Scanner:  DefaultScanner(),
		SNRWidth: 3,
		MinSNR:   25.0,
	}
}

func (d *Decoder) uniq(list []int, freq int) (result []int) {
	for _, m := range d.History {
		lower := m.Freq >= freq-d.Scatter
		upper := m.Freq <= freq+d.Scatter
		if lower && upper {
			return append(list, m.Freq)
		}
	}
	return append(list, freq)
}

func (d *Decoder) Next(signal []float64) (result []Message) {
	data := d.Monitor.Next(signal)
	pows := make([]float64, len(data))
	for n, msg := range data {
		pows[n] = pow64(msg.Data)
	}

	snrs := localSNR(pows, d.SNRWidth)

	var list []int
	for _, n := range top64(pows) {
		if pows[n] > d.Squelch {
			if snrs[n] > d.MinSNR {
				if isLocalPeak(pows, n, d.SNRWidth) {
					list = d.uniq(list, n)
				}
			}
		}
	}
	for _, msg := range d.History {
		list = append(list, msg.Freq)
	}
	list = funk.UniqInt(list)
	for _, n := range list {
		result = append(result, d.Scanner.Scan(data[n]))
	}
	return
}

func (d *Decoder) Read(signal []float64) (result []Message) {
	for _, next := range d.Next(signal) {
		for _, prev := range d.History {
			if prev.Freq == next.Freq {
				next = d.Scanner.Scan(prev.Merge(next))
				if next.Code == prev.Code {
					next.Miss = prev.Miss + 1
				}
			}
		}
		if d.Program != nil {
			next = d.Program(next)
		}
		if next.Miss <= d.MaxMiss {
			result = append(result, next)
		}
	}
	d.History = result
	return
}

func localSNR(pows []float64, halfWidth int) []float64 {
	out := make([]float64, len(pows))
	for i := range pows {
		floor := localMedianNoise(pows, i, halfWidth)
		if floor <= 0 {
			out[i] = 0
			continue
		}
		out[i] = pows[i] / floor
	}
	return out
}

func localMedianNoise(pows []float64, center, halfWidth int) float64 {
	var xs []float64
	lo := max(0, center-halfWidth)
	hi := min(len(pows)-1, center+halfWidth)

	for i := lo; i <= hi; i++ {
		if i == center {
			continue
		}
		xs = append(xs, pows[i])
	}

	if len(xs) == 0 {
		return 0
	}
	sort.Float64s(xs)
	n := len(xs)
	if n%2 == 1 {
		return xs[n/2]
	}
	return 0.5 * (xs[n/2-1] + xs[n/2])
}

func isLocalPeak(pows []float64, center, halfWidth int) bool {
	lo := max(0, center-halfWidth)
	hi := min(len(pows)-1, center+halfWidth)
	v := pows[center]
	for i := lo; i <= hi; i++ {
		if i == center {
			continue
		}
		if pows[i] > v {
			return false
		}
	}
	return true
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}

func max(a, b int) int {
	if a > b {
		return a
	}
	return b
}

[jucky154]

ありがとうございます
realistic_morse_1db.mp3
のシグナル(ZLO)が1[dB]で入っている。（YWXはその0.4倍)　と
realistic_morse_m5db.mp3
のシグナル(ZLO)が-5[dB]で入っている。（YWXはその0.4倍)　
の音声で試しました

1[dB]が

-5[dB]が

となりました。どちらも、ZLO側は問題ないですが、YWX側ができてないです(あとJがうまくいっていない時も多い）
ZLO側の音声が終わりかけたときにYWX側が解読され始めるので、ZLOが強くいる時はYWX側が無視されてしまう傾向にありそうです

[jucky154]

あと綺麗なホワイトノイズ/綺麗な信号(JA1ZLOでホワイトノイズのみ)、-10[dB](自分のコード的には… 本当は分散比ではなく振幅比にするべきな気がしてきた）の信号も最初の立ち上がりの文字以外は解読できています

m10dB.mp3

[jucky154]

古いversionで同じ-10[dB]の音声を解読させてみました

スケルチを調整すれば解読できました

一方で、スケルチを開く方向（左に寄せる）とぐちゃぐちゃになりました

新しいversionはスケルチを左に寄せてもおかしな結果が表示されないので、そういう意味では解読がやりやすくなったように思います

[jucky154]

今回の調整は周波数選択を突出度にすることによって、スケルチが開き気味でも広帯域騒音/ホワイトノイズは削除されたので、シグナルの解読が綺麗に見えるようになった。その結果、低SN比でも解読が読みやすくなったという感じですかね（過去でも解読は行われていたんでしょうけど、一瞬でどこかにいっていたと考えられる）

ただし、頭の文字が解読されにくいという課題がある

[jucky154]

問題意識
色々とモールス解読のコードを試しているが、検証が実際のコンテスト音声だったりなので、自分たちが気にしている問題を解決できたかが明らかではない。よって、検証の音声・モデルで自分たちが好きなようにできる音声を作る必要がある。
解決方法
いかにフェージング・QRM・納豆・空電・低SN比の音声を作るコードを用意した

import numpy as np
from scipy.signal import butter, lfilter
from pydub import AudioSegment
from scipy.signal import butter, sosfilt 

def bandpass_filter(data, lowcut, highcut, fs, order=5):
    """
    数値的に安定したSOS(Second-Order Sections)形式のフィルタ
    lfilterの代わりにsosfiltを使用することで、発散を防ぎます
    """
    nyq = 0.5 * fs
    low = lowcut / nyq
    high = highcut / nyq
    # output='sos' を指定するのがポイントです
    sos = butter(order, [low, high], btype='band', output='sos')
    return sosfilt(sos, data)

def add_fullband_noise_with_band_limited_snr(signal, snr_db, peak_ref_signal, fa, fb, fs):
    """
    信号のピーク値と、帯域(fa-fb)内のノイズRMSに基づいてノイズを付加
    """
    sig_peak = np.max(np.abs(peak_ref_signal))
    # 安全のためsig_peakが0の場合の処理を追加
    if sig_peak == 0: sig_peak = 1e-6

    raw_noise = np.random.normal(0, 1.0, len(signal))
    
    # 安定化したフィルタを使用
    filtered_component = bandpass_filter(raw_noise, fa, fb, fs)
    
    # オーバーフロー対策：計算前に値が異常に大きくないかチェック
    band_limited_rms = np.sqrt(np.mean(np.square(filtered_component)))
    
    # ゼロ除算防止
    if band_limited_rms == 0: band_limited_rms = 1e-10

    target_band_rms = sig_peak / (10 ** (snr_db / 20))
    scaling_factor = target_band_rms / band_limited_rms
    final_fullband_noise = raw_noise * scaling_factor
    
    return signal + final_fullband_noise

def generate_morse_wave(text, fs=44100, freq=600, wpm=20, amplitude=1.0, 
                        natto_mode=False, natto_offset=1.5, natto_swing=1.0):
    """
    モールス信号生成
    natto_offset: 長音比率の底上げ
    natto_swing: 粘りの時間変動幅
    """
    MORSE_CODE = {
        'J': '.---', 'A': '.-', '1': '.----', 'Z': '--..', 'L': '.-..', 'O': '---',
        'Y': '-.--', 'W': '.--', 'X': '-..-'
    }
    dot_duration = 1.2 / wpm
    signal = np.array([])
    elapsed_time = 0

    for char in text.upper():
        if char not in MORSE_CODE: continue
        for symbol in MORSE_CODE[char]:
            dash_ratio = 3.0
            if natto_mode:
                # 粘りの中心値 + 変動幅
                center = 3.0 + natto_offset
                dash_ratio = center + natto_swing * np.sin(2 * np.pi * 0.1 * elapsed_time)
            
            duration = dot_duration if symbol == '.' else dot_duration * dash_ratio
            t = np.linspace(0, duration, int(fs * duration), endpoint=False)
            
            # リクエスト：各音ごとに初期位相をランダム化
            phase = np.random.uniform(0, 2 * np.pi)
            wave = amplitude * np.sin(2 * np.pi * freq * t + phase)
            
            signal = np.concatenate([signal, wave, np.zeros(int(fs * dot_duration))])
            elapsed_time += (duration + dot_duration)
            
        signal = np.concatenate([signal, np.zeros(int(fs * dot_duration * 2))])
        elapsed_time += (dot_duration * 2)
    return signal

def apply_fading(signal, fs, speed=0.5, depth=0.8):
    """信号にフェージング（QSB）を付加"""
    if depth == 0: return signal
    t = np.linspace(0, len(signal)/fs, len(signal), endpoint=False)
    # 振幅を (1-depth) から 1.0 の間で周期変動させる
    fading_envelope = (1.0 - depth) + depth * (0.5 * (1 + np.sin(2 * np.pi * speed * t)))
    return signal * fading_envelope

def add_impulse_noise(signal, intensity=0.5, probability=0.0005):
    """バリバリという空電ノイズの付加"""
    if intensity == 0: return signal
    noise_mask = np.random.choice([0, 1], size=len(signal), p=[1-probability, probability])
    impulse = noise_mask * np.random.normal(0, intensity, len(signal))
    return signal + impulse

# ==========================================
# --- 無線機設定パネル (Parameters) ---
# ==========================================
FS = 44100
FILENAME = "test_morse_15db.mp3"

# 1. 環境・ノイズ設定
SNR_DB_SPEC = 15       # 指定帯域内でのS/N比 (dB)
FA = 550              # SNR計算用の帯域下限 (Hz)
FB = 650              # SNR計算用の帯域上限 (Hz)
PRE_NOISE_SEC = 2.0   # 冒頭のノイズのみの秒数

# 2. 目的信号 (JA1ZLO)
TARGET_FREQ = 600
TARGET_WPM = 20

# 【納豆設定】
ENABLE_NATTO = False  # 納豆モードON
NATTO_OFFSET = 2.0    # 長音比率の底上げ (1:5中心)
NATTO_SWING = 1.5     # 粘りの揺れ幅 (1:3.5 ～ 1:6.5)

# 【フェージング設定】
ENABLE_FADING = False
FADING_DEPTH = 0.8    # 変動の深さ
FADING_SPEED = 0.2    # 変動の速さ (Hz)

# 3. 混信局 (JA1YWX)
ENABLE_QRM = False
QRM_FREQ = 500
QRM_WPM = 18
QRM_AMPLITUDE = 0.3

# 4. 突発ノイズ (バリバリ)
ENABLE_IMPULSE = True
IMPULSE_INTENSITY = 0.7

# ==========================================
# --- 生成・合成プロセス ---
# ==========================================

# 目的信号の生成
sig_target = generate_morse_wave(
    "JA1ZLO", FS, freq=TARGET_FREQ, wpm=TARGET_WPM, 
    natto_mode=ENABLE_NATTO, 
    natto_offset=NATTO_OFFSET, 
    natto_swing=NATTO_SWING
)

# 目的信号にフェージング適用
if ENABLE_FADING:
    sig_target = apply_fading(sig_target, FS, speed=FADING_SPEED, depth=FADING_DEPTH)

# 混信局の生成
if ENABLE_QRM:
    sig_interfere = generate_morse_wave("JA1YWX", FS, freq=QRM_FREQ, wpm=QRM_WPM, amplitude=QRM_AMPLITUDE)
else:
    sig_interfere = np.zeros(1)

# 配列長の調整
max_sig_len = max(len(sig_target), len(sig_interfere))
sig_target = np.pad(sig_target, (0, max_sig_len - len(sig_target)))
sig_interfere = np.pad(sig_interfere, (0, max_sig_len - len(sig_interfere)))

# 冒頭ノイズ用パディング（2秒間の無音配列を結合）
pre_padding = np.zeros(int(FS * PRE_NOISE_SEC))
combined_signal = np.concatenate([pre_padding, sig_target + sig_interfere])

# ホワイトノイズ付加 (全帯域に付加しつつ、強度は指定帯域SNRで決定)
# 基準となるピークはフェージング前の元の信号強度が望ましいが、ここでは現在のsig_targetを使用
final_signal = add_fullband_noise_with_band_limited_snr(
    combined_signal, 
    SNR_DB_SPEC, 
    sig_target, 
    FA, FB, FS
)

# 突発ノイズ付加
if ENABLE_IMPULSE:
    final_signal = add_impulse_noise(final_signal, intensity=IMPULSE_INTENSITY)

# 書き出し（クリッピング防止のための正規化）
max_val = np.max(np.abs(final_signal))
if max_val > 0:
    final_signal = (final_signal / max_val) * 0.9

int_signal = (final_signal * 32767).astype(np.int16)
AudioSegment(
    int_signal.tobytes(), 
    frame_rate=FS, 
    sample_width=2, 
    channels=1
).export(FILENAME, format="mp3", bitrate="192k")

print(f"--- 統合シミュレーション完了 ---")
print(f"File: {FILENAME}")
print(f"SNR Definition: Peak vs RMS in {FA}-{FB}Hz")
print(f"Settings: SNR={SNR_DB_SPEC}dB, Natto={'ON' if ENABLE_NATTO else 'OFF'}, QRM={'ON' if ENABLE_QRM else 'OFF'}")