wsy182/tdEngine_mqtt_mock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(core): 重构主应用启动逻辑并改进WITS数据发送

Browse Source 
- 将主应用改为同时启动MQTT订阅入库和WITS数据发送两个服务
- 实现WITS发送器的自动重连机制和连接状态管理
- 添加日志记录到log/app.log和错误日志到log/error.log
- 更新WITS通道映射定义并支持字符串类型的日期时间字段
- 修改数据入库逻辑以支持空值处理和类型转换容错
- 移除命令行子命令模式，改为配置文件驱动的参数设置
- 添加.vscode和log目录到.gitignore忽略列表
This commit is contained in:
wsy182
2026-03-12 13:32:27 +08:00
parent 6557479a2f
commit 0a123ba210
7 changed files with 314 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@@ -9,4 +9,6 @@
.pytest_cache/ .pytest_cache/
.pytest_ .pytest_
.vscode .vscode
log/*

217
app/wits_sender.py
View File

@@ -7,74 +7,79 @@ from datetime import datetime
from pathlib import Path from pathlib import Path
from config import build_wits_sender_dependencies from config import build_wits_sender_dependencies
from model import WITS_FIELD_MAPPING, WitsData from model import WITS_CHANNEL_MAPPING, WitsData
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
BEGIN_MARK = "&&\r\n" BEGIN_MARK = "&&\r\n"
END_MARK = "!!\r\n" END_MARK = "!!\r\n"
RECORD_TERMINATOR = "*\r\n"
RECONNECT_DELAY = 3
def rand_int(a, b): def rand_int(a, b):
return random.randint(a, b) return random.randint(a, b)
def rand_float(a, b, digits=2): def rand_float(a, b, digits=6):
return round(random.uniform(a, b), digits) return round(random.uniform(a, b), digits)
def build_random_wits_data(device_code): def build_random_wits_data(device_code):
now = datetime.now() now = datetime.now()
ts_ms = int(time.time() * 1000) ts_ms = int(time.time() * 1000)
hook_load = rand_float(17.3, 18.8)
standpipe_pressure = rand_float(990.0, 1012.0)
mud_density = rand_float(1069.8, 1070.1)
return WitsData( return WitsData(
ts=ts_ms, ts=ts_ms,
wellid=device_code, wellid="???1",
stknum=rand_int(0, 500), stknum=0,
recid=0, recid=1,
seqid=rand_int(1, 999999), seqid=rand_int(1600, 9999),
actual_date=float(now.strftime("%Y%m%d")), actual_date=now.strftime("%y%m%d"),
actual_time=float(now.strftime("%H%M%S")), actual_time=now.strftime("%H%M%S"),
actual_ts=ts_ms, actual_ts=ts_ms,
actcod=rand_int(0, 9), actcod=37,
actod_label="AUTO", actod_label="AUTO",
deptbitm=rand_float(0, 5000), deptbitm=200.0,
deptbitv=rand_float(0, 5000), deptbitv=198.551422,
deptmeas=rand_float(0, 5000), deptmeas=200.0,
deptvert=rand_float(0, 5000), deptvert=198.551422,
blkpos=rand_float(0, 100), blkpos=6.001850,
ropa=rand_float(0, 200), ropa=0.0,
hkla=rand_float(0, 500), hkla=hook_load,
hklx=rand_float(0, 500), hklx=hook_load,
woba=rand_float(0, 200), woba=0.0,
wobx=rand_float(0, 200), wobx=-hook_load,
torqa=rand_float(0, 200), torqa=0.0,
torqx=rand_float(0, 200), torqx=0.0,
rpma=rand_int(0, 300), rpma=0,
sppa=rand_float(0, 5000), sppa=standpipe_pressure,
chkp=rand_float(0, 5000), chkp=0.0,
spm1=rand_int(0, 200), spm1=0,
spm2=rand_int(0, 200), spm2=0,
spm3=rand_int(0, 200), spm3=0,
tvolact=rand_float(0, 20000), tvolact=0.0,
tvolcact=rand_float(0, 20000), tvolcact=0.0,
mfop=rand_int(0, 1000), mfop=0,
mfoa=rand_float(0, 1000), mfoa=0.0,
mfia=rand_float(0, 1000), mfia=0.0,
mdoa=rand_float(0, 1000), mdoa=mud_density,
mdia=rand_float(0, 1000), mdia=26.846003,
mtoa=rand_float(0, 1000), mtoa=29.113855,
mtia=rand_float(0, 1000), mtia=346.874634,
mcoa=rand_float(0, 1000), mcoa=241.874634,
mcia=rand_float(0, 1000), mcia=0.0,
stkc=rand_int(0, 200), stkc=0,
lagstks=rand_int(0, 200), lagstks=0,
deptretm=rand_float(0, 5000), deptretm=200.0,
gasa=rand_float(0, 100), gasa=0.0,
space1=rand_float(0, 10), space1=0.0,
space2=rand_float(0, 10), space2=0.0,
space3=rand_float(0, 10), space3=0.0,
space4=rand_float(0, 10), space4=0.0,
space5=rand_float(0, 10), space5=0.0,
) )
@@ -83,15 +88,14 @@ def format_wits_value(value, kind):
return str(value) return str(value)
if kind == "int": if kind == "int":
return str(int(value)) return str(int(value))
return f"{float(value):.2f}" if kind == "float6":
return f"{float(value):.6f}"
return str(value)
def build_wits_packet(data): def build_wits_packet(data):
lines = [] lines = [f"{channel}{format_wits_value(getattr(data, field_name), kind)}" for channel, field_name, kind in WITS_CHANNEL_MAPPING]
for index, field_name, kind in WITS_FIELD_MAPPING: return BEGIN_MARK + "\r\n".join(lines) + "\r\n" + END_MARK + RECORD_TERMINATOR
value = getattr(data, field_name)
lines.append(f"{index:02d}{format_wits_value(value, kind)}")
return BEGIN_MARK + "\r\n".join(lines) + "\r\n" + END_MARK
def normalize_packet(text): def normalize_packet(text):
@@ -99,49 +103,116 @@ def normalize_packet(text):
lines = [line.rstrip() for line in body.split("\n") if line.strip()] lines = [line.rstrip() for line in body.split("\n") if line.strip()]
if lines and lines[0] == "&&": if lines and lines[0] == "&&":
lines = lines[1:] lines = lines[1:]
if lines and lines[-1] == "*":
lines = lines[:-1]
if lines and lines[-1] == "!!": if lines and lines[-1] == "!!":
lines = lines[:-1] lines = lines[:-1]
return BEGIN_MARK + "\r\n".join(lines) + "\r\n" + END_MARK return BEGIN_MARK + "\r\n".join(lines) + "\r\n" + END_MARK + RECORD_TERMINATOR
def load_packet_from_file(path): def load_packet_from_file(path):
return normalize_packet(Path(path).read_text(encoding="utf-8-sig")) return normalize_packet(Path(path).read_text(encoding="utf-8-sig"))
def send_packet(host, port, timeout, packet): def open_connection(host, port, timeout):
with socket.create_connection((host, port), timeout=timeout) as sock: sock = socket.create_connection((host, port), timeout=timeout)
sock.sendall(packet.encode("ascii", errors="strict")) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
sock.settimeout(timeout)
return sock
def send_packet(sock, packet):
sock.sendall(packet.encode("ascii", errors="strict"))
def run_wits_sender(args, deps): def run_wits_sender(args, deps):
wits_config = deps.config.wits wits_config = deps.config.wits
device_code = deps.config.tms.device_code
source_file = args.source_file or wits_config.source_file source_file = args.source_file or wits_config.source_file
host = args.host or wits_config.host host = args.host or wits_config.host
port = args.port or wits_config.port port = args.port or wits_config.port
timeout = args.timeout or wits_config.timeout timeout = args.timeout or wits_config.timeout
interval = args.interval or 2.0
if not host or not port: if not host or not port:
raise ValueError("WITS target host/port is empty. Configure wits.host/wits.port or tms.server-ip/tms.server-port") raise ValueError("WITS target host/port is empty. Configure wits.host/wits.port or tms.server-ip/tms.server-port")
logger.info("WITS sender config host=%s port=%s timeout=%ss source_file=%s interval=%ss count=%s", host, port, timeout, source_file or "(generated)", args.interval, args.count or "forever") logger.info(
"WITS sender config host=%s port=%s timeout=%ss source_file=%s interval=%ss count=%s",
host,
port,
timeout,
source_file or "(generated)",
interval,
args.count or "forever",
)
seq = 0 seq = 0
sock = None
try: try:
while True: while True:
seq += 1 if sock is None:
if source_file: try:
packet = load_packet_from_file(source_file) sock = open_connection(host, port, timeout)
else: logger.info("WITS connected %s:%s", host, port)
packet = build_wits_packet(build_random_wits_data(device_code)) except ConnectionRefusedError:
send_packet(host, port, timeout, packet) logger.warning("WITS target refused connection %s:%s, retry in %ss", host, port, RECONNECT_DELAY)
logger.info("TX WITS #%s -> %s:%s", seq, host, port) time.sleep(RECONNECT_DELAY)
if logger.isEnabledFor(logging.DEBUG): continue
logger.debug("WITS packet:\n%s", packet) except TimeoutError:
if args.count and seq >= args.count: logger.warning("WITS connect timeout %s:%s, retry in %ss", host, port, RECONNECT_DELAY)
break time.sleep(RECONNECT_DELAY)
time.sleep(args.interval) continue
except OSError as exc:
logger.error("WITS connect failed %s:%s (%s), retry in %ss", host, port, exc, RECONNECT_DELAY)
time.sleep(RECONNECT_DELAY)
continue
try:
seq += 1
if source_file:
packet = load_packet_from_file(source_file)
else:
packet = build_wits_packet(build_random_wits_data(deps.config.tms.device_code))
send_packet(sock, packet)
logger.info("TX WITS #%s -> %s:%s", seq, host, port)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("WITS packet:\n%s", packet)
if args.count and seq >= args.count:
break
time.sleep(interval)
except (BrokenPipeError, ConnectionResetError):
logger.warning("WITS connection dropped by remote host, reconnecting in %ss", RECONNECT_DELAY)
try:
sock.close()
except OSError:
pass
sock = None
time.sleep(RECONNECT_DELAY)
except TimeoutError:
logger.warning("WITS send timeout, reconnecting in %ss", RECONNECT_DELAY)
try:
sock.close()
except OSError:
pass
sock = None
time.sleep(RECONNECT_DELAY)
except OSError as exc:
logger.error("WITS send failed (%s), reconnecting in %ss", exc, RECONNECT_DELAY)
try:
sock.close()
except OSError:
pass
sock = None
time.sleep(RECONNECT_DELAY)
except KeyboardInterrupt: except KeyboardInterrupt:
logger.info("WITS sender interrupted") logger.info("WITS sender interrupted")
finally:
if sock is not None:
try:
sock.close()
except OSError:
pass
logger.info("WITS disconnected")
def add_arguments(parser): def add_arguments(parser):
@@ -150,8 +221,8 @@ def add_arguments(parser):
parser.add_argument("--port", type=int, default=0, help="Override target port") parser.add_argument("--port", type=int, default=0, help="Override target port")
parser.add_argument("--timeout", type=int, default=0, help="Override socket timeout") parser.add_argument("--timeout", type=int, default=0, help="Override socket timeout")
parser.add_argument("--source-file", default="", help="Send raw WITS packet from file") parser.add_argument("--source-file", default="", help="Send raw WITS packet from file")
parser.add_argument("--interval", type=float, default=3.0, help="Send interval in seconds") parser.add_argument("--interval", type=float, default=2.0, help="Send interval in seconds")
parser.add_argument("--count", type=int, default=1, help="Send count (0 = forever)") parser.add_argument("--count", type=int, default=0, help="Send count (0 = forever)")
def main(argv=None): def main(argv=None):

29
db/orm.py
View File

@@ -1,4 +1,5 @@
import re import logging
import re
from model import DrillingRealtimeData from model import DrillingRealtimeData
@@ -88,14 +89,34 @@ class DrillingRealtimeORM:
table_name = sanitize_identifier(f"drilling_realtime_{equipment_code}", "drilling_realtime_default") table_name = sanitize_identifier(f"drilling_realtime_{equipment_code}", "drilling_realtime_default")
values = [] values = []
for column in DB_COLUMNS: for column in DB_COLUMNS:
raw = getattr(entity, column) raw = getattr(entity, column, None)
if raw is None:
logging.debug("Column %s is None, default=0", column)
if column in INT_COLUMNS or column == "ts": if column in INT_COLUMNS or column == "ts":
values.append(str(int(raw))) values.append(str(to_int(raw)))
else: else:
values.append(str(float(raw))) values.append(str(to_float(raw)))
columns_sql = ", ".join([f"`{column}`" for column in DB_COLUMNS]) columns_sql = ", ".join([f"`{column}`" for column in DB_COLUMNS])
values_sql = ", ".join(values) values_sql = ", ".join(values)
return ( return (
f"INSERT INTO `{self.database}`.`{table_name}` USING `{self.database}`.`{self.stable}` " f"INSERT INTO `{self.database}`.`{table_name}` USING `{self.database}`.`{self.stable}` "
f"TAGS ({sql_quote(equipment_code)}) ({columns_sql}) VALUES ({values_sql})" f"TAGS ({sql_quote(equipment_code)}) ({columns_sql}) VALUES ({values_sql})"
) )
def to_int(value, default=0):
try:
if value is None or value == "":
return default
return int(value)
except Exception:
return default
def to_float(value, default=0.0):
try:
if value is None or value == "":
return default
return float(value)
except Exception:
return default

125
main.py
View File

@@ -1,54 +1,117 @@
import argparse import argparse
import logging import logging
import os
import threading
import time
from pathlib import Path
from types import SimpleNamespace
from app import mqtt_mock, mqtt_sender, mqtt_subscriber, wits_sender from app import mqtt_mock, wits_sender
from config import build_mock_dependencies, build_wits_sender_dependencies
logger = logging.getLogger(__name__)
BASE_DIR = Path(__file__).resolve().parent
def configure_logging(level_name): def configure_logging(level_name):
log_dir = BASE_DIR / "log"
os.makedirs(log_dir, exist_ok=True)
level = getattr(logging, str(level_name).upper(), logging.INFO) level = getattr(logging, str(level_name).upper(), logging.INFO)
logging.basicConfig( formatter = logging.Formatter("%(asctime)s %(levelname)s %(name)s - %(message)s")
level=level,
format="%(asctime)s %(levelname)s %(name)s - %(message)s", root_logger = logging.getLogger()
) root_logger.handlers.clear()
root_logger.setLevel(level)
console_handler = logging.StreamHandler()
console_handler.setLevel(level)
console_handler.setFormatter(formatter)
app_file_handler = logging.FileHandler(log_dir / "app.log", encoding="utf-8")
app_file_handler.setLevel(level)
app_file_handler.setFormatter(formatter)
error_file_handler = logging.FileHandler(log_dir / "error.log", encoding="utf-8")
error_file_handler.setLevel(logging.ERROR)
error_file_handler.setFormatter(formatter)
root_logger.addHandler(console_handler)
root_logger.addHandler(app_file_handler)
root_logger.addHandler(error_file_handler)
def build_parser(): def build_parser():
parser = argparse.ArgumentParser(description="tdEngine mqtt/wits mock application") parser = argparse.ArgumentParser(description="tdEngine mqtt/wits mock application")
parser.add_argument("--config", default="config.yaml", help="Path to config yaml")
parser.add_argument("--log-level", default="INFO", help="Logging level") parser.add_argument("--log-level", default="INFO", help="Logging level")
subparsers = parser.add_subparsers(dest="command", required=True) parser.add_argument("--data-file", default="", help="Override data-file in config")
parser.add_argument("--wits-host", default="", help="Override WITS target host")
mqtt_mock_parser = subparsers.add_parser("mqtt-mock", help="Run MQTT mock service") parser.add_argument("--wits-port", type=int, default=0, help="Override WITS target port")
mqtt_mock.add_arguments(mqtt_mock_parser) parser.add_argument("--wits-timeout", type=int, default=0, help="Override WITS socket timeout")
parser.add_argument("--wits-source-file", default="", help="Send raw WITS packet from file")
mqtt_sender_parser = subparsers.add_parser("mqtt-sender", help="Run MQTT sender")
mqtt_sender.add_arguments(mqtt_sender_parser)
mqtt_subscriber_parser = subparsers.add_parser("mqtt-subscriber", help="Run MQTT subscriber")
mqtt_subscriber.add_arguments(mqtt_subscriber_parser)
wits_sender_parser = subparsers.add_parser("wits-sender", help="Run WITS sender")
wits_sender.add_arguments(wits_sender_parser)
return parser return parser
def build_mqtt_args(args):
return SimpleNamespace(
config=args.config,
mode="listen",
interval=2.0,
count=0,
data_file=args.data_file,
)
def build_wits_args(args):
return SimpleNamespace(
config=args.config,
host=args.wits_host,
port=args.wits_port,
timeout=args.wits_timeout,
source_file=args.wits_source_file,
interval=2.0,
count=0,
)
def start_worker(name, target, args, deps):
thread = threading.Thread(target=target, args=(args, deps), name=name, daemon=True)
thread.start()
return thread
def main(argv=None): def main(argv=None):
parser = build_parser() parser = build_parser()
args = parser.parse_args(argv) args = parser.parse_args(argv)
configure_logging(args.log_level) configure_logging(args.log_level)
logging.getLogger(__name__).info("start app command=%s", args.command)
if args.command == "mqtt-mock": logger.info("start app")
deps = mqtt_mock.build_mock_dependencies(args.config, data_file_override=args.data_file)
mqtt_mock.run_mock_service(args, deps) mqtt_args = build_mqtt_args(args)
elif args.command == "mqtt-sender": wits_args = build_wits_args(args)
deps = mqtt_sender.build_sender_dependencies(args.config)
mqtt_sender.run_sender(args, deps) mqtt_deps = build_mock_dependencies(mqtt_args.config, data_file_override=mqtt_args.data_file)
elif args.command == "mqtt-subscriber": wits_deps = build_wits_sender_dependencies(wits_args.config)
deps = mqtt_subscriber.build_subscriber_dependencies(args.config)
mqtt_subscriber.run_subscriber(args, deps) threads = [
elif args.command == "wits-sender": start_worker("mqtt-mock", mqtt_mock.run_mock_service, mqtt_args, mqtt_deps),
deps = wits_sender.build_wits_sender_dependencies(args.config) start_worker("wits-sender", wits_sender.run_wits_sender, wits_args, wits_deps),
wits_sender.run_wits_sender(args, deps) ]
logger.info("services started: mqtt subscription+ingest and wits sender")
try:
while True:
for thread in threads:
if not thread.is_alive():
raise RuntimeError(f"worker stopped unexpectedly: {thread.name}")
time.sleep(1)
except KeyboardInterrupt:
logger.info("shutdown requested")
except Exception:
logger.exception("application stopped unexpectedly")
raise
if __name__ == "__main__": if __name__ == "__main__":

6
model/__init__.py
View File

@@ -1,6 +1,9 @@
from model.config import AppConfig, MqttConfig, TdengineConfig, TmsConfig, WitsConfig from model.config import AppConfig, MqttConfig, TdengineConfig, TmsConfig, WitsConfig
from model.drilling import DrillingRealtimeData from model.drilling import DrillingRealtimeData
from model.wits import WITS_FIELD_MAPPING, WitsData from model.wits import WITS_CHANNEL_MAPPING, WitsData
# Backward-compatible alias for older imports.
WITS_FIELD_MAPPING = WITS_CHANNEL_MAPPING
__all__ = [ __all__ = [
"AppConfig", "AppConfig",
@@ -8,6 +11,7 @@ __all__ = [
"MqttConfig", "MqttConfig",
"TdengineConfig", "TdengineConfig",
"TmsConfig", "TmsConfig",
"WITS_CHANNEL_MAPPING",
"WITS_FIELD_MAPPING", "WITS_FIELD_MAPPING",
"WitsConfig", "WitsConfig",
"WitsData", "WitsData",

86
model/wits.py
View File

@@ -8,8 +8,8 @@ class WitsData:
stknum: int stknum: int
recid: int recid: int
seqid: int seqid: int
actual_date: float actual_date: str
actual_time: float actual_time: str
actual_ts: int actual_ts: int
actcod: int actcod: int
actod_label: str actod_label: str
@@ -53,50 +53,40 @@ class WitsData:
space5: float space5: float
WITS_FIELD_MAPPING = [ WITS_CHANNEL_MAPPING = [
(1, "wellid", "string"), ("0101", "wellid", "string"),
(2, "stknum", "int"), ("0102", "stknum", "int"),
(3, "recid", "int"), ("0103", "recid", "int"),
(4, "seqid", "int"), ("0104", "seqid", "int"),
(5, "actual_date", "float"), ("0105", "actual_date", "string"),
(6, "actual_time", "float"), ("0106", "actual_time", "string"),
(7, "actcod", "int"), ("0107", "actcod", "int"),
(8, "deptbitm", "float"), ("0108", "deptbitm", "float6"),
(9, "deptbitv", "float"), ("0109", "deptbitv", "float6"),
(10, "deptmeas", "float"), ("0110", "deptmeas", "float6"),
(11, "deptvert", "float"), ("0111", "deptvert", "float6"),
(12, "blkpos", "float"), ("0112", "blkpos", "float6"),
(13, "ropa", "float"), ("0113", "ropa", "float6"),
(14, "hkla", "float"), ("0114", "hkla", "float6"),
(15, "hklx", "float"), ("0116", "woba", "float6"),
(16, "woba", "float"), ("0117", "wobx", "float6"),
(17, "wobx", "float"), ("0118", "torqa", "float6"),
(18, "torqa", "float"), ("0119", "torqx", "float6"),
(19, "torqx", "float"), ("0120", "rpma", "int"),
(20, "rpma", "int"), ("0121", "sppa", "float6"),
(21, "sppa", "float"), ("0123", "spm1", "int"),
(22, "chkp", "float"), ("0124", "spm2", "int"),
(23, "spm1", "int"), ("0125", "spm3", "int"),
(24, "spm2", "int"), ("0126", "tvolact", "float6"),
(25, "spm3", "int"), ("0127", "tvolcact", "float6"),
(26, "tvolact", "float"), ("0128", "mfop", "int"),
(27, "tvolcact", "float"), ("0130", "mfoa", "float6"),
(28, "mfop", "int"), ("0131", "mfia", "float6"),
(29, "mfoa", "float"), ("0132", "mdoa", "float6"),
(30, "mfia", "float"), ("0133", "mdia", "float6"),
(31, "mdoa", "float"), ("0134", "mtoa", "float6"),
(32, "mdia", "float"), ("0135", "mtia", "float6"),
(33, "mtoa", "float"), ("0136", "mcoa", "float6"),
(34, "mtia", "float"), ("0137", "stkc", "int"),
(35, "mcoa", "float"), ("0139", "deptretm", "float6"),
(36, "mcia", "float"),
(37, "stkc", "int"),
(38, "lagstks", "int"),
(39, "deptretm", "float"),
(40, "gasa", "float"),
(41, "space1", "float"),
(42, "space2", "float"),
(43, "space3", "float"),
(44, "space4", "float"),
(45, "space5", "float"),
] ]

6
requirements.md
View File

@@ -21,6 +21,10 @@
程序启动入口main.py,使用logging记录日志。 程序启动入口main.py,使用logging记录日志。
启动是同时启动wits数据模拟和mqtt消息订阅和入库。不需要选择功能。
将业务对象抽取到model包下。 将业务对象抽取到model包下。
数据库实体放在model包下。 数据库实体放在model包下。
将日志记录到log/app.log,错误日志记录到log/error.log