SDR スクリプト

#!/usr/bin/env python3
"""
dxcc_notify.py
WSJT-X の UDP 出力を監視し、バンド別に未交信（または未コンファーム）の
DXCC エンティティがデコードされたらスマートフォンへプッシュ通知する。

構成: SDRconnect -> WSJT-X -> (UDPマルチキャスト) -> JTAlert
+-> 本スクリプト -> Gmail (主) / ntfy (予備) -> スマホ

必要なもの:
- cty.dat (https://www.country-files.com/ から最新版を取得)
- 交信ログの ADIF エクスポート (JTAlert/Log4OM/HAMLOG等から)
- 送信用Gmail (2段階認証を有効化し、アプリパスワードを発行)
- 予備にntfyを使う場合はスマホに ntfy アプリを入れトピックを購読

電離層の高さについて:
NICT (https://wdc.nict.go.jp/Ionosphere/) のイオノゾンデ自動読取値(15分値)から、
h'F2（求まらない時間帯はh'Fで代用）とfoF2を取得し、通知に添える(--iono-station、既定は国分寺)。
数値データそのものの利用であり、ページの見た目・図表の複製・再配布は行わない。
HF-STARTの伝搬経路・出力ガイドは、再配布に関する利用条件があるため未実装（要NICT確認）。
標準ライブラリのみで動作 (Python 3.8+)
"""
import argparse
import glob
import json
import os
import re
import socket
import struct
import time
import urllib.request
import urllib.error
import smtplib
import ssl
import threading
import queue
import select
import gzip
import calendar
from datetime import datetime, timezone, timedelta
from email.message import EmailMessage

MAGIC = 0xADBCCBDA

# ---------------------------------------------------------------- UDP 解析
class Reader:
"""Qt QDataStream (ビッグエンディアン) の簡易リーダ"""
def __init__(self, data):
self.d, self.p = data, 0

def _u(self, fmt):
v = struct.unpack_from(fmt, self.d, self.p)[0]
self.p += struct.calcsize(fmt)
return v

def u8(self): return self._u('>B')
def bool(self): return self._u('>?')
def u32(self): return self._u('>I')
def i32(self): return self._u('>i')
def u64(self): return self._u('>Q')
def f64(self): return self._u('>d')

def qstr(self):
n = self.u32()
if n == 0xFFFFFFFF:
return None
s = self.d[self.p:self.p + n].decode('utf-8', 'replace')
self.p += n
return s

BANDS = [ # (下限Hz, 上限Hz, 名称)
(1_800_000, 2_000_000, '160m'), (3_500_000, 4_000_000, '80m'),
(5_250_000, 5_450_000, '60m'), (7_000_000, 7_300_000, '40m'),
(10_100_000, 10_150_000, '30m'), (14_000_000, 14_350_000, '20m'),
(18_068_000, 18_168_000, '17m'), (21_000_000, 21_450_000, '15m'),
(24_890_000, 24_990_000, '12m'), (28_000_000, 29_700_000, '10m'),
(50_000_000, 54_000_000, '6m'),
]

def freq_to_band(hz):
for lo, hi, name in BANDS:
if lo <= hz <= hi:
return name
return None

# ---------------------------------------------------------------- DXCC 判定
class Cty:
"""cty.dat による最長一致プレフィックス判定。
'*' 付きエンティティ (WAE専用, DXCC外) は除外するので、
例えば IT9 は親の Italy に落ちる。"""
STRIP = re.compile(r'$.*?$|\[.*?\]|<.*?>|\{.*?\}|~.*?~')
SUFFIXES = {'P', 'M', 'QRP', 'A', 'AM', 'B', 'LH'}

def __init__(self, path):
self.prefix, self.exact, self.pos = {}, {}, {}
with open(path, encoding='latin-1') as f:
text = f.read()
for block in text.split(';'):
parts = block.strip().split(':')
if len(parts) < 9:
continue
name, primary = parts[0].strip(), parts[7].strip()
if primary.startswith('*'):
continue
try: # 代表位置（cty.dat は西経を正で記載 → 東経に直す）
self.pos[name] = (float(parts[4]), -float(parts[5]))
except ValueError:
pass
for p in parts[8].split(','):
p = self.STRIP.sub('', p).strip().upper()
if not p:
continue
if p.startswith('='):
self.exact[p[1:]] = name
else:
self.prefix[p] = name

@classmethod
def normalize(cls, call):
call = call.upper().strip('<>')
if '/' not in call:
return call
parts = [x for x in call.split('/') if x]
if 'MM' in parts: # /MM は DXCC 対象外
return None
parts = [x for x in parts if x not in cls.SUFFIXES and not x.isdigit()]
if not parts:
return None
if len(parts) == 1:
return parts[0]
return min(parts, key=len) # 短い側を運用地プレフィックスとみなす（簡易）

def entity(self, call):
raw = call.upper().strip('<>')
if raw in self.exact:
return self.exact[raw]
c = self.normalize(raw)
if not c:
return None
if c in self.exact:
return self.exact[c]
for i in range(len(c), 0, -1):
e = self.prefix.get(c[:i])
if e:
return e
return None

# ---------------------------------------------------------------- ログ読込
def load_worked(adif_path, cty, confirmed_only):
with open(adif_path, encoding='utf-8', errors='replace') as f:
text = f.read()
m = re.search(r'<eoh>', text, re.I)
body = text[m.end():] if m else text
worked = set()
for rec in re.split(r'<eor>', body, flags=re.I):
fld = {}
for t in re.finditer(r'<(\w+):(\d+)(?::\w+)?>', rec):
fld[t.group(1).upper()] = rec[t.end():t.end() + int(t.group(2))]
call, band = fld.get('CALL'), fld.get('BAND', '').lower()
if not call or not band:
continue
if confirmed_only:
ok = (fld.get('QSL_RCVD', '').upper() in ('Y', 'V') or
fld.get('LOTW_QSL_RCVD', '').upper() in ('Y', 'V'))
if not ok:
continue
ent = cty.entity(call)
if ent:
worked.add((ent, band))
return worked

# ---------------------------------------------------------------- メッセージ解析
# コールサインの形: プレフィックス(英字を含む1〜3文字) + 数字 + サフィックス(英字1〜4文字)
# 「/P」などの付加や「VP2E/K1ABC」形式にも対応。FT8の偽デコード(例: Z64EGXM9SNU)を除外する
_BASE = re.compile(r'^([A-Z0-9]{1,3})([0-9])([A-Z]{1,4})$')

def _is_base_call(tok):
m = _BASE.match(tok)
return bool(m) and bool(re.search(r'[A-Z]', m.group(1)))

def _is_call(tok):
tok = tok.strip('<>')
if not tok or '...' in tok:
return False
parts = tok.split('/')
if len(parts) > 3 or any(not p or not re.fullmatch(r'[A-Z0-9]{1,8}', p) for p in parts):
return False
return any(_is_base_call(p) for p in parts)

class _CallRE:
"""既存コードの CALL_RE.match(x) 呼び出しをそのまま使えるようにする"""
@staticmethod
def match(tok):
return _is_call(tok)

CALL_RE = _CallRE()

def sender_of(msg, cq_only):
t = msg.upper().split()
if not t:
return None
if t[0] == 'CQ':
for x in t[1:]: # "CQ DX K1ABC FN42" の修飾語を読み飛ばす
if CALL_RE.match(x):
return None if '...' in x else x.strip('<>')
return None
if cq_only or len(t) < 2:
return None
x = t[1] # "JA1BBE K1ABC -10" の送信局側
if CALL_RE.match(x) and '...' not in x:
return x.strip('<>')
return None

def called_of(msg):
"""「DXCALL JACALL ...」形式から (呼ばれている局, 呼んでいる局) を返す"""
t = msg.upper().split()
if len(t) < 2 or t[0] in ('CQ', 'QRZ', 'DE'):
return None
a, b = t[0], t[1]
if not (CALL_RE.match(a) and CALL_RE.match(b)) or '...' in a or '...' in b:
return None
return a.strip('<>'), b.strip('<>')

GRID_RE = re.compile(r'^[A-R]{2}[0-9]{2}$')

def grid_of(msg):
"""メッセージ末尾の4桁グリッド（送信局の位置）。RR73 は除外"""
t = msg.upper().split()
if len(t) >= 3 and GRID_RE.match(t[-1]) and t[-1] != 'RR73':
return t[-1]
return None

def grid_center(g):
lon = (ord(g[0]) - 65) * 20 - 180 + int(g[2]) * 2 + 1
lat = (ord(g[1]) - 65) * 10 - 90 + int(g[3]) + 0.5
return lat, lon

def distance_km(p, q):
import math
la1, lo1, la2, lo2 = map(math.radians, (p[0], p[1], q[0], q[1]))
c = (math.sin(la1) * math.sin(la2) + math.cos(la1) * math.cos(la2) * math.cos(lo1 - lo2))
return 6371 * math.acos(max(-1.0, min(1.0, c)))

def bearing_deg(p, q):
"""pからqへの大圏方位（真北=0、真東=90、真南=180、真西=270）"""
import math
la1, lo1, la2, lo2 = map(math.radians, (p[0], p[1], q[0], q[1]))
y = math.sin(lo2 - lo1) * math.cos(la2)
x = math.cos(la1) * math.sin(la2) - math.sin(la1) * math.cos(la2) * math.cos(lo2 - lo1)
return (math.degrees(math.atan2(y, x)) + 360) % 360

def utc_day(t):
return time.strftime('%Y-%m-%d', time.gmtime(t))

def adif_records(text):
"""ADIFテキストからレコード（フィールド名を大文字にしたdict）の一覧を返す"""
m = re.search(r'<eoh>', text, re.I)
body = text[m.end():] if m else text
out = []
for rec in re.split(r'<eor>', body, flags=re.I):
fld = {}
for t in re.finditer(r'<(\w+):(\d+)(?::\w+)?>', rec):
fld[t.group(1).upper()] = rec[t.end():t.end() + int(t.group(2))]
if fld.get('CALL'):
out.append(fld)
return out

def load_dxcc_numbers(path):
"""cty.csv からエンティティ名 → ADIFのDXCC番号の対応表を作る"""
nums = {}
with open(path, encoding='latin-1') as f:
for line in f:
p = line.split(',')
if len(p) > 2 and not p[0].startswith('*') and p[2].strip().isdigit():
nums[p[1].strip()] = p[2].strip()
return nums

def append_qso(path, fld, band, dxcc):
"""受け取った交信を統合ログに追記する（同じ交信が既にあれば追記しない）"""
call = fld['CALL'].upper()
date, ton = fld.get('QSO_DATE', ''), fld.get('TIME_ON', '')[:4]
try:
with open(path, encoding='ascii', errors='replace') as f:
for r in adif_records(f.read()):
if (r.get('CALL', '').upper(), r.get('QSO_DATE', ''), r.get('TIME_ON', '')[:4],
r.get('BAND', '').lower()) == (call, date, ton, band):
return 'dup'
except OSError:
pass
rec = {'CALL': call, 'QSO_DATE': date, 'TIME_ON': ton, 'BAND': band,
'MODE': fld.get('MODE', ''), 'FREQ': fld.get('FREQ', ''),
'GRIDSQUARE': fld.get('GRIDSQUARE', ''), 'DXCC': dxcc or ''}
line = ''.join(f'<{k}:{len(v)}>{v}' for k, v in rec.items() if v and v.isascii()) + '<EOR>\r\n'
for _ in range(5): # JTAlert が書き込み中なら少し待って再試行
try:
with open(path, 'a', encoding='ascii', newline='') as f:
f.write(line)
return 'ok'
except PermissionError:
time.sleep(1)
return 'fail'

# ---------------------------------------------------------------- 宇宙天気（NOAA SWPC）
SWPC = 'https://services.swpc.noaa.gov'
SW_URLS = {
'kp': SWPC + '/products/noaa-planetary-k-index.json',
'mag': SWPC + '/json/rtsw/rtsw_mag_1m.json',
'wind': SWPC + '/json/rtsw/rtsw_wind_1m.json',
'scales': SWPC + '/products/noaa-scales.json',
'sfi': SWPC + '/products/10cm-flux-30-day.json',
}

def _ts(s):
"""'2026-09-27T00:13:00' などをUTCの時刻（秒）に"""
return datetime.fromisoformat(s.replace('Z', '')[:19]).replace(tzinfo=timezone.utc).timestamp()

def fetch_json(url, timeout=30):
req = urllib.request.Request(url, headers={'Accept-Encoding': 'gzip',
'User-Agent': 'dxcc_notify (amateur radio, JA1BBE)'})
with urllib.request.urlopen(req, timeout=timeout) as r:
data = r.read()
if r.headers.get('Content-Encoding', '').lower() == 'gzip' or data[:2] == b'\x1f\x8b':
data = gzip.decompress(data)
return json.loads(data.decode('utf-8'))

class SpaceWeather:
"""NOAAの宇宙天気を定期的に取得し、通知に添える1行と、好転・悪化の知らせを作る"""
def __init__(self, a, notifier, site, fetch=fetch_json):
self.a, self.notifier, self.site, self.fetch = a, notifier, site, fetch
self.lock = threading.Lock()
self.v = {} # 最新の値
self.kp_hist = [] # [(時刻, Kp)]
self.disturbed = False # Kp 4以上の乱れを経験中か
self.last_alert = {} # 種類 -> 送った時刻
self.last_s = 0
self.errors = set()
self.first = True

# ---- 取得と解析
def update(self):
got = {}
for key, url in SW_URLS.items():
if key == 'sfi' and self.v.get('sfi_at', 0) > time.time() - 3 * 3600:
continue # 1日1回の値なので、3時間に1回で十分
try:
got[key] = self.fetch(url)
self.errors.discard(key)
except Exception as e:
if key not in self.errors:
print(f'[宇宙天気] {key} を取得できませんでした: {e}')
self.errors.add(key)
with self.lock:
try:
if 'kp' in got:
self._parse_kp(got['kp'])
if 'mag' in got:
self._parse_rtsw(got['mag'], {'bz': 'bz_gsm', 'bt': 'bt'})
if 'wind' in got:
self._parse_rtsw(got['wind'], {'speed': 'proton_speed', 'density': 'proton_density'})
if 'scales' in got:
self._parse_scales(got['scales'])
if 'sfi' in got:
rows = [r for r in got['sfi'] if isinstance(r, dict) and r.get('flux') is not None]
if rows:
self.v['sfi'] = float(rows[-1]['flux'])
self.v['sfi_at'] = time.time()
except Exception as e:
print(f'[宇宙天気] データの形式を読み取れませんでした: {e}')
self._check_alerts()
self.first = False

def _parse_kp(self, data):
rows = []
if data and isinstance(data[0], list): # 旧形式（1行目が見出し）にも対応
head = data[0]
data = [dict(zip(head, r)) for r in data[1:]]
for r in data:
kp = r.get('Kp', r.get('kp'))
if kp is not None:
rows.append((_ts(r['time_tag']), float(kp)))
rows.sort()
if rows:
self.kp_hist = rows[-16:] # 直近48時間
self.v['kp'], self.v['kp_at'] = rows[-1][1], rows[-1][0]

def _parse_rtsw(self, data, fields):
act = [r for r in data if isinstance(r, dict) and r.get('active') is True]
if not act:
return
latest_t = max(_ts(r['time_tag']) for r in act)
for name, col in fields.items():
vals = [(_ts(r['time_tag']), r.get(col)) for r in act if r.get(col) is not None]
if not vals:
continue
vals.sort()
self.v[name] = float(vals[-1][1])
win = [float(x) for t, x in vals if t > latest_t - 1800]
if win:
self.v[name + '30'] = sum(win) / len(win)
self.v['rtsw_at'] = latest_t

def _parse_scales(self, d):
cur = d.get('0', {})
for k in ('S', 'G', 'R'):
sc = (cur.get(k) or {}).get('Scale')
if sc not in (None, ''):
self.v[k] = int(sc)
prob = ((d.get('1') or {}).get('S') or {}).get('Prob')
if prob not in (None, ''):
self.v['S_prob'] = int(prob)

# ---- 表示
def line(self):
with self.lock:
v = dict(self.v)
if not v:
return '宇宙天気: 取得できていません'
parts = []
if 'kp' in v:
parts.append(f"Kp {v['kp']:.1f}")
if 'speed' in v:
parts.append(f"太陽風 {v['speed']:.0f}km/s")
if 'density' in v:
parts.append(f"密度 {v['density']:.1f}")
if 'bz' in v:
parts.append(f"Bz {v['bz']:+.1f}nT（{'南向き' if v['bz'] < 0 else '北向き'}）")
if 'bt' in v:
parts.append(f"Bt {v['bt']:.1f}nT")
if 'S' in v:
parts.append(f"S{v['S']}")
at = v.get('rtsw_at') or v.get('kp_at')
tail = f" [{time.strftime('%H:%M', time.gmtime(at))} UTC]" if at else ''
return '宇宙天気: ' + ' / '.join(parts) + tail

def short(self):
with self.lock:
v = dict(self.v)
out = []
if 'kp' in v:
out.append(f"Kp{v['kp']:.1f}")
if 'bz' in v:
out.append(f"Bz{v['bz']:+.0f}")
if 'speed' in v:
out.append(f"V{v['speed']:.0f}")
return ' '.join(out)

def start_text(self):
with self.lock:
v = dict(self.v)
extra = []
if 'sfi' in v:
extra.append(f"SFI {v['sfi']:.0f}")
if 'S_prob' in v:
extra.append(f"プロトン現象の確率（今日） {v['S_prob']}%")
if 'G' in v:
extra.append(f"地磁気嵐 G{v['G']}")
return self.line() + ('\n' + ' / '.join(extra) if extra else '')

# ---- 好転・悪化の知らせ
def _alert(self, kind, subject, body, min_interval=3 * 3600):
now = time.time()
if now - self.last_alert.get(kind, 0) < min_interval:
return
self.last_alert[kind] = now
print(time.strftime('%H:%M:%S'), subject, '|', body.replace('\n', ' | '))
self.notifier.send(subject, body)

def _check_alerts(self):
with self.lock:
v = dict(self.v)
hist = list(self.kp_hist)
kp, bz30, sp30 = v.get('kp'), v.get('bz30'), v.get('speed30')
if kp is None:
return
# 起動直後は、直近24時間に Kp 4以上があれば「乱れを経験中」とみなす（起動しただけでは知らせない）
if self.first:
self.disturbed = any(k >= 4 for t, k in hist if t > time.time() - 86400) and kp > 2
self.last_s = v.get('S', 0)
return
if kp >= 4:
self.disturbed = True
# 好転: Kp 4以上 → 2以下、Bz 30分平均が北向き、太陽風 450km/s 未満
if (self.disturbed and kp <= 2 and bz30 is not None and bz30 > 0
and sp30 is not None and sp30 < 450):
self.disturbed = False
self._alert('better', f'[DXCC] 宇宙天気好転 {self.site}',
'地磁気の乱れが収まり、ローバンドのコンディションが良くなる可能性があります。\n'
+ self.line(), min_interval=0)
# 悪化の予告: Bz 30分平均が -10nT より南向き、かつ太陽風 500km/s 以上
if bz30 is not None and sp30 is not None and bz30 < -10 and sp30 >= 500:
self._alert('worse', f'[DXCC] 宇宙天気悪化の予告 {self.site}',
'強い南向きのBzと高速の太陽風が続いています。1時間ほどのうちに地磁気が乱れ、'
'北米・ヨーロッパ方面（極地方を通る経路）が悪くなる可能性があります。\n' + self.line())
# プロトン現象: Sスケールが上がった
s = v.get('S', 0)
if s >= 1 and s > self.last_s:
self._alert(f'proton{s}', f'[DXCC] 宇宙天気プロトン現象 S{s} {self.site}',
'太陽プロトン現象が起きています。極地方で電波の吸収（PCA）が強まり、'
'北米・ヨーロッパ方面のローバンドは数日悪くなることがあります。\n' + self.line(),
min_interval=0)
self.last_s = s

def run(self):
while True:
time.sleep(self.a.sw_interval)
try:
self.update()
except Exception as e:
print(f'[宇宙天気] エラー: {e}')

# ---------------------------------------------------------------- 電離層の高さ（NICT イオノゾンデ）
# データ元: https://wdc.nict.go.jp/Ionosphere/ の自動読取15分値（factor-auto-*.sjis.txt）
# 列の並び・単位は、実データとの突き合わせで確認したもの（周波数系は100分の1MHz、高さ系は10分の1km）。
IONO_BASE = 'https://wdc.nict.go.jp/Ionosphere/archive/observation-history/factor-auto-{station}-{year}.sjis.txt'
IONO_REFERER = 'https://wdc.nict.go.jp/Ionosphere/archive/isdj_auto_txt.html'
IONO_STALE_SEC = 90 * 60 # 観測からこれ以上経った値は、通知に「約○時間前の値」と書き添える
IONO_STATIONS = { # 局コード: 表示名
'wk546': '稚内', 'to536': '国分寺', 'yg431': '山川', 'ok426': '沖縄大宜味',
}
# stationの後、日時と最初の値(fmin)が「日時:fmin」の形でひとつのカンマ区切り欄に入っている。
# 残りは、この順でカンマ区切り。
IONO_FIELDS = ["foE", "h'E", 'foEs', "h'Es", 'fbEs', 'foF1', "h'F", "h'F2", 'foF2',
'fxF2', 'M3F2', 'fxI', 'hpF2', 'fxEs', 'ftEs', 'fxF1', 'FSPR', 'MUF',
'Es1', 'Es2', 'Es', 'W']
IONO_HEIGHT_FIELDS = {"h'E", "h'Es", "h'F", "h'F2", 'hpF2'} # 10分の1kmで書かれている項目
# 限定記号・説明記号のうち、値を信用しないほうがよいもの（解説ページより）。
# F(周波数スプレッドの影響)を追加（従来漏れていた）。
IONO_BAD_FLAGS = set('CDEFGHNQRSVWB')
# このうち F・Q は、値が使えないだけでなく、電離層が乱れている(Spread-F)兆候そのものでもある
IONO_SPREAD_FLAGS = set('FQ')
_IONO_FIELD_RE = re.compile(r'(\d+)?-*([A-Za-z]{0,2})$')

def _parse_iono_token(tok):
"""'3220----' -> (3220, ''), '----G' -> (None, 'G') のように、値と記号に分ける"""
m = _IONO_FIELD_RE.match(tok.strip())
if not m:
return None, ''
val = int(m.group(1)) if m.group(1) else None
return val, m.group(2).upper()

def parse_iono_line(line):
"""factor-auto-*.sjis.txt の1行を (局コード, 日時文字列(JST), {項目名: (値, 記号)}) に分解する。
形式に合わない行（見出し行や壊れた行）は None を返す"""
parts = line.rstrip('\r\n').split(',')
if len(parts) < len(IONO_FIELDS) + 2 or ':' not in parts[1]:
return None
station = parts[0].strip().lower()
dt_str, first = parts[1].split(':', 1)
fields = {'fmin': _parse_iono_token(first)}
for name, tok in zip(IONO_FIELDS, parts[2:]):
fields[name] = _parse_iono_token(tok)
return station, dt_str.strip(), fields

def _iono_value(fields, name, scale):
v, flag = fields.get(name, (None, ''))
if v is None or flag in IONO_BAD_FLAGS:
return None
return v / scale

class Ionosonde:
"""NICTのイオノゾンデ自動読取値(15分値)を定期的に取得し、電離層の高さ(h'F2)を通知に添える。
再配布・複製の条件が明記されているHF-STARTとは別サービス。ページ全体の著作権表示はあるが、
本クラスが使うのは観測値そのもの（数値）のみで、ページの見た目・図表は取得・保存しない"""

def __init__(self, a, notifier, site, station, fetch=None):
self.a, self.notifier, self.site = a, notifier, site
self.station = station.lower()
self.fetch = fetch or self._http_full
self.lock = threading.Lock()
self.v = {}
self.errors = set()
self.spread = False # Spread-Fの兆候が出ている状態かどうか
self.last_alert = 0
self.first = True # 起動直後は、既に乱れていても知らせない（起動時に急に鳴るのを防ぐ）

def _http_full(self, year):
# この年間ファイルは1年分の枠があらかじめ用意されており、まだ来ていない時刻は空欄のまま埋め込まれている。
# そのため「末尾(Range)だけ取得」では大晦日の空欄にしか当たらず、今日のデータには届かない。
# 多少重くても、毎回ファイル全体を取得する。
url = IONO_BASE.format(station=self.station.upper(), year=year)
# 独自のUser-Agentだと、NICT側で機械的なアクセスとして404を返される様子だったため、
# 一般的なブラウザに近いヘッダーに変更した
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
'(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36',
'Accept': 'text/plain,text/html,*/*',
'Accept-Language': 'ja,en;q=0.8',
'Referer': IONO_REFERER, # サイト内の一覧ページから辿った体裁にする
}
req = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(req, timeout=60) as r:
return r.read()

@staticmethod
def _has_value(lines):
for ln in reversed(lines):
p = parse_iono_line(ln)
if p and (_iono_value(p[2], "h'F2", 10) is not None or _iono_value(p[2], "h'F", 10) is not None
or _iono_value(p[2], 'foF2', 100) is not None):
return True
return False

def update(self):
# 年間ファイルは日本時間の年ごと。年明け直後は新しい年のファイルがまだ無いか、
# 公開の時間差でまだ値が入っていないことがあるため、1月1〜3日は前年のファイルも使う
jst = datetime.now(timezone(timedelta(hours=9)))
year = jst.year
lines, err = [], None
try:
lines = [ln for ln in self.fetch(year).decode('cp932', 'replace').splitlines() if ln.strip()]
except Exception as e:
err = e
if jst.month == 1 and jst.day <= 3 and (err is not None or not self._has_value(lines)):
try:
lines = [ln for ln in self.fetch(year - 1).decode('cp932', 'replace').splitlines() if ln.strip()]
year, err = year - 1, None
except Exception as e:
err = err or e
if err is not None:
if 'fetch' not in self.errors:
url = IONO_BASE.format(station=self.station.upper(), year=year)
print(f'[電離層] {self.station} を取得できませんでした: {err}\n'
f' 試したURL: {url}\n'
f' （ブラウザでこのURLが開けるか確認してください）')
self.errors.add('fetch')
return
self.errors.discard('fetch')
raw = '\n'.join(lines).encode('cp932', 'replace')
parsed = None # h'F2/h'F/foF2のどれかに使える値がある、一番新しい行
parsed_any = None # 形式としては解析できた、一番新しい行（値が空でも可）
for ln in reversed(lines):
p = parse_iono_line(ln)
if not p:
continue
if parsed_any is None:
parsed_any = p
_, _, f = p
if (_iono_value(f, "h'F2", 10) is not None or _iono_value(f, "h'F", 10) is not None
or _iono_value(f, 'foF2', 100) is not None):
parsed = p
break
if parsed is None:
parsed = parsed_any # 使える値が無くても、形式が解析できた最新行は表示・記号検出に使う
if not parsed:
if 'parse' not in self.errors:
sample = [repr(ln[:160]) for ln in lines[-3:]]
print(f'[電離層] {self.station}: 取得はできましたが、data行を解釈できませんでした\n'
f' 取得バイト数: {len(raw)}　行数: {len(lines)}\n'
f' 末尾の行（最大3行、各160文字まで）:\n ' + '\n '.join(sample or ['(行なし)']))
self.errors.add('parse')
return
self.errors.discard('parse')
_, dt_str, fields = parsed
with self.lock:
h = _iono_value(fields, "h'F2", 10)
kind = "h'F2"
if h is None: # 夜間などh'F2が求まらない時間帯はh'Fで代用
h = _iono_value(fields, "h'F", 10)
kind = "h'F"
if h is not None:
self.v['h'], self.v['h_kind'] = h, kind
else:
self.v.pop('h', None); self.v.pop('h_kind', None)
fof2 = _iono_value(fields, 'foF2', 100)
if fof2 is not None:
self.v['fof2'] = fof2
else:
self.v.pop('fof2', None)
if h is None and fof2 is None and 'empty' not in self.errors:
raw_hf2 = fields.get("h'F2")
raw_hf = fields.get("h'F")
raw_fof2 = fields.get('foF2')
print(f'[電離層] {self.station}: {dt_str} の行は解析できましたが、'
"h'F2/h'F/foF2がいずれも値なしでした\n"
f" h'F2={raw_hf2} h'F={raw_hf} foF2={raw_fof2}")
self.errors.add('empty')
if h is not None or fof2 is not None:
self.errors.discard('empty')
try:
t = time.strptime(dt_str, '%Y%m%d%H%M%S') # dt_strはJST
self.v['at'] = calendar.timegm(t) - 9 * 3600 # UTC epoch秒に変換
except ValueError:
self.v['at'] = None
# Spread-F（電離層の乱れ）の兆候: foF2・h'F2・h'Fのいずれかに F/Q の記号が付いているか
spread_now = any(fields.get(name, (None, ''))[1] in IONO_SPREAD_FLAGS
for name in ('foF2', "h'F2", "h'F"))
self._check_spread(spread_now)

def _check_spread(self, spread_now):
if self.first:
self.spread = spread_now
self.first = False
return
name = IONO_STATIONS.get(self.station, self.station)
if spread_now and not self.spread:
self._alert(f'[DXCC] 電離層乱れの兆候(Spread-F) {self.site}',
f'{name}上空の電離層観測で、周波数方向の乱れ（Spread-F）の兆候が見られます。\n'
'電離層が不安定になっている可能性があります（コンディションが悪化することも、'
'逆に飛びやすくなることもあります）。\n'
'これはあくまで観測点上空の状態で、交信したい方向の経路上の状態を'
'直接示すものではありません。')
elif not spread_now and self.spread:
self._alert(f'[DXCC] 電離層乱れが収まった様子 {self.site}',
f'{name}上空の電離層観測で、周波数方向の乱れ（Spread-F）の兆候が収まりました。')
self.spread = spread_now

def _alert(self, subject, body, min_interval=1800):
now = time.time()
if now - self.last_alert < min_interval:
return
self.last_alert = now
print(time.strftime('%H:%M:%S'), subject, '|', body.replace('\n', ' | '))
self.notifier.send(subject, body)

def line(self):
with self.lock:
v = dict(self.v)
if 'h' not in v and 'fof2' not in v:
return ''
parts = []
if 'h' in v:
parts.append(f"{v['h_kind']} {v['h']:.0f}km")
if 'fof2' in v:
parts.append(f"foF2 {v['fof2']:.1f}MHz")
at = v.get('at')
tail = ''
if at:
g = time.gmtime(at)
# NICTの公開ファイルは更新に時間差があるため、日付を付けて「いつの値か」を明示する
tail = f" [{g.tm_mon}/{g.tm_mday} {time.strftime('%H:%M', g)} UTC]"
age = time.time() - at
if age >= IONO_STALE_SEC:
tail += f'（約{age/60:.0f}分前の値）' if age < 3 * 3600 else f'（約{age/3600:.0f}時間前の値）'
name = IONO_STATIONS.get(self.station, self.station)
return f'電離層({name}): ' + ' / '.join(parts) + tail

def short(self):
"""セッション記録の1行に添える簡潔な表記。電離層は15分毎の値なので、
いつ時点の値かを示す測定日時（データ自体の時刻。取得・受信した時刻ではない）を必ず添える"""
with self.lock:
v = dict(self.v)
if 'h' not in v and 'fof2' not in v:
return ''
parts = []
if 'h' in v:
parts.append(f"{v['h_kind']}{v['h']:.0f}km")
if 'fof2' in v:
parts.append(f"foF2{v['fof2']:.1f}")
at = v.get('at')
if at:
g = time.gmtime(at)
when = f"{g.tm_mon}/{g.tm_mday} {time.strftime('%H:%M', g)}"
else:
when = '--:--'
return ' '.join(parts) + f'({when})'

def run(self):
while True:
try:
self.update()
except Exception as e:
print(f'[電離層] エラー: {e}')
time.sleep(self.a.iono_interval)

# ---------------------------------------------------------------- 通知
def notify_ntfy(url, title, body, priority='high'):
req = urllib.request.Request(url, data=body.encode('utf-8'), method='POST')
req.add_header('Title', title) # HTTPヘッダはASCIIのみ
req.add_header('Priority', priority)
req.add_header('Tags', 'satellite_antenna')
urllib.request.urlopen(req, timeout=10).read()

def notify_gmail(user, password, to, subject, body):
msg = EmailMessage()
msg['From'] = user
msg['To'] = to
msg['Subject'] = subject
msg.set_content(body)
ctx = ssl.create_default_context()
with smtplib.SMTP_SSL('smtp.gmail.com', 465, context=ctx, timeout=20) as s:
s.login(user, password)
s.send_message(msg)

class Notifier:
"""送信は別スレッドで行う。SMTPの数秒の待ちでUDPパケットを取りこぼさないため。"""
def __init__(self, a):
self.a = a
self.password = None
if a.mail_to:
self.password = os.environ.get('DXCC_MAIL_PASS')
if not self.password and a.pass_file:
with open(a.pass_file, encoding='utf-8') as f:
self.password = f.read().strip()
if not self.password:
raise SystemExit('Gmailのアプリパスワードが見つかりません '
'(環境変数 DXCC_MAIL_PASS または --pass-file)')
self.q = queue.Queue()
self.sent = [] # 送信時刻（1時間あたりの上限管理）
threading.Thread(target=self._worker, daemon=True).start()

def send(self, subject, body):
now = time.time()
self.sent = [t for t in self.sent if now - t < 3600]
if len(self.sent) >= self.a.max_per_hour:
print('[notify] 1時間あたりの上限に達したため送信を見送り')
return
self.sent.append(now)
self.q.put((subject, body))

def _worker(self):
while True:
subject, body = self.q.get()
if self.a.mail_to:
try:
notify_gmail(self.a.mail_from, self.password, self.a.mail_to, subject, body)
except Exception as e:
print(f'[gmail error] {e}')
if self.a.ntfy:
try:
notify_ntfy(self.a.ntfy, subject.encode('ascii', 'ignore').decode() or 'DXCC alert', body)
except Exception as e:
print(f'[ntfy error] {e}')

# ---------------------------------------------------------------- メイン
def main():
ap = argparse.ArgumentParser()
ap.add_argument('--site', default=os.environ.get('COMPUTERNAME', 'PC'),
help='このPCのID（例: home-sdr、home-ic7850）。メールの件名には対応する日本語名が入る')
ap.add_argument('--cty', help='cty.dat のパス（既定: スクリプトと同じフォルダの common\\cty_latest.dat）')
ap.add_argument('--adif', nargs='*',
help='交信ログADIF（既定: スクリプトと同じフォルダの logs の中の *.adi をすべて）')
ap.add_argument('--mail-from', default='ja1bbe.dxcc@gmail.com', help='送信用Gmailアドレス')
ap.add_argument('--mail-to', help='通知の宛先（スマホで受信するアドレス）')
ap.add_argument('--pass-file', help='アプリパスワードを1行で書いたファイル')
ap.add_argument('--ntfy', help='予備: https://ntfy.sh/推測されにくいトピック名')
ap.add_argument('--max-per-hour', type=int, default=20, help='通知の上限（件/時）')
ap.add_argument('--group', default='239.255.0.1', help='マルチキャストアドレス')
ap.add_argument('--port', type=int, default=2237)
ap.add_argument('--bands', default='160m,80m,60m,40m,30m,20m,17m,15m,12m,10m,6m', help='通知対象バンド')
ap.add_argument('--confirmed', action='store_true',
help='コンファーム済みのみ「済」扱い (Honor Roll向け)')
ap.add_argument('--cq-only', action='store_true', help='CQ局のみ通知')
ap.add_argument('--confirm', type=int, default=2,
help='同じ局をこの回数デコードしてから通知（偽デコード対策、既定2。1で即時通知）')
ap.add_argument('--confirm-window', type=int, default=180, help='--confirm の回数を数える時間（秒）')
ap.add_argument('--grid-km', type=int, default=6000,
help='グリッドとエンティティの位置がこの距離以上離れていたら「要確認」を付ける')
ap.add_argument('--my-grid',
help='自局のグリッドロケータ（例: PM95）。指定すると、通知にローテーターの目安方位（真北=0、'
'真東=90、真南=180、真西=270）を添える')
ap.add_argument('--log-port', type=int, default=0,
help='別PCのWSJT-X（セカンドUDPサーバー）から交信記録を受け取るポート。0で無効')
ap.add_argument('--log-append', help='受け取った交信を追記するADIF（既定: --adif の最初のファイル）。none で追記しない')
ap.add_argument('--cty-csv', help='DXCC番号の付与に使う cty.csv（既定: スクリプトと同じフォルダの cty_latest.csv）')
ap.add_argument('--no-space-weather', action='store_true', help='宇宙天気（NOAA）を使わない')
ap.add_argument('--sw-interval', type=int, default=900, help='宇宙天気を取りに行く間隔（秒、既定15分）')
ap.add_argument('--iono-station', choices=['wk546', 'to536', 'yg431', 'ok426', 'off'], default='to536',
help='電離層の高さ(NICT)に使う観測局: wk546=稚内 to536=国分寺(既定) yg431=山川 ok426=沖縄大宜味 off=使わない')
ap.add_argument('--iono-interval', type=int, default=1800,
help='電離層データを取りに行く間隔（秒、既定30分。データ自体の更新も15分毎）')
ap.add_argument('--called', choices=['off', 'ja', 'any'], default='ja',
help='他局が呼んでいる相手局も判定: ja=呼んでいるのが日本の局のときのみ(既定), any=誰でも, off=しない')
ap.add_argument('--min-snr', type=int, default=-30)
ap.add_argument('--debug-udp', action='store_true',
help='WSJT-Xからのデコードを受信するたびに画面に表示する（受信できているかの確認用）')
ap.add_argument('--repeat', choices=['utcday', 'cooldown'], default='utcday',
help='同じ局の再通知: utcday=UTCの1日に1回(既定), cooldown=--cooldown 秒ごと')
ap.add_argument('--cooldown', type=int, default=600, help='--repeat cooldown のときの再通知抑止秒')
ap.add_argument('--state', help='通知済み記録ファイル（既定: sessions\\state_<PCの名前>.json）')
ap.add_argument('--test', action='store_true', help='テスト通知を送って終了')
a = ap.parse_args()
base = os.path.dirname(os.path.abspath(__file__))
# batファイルに日本語を書くと文字化けしやすいので、--site には英数字のIDを書き、表示名はここで対応させる
SITE_NAMES = {'home-sdr': '自宅SDR', 'home-ic7850': '自宅IC-7850',
'yugawara-sdr': '湯河原SDR', 'yugawara-ic7851': '湯河原IC-7851'}
site_id = a.site
a.site = SITE_NAMES.get(site_id.lower(), site_id)
a.cty = a.cty or os.path.join(base, 'common', 'cty_latest.dat')
my_pos = None
if a.my_grid:
g = a.my_grid.strip().upper()
if re.fullmatch(r'[A-R]{2}\d{2}([A-X]{2})?', g):
my_pos = grid_center(g)
else:
print(f'[注意] --my-grid の形式が不正なため無視します: {a.my_grid}')
sess_dir = os.path.join(base, 'sessions')
os.makedirs(sess_dir, exist_ok=True)

def log_files():
return a.adif if a.adif else sorted(glob.glob(os.path.join(base, 'logs', '*.adi')))

if not a.mail_to and not a.ntfy:
raise SystemExit('--mail-to か --ntfy の少なくとも一方を指定してください')
notifier = Notifier(a)
if a.test:
notifier.send('[DXCC] テスト通知', 'dxcc_notify.py からのテスト通知です。')
time.sleep(30) # 送信スレッドの完了を待つ
return

targets = set(a.bands.lower().split(','))
cty = Cty(a.cty)
worked, mtimes = set(), None

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('', a.port))
if ipaddr_is_multicast(a.group):
# WSJT-X が送信に使うインターフェース（ループバック/LAN）のどちらでも受信できるよう両方で参加
joined = 0
for iface in ('127.0.0.1', '0.0.0.0'):
try:
mreq = struct.pack('4s4s', socket.inet_aton(a.group), socket.inet_aton(iface))
sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
joined += 1
except OSError as e:
print(f'[multicast] {iface} で参加できませんでした: {e}')
if not joined:
raise SystemExit('マルチキャストグループに参加できませんでした')
print(f'listening {a.group}:{a.port} bands={sorted(targets)}')

# 別PCのWSJT-Xからの交信記録の受信口
lsock, remote_worked = None, set()
if a.log_port:
lsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
lsock.bind(('', a.log_port))
print(f'QSO log receiver on UDP port {a.log_port}')
append_path = None if (a.log_append or '').lower() == 'none' else (a.log_append or (log_files() or [None])[0])
csv_path = a.cty_csv or os.path.join(base, 'common', 'cty_latest.csv')
dxcc_nums = {}
if lsock and os.path.exists(csv_path):
dxcc_nums = load_dxcc_numbers(csv_path)
elif lsock:
print(f'[注意] {csv_path} がないため、追記する交信にDXCC番号を付けられません（JTAlertのRebuildに反映されません）')

bands, last_sent, last_check, seen = {}, {}, 0, {} # bands: WSJT-Xの名前(id)ごとのバンド

# 通知済みの記録（UTCの日付ごと）。PCを再起動しても同じ日に再通知しないようファイルに保存
state_path = a.state or os.path.join(sess_dir, f'state_{site_id}.json')
sent_day = {}
try:
with open(state_path, encoding='utf-8') as f:
today = utc_day(time.time())
sent_day = {tuple(k.split('|')): v for k, v in json.load(f).items() if v == today}
except (OSError, ValueError):
pass

def already_sent(key, now):
if a.repeat == 'utcday':
return sent_day.get(key) == utc_day(now)
return now - last_sent.get(key, 0) < a.cooldown

def mark_sent(key, now):
last_sent[key] = now
today = utc_day(now)
sent_day[key] = today
for k in [k for k, v in sent_day.items() if v != today]: # 前日以前は削除
del sent_day[k]
try:
with open(state_path, 'w', encoding='utf-8') as f:
json.dump({'|'.join(k): v for k, v in sent_day.items()}, f)
except OSError as e:
print(f'[state] 保存できませんでした: {e}')
# ---------------- ログの読み込み（起動時と、ファイルが更新されたとき）
def reload_logs(force=False):
nonlocal worked, mtimes
files = log_files()
m = tuple((p, os.path.getmtime(p)) for p in files if os.path.exists(p))
if not force and m == mtimes:
return
worked, mtimes = set(remote_worked), m
info = []
for p, _ in m:
w = load_worked(p, cty, a.confirmed)
worked |= w
info.append(f'{os.path.basename(p)}: {len(w)} pairs')
print(f'log loaded: {info[-1]}')
if not m:
print('[注意] 交信ログが見つかりません:', os.path.join(base, 'logs'))
print(f'log total: {len(worked)} entity-band pairs')
return info

log_info = reload_logs(force=True) or []

# 宇宙天気: 最初の1回はその場で取得し、以後は別スレッドで定期的に取得する
sw = None
if not a.no_space_weather:
sw = SpaceWeather(a, notifier, a.site)
print('宇宙天気を取得しています…')
sw.update()
print(sw.line())
threading.Thread(target=sw.run, daemon=True).start()

# 電離層の高さ（NICT）: 同様に、最初の1回はその場で取得し、以後は別スレッドで定期的に取得する
iono = None
if a.iono_station != 'off':
iono = Ionosonde(a, notifier, a.site, a.iono_station)
print('電離層データを取得しています…')
iono.update()
print(iono.line() or '（まだ値が得られていません）')
threading.Thread(target=iono.run, daemon=True).start()

def extra_lines():
out = []
if sw:
out.append(sw.line())
if iono and iono.line():
out.append(iono.line())
return ('\n' + '\n'.join(out)) if out else ''
swline = extra_lines
last_check = time.time()

# ---------------- この運用で自局が直接受信した未交信局の記録
t_start = time.time()
stamp = time.strftime('%Y%m%d_%H%M', time.gmtime(t_start))
sess_path = os.path.join(sess_dir, f'{site_id}_{stamp}UTC.txt')
sess = {} # (call, band) -> 記録
sess_dirty, sess_saved = False, 0
n_notified = [0]
band_order = {b: i for i, b in enumerate(['160m', '80m', '60m', '40m', '30m', '20m', '17m', '15m',
'12m', '10m', '6m'])}

def hm(t):
return time.strftime('%H:%M', time.gmtime(t))

def session_text(final=False):
rows = [r for r in sess.values() if r['count'] >= a.confirm]
rows.sort(key=lambda r: (band_order.get(r['band'], 99), r['first']))
t_end = time.time()
completed = [r for r in rows if (r['ent'], r['band']) in worked] if final else []
lines = []
if completed:
lines += [f'この運用中に、次の{len(completed)}エンティティと新たに交信できました。',
'バンドコールエンティティ']
for r in completed:
lines.append(f"{r['band']:<6} {r['call']:<12} {r['ent'][:26]}")
lines += ['', '-' * 60, '']
lines += [f'PC: {a.site}',
f'運用時間: {time.strftime("%Y-%m-%d %H:%M", time.gmtime(t_start))} 〜 '
f'{time.strftime("%Y-%m-%d %H:%M", time.gmtime(t_end))} UTC'
f'（{(t_end - t_start) / 3600:.1f}時間）' + ('' if final else '（運用中）'),
f'送った通知: {n_notified[0]}通',
'',
f'自局の設備で直接受信した未交信局: {len(rows)}局（{a.confirm}回以上デコードした局）',
'バンドコールエンティティ受信(UTC) 回数最大SNR '
'宇宙天気(初受信時) 電離層(初受信時・測定日時はUTC) 備考']
for r in rows:
note = []
if r['notified']:
note.append('通知済み')
if (r['ent'], r['band']) in worked:
note.append('この運用中に交信')
if r['warn']:
note.append('要確認(グリッド)')
lines.append(f"{r['band']:<6} {r['call']:<12} {r['ent'][:26]:<26} "
f"{hm(r['first'])}-{hm(r['last'])} {r['count']:>4} {r['best']:>+5d}dB "
f"{r.get('sw', ''):<18} {r.get('iono', ''):<28} {'、'.join(note)}")
if not rows:
lines.append('（なし）')
if final:
tail = []
if sw:
tail.append('終了時の' + sw.line())
if iono and iono.line():
tail.append(iono.line())
if tail:
lines += [''] + tail
return '\n'.join(lines)

def save_session(final=False):
nonlocal sess_dirty, sess_saved
try:
with open(sess_path, 'w', encoding='utf-8') as f:
f.write(session_text(final) + '\n')
sess_dirty, sess_saved = False, time.time()
except OSError as e:
print(f'[session] 保存できませんでした: {e}')

def send_now(subject, body):
"""開始・終了のメールは上限に関係なく、その場で送る"""
if a.mail_to:
try:
notify_gmail(a.mail_from, notifier.password, a.mail_to, subject, body)
return True
except Exception as e:
print(f'[gmail error] {e}')
return False

stopped = [False]

def stop(reason):
if stopped[0]:
return
stopped[0] = True
save_session(final=True)
print(f'停止します（{reason}）。終了のメールを送ります…')
ok = send_now(f'[DXCC] STOP {a.site}', session_text(final=True))
print('終了のメールを送りました。' if ok else '終了のメールを送れませんでした。', '記録:', sess_path)

if os.name == 'nt': # 黒い画面を×で閉じたときも、できるだけ終了のメールを送る
import ctypes
HANDLER = ctypes.WINFUNCTYPE(ctypes.c_int, ctypes.c_uint)

def _on_console(ev):
if ev in (1, 2, 5, 6): # Ctrl+Break、ウィンドウを閉じる、ログオフ、シャットダウン
stop('画面を閉じた')
return 1
return 0 # Ctrl+C は通常の処理（KeyboardInterrupt）に任せる
_console_handler = HANDLER(_on_console)
ctypes.windll.kernel32.SetConsoleCtrlHandler(_console_handler, True)

def main_loop():
nonlocal last_check, sess_dirty
while True:
ready, _, _ = select.select([sock] + ([lsock] if lsock else []), [], [], 5.0)
if lsock in ready:
ldata, laddr = lsock.recvfrom(65535)
try:
if len(ldata) > 12 and struct.unpack('>I', ldata[:4])[0] == MAGIC:
lr = Reader(ldata); lr.u32(); lr.u32()
if lr.u32() != 12: # WSJT-X の「Logged ADIF」メッセージ以外は無視
continue
lr.qstr(); text = lr.qstr() or ''
else:
text = ldata.decode('utf-8', 'replace')
for fld in adif_records(text):
call = fld['CALL'].upper()
band = (fld.get('BAND') or '').lower()
if not band and fld.get('FREQ'):
try:
band = freq_to_band(float(fld['FREQ']) * 1e6) or ''
except ValueError:
band = ''
ent = cty.entity(call)
if not ent or not band:
print(f'[remote QSO] 判定できませんでした: {call} {band}')
continue
worked.add((ent, band)); remote_worked.add((ent, band))
res = append_qso(append_path, fld, band, dxcc_nums.get(ent)) if append_path else 'skip'
print(time.strftime('%H:%M:%S'), f'[remote QSO from {laddr[0]}] {call} ({ent}) {band} '
f'→ 交信済みに反映（ログ追記: {res}）')
except (struct.error, IndexError, ValueError) as e:
print(f'[remote QSO] 読み取れませんでした: {e}')
continue
# ログの読み直し（交信直後に通知が止まるように）と、運用中の記録の保存
if time.time() - last_check > 60:
last_check = time.time()
reload_logs()
if sess_dirty:
save_session()
if sock not in ready:
continue
data, _ = sock.recvfrom(4096)

try:
r = Reader(data)
if r.u32() != MAGIC:
continue
r.u32() # schema
mtype = r.u32()
cid = r.qstr() # WSJT-X の名前（--rig-name ごとに異なる）
if mtype == 1: # Status
bands[cid] = freq_to_band(r.u64())
elif mtype == 2 and bands.get(cid) in targets: # Decode
band = bands[cid]
is_new = r.bool(); r.u32()
snr = r.i32(); dt = r.f64(); df = r.u32()
mode = r.qstr(); msg = r.qstr() or ''
low_conf = r.bool() # WSJT-X が「?」を付ける低信頼デコード
if a.debug_udp:
print(f'[UDP受信] {cid} {band} SNR{snr:+d} "{msg}"'
+ ('' if is_new else ' (再デコード)')
+ (' (低信頼)' if low_conf else ''))
if not is_new or snr < a.min_snr or low_conf:
continue
cands = [] # (コール, 種別)
call = sender_of(msg, a.cq_only)
if call:
cands.append((call, 'direct'))
if a.called != 'off':
res = called_of(msg)
if res:
target, caller = res
if a.called == 'any' or cty.entity(caller) == 'Japan':
cands.append((target, 'called'))
for call, kind in cands:
ent = cty.entity(call)
if not ent or (ent, band) in worked:
continue
if kind == 'direct': # 自局が直接受信した未交信局を、この運用の記録に残す
r0 = sess.get((call, band))
t0 = time.time()
if r0 is None:
r0 = sess[(call, band)] = {'call': call, 'band': band, 'ent': ent, 'first': t0,
'last': t0, 'count': 0, 'best': snr,
'notified': False, 'warn': False,
'sw': sw.short() if sw else '',
'iono': iono.short() if iono else ''}
r0['last'] = t0; r0['count'] += 1; r0['best'] = max(r0['best'], snr)
if r0['count'] == a.confirm:
sess_dirty = True
key = (call, band, kind) # 直接受信は「呼ばれ中」と別に通知する
now = time.time()
# 偽デコード対策: 同じ局を一定時間内に所定回数デコードしてから通知
hist = [t for t in seen.get(key, []) if now - t < a.confirm_window] + [now]
seen[key] = hist
if len(hist) < a.confirm:
continue
if already_sent(key, now):
continue
dkey = (call, band, 'direct')
if kind == 'called':
# 自局を優先: 自局で受信済み（通知済み、またはデコード中）の局は他局情報を通知しない
if already_sent(dkey, now):
continue
if any(now - t < a.confirm_window for t in seen.get(dkey, [])):
continue
after_called = (kind == 'direct' and already_sent((call, band, 'called'), now))
mark_sent(key, now)
warn = ''
g = grid_of(msg) if kind == 'direct' else None
if g and ent in cty.pos:
d = distance_km(grid_center(g), cty.pos[ent])
if d > a.grid_km:
warn = f'【要確認】グリッド{g}が{ent}から約{d:,.0f}km離れています（偽デコードの可能性）\n'
bearing_line = ''
if my_pos:
tgt_pos = grid_center(g) if g else cty.pos.get(ent)
if tgt_pos:
src = 'グリッドより' if g else f'{ent}の目安位置より（概算）'
bearing_line = f"\nローテーター目安: {bearing_deg(my_pos, tgt_pos):.0f}°（{src}）"
if kind == 'direct':
subject = f'[DXCC] {band} {call} ({ent})' + (' 要確認' if warn else '')
if after_called:
subject += ' 自局でも受信'
warn += '先に「他局が呼び出し中」で通知した局を、自局の設備でも受信できました\n'
body = (warn + f'{call} ({ent}) {band} SNR {snr:+d}dB DT {dt:+.1f}s DF {df}Hz\n{msg}'
+ bearing_line + swline())
else:
subject = f'[DXCC] {band} {call} ({ent}) 他局が呼び出し中'
body = (f'{call} ({ent}) {band} を他局が呼んでいます（当局では未受信）\n'
f'呼んでいる局の信号: SNR {snr:+d}dB DT {dt:+.1f}s DF {df}Hz\n{msg}'
+ bearing_line + swline())
print(time.strftime('%H:%M:%S'), body.replace('\n', ' | '))
notifier.send(subject, body)
n_notified[0] += 1
if kind == 'direct' and (call, band) in sess:
sess[(call, band)]['notified'] = True
sess[(call, band)]['warn'] = bool(warn and '要確認' in warn)
sess_dirty = True
except (struct.error, IndexError):
continue

ok = send_now(f'[DXCC] START {a.site}',
f'PC: {a.site}\n開始: {time.strftime("%Y-%m-%d %H:%M", time.gmtime(t_start))} UTC\n'
f'対象バンド: {", ".join(sorted(targets, key=lambda b: band_order.get(b, 99)))}\n'
f'交信ログ: ' + ('、'.join(log_info) if log_info else '見つかりません') +
f'\n交信済み（エンティティ×バンド）: {len(worked)}\n'
+ (sw.start_text() + '\n' if sw else '')
+ (iono.line() + '\n' if iono and iono.line() else '') +
f'運用中の記録: {sess_path}')
print('開始のメールを送りました' if ok else '[注意] 開始のメールを送れませんでした（上のエラーを確認してください）')
try:
main_loop()
except KeyboardInterrupt:
stop('Ctrl+C')

def ipaddr_is_multicast(addr):
try:
return 224 <= int(addr.split('.')[0]) <= 239
except ValueError:
return False

if __name__ == '__main__':
main()