SunYF/bj_power
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
380715f8a9d8e5223577143fa6ddd8df5cbbfa85
bj_power/bj_power_mes/docs/superpowers/plans/2026-05-16-machine-actor-plan.md
T

1284 lines
32 KiB
Markdown
Raw Normal View History

chore: init bj_power monorepo (dashboard/mes/wms/wms_client/workstation), clear subproject git metadata and align naming to folder names
2026-08-27 10:09:54 +08:00
# Machine Actor 实现计划
> **For agentic workers:** REQUIRED: Use superpowers:executing-plans to implement this plan. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** 用 Actor 模型（goroutine + channel）替换 Station/SignalWatcher/TempSlotAllocator，每个设备自治管理槽位状态。
**Architecture:** 新建 `internal/actor/` 包，包含 MachineActor、TempStoreActor、SignalRouter 和共享类型。EventLoop 移除 equipment_slot 写入逻辑，调度器从 Actor 快照读设备/暂存台状态。
**Tech Stack:** Go 1.25, ent ORM, go-zero, PostgreSQL
**Spec:** `docs/superpowers/specs/2026-05-16-machine-actor-design.md`
---
## 文件结构
| 文件 | 操作 | 职责 |
|------|------|------|
| `internal/actor/types.go` | 新建 | 共享类型（MachineMsg, MachineSnapshot, ActorStatus, SignalEvent） |
| `internal/actor/signal_router.go` | 新建 | SignalRouter 集中 PLC 轮询 + 路由 |
| `internal/actor/machine_actor.go` | 新建 | MachineActor 状态机 |
| `internal/actor/machine_actor_test.go` | 新建 | MachineActor 单元测试 |
| `internal/actor/tempstore_actor.go` | 新建 | TempStoreActor 暂存台管理 |
| `internal/actor/tempstore_actor_test.go` | 新建 | TempStoreActor 单元测试 |
| `internal/eventloop/dbstate.go` | 修改 | 新增 SetJobTempSlot / ClearJobTempSlot |
| `internal/eventloop/loop.go` | 修改 | 移除 updateStationDone/batchMachines/inspectionMachines；handleMachineDone 精简 |
| `internal/eventloop/scheduler_bridge.go` | 修改 | buildSystemState 改读 Actor 快照 |
| `internal/eventloop/worker_dispatch.go` | 修改 | 上料/下料完成后通知 Actor |
| `internal/svc/service_context.go` | 修改 | 用 Actor 替换 Station/SignalWatcher |
| `internal/station/*.go` | 删除 | 保留 HandheldTool 接口移到新位置 |
| `internal/processor/signal_watcher.go` | 删除 | 被 SignalRouter 替代 |
| `internal/processor/replenisher.go` | 修改 | 暂存台查询改为读 TempStoreActor 快照 |
---
## Chunk 1: Actor 类型 + DBState 扩展
### Task 1.1: 定义 Actor 共享类型
**Files:**
- Create: `internal/actor/types.go`
- [ ] **Step 1: 创建 types.go**
```go
package actor
import "time"
// ActorStatus 工站状态
type ActorStatus string
const (
ActorIdle ActorStatus = "Idle"
ActorProcessing ActorStatus = "Processing"
ActorFull ActorStatus = "Full"
ActorWaiting ActorStatus = "Waiting"
ActorFault ActorStatus = "Fault"
)
// SlotStatus 槽位状态
type SlotStatus string
const (
SlotEmpty SlotStatus = "Empty"
SlotOccupied SlotStatus = "Occupied"
SlotDone SlotStatus = "Done"
)
// MachineMsg 外部发给 Actor 的命令
type MachineMsg struct {
Type string // LOAD_COMPLETE / UNLOAD_COMPLETE / RESTORE / ALLOCATE / RELEASE
JobID int
SlotNo int
Params map[string]any
}
// MachineSnapshot 设备 Actor 只读快照
type MachineSnapshot struct {
ID int
Type string
Status ActorStatus
Slots []SlotSnapshot
UpdatedAt time.Time
}
// SlotSnapshot 槽位快照
type SlotSnapshot struct {
SlotNo int
Status SlotStatus
JobID int
OccupiedAt time.Time
}
// TempStoreSnapshot 暂存台快照
type TempStoreSnapshot struct {
Capacity int
Slots []int // slotNo-1 → jobID (0 = 空), 复制
}
// SignalEvent PLC 信号事件
type SignalEvent struct {
Type SignalType
Value bool
}
type SignalType int
const (
SignalDone SignalType = iota
SignalNG
)
```
- [ ] **Step 2: 验证编译**
```bash
go build ./internal/actor/...
```
### Task 1.2: DBState 新增暂存台 DB 方法
**Files:**
- Modify: `internal/eventloop/dbstate.go`
- [ ] **Step 1: 添加 SetJobTempSlot / ClearJobTempSlot**
在 `internal/eventloop/dbstate.go` 末尾添加：
```go
// SetJobTempSlot 设置 job.temp_slot_no
func (d *DBState) SetJobTempSlot(ctx context.Context, jobID, slotNo int) error {
_, err := d.client.Job.UpdateOneID(jobID).
SetTempSlotNo(slotNo).
AddVersion(1).
Save(ctx)
if err != nil {
return fmt.Errorf("set job %d temp slot %d: %w", jobID, slotNo, err)
}
return nil
}
// ClearJobTempSlot 清除 job.temp_slot_no
func (d *DBState) ClearJobTempSlot(ctx context.Context, jobID int) error {
_, err := d.client.Job.UpdateOneID(jobID).
ClearTempSlotNo().
AddVersion(1).
Save(ctx)
if err != nil {
return fmt.Errorf("clear job %d temp slot: %w", jobID, err)
}
return nil
}
```
- [ ] **Step 2: 验证编译和已有测试**
```bash
go build ./...
go test ./internal/eventloop/...
```
- [ ] **Step 3: 提交**
```bash
git add internal/actor/types.go internal/eventloop/dbstate.go
git commit -m "feat: add Actor types and DBState temp slot methods"
```
---
## Chunk 2: SignalRouter
### Task 2.1: 实现 SignalRouter
**Files:**
- Create: `internal/actor/signal_router.go`
- [ ] **Step 1: 实现 SignalRouter**
```go
package actor
import (
"context"
"sync"
goplc "bjhardman.cn/bjhardman/goplc"
)
type SignalRouter struct {
plc goplc.Client
mu sync.RWMutex
watches map[int]*SignalWatch
}
type SignalWatch struct {
DoneAddr string
NGAddr string
Ch chan SignalEvent
}
func NewSignalRouter(plc goplc.Client) *SignalRouter {
return &SignalRouter{
plc: plc,
watches: make(map[int]*SignalWatch),
}
}
// Watch 注册设备信号监听，返回 Actor 用于接收信号的 channel
func (r *SignalRouter) Watch(machineID int, doneAddr, ngAddr string) chan SignalEvent {
r.mu.Lock()
defer r.mu.Unlock()
ch := make(chan SignalEvent, 1)
r.watches[machineID] = &SignalWatch{
DoneAddr: doneAddr,
NGAddr: ngAddr,
Ch: ch,
}
return ch
}
// Run 轮询 PLC 信号，阻塞直到 ctx 取消
func (r *SignalRouter) Run(ctx context.Context, interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
lastState := make(map[int]bool) // machineID → 上次Done值
lastNG := make(map[int]bool) // -machineID → 上次NG值
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
r.poll(lastState, lastNG)
}
}
}
func (r *SignalRouter) poll(lastState, lastNG map[int]bool) {
if r.plc == nil || !r.plc.IsConnected() {
return
}
r.mu.RLock()
defer r.mu.RUnlock()
for machineID, w := range r.watches {
if w.DoneAddr == "" {
continue
}
current, err := r.plc.ReadBool(w.DoneAddr)
if err != nil {
continue
}
prev := lastState[machineID]
lastState[machineID] = current
// 上升沿
if current && !prev {
r.sendNonBlock(w.Ch, SignalEvent{Type: SignalDone, Value: true})
}
// NG 信号
if w.NGAddr != "" {
ng, err := r.plc.ReadBool(w.NGAddr)
if err != nil {
continue
}
key := -machineID
ngPrev := lastNG[key]
lastNG[key] = ng
if ng && !ngPrev {
r.sendNonBlock(w.Ch, SignalEvent{Type: SignalNG, Value: true})
}
}
}
}
func (r *SignalRouter) sendNonBlock(ch chan SignalEvent, evt SignalEvent) {
select {
case ch <- evt:
default:
// 缓冲区满，丢弃（表示上一事件未被消费）
}
}
```
- [ ] **Step 2: 验证编译**
```bash
go build ./internal/actor/...
```
---
## Chunk 3: MachineActor
### Task 3.1: 实现 MachineActor
**Files:**
- Create: `internal/actor/machine_actor.go`
- Create: `internal/actor/machine_actor_test.go`
- [ ] **Step 1: 写失败测试**
```go
// internal/actor/machine_actor_test.go
package actor
import (
"context"
"testing"
"time"
)
func TestMachineActor_IdleToProcessing(t *testing.T) {
cfg := MachineActorConfig{
ID: 1,
Type: "CNC",
Capacity: 2,
SignalCh: make(chan SignalEvent),
}
actor := NewMachineActor(cfg)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go actor.Run(ctx)
// Load first job
actor.Send(MachineMsg{Type: "LOAD_COMPLETE", JobID: 100, SlotNo: 1})
time.Sleep(10 * time.Millisecond)
snap := actor.Snapshot()
if snap.Status != ActorProcessing {
t.Fatalf("expected Processing, got %s", snap.Status)
}
if snap.Slots[0].Status != SlotOccupied || snap.Slots[0].JobID != 100 {
t.Fatalf("expected slot 1 Occupied with job 100, got %v", snap.Slots[0])
}
}
func TestMachineActor_FullOnAllOccupied(t *testing.T) {
cfg := MachineActorConfig{
ID: 2,
Type: "CNC",
Capacity: 1,
SignalCh: make(chan SignalEvent),
}
actor := NewMachineActor(cfg)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go actor.Run(ctx)
actor.Send(MachineMsg{Type: "LOAD_COMPLETE", JobID: 200, SlotNo: 1})
time.Sleep(10 * time.Millisecond)
snap := actor.Snapshot()
if snap.Status != ActorFull {
t.Fatalf("expected Full for 1-slot CNC, got %s", snap.Status)
}
}
func TestMachineActor_DoneSignal(t *testing.T) {
signalCh := make(chan SignalEvent, 1)
cfg := MachineActorConfig{
ID: 3,
Type: "CNC",
Capacity: 1,
SignalCh: signalCh,
}
actor := NewMachineActor(cfg)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go actor.Run(ctx)
actor.Send(MachineMsg{Type: "LOAD_COMPLETE", JobID: 300, SlotNo: 1})
time.Sleep(10 * time.Millisecond)
// Simulate PLC done signal
signalCh <- SignalEvent{Type: SignalDone, Value: true}
time.Sleep(10 * time.Millisecond)
snap := actor.Snapshot()
if snap.Status != ActorWaiting {
t.Fatalf("expected Waiting, got %s", snap.Status)
}
if snap.Slots[0].Status != SlotDone {
t.Fatalf("expected slot Done, got %s", snap.Slots[0].Status)
}
}
func TestMachineActor_BatchDoneSignal(t *testing.T) {
signalCh := make(chan SignalEvent, 1)
cfg := MachineActorConfig{
ID: 5,
Type: "WASHER_HP",
Capacity: 2,
Batch: true,
SignalCh: signalCh,
}
actor := NewMachineActor(cfg)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go actor.Run(ctx)
actor.Send(MachineMsg{Type: "LOAD_COMPLETE", JobID: 400, SlotNo: 1})
actor.Send(MachineMsg{Type: "LOAD_COMPLETE", JobID: 401, SlotNo: 2})
time.Sleep(10 * time.Millisecond)
signalCh <- SignalEvent{Type: SignalDone, Value: true}
time.Sleep(10 * time.Millisecond)
snap := actor.Snapshot()
if snap.Slots[0].Status != SlotDone {
t.Fatalf("expected slot 1 Done, got %s", snap.Slots[0].Status)
}
if snap.Slots[1].Status != SlotDone {
t.Fatalf("expected slot 2 Done, got %s", snap.Slots[1].Status)
}
}
```
- [ ] **Step 2: 运行测试验证失败**
```bash
go test ./internal/actor/... -run TestMachineActor -v
```
Expected: FAIL (MachineActor not defined)
- [ ] **Step 3: 实现 MachineActor**
```go
// internal/actor/machine_actor.go
package actor
import (
"context"
"sync/atomic"
"time"
"hougai/constants"
"hougai/internal/eventbus"
)
type MachineActorConfig struct {
ID int
Type string
Capacity int
Batch bool
Inspection bool
SignalCh chan SignalEvent
DB interface { // DBState 的子集，避免循环依赖
SetEquipmentSlot(ctx context.Context, equipmentID, slotNo int, expected, next constants.SlotStatus, jobID int) error
}
Bus eventbus.Bus
}
type machineActor struct {
id int
typ string
capacity int
batch bool
inspection bool
signalCh chan SignalEvent
msgCh chan MachineMsg
slots []SlotStatus
slotJobs []int
slotOccupiedAt []time.Time
status atomic.Value // ActorStatus
db interface {
SetEquipmentSlot(ctx context.Context, equipmentID, slotNo int, expected, next constants.SlotStatus, jobID int) error
}
bus eventbus.Bus
}
func NewMachineActor(cfg MachineActorConfig) *machineActor {
slots := make([]SlotStatus, cfg.Capacity)
slotJobs := make([]int, cfg.Capacity)
slotOccupiedAt := make([]time.Time, cfg.Capacity)
for i := range slots {
slots[i] = SlotEmpty
}
a := &machineActor{
id: cfg.ID,
typ: cfg.Type,
capacity: cfg.Capacity,
batch: cfg.Batch,
inspection: cfg.Inspection,
signalCh: cfg.SignalCh,
msgCh: make(chan MachineMsg, 8),
slots: slots,
slotJobs: slotJobs,
slotOccupiedAt: slotOccupiedAt,
db: cfg.DB,
bus: cfg.Bus,
}
a.status.Store(ActorIdle)
return a
}
func (m *machineActor) ID() int { return m.id }
func (m *machineActor) Type() string { return m.typ }
func (m *machineActor) Send(msg MachineMsg) {
select {
case m.msgCh <- msg:
default:
}
}
func (m *machineActor) Snapshot() MachineSnapshot {
slots := make([]SlotSnapshot, m.capacity)
for i := range m.slots {
slots[i] = SlotSnapshot{
SlotNo: i + 1,
Status: m.slots[i],
JobID: m.slotJobs[i],
OccupiedAt: m.slotOccupiedAt[i],
}
}
return MachineSnapshot{
ID: m.id,
Type: m.typ,
Status: m.status.Load().(ActorStatus),
Slots: slots,
UpdatedAt: time.Now(),
}
}
func (m *machineActor) Run(ctx context.Context) {
for {
select {
case msg := <-m.msgCh:
m.handleMessage(ctx, msg)
case evt := <-m.signalCh:
m.handleSignal(ctx, evt)
case <-ctx.Done():
return
}
}
}
func (m *machineActor) handleMessage(ctx context.Context, msg MachineMsg) {
switch msg.Type {
case "LOAD_COMPLETE":
m.slots[msg.SlotNo-1] = SlotOccupied
m.slotJobs[msg.SlotNo-1] = msg.JobID
m.slotOccupiedAt[msg.SlotNo-1] = time.Now()
m.updateStatus()
if m.db != nil {
m.db.SetEquipmentSlot(ctx, m.id, msg.SlotNo, constants.SlotStatus_Empty, constants.SlotStatus_Occupied, msg.JobID)
}
case "UNLOAD_COMPLETE":
if m.db != nil {
m.db.SetEquipmentSlot(ctx, m.id, msg.SlotNo, constants.SlotStatus_Done, constants.SlotStatus_Empty, 0)
}
m.slots[msg.SlotNo-1] = SlotEmpty
m.slotJobs[msg.SlotNo-1] = 0
m.slotOccupiedAt[msg.SlotNo-1] = time.Time{}
m.updateStatus()
case "RESTORE":
m.slots[msg.SlotNo-1] = parseSlotStatus(msg.Params["status"])
m.slotJobs[msg.SlotNo-1] = msg.JobID
m.updateStatus()
}
}
func (m *machineActor) handleSignal(ctx context.Context, evt SignalEvent) {
if evt.Type == SignalNG && evt.Value && m.inspection {
if m.bus != nil {
m.bus.Publish(eventbus.Event{
Type: eventbus.EventInspectionResult,
Data: map[string]any{"machineId": m.id, "pass": false},
})
}
return
}
if evt.Type == SignalDone && evt.Value {
if m.inspection {
for i := range m.slots {
if m.slots[i] == SlotOccupied {
m.slots[i] = SlotDone
if m.db != nil {
m.db.SetEquipmentSlot(ctx, m.id, i+1, constants.SlotStatus_Occupied, constants.SlotStatus_Done, m.slotJobs[i])
}
}
}
m.updateStatus()
if m.bus != nil {
m.bus.Publish(eventbus.Event{
Type: eventbus.EventInspectionResult,
Data: map[string]any{"machineId": m.id, "pass": true},
})
}
return
}
var doneSlots []int
if m.batch {
for i := range m.slots {
if m.slots[i] == SlotOccupied {
m.slots[i] = SlotDone
doneSlots = append(doneSlots, i+1)
}
}
} else {
slotNo := m.findFirstOccupied()
if slotNo > 0 {
m.slots[slotNo-1] = SlotDone
doneSlots = append(doneSlots, slotNo)
}
}
m.updateStatus()
for _, slotNo := range doneSlots {
if m.db != nil {
m.db.SetEquipmentSlot(ctx, m.id, slotNo, constants.SlotStatus_Occupied, constants.SlotStatus_Done, m.slotJobs[slotNo-1])
}
}
if m.bus != nil {
for _, slotNo := range doneSlots {
m.bus.Publish(eventbus.Event{
Type: eventbus.EventMachineDone,
Data: map[string]any{"machineId": m.id, "slotNo": slotNo, "jobId": m.slotJobs[slotNo-1]},
})
}
}
}
}
func (m *machineActor) findFirstOccupied() int {
earliest := time.Time{}
earliestSlot := 0
for i, s := range m.slots {
if s != SlotOccupied {
continue
}
if t := m.slotOccupiedAt[i]; earliestSlot == 0 || t.Before(earliest) {
earliest = t
earliestSlot = i + 1
}
}
return earliestSlot
}
func (m *machineActor) updateStatus() {
hasEmpty := false
hasOccupied := false
hasDone := false
for _, s := range m.slots {
switch s {
case SlotEmpty:
hasEmpty = true
case SlotOccupied:
hasOccupied = true
case SlotDone:
hasDone = true
}
}
var s ActorStatus
switch {
case !hasOccupied && !hasDone:
s = ActorIdle
case hasDone && !hasOccupied:
s = ActorWaiting
case !hasEmpty:
s = ActorFull
default:
s = ActorProcessing
}
m.status.Store(s)
}
func parseSlotStatus(s string) SlotStatus {
if s == "OCCUPIED" {
return SlotOccupied
}
if s == "DONE" {
return SlotDone
}
return SlotEmpty
}
```
- [ ] **Step 4: 运行测试**
```bash
go test ./internal/actor/... -run TestMachineActor -v
```
Expected: PASS (4 tests)
- [ ] **Step 5: 提交**
```bash
git add internal/actor/machine_actor.go internal/actor/machine_actor_test.go
git commit -m "feat: add MachineActor with state machine and PLC signal handling"
```
---
## Chunk 4: TempStoreActor
### Task 4.1: 实现 TempStoreActor
**Files:**
- Create: `internal/actor/tempstore_actor.go`
- Create: `internal/actor/tempstore_actor_test.go`
- [ ] **Step 1: 写失败测试**
```go
// internal/actor/tempstore_actor_test.go
package actor
import (
"context"
"testing"
"time"
)
func TestTempStoreActor_AllocateRelease(t *testing.T) {
actor := NewTempStoreActor(8, nil)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go actor.Run(ctx)
actor.Send(MachineMsg{Type: "ALLOCATE", JobID: 100})
time.Sleep(10 * time.Millisecond)
snap := actor.Snapshot()
found := false
for slotNo, jobID := range snap.Slots {
if jobID == 100 {
if slotNo+1 < 1 || slotNo+1 > 8 {
t.Fatalf("invalid slot %d", slotNo+1)
}
found = true
break
}
}
if !found {
t.Fatalf("job 100 not allocated in temp store")
}
actor.Send(MachineMsg{Type: "RELEASE", JobID: 100})
time.Sleep(10 * time.Millisecond)
snap = actor.Snapshot()
for _, jobID := range snap.Slots {
if jobID == 100 {
t.Fatalf("job 100 should have been released")
}
}
}
func TestTempStoreActor_Full(t *testing.T) {
actor := NewTempStoreActor(2, nil)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go actor.Run(ctx)
actor.Send(MachineMsg{Type: "ALLOCATE", JobID: 1})
actor.Send(MachineMsg{Type: "ALLOCATE", JobID: 2})
time.Sleep(10 * time.Millisecond)
snap := actor.Snapshot()
free := 0
for _, jobID := range snap.Slots {
if jobID == 0 {
free++
}
}
if free != 0 {
t.Fatalf("expected 0 free slots, got %d", free)
}
}
```
- [ ] **Step 2: 运行测试验证失败**
```bash
go test ./internal/actor/... -run TestTempStore -v
```
Expected: FAIL
- [ ] **Step 3: 实现 TempStoreActor**
```go
// internal/actor/tempstore_actor.go
package actor
import "context"
type TempStoreActor struct {
capacity int
slots []int
msgCh chan MachineMsg
db interface {
SetJobTempSlot(ctx context.Context, jobID, slotNo int) error
ClearJobTempSlot(ctx context.Context, jobID int) error
}
}
func NewTempStoreActor(capacity int, db interface {
SetJobTempSlot(ctx context.Context, jobID, slotNo int) error
ClearJobTempSlot(ctx context.Context, jobID int) error
}) *TempStoreActor {
return &TempStoreActor{
capacity: capacity,
slots: make([]int, capacity),
msgCh: make(chan MachineMsg, 8),
db: db,
}
}
func (t *TempStoreActor) Send(msg MachineMsg) {
select {
case t.msgCh <- msg:
default:
}
}
func (t *TempStoreActor) Snapshot() TempStoreSnapshot {
slots := make([]int, t.capacity)
copy(slots, t.slots)
return TempStoreSnapshot{
Capacity: t.capacity,
Slots: slots,
}
}
func (t *TempStoreActor) Run(ctx context.Context) {
for {
select {
case msg := <-t.msgCh:
switch msg.Type {
case "ALLOCATE":
t.allocate(ctx, msg.JobID)
case "RELEASE":
t.release(ctx, msg.JobID)
case "RESTORE":
t.slots[msg.SlotNo-1] = msg.JobID
}
case <-ctx.Done():
return
}
}
}
func (t *TempStoreActor) allocate(ctx context.Context, jobID int) {
for i := range t.slots {
if t.slots[i] == 0 {
t.slots[i] = jobID
if t.db != nil {
t.db.SetJobTempSlot(ctx, jobID, i+1)
}
return
}
}
}
func (t *TempStoreActor) release(ctx context.Context, jobID int) {
for i := range t.slots {
if t.slots[i] == jobID {
t.slots[i] = 0
if t.db != nil {
t.db.ClearJobTempSlot(ctx, jobID)
}
return
}
}
}
```
- [ ] **Step 4: 运行测试**
```bash
go test ./internal/actor/... -run TestTempStore -v
```
Expected: PASS (2 tests)
- [ ] **Step 5: 提交**
```bash
git add internal/actor/tempstore_actor.go internal/actor/tempstore_actor_test.go
git commit -m "feat: add TempStoreActor for temp slot allocation"
```
---
## Chunk 5: EventLoop 精简
### Task 5.1: 更新 ProductionEventLoop 结构
**Files:**
- Modify: `internal/eventloop/loop.go`
- [ ] **Step 1: 替换 struct 字段**
将 `registry`, `batchMachines`, `inspectionMachines` 替换为：
```go
machineActors map[int]actor.MachineActor // 设备 Actor map
tempStoreActor *actor.TempStoreActor // 暂存台 Actor
```
更新 `NewProductionEventLoop` 签名，接收 `machineActors` 和 `tempStoreActor` 替代旧的三个参数。
- [ ] **Step 2: 删除 `updateStationDone` 方法**
删除 `loop.go:363-386` 的 `updateStationDone` 方法。
- [ ] **Step 3: 精简 `handleMachineDone`**
移除 DB equipment_slot 更新（Actor 已写）、移除 `updateStationDone` 调用、移除 batch/non-batch slot 查找。精简为：
```go
func (l *ProductionEventLoop) handleMachineDone(ctx context.Context, msg EventLoopMessage) {
machineID := intFromPayload(msg.Payload, "machineId")
if machineID == 0 {
return
}
// 检测设备走 handleInspectionDone
if l.inspectionActors[machineID] {
return
}
slotNo := intFromPayload(msg.Payload, "slotNo")
jobID := intFromPayload(msg.Payload, "jobId")
job, err := l.entClient.Job.Get(ctx, jobID)
if err != nil || job.Status != constants.JobStatus_Processing {
return
}
if l.isOperationStep(jobID) {
l.db.SetJobWaitingUnload(ctx, jobID)
} else {
l.db.AdvanceStep(ctx, jobID, nil, "", "")
l.entClient.Job.UpdateOneID(jobID).
SetStatus(constants.JobStatus_WaitingUnload).
Save(ctx)
}
if l.jobOps != nil {
l.jobOps.WakeJob(jobID)
}
l.trySchedule(ctx)
}
```
- [ ] **Step 4: 处理 Worker 结果时通知 Actor**
在 `handleWorkerResult` 的 Load 成功路径添加：
```go
// 通知 Actor 上料完成
actor, ok := l.machineActors[targetID]
if ok {
actor.Send(actor.MachineMsg{
Type: "LOAD_COMPLETE",
JobID: jobID,
SlotNo: slotNo,
})
}
```
在 Unload 成功路径添加：
```go
// 通知 Actor 下料完成
actor, ok := l.machineActors[targetID]
if ok {
actor.Send(actor.MachineMsg{
Type: "UNLOAD_COMPLETE",
JobID: jobID,
SlotNo: slotNo,
})
}
```
- [ ] **Step 5: 验证编译**
```bash
go build ./internal/eventloop/...
```
### Task 5.2: 更新调度桥接
**Files:**
- Modify: `internal/eventloop/scheduler_bridge.go`
- [ ] **Step 1: 更新 `buildSystemState`**
改为从 Actor 快照读：
```go
func (l *ProductionEventLoop) buildSystemState(ctx context.Context) scheduler.SystemState {
state := scheduler.SystemState{
MachineBusy: make(map[int]bool),
MachineHasJob: make(map[int]int),
TempSlotJobs: make(map[int]int),
}
for _, a := range l.machineActors {
snap := a.Snapshot()
full := snap.Status == actor.ActorFull || snap.Status == actor.ActorWaiting
state.MachineBusy[snap.ID] = full
for _, slot := range snap.Slots {
if slot.JobID > 0 && slot.Status != actor.SlotEmpty {
state.MachineHasJob[snap.ID] = slot.JobID
}
}
}
// 暂存台
if l.tempStoreActor != nil {
tsSnap := l.tempStoreActor.Snapshot()
occupied := 0
for slotNo, jobID := range tsSnap.Slots {
if jobID > 0 {
state.TempSlotJobs[slotNo+1] = jobID
occupied++
}
}
state.TempSlotFree = tsSnap.Capacity - occupied
}
// ... 后续逻辑不变（OrderPaused, JobSuspended）
return state
}
```
- [ ] **Step 2: 更新 `findIdleMachine`**
改为从 Actor 快照查：
```go
func (l *ProductionEventLoop) findIdleMachine(resourceType string, productTypeID int) int {
for _, a := range l.machineActors {
snap := a.Snapshot()
if snap.Type != resourceType {
continue
}
if snap.Status == actor.ActorFault {
continue
}
// 有空槽位即可接受
for _, slot := range snap.Slots {
if slot.Status == actor.SlotEmpty {
return snap.ID
}
}
}
return 0
}
```
- [ ] **Step 3: 移除 `findDoneJobIDsOnMachine` DB 查询**
```go
func (l *ProductionEventLoop) findDoneJobIDsOnMachine(ctx context.Context, machineID int) []int {
a, ok := l.machineActors[machineID]
if !ok {
return nil
}
snap := a.Snapshot()
var ids []int
for _, slot := range snap.Slots {
if slot.Status == actor.SlotDone && slot.JobID > 0 {
ids = append(ids, slot.JobID)
}
}
return ids
}
```
- [ ] **Step 4: 验证编译和已有测试**
```bash
go build ./...
go test ./internal/eventloop/... -v
```
- [ ] **Step 5: 提交**
```bash
git add internal/eventloop/
git commit -m "refactor: EventLoop reads from Actor snapshots, removes equipment_slot writes"
```
---
## Chunk 6: ServiceContext 组装
### Task 6.1: 更新服务初始化
**Files:**
- Modify: `internal/svc/service_context.go`
- [ ] **Step 1: 创建 Actor 并在 ServiceContext 中替换 Station/SignalWatcher**
```go
// 替换 BuildProductionLine 调用为 BuildMachineActors
import "hougai/internal/actor"
// 创建 SignalRouter
signalRouter := actor.NewSignalRouter(plcManager.GetPlc())
// 从 DB 构建信号映射并注册
doneSignals, ngSignals := buildDoneSignalMaps(ctx, entClient)
for machineID, addr := range doneSignals {
ngAddr := ngSignals[machineID]
signalRouter.Watch(machineID, addr, ngAddr)
}
// 创建 MachineActor
machineActors := make(map[int]actor.MachineActor)
equipments, _ := entClient.Equipment.Query().WithEquipmentType().All(ctx)
for _, eq := range equipments {
cfg := actor.MachineActorConfig{
ID: eq.ID,
Type: eq.Edges.EquipmentType.Code,
Capacity: eq.SlotCount,
Batch: eq.Batch,
Inspection: eq.Edges.EquipmentType.Code == "INSPECTION" || eq.Edges.EquipmentType.Code == "SAMPLING",
SignalCh: signalRouter.Watch(eq.ID, doneSignals[eq.ID], ngSignals[eq.ID]),
DB: dbState,
Bus: eventBus,
}
machineActors[eq.ID] = actor.NewMachineActor(cfg)
}
// 创建 TempStoreActor
tempStoreActor := actor.NewTempStoreActor(dbState.TempSlotCapacity(ctx), dbState)
// 启动 SignalRouter
go signalRouter.Run(ctx, 1*time.Second)
// 启动所有 Actor
for _, a := range machineActors {
go a.Run(ctx)
}
go tempStoreActor.Run(ctx)
// 恢复 Actor 状态
dbState.RecoverMachineActors(ctx, machineActors, tempStoreActor)
// 更新 NewProductionEventLoop 调用
eventLoop := eventloop.NewProductionEventLoop(entClient, worker, machineActors, tempStoreActor, sched)
```
- [ ] **Step 2: 移除 Station 初始化**
删除 `BuildProductionLine`、`RestoreStationSlots`、`configureCNCProductTypes` 调用。
- [ ] **Step 3: 移除 SignalWatcher 初始化**
删除 `NewSignalWatcher` 调用和回调绑定。
- [ ] **Step 4: 移除 queryBatchMachines / queryInspectionMachines**
删除这些辅助函数，逻辑已内化到 Actor 配置中。
- [ ] **Step 5: 验证编译**
```bash
go build ./...
```
- [ ] **Step 6: 提交**
```bash
git add internal/svc/service_context.go
git commit -m "refactor: wire Actors into ServiceContext, remove Station/SignalWatcher"
```
---
## Chunk 7: 清理旧代码
### Task 7.1: 删除 Station 和 SignalWatcher
- [ ] **Step 1: 删除文件**
```bash
rm internal/station/base.go
rm internal/station/cnc.go
rm internal/station/cleaning.go
rm internal/station/washer.go
rm internal/station/inspection.go
rm internal/station/deburr.go
rm internal/station/sampling.go
rm internal/station/interface.go # Station 接口（保留 HandheldTool 到 action 包）
rm internal/station/registry.go
rm internal/station/helpers.go
rm internal/station/scanner.go
rm internal/station/laser_marker.go
# 保留: internal/station/interface_test.go 中 HandheldTool 相关
rm internal/processor/signal_watcher.go
rm internal/processor/step_timeout.go
```
- [ ] **Step 2: 将 HandheldTool 接口移到 action 包**
```go
// internal/action/tool.go (或保留在 station 包的最小文件中)
type HandheldTool interface {
ID() string
Type() string
Name() string
Execute(cmd StationCommand) error
}
```
更新所有 import 路径。
- [ ] **Step 3: 清理 processor 包引用**
移除 `internal/processor/job_processor.go` 中的 `registry *station.StationRegistry` 字段。移除 `internal/processor/robot_worker.go` 中的 registry 引用（上料卸载涉及 station 的逻辑改为通过 Actor 发送消息）。
- [ ] **Step 4: 更新 Replenisher**
`internal/processor/replenisher.go` 中的 `findFreeTempSlot` / `freeSlotsFromDB` 改为从 TempStoreActor 快照查。
- [ ] **Step 5: 验证全量编译**
```bash
go build ./...
```
- [ ] **Step 6: 运行所有测试**
```bash
go test ./internal/eventloop/... ./internal/actor/... ./internal/scheduler/... ./internal/processor/...
```
- [ ] **Step 7: 提交**
```bash
git add -A
git commit -m "refactor: remove Station/SignalWatcher, migrate to Actor model"
```
---
## Chunk 8: 端到端验证
### Task 8.1: Mockrun 集成验证
- [ ] **Step 1: 确认 Mock.Enable=true**
```bash
grep "Enable:" etc/spherical-api.yaml
```
- [ ] **Step 2: 运行端到端测试**
```bash
go run cmd/mockrun/main.go -jobs 2 -timeout 5m
```
Expected: 2 个工件全部完成，无超时、无报错。
- [ ] **Step 3: 修复任何集成问题**
根据 mockrun 输出修复 Actor 与 EventLoop 之间的消息传递、事件路由等问题。
- [ ] **Step 4: 最终提交**
```bash
git add -A
git commit -m "fix: Actor integration issues from mockrun verification"
```
---
## 回滚策略
如果遇到不可逾越的阻塞，所有变更在 `actor模型` 分支上，可随时回到 master：
```bash
git checkout master
```
Reference in New Issue Copy Permalink