cacto
abc83faf05
Слушает telemetry.new_reading (ручной ack/nack, реквизишн только при транспортных ошибках gRPC), кэширует активные automation_rules в памяти с периодическим обновлением из PostgreSQL (JOIN с devices — резолвит внутренние ID в external_id, которым оперируют MQTT/Redis/gRPC). Условия правил (>,<,>=,<=,=,!=) оцениваются обобщённо, без привязки к конкретным типам устройств. При срабатывании вызывает device-control-service по gRPC и публикует automation.rule_triggered в RabbitMQ. Проверено сквозным тестом через docker compose на полном пайплайне: mosquitto_pub → ingest-service (ClickHouse + telemetry.new_reading) → rule-engine-service (совпадение правила) → device-control-service (gRPC → Redis desired_state + MQTT-команда) → automation.rule_triggered. Показание ниже порога проверено отдельно — правило корректно не срабатывает.
96 lines
3.0 KiB
Go
96 lines
3.0 KiB
Go
// Package devicecontrolclient wraps the generated DeviceControl gRPC client,
|
|
// translating an automation_rules row's generic action_type/action_params
|
|
// into the specific TurnOn/TurnOff/SetLevel RPC.
|
|
package devicecontrolclient
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
|
|
devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control"
|
|
)
|
|
|
|
type Client struct {
|
|
conn *grpc.ClientConn
|
|
client devicecontrol.DeviceControlClient
|
|
}
|
|
|
|
func Connect(target string) (*Client, error) {
|
|
conn, err := grpc.NewClient(target, grpc.WithTransportCredentials(insecure.NewCredentials()))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("connect to device-control-service at %s: %w", target, err)
|
|
}
|
|
return &Client{conn: conn, client: devicecontrol.NewDeviceControlClient(conn)}, nil
|
|
}
|
|
|
|
func (c *Client) Close() error {
|
|
return c.conn.Close()
|
|
}
|
|
|
|
// newWithClient builds a Client around an already-constructed gRPC client,
|
|
// bypassing Connect. Used by tests to inject a fake devicecontrol.DeviceControlClient.
|
|
func newWithClient(client devicecontrol.DeviceControlClient) *Client {
|
|
return &Client{client: client}
|
|
}
|
|
|
|
// Result mirrors devicecontrol.CommandResult: Success/Error come from the
|
|
// device-control-service's business logic, not a transport-level failure.
|
|
type Result struct {
|
|
Success bool
|
|
Error string
|
|
}
|
|
|
|
// Dispatch issues actionType against deviceID, using params for actions that
|
|
// need them (currently just set_level's "level"). The returned error is only
|
|
// set for transport/protocol failures; a rejected command comes back as
|
|
// Result{Success: false, Error: "..."}.
|
|
func (c *Client) Dispatch(ctx context.Context, deviceID, actionType string, params map[string]any) (Result, error) {
|
|
switch actionType {
|
|
case "turn_on":
|
|
res, err := c.client.TurnOn(ctx, &devicecontrol.TurnOnRequest{DeviceId: deviceID})
|
|
if err != nil {
|
|
return Result{}, fmt.Errorf("turn_on rpc: %w", err)
|
|
}
|
|
return Result{Success: res.GetSuccess(), Error: res.GetError()}, nil
|
|
|
|
case "turn_off":
|
|
res, err := c.client.TurnOff(ctx, &devicecontrol.TurnOffRequest{DeviceId: deviceID})
|
|
if err != nil {
|
|
return Result{}, fmt.Errorf("turn_off rpc: %w", err)
|
|
}
|
|
return Result{Success: res.GetSuccess(), Error: res.GetError()}, nil
|
|
|
|
case "set_level":
|
|
level, ok := numericParam(params, "level")
|
|
if !ok {
|
|
return Result{}, fmt.Errorf("set_level requires a numeric %q param, got %+v", "level", params)
|
|
}
|
|
res, err := c.client.SetLevel(ctx, &devicecontrol.SetLevelRequest{DeviceId: deviceID, Level: level})
|
|
if err != nil {
|
|
return Result{}, fmt.Errorf("set_level rpc: %w", err)
|
|
}
|
|
return Result{Success: res.GetSuccess(), Error: res.GetError()}, nil
|
|
|
|
default:
|
|
return Result{}, fmt.Errorf("unknown action_type %q", actionType)
|
|
}
|
|
}
|
|
|
|
func numericParam(params map[string]any, key string) (float64, bool) {
|
|
v, ok := params[key]
|
|
if !ok {
|
|
return 0, false
|
|
}
|
|
switch n := v.(type) {
|
|
case float64:
|
|
return n, true
|
|
case int:
|
|
return float64(n), true
|
|
default:
|
|
return 0, false
|
|
}
|
|
}
|