This repository has been archived on 2025-08-02. You can view files and clone it, but cannot push or open issues or pull requests.
Files
Cryville.EEW.Unity/Assets/Cryville.EEW.Unity/Worker.cs
2025-07-05 01:10:15 +08:00

361 lines
17 KiB
C#

using Cryville.EEW.BMKGOpenData;
using Cryville.EEW.BMKGOpenData.TTS;
using Cryville.EEW.Core;
using Cryville.EEW.CWAOpenData;
using Cryville.EEW.CWAOpenData.Model;
using Cryville.EEW.CWAOpenData.TTS;
using Cryville.EEW.EMSC;
using Cryville.EEW.FANStudio;
using Cryville.EEW.FANStudio.TTS;
using Cryville.EEW.GeoNet;
using Cryville.EEW.GeoNet.TTS;
using Cryville.EEW.GlobalQuake;
using Cryville.EEW.JMAAtom;
using Cryville.EEW.JMAAtom.TTS;
using Cryville.EEW.NOAA;
using Cryville.EEW.NOAA.TTS;
using Cryville.EEW.QuakeML;
using Cryville.EEW.Report;
using Cryville.EEW.Unity.Map;
using Cryville.EEW.Unity.UI;
using Cryville.EEW.UpdateChecker;
using Cryville.EEW.USGS;
using Cryville.EEW.Wolfx;
using Cryville.EEW.Wolfx.TTS;
using System;
using System.Collections.Concurrent;
using System.Globalization;
using System.Linq;
using System.Net;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
using UnityEngine;
namespace Cryville.EEW.Unity {
sealed class Worker : MonoBehaviour {
public static Worker Instance { get; private set; }
[SerializeField] CameraController m_cameraController;
[SerializeField] MapElementManager m_mapElementManager;
[SerializeField] EventOngoingListView m_ongoingEventList;
[SerializeField] EventGroupListView m_historyEventGroupList;
[SerializeField] GameObject m_connectingHint;
CoreWorker _worker;
ReportGrouper _grouper;
CancellationTokenSource _cancellationTokenSource;
void Awake() {
if (Instance != null) {
Destroy(this);
throw new InvalidOperationException("Duplicate worker.");
}
Instance = this;
try {
App.Init();
_worker = new(new TTSWorker());
_grouper = new ReportGrouper();
_cancellationTokenSource = new();
}
catch (Exception ex) {
Dialog.Instance.Show("FATAL ERROR", ex.ToString());
throw;
}
}
void Start() {
try {
App.MainLogger.Log(1, "App", null, "Initializing localized resources manager");
LocalizedResources.Init(new LocalizedResourcesManager());
RegisterViewModelGenerators(_worker);
RegisterTTSMessageGenerators(_worker);
BuildWorkers();
_worker.RVMGeneratorContext = SharedSettings.Instance;
_worker.TTSMessageGeneratorContext = SharedSettings.Instance;
_worker.RVMCulture = SharedSettings.Instance.RVMCulture;
_worker.SetTTSCultures(SharedSettings.Instance.TTSCultures ?? new TTSCultureConfig[0]);
_worker.IgnoreLanguageVariant = SharedSettings.Instance.DoIgnoreLanguageVariant;
_ongoingReportManager.Changed += OnOngoingReported;
_worker.Reported += OnReported;
_grouper.GroupUpdated += OnGroupUpdated;
_grouper.GroupRemoved += OnGroupRemoved;
App.MainLogger.Log(1, "App", null, "Worker ready");
Task.Run(() => GatewayVerify(_cancellationTokenSource.Token)).ContinueWith(task => {
if (task.IsFaulted) {
OnReported(this, new() { Title = task.Exception.Message });
return;
}
_verified = true;
_uiActionQueue.Enqueue(() => m_connectingHint.SetActive(false));
Task.Run(() => ScheduledGatewayVerify(_cancellationTokenSource, _cancellationTokenSource.Token));
Task.Run(() => _worker.RunAsync(_cancellationTokenSource.Token));
Task.Run(() => _ongoingReportManager.RunAsync(_cancellationTokenSource.Token));
}, TaskScheduler.Current);
}
catch (Exception ex) {
Dialog.Instance.Show("FATAL ERROR", ex.ToString());
throw;
}
}
void OnDestroy() {
_cancellationTokenSource.Cancel();
_worker.Dispose();
_ongoingReportManager.Dispose();
}
CENCEarthquakeRVMGenerator _cencEarthquakeRVMGenerator;
void RegisterViewModelGenerators(CoreWorker worker) {
worker.RegisterViewModelGenerator(new BMKGEarthquakeRVMGenerator());
worker.RegisterViewModelGenerator(new CEAEEWRVMGenerator());
worker.RegisterViewModelGenerator(_cencEarthquakeRVMGenerator = new CENCEarthquakeRVMGenerator());
worker.RegisterViewModelGenerator(new CENCEEWRVMGenerator());
worker.RegisterViewModelGenerator(new CWAEarthquakeRVMGenerator());
worker.RegisterViewModelGenerator(new CWAEEWRVMGenerator());
worker.RegisterViewModelGenerator(new CWATsunamiRVMGenerator());
worker.RegisterViewModelGenerator(new EMSCRealTimeEventRVMGenerator());
worker.RegisterViewModelGenerator(new FANStudio.FujianEEWRVMGenerator());
worker.RegisterViewModelGenerator(new Wolfx.FujianEEWRVMGenerator());
worker.RegisterViewModelGenerator(new GeoNetQuakeHistoryRVMGenerator());
worker.RegisterViewModelGenerator(new GeoNetQuakeRVMGenerator());
worker.RegisterViewModelGenerator(new GeoNetStrongRVMGenerator());
worker.RegisterViewModelGenerator(new GlobalQuakeRVMGenerator());
worker.RegisterViewModelGenerator(new ICLEEWRVMGenerator());
worker.RegisterViewModelGenerator(new JMAAtomRVMGenerator());
worker.RegisterViewModelGenerator(new JMAEEWRVMGenerator());
worker.RegisterViewModelGenerator(new NOAAAtomRVMGenerator());
var quakemlEventRVMGenerator = new QuakeMLEventRVMGenerator();
quakemlEventRVMGenerator.AddExtension(new USGSQuakeMLExtension());
worker.RegisterViewModelGenerator(quakemlEventRVMGenerator);
worker.RegisterViewModelGenerator(new FANStudio.SichuanEEWRVMGenerator());
worker.RegisterViewModelGenerator(new Wolfx.SichuanEEWRVMGenerator());
worker.RegisterViewModelGenerator(new USGSContoursRVMGenerator());
worker.RegisterViewModelGenerator(new VersionRVMGenerator());
}
CENCEarthquakeTTSMessageGenerator _cencEarthquakeTTSMessageGenerator;
void RegisterTTSMessageGenerators(CoreWorker worker) {
worker.RegisterTTSMessageGenerator(new BMKGEarthquakeTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new CEAEEWTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(_cencEarthquakeTTSMessageGenerator = new CENCEarthquakeTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new CENCEEWTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new CWAEarthquakeTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new CWAEEWTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new CWATsunamiTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new FANStudio.TTS.FujianEEWTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new Wolfx.TTS.FujianEEWTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new GeoNetQuakeHistoryTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new GeoNetQuakeTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new GeoNetStrongTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new ICLEEWTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new JMAAtomTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new JMAEEWTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new NOAATTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new FANStudio.TTS.SichuanEEWTTSMessageGenerator());
worker.RegisterTTSMessageGenerator(new Wolfx.TTS.SichuanEEWTTSMessageGenerator());
}
bool _verified;
void BuildWorkers() {
App.MainLogger.Log(1, "App", null, "Building workers");
#if false//UNITY_EDITOR
_worker.AddWorker(new WolfxWorker(new Uri("ws://localhost:9995/wolfx")));
_worker.AddWorker(new JMAAtomWorker(new Uri("http://localhost:9095/eqvol.xml")));
_worker.AddWorker(new FANStudioWorker<FANStudio.Model.CEAEEW>(new("ws://localhost:9995/fan/cea")));
_worker.AddWorker(new CWAReportWorker<Tsunami>(new Uri("http://localhost:9095/E-A0014-001.json"), "1"));
_worker.AddWorker(new CWAReportWorker<Earthquake>(new Uri("http://localhost:9095/E-A0015-001.json"), "1"));
_worker.AddWorker(new CWAReportWorker<Earthquake>(new Uri("http://localhost:9095/E-A0016-001.json"), "1"));
BMKGOpenDataWorker bmkgWorker = new(new Uri("http://localhost:9095/autogempa.json"));
bmkgWorker.SetDataUris(new Uri[] { new("http://localhost:9095/gempadirasakan.json") });
_worker.AddWorker(bmkgWorker);
_worker.AddWorker(new NOAAAtomWorker(new("http://localhost:9095/PAAQAtom.xml"), forceHttps: false));
_worker.AddWorker(new UpdateCheckerWorker(typeof(Worker).Assembly.GetName().Version?.ToString(3) ?? "", "unity"));
#else
foreach (var source in SharedSettings.Instance.EventSources) {
_worker.AddWorker(source switch {
BMKGOpenDataEventSourceConfig bmkgOpenData => BuildBMKGOpenDataWorkerUris(new(new("https://data.bmkg.go.id/DataMKG/TEWS/autogempa.json")), bmkgOpenData),
CWAOpenDataEventSourceConfig cwaOpenData => cwaOpenData.Subtype switch {
"E-A0014-001" => new CWAReportWorker<Tsunami>(new("https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0014-001"), cwaOpenData.Token, 1440, 17280),
"E-A0015-001" => new CWAReportWorker<Earthquake>(new("https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001"), cwaOpenData.Token),
"E-A0016-001" => new CWAReportWorker<Earthquake>(new("https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0016-001"), cwaOpenData.Token),
_ => throw new InvalidOperationException("Unknown CWA open data sub-type."),
},
EMSCRealTimeEventSourceConfig => new EMSCRealTimeWorker(new("wss://www.seismicportal.eu/standing_order/websocket")),
FANStudioEventSourceConfig fanStudio => fanStudio.Subtype switch {
"cea" => new FANStudioWorker<FANStudio.Model.CEAEEW>(new("wss://websocket.fanstudio.hk/cea")),
"sichuan" => new FANStudioWorker<FANStudio.Model.SichuanEEW>(new("wss://websocket.fanstudio.hk/sichuan")),
"fujian" => new FANStudioWorker<FANStudio.Model.FujianEEW>(new("wss://websocket.fanstudio.hk/fujian")),
_ => throw new InvalidOperationException("Unknown FAN Studio sub-type."),
},
GeoNetEventSourceConfig geoNet => BuildGeoNetWorker(new(new("https://api.geonet.org.nz/quake"), new("https://api.geonet.org.nz/quake/history/index"), new("https://api.geonet.org.nz/intensity/strong/processed/index")), geoNet),
GlobalQuakeServer15EventSourceConfig gq => new GlobalQuakeWorker15(gq.Host, gq.Port),
GlobalQuakeServerEventSourceConfig gq => new GlobalQuakeWorker(gq.Host, gq.Port),
JMAAtomEventSourceConfig jmaAtom => BuildJMAAtomWorkerFilter(new(new("https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml")), jmaAtom),
NOAAEventSourceConfig noaaAtom => noaaAtom.Subtype switch {
"PAAQ" => new NOAAAtomWorker(new("https://www.tsunami.gov/events/xml/PAAQAtom.xml"), new("https://www.tsunami.gov/"), new("/php/esri.php?e=t", UriKind.Relative), "PAAQ"),
"PHEB" => new NOAAAtomWorker(new("https://www.tsunami.gov/events/xml/PHEBAtom.xml"), new("https://www.tsunami.gov/"), new("/php/esri.php?e=t", UriKind.Relative), "PHEB"),
_ => throw new InvalidOperationException("Unknown NOAA sub-type."),
},
UpdateCheckerEventSourceConfig => new UpdateCheckerWorker(typeof(Worker).Assembly.GetName().Version?.ToString(3) ?? "", "unity"),
USGSEventSourceConfig usgs => BuildUSGSWorker(usgs.UseGeoJSONFeeds
? new USGSGeoJSONWorker(new Uri("https://earthquake.usgs.gov/earthquakes/feed/v1.0/geojson.php"))
: new USGSQuakeMLWorker(new Uri("https://earthquake.usgs.gov/earthquakes/feed/v1.0/quakeml.php"))
, usgs),
WolfxEventSourceConfig wolfx => BuildWolfxWorkerFilter(new WolfxWorker(new("wss://ws-api.wolfx.jp/all_eew")), wolfx),
_ => throw new InvalidOperationException("Unknown event source type."),
});
}
#endif
}
static JMAAtomWorker BuildJMAAtomWorkerFilter(JMAAtomWorker worker, JMAAtomEventSourceConfig config) {
if (config.Filter != null) worker.SetFilter(config.Filter);
worker.IsFilterWhitelist = config.IsFilterWhitelist;
return worker;
}
WolfxWorker BuildWolfxWorkerFilter(WolfxWorker worker, WolfxEventSourceConfig config) {
if (config.Filter != null) worker.SetFilter(config.Filter.Select(i => i switch {
"cenc_eew" => typeof(Wolfx.Model.CENCEEW),
"cenc_eqlist" => typeof(Wolfx.Model.WolfxEarthquakeList<Wolfx.Model.CENCEarthquake>),
"cwa_eew" => typeof(Wolfx.Model.CWAEEW),
"fj_eew" => typeof(Wolfx.Model.FujianEEW),
"jma_eew" => typeof(Wolfx.Model.JMAEEW),
"sc_eew" => typeof(Wolfx.Model.SichuanEEW),
_ => throw new InvalidOperationException("Unknown Wolfx event type."),
}));
worker.IsFilterWhitelist = config.IsFilterWhitelist;
_cencEarthquakeRVMGenerator.UseRawLocationName
= _cencEarthquakeTTSMessageGenerator.UseRawLocationName
= config.UseRawCENCLocationName;
return worker;
}
static BMKGOpenDataWorker BuildBMKGOpenDataWorkerUris(BMKGOpenDataWorker worker, BMKGOpenDataEventSourceConfig config) {
worker.SetDataUris(config.Subtypes.Select(i => new Uri(string.Format(CultureInfo.InvariantCulture, "https://data.bmkg.go.id/DataMKG/TEWS/{0}.json", i))));
return worker;
}
static GeoNetWorker BuildGeoNetWorker(GeoNetWorker worker, GeoNetEventSourceConfig pref) {
worker.MinimumMMI = pref.MinimumMMI;
worker.DoGetFullHistory = pref.DoGetFullHistory;
worker.DoGetStrongMotionInfo = pref.DoGetStrongMotionInfo;
return worker;
}
static USGSWorker BuildUSGSWorker(USGSWorker worker, USGSEventSourceConfig config) {
worker.SetFeedRelativeUri(new(string.Format(CultureInfo.InvariantCulture, "/earthquakes/feed/v1.0/summary/{0}{1}", config.Subtype, config.UseGeoJSONFeeds ? ".geojson" : ".quakeml"), UriKind.Relative));
if (worker is USGSGeoJSONWorker geojsonWorker) {
geojsonWorker.SetFilter(
config.Products
.Select(i => i.Split('/', 2))
.GroupBy(i => i[0])
.ToDictionary(g => g.Key, g => g.Select(i => i[1]))
);
}
return worker;
}
readonly OngoingReportManager _ongoingReportManager = new();
readonly ConcurrentQueue<Action> _uiActionQueue = new();
ReportViewModel _latestHistoryReport;
void OnReported(object sender, ReportViewModel e) {
if (e.Model is Exception && e.Model is not SourceWorkerNetworkException)
App.MainLogger.Log(4, "App", null, "Received an error from {0}: {1}", sender.GetType(), e.Model);
_grouper.Report(e);
_ongoingReportManager.Report(e);
_uiActionQueue.Enqueue(() => {
if (m_mapElementManager.OngoingCount == 0) {
m_mapElementManager.SetSelected(e);
m_cameraController.OnMapElementUpdated();
}
if (e.InvalidatedTime == null && (!(e.RevisionKey?.IsCancellation ?? false))) {
_latestHistoryReport = e;
}
});
}
void OnOngoingReported(ReportViewModel item, CollectionChangeAction action) {
if (action == CollectionChangeAction.Add) {
_uiActionQueue.Enqueue(() => {
if (m_mapElementManager.Add(item)) {
m_mapElementManager.SetSelected(null);
}
m_ongoingEventList.Add(item);
});
}
else if (action == CollectionChangeAction.Remove) {
_uiActionQueue.Enqueue(() => {
m_mapElementManager.Remove(item);
if (m_mapElementManager.Count == 0 && SharedSettings.Instance.DoSwitchBackToHistory && _latestHistoryReport is not null) {
m_mapElementManager.SetSelected(_latestHistoryReport, true);
}
m_ongoingEventList.Remove(item);
});
}
}
void OnGroupUpdated(object sender, ReportGroup e) {
_uiActionQueue.Enqueue(() => m_historyEventGroupList.UpdateGroup(e));
}
void OnGroupRemoved(object sender, ReportGroup e) {
_uiActionQueue.Enqueue(() => m_historyEventGroupList.RemoveGroup(e));
}
public void SetSelected(ReportViewModel e) {
m_mapElementManager.SetSelected(e);
m_cameraController.OnMapElementUpdated();
}
public void SetCurrent(ReportViewModel e) {
if (m_mapElementManager.SetCurrent(e)) {
m_cameraController.OnMapElementUpdated();
}
}
void Update() {
while (_uiActionQueue.TryDequeue(out var action)) {
action();
}
}
async Task ScheduledGatewayVerify(CancellationTokenSource source, CancellationToken cancellationToken) {
Exception lastEx = null;
for (int i = 0; i < 8; i++) {
await Task.Delay(TimeSpan.FromHours(3), cancellationToken).ConfigureAwait(true);
try {
await GatewayVerify(cancellationToken).ConfigureAwait(true);
i = -1;
}
catch (HttpRequestException ex) {
lastEx = ex;
}
catch (WebException ex) {
lastEx = ex;
}
catch (InvalidOperationException ex) {
lastEx = ex;
break;
}
}
if (lastEx != null) {
OnReported(this, new() { Title = lastEx.Message });
_verified = false;
}
source.Cancel();
}
#if UNITY_EDITOR
#pragma warning disable CS1998
#endif
static async Task GatewayVerify(CancellationToken cancellationToken) {
#if !UNITY_EDITOR
using var client = new HttpClient();
client.DefaultRequestHeaders.UserAgent.ParseAdd(SharedSettings.Instance.UnityUserAgent);
using var response = await client.GetAsync(new Uri("https://gateway.cryville.world/?rin=" + SharedSettings.Instance.Id), cancellationToken).ConfigureAwait(true);
if (response.StatusCode is >= ((HttpStatusCode)400) and < ((HttpStatusCode)500)) {
throw new InvalidOperationException(response.ReasonPhrase);
}
response.EnsureSuccessStatusCode();
#endif
}
#if UNITY_EDITOR
#pragma warning restore CS1998
#endif
}
}