AI/HPC帶動台灣封測廠8月營收創高：製程、冷卻、氣體、設備都需要溫壓監測完全指南

import logging
import time
from pymodbus.client import ModbusSerialClient

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("atlantis_gateway")

# ------------------------------------------------------------
# 設定區塊：請依現場實際配置調整
# ------------------------------------------------------------
SERIAL_PORT = "/dev/ttyUSB0"      # RS485轉USB埠號，Windows請改為 "COM3" 等
BAUDRATE = 9600
PARITY = "N"
STOPBITS = 1
BYTESIZE = 8

# 感測器對應表：{Modbus從站位址: (名稱, 暫存器位址, 縮放係數, 單位)}
SENSOR_MAP = {
    1: ("冷卻水主幹壓力",   0, 0.01, "MPa"),
    2: ("迴焊爐第一區溫度", 0, 0.1,  "°C"),
    3: ("特殊氣體管路壓力", 0, 0.01, "MPa"),
}

client = ModbusSerialClient(
    port=SERIAL_PORT,
    baudrate=BAUDRATE,
    parity=PARITY,
    stopbits=STOPBITS,
    bytesize=BYTESIZE,
    timeout=1,
)


def read_sensor(slave_id: int, register: int, scale: float):
    """讀取單一儀表的16位元有號整數暫存器並套用縮放係數"""
    result = client.read_holding_registers(address=register, count=1, slave=slave_id)
    if result.isError():
        logger.warning(f"讀取從站 {slave_id} 失敗：{result}")
        return None
    raw_value = result.registers[0]
    # 若暫存器為有號數，需視儀表規格判斷是否要做補數轉換
    if raw_value > 32767:
        raw_value -= 65536
    return round(raw_value * scale, 3)


def poll_all_sensors():
    """輪詢所有已配置的儀表，回傳目前讀值字典"""
    readings = {}
    for slave_id, (name, register, scale, unit) in SENSOR_MAP.items():
        value = read_sensor(slave_id, register, scale)
        if value is not None:
            readings[name] = {"value": value, "unit": unit}
            logger.info(f"{name}: {value} {unit}")
    return readings


if __name__ == "__main__":
    if not client.connect():
        logger.error("無法連接RS485介面，請檢查接線與埠號設定")
        raise SystemExit(1)

    try:
        while True:
            data = poll_all_sensors()
            # 此處可接續呼叫下方 mqtt_publisher.py 的發佈函式
            time.sleep(10)  # 每10秒輪詢一次，可依需求調整
    except KeyboardInterrupt:
        logger.info("程式中止，關閉連線")
    finally:
        client.close()

import json
import logging
import paho.mqtt.client as mqtt

logger = logging.getLogger("atlantis_mqtt")

MQTT_BROKER_HOST = "192.168.1.100"   # 請改為客戶端自建或既有的MQTT broker位址
MQTT_BROKER_PORT = 1883
MQTT_TOPIC_PREFIX = "factory/osat/sensors"


def on_connect(client, userdata, flags, rc, properties=None):
    if rc == 0:
        logger.info("已連線至MQTT Broker")
    else:
        logger.error(f"MQTT連線失敗，錯誤碼：{rc}")


def on_publish(client, userdata, mid, reason_code=None, properties=None):
    logger.debug(f"訊息已發佈，mid={mid}")


def build_mqtt_client():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    client.on_connect = on_connect
    client.on_publish = on_publish
    client.connect(MQTT_BROKER_HOST, MQTT_BROKER_PORT, keepalive=60)
    client.loop_start()
    return client


def publish_readings(client, readings: dict):
    """將輪詢取得的儀表數據以JSON格式發佈至對應主題"""
    for sensor_name, payload in readings.items():
        topic = f"{MQTT_TOPIC_PREFIX}/{sensor_name}"
        message = json.dumps(payload, ensure_ascii=False)
        client.publish(topic, message, qos=1)


# 使用範例（與 modbus_reader.py 整合）：
#
# from modbus_reader import poll_all_sensors
# mqtt_client = build_mqtt_client()
# while True:
#     data = poll_all_sensors()
#     publish_readings(mqtt_client, data)
#     time.sleep(10)

import json
import logging
from awsiot import mqtt_connection_builder
from awscrt import io, mqtt

logger = logging.getLogger("atlantis_aws_bridge")

# ------------------------------------------------------------
# 以下AWS相關參數需由客戶端自行於AWS IoT Core主控台建立並取得，
# 昶特ATLANTIS不提供AWS帳號、憑證或雲端資源建置服務。
# ------------------------------------------------------------
AWS_IOT_ENDPOINT = "your-endpoint.iot.ap-northeast-1.amazonaws.com"
CERT_PATH = "/path/to/certificate.pem.crt"
PRIVATE_KEY_PATH = "/path/to/private.pem.key"
ROOT_CA_PATH = "/path/to/AmazonRootCA1.pem"
CLIENT_ID = "osat-factory-gateway-01"
AWS_TOPIC = "factory/osat/sensors/cloud"

event_loop_group = io.EventLoopGroup(1)
host_resolver = io.DefaultHostResolver(event_loop_group)
client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)


def build_aws_connection():
    """建立與AWS IoT Core的MQTT連線（前提：客戶已完成憑證與Thing註冊）"""
    connection = mqtt_connection_builder.mtls_from_path(
        endpoint=AWS_IOT_ENDPOINT,
        cert_filepath=CERT_PATH,
        pri_key_filepath=PRIVATE_KEY_PATH,
        ca_filepath=ROOT_CA_PATH,
        client_bootstrap=client_bootstrap,
        client_id=CLIENT_ID,
        clean_session=False,
        keep_alive_secs=30,
    )
    connect_future = connection.connect()
    connect_future.result()
    logger.info("已連線至AWS IoT Core")
    return connection


def forward_to_cloud(connection, sensor_name: str, payload: dict):
    """將單筆感測器數據轉發至AWS IoT Core對應主題"""
    message = json.dumps({"sensor": sensor_name, **payload}, ensure_ascii=False)
    connection.publish(
        topic=f"{AWS_TOPIC}/{sensor_name}",
        payload=message,
        qos=mqtt.QoS.AT_LEAST_ONCE,
    )


# 使用範例：將本地MQTT訂閱到的訊息轉發至AWS
#
# aws_conn = build_aws_connection()
# for sensor_name, payload in readings.items():
#     forward_to_cloud(aws_conn, sensor_name, payload)