MetzNet
/
vex_mqtt_rust
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Added score publishing

master
noah metz 2024-01-21 23:15:59 -07:00
parent ea21d4728a
commit 5a4e7b0a24
1 changed files with 256 additions and 167 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

189
src/main.rs
Unescape Escape View File

@ -1,3 +1,4 @@
use log::warn;
use rumqttc::{MqttOptions, Client, QoS, LastWill};
use bytes::Bytes;
use std::time::Duration;
@ -19,23 +20,23 @@ use std::sync::mpsc;
// MQTT Topics:
// - division/{division_id}/{round}/{match}/score
// - division/{division_id}/ranking
// - arena/{arena_id}/score
// - arena/{arena_id}/state
// - arena/{arena_id}
// - field/{fieldset_id}/{field_id}/score
// - field/{fieldset_id}/{field_id}/state
// - field/{fieldset_id}/{field_id}
// - team/{team_string}
pub mod tm {
include!(concat!(env!("OUT_DIR"), "/tm.rs"));
}
#[derive(Serialize, Deserialize, Debug)]
#[derive(Serialize, Deserialize, Debug, Clone)]
enum GameSide {
Red,
Blue,
Tie,
}
#[derive(Serialize, Deserialize, Debug)]
#[derive(Serialize, Deserialize, Debug, Clone)]
struct AllianceScore {
auton_wp: bool,
team_goal: i32,
@ -45,8 +46,8 @@ struct AllianceScore {
elevation_tiers: [i32; 2],
}
#[derive(Serialize, Deserialize, Debug)]
struct GameScore {
#[derive(Serialize, Deserialize, Debug, Clone)]
struct MatchScore {
autonomous_winner: Option<GameSide>,
red_score: AllianceScore,
red_total: i32,
@ -157,11 +158,10 @@ impl Event {
// TODO: remove extra entries instead of just adding new ones
fn parse_field_sets(self: &mut Event, sets: tm::FieldSetList) {
for set in sets.field_sets {
let mut fields = Vec::new();
let mut fields = HashMap::new();
for field in &set.fields {
fields.push(Field{
fields.insert(field.id(), Field{
name: String::from(field.name()),
id: field.id(),
last_known_match: None,
});
}
@ -196,6 +196,7 @@ impl Event {
match_list.push(Match{
state: None,
info: None,
score: None,
tuple: tuple.clone(),
})
},
@ -204,6 +205,7 @@ impl Event {
new_match_list.push(Match{
state: None,
info: None,
score: None,
tuple: tuple.clone(),
});
matches.insert(tuple.division, new_match_list);
@ -221,7 +223,7 @@ impl Event {
#[derive(Serialize, Deserialize, Debug, Clone)]
struct DivisionState {
current_arena: u32,
current_field: FieldTuple,
current_match: MatchTuple,
}
@ -234,21 +236,19 @@ struct Division {
#[derive(Serialize, Deserialize, Debug, Clone)]
struct FieldSet {
fields: Vec<Field>,
fields: HashMap<i32, Field>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
struct Field {
name: String,
id: i32,
last_known_match: Option<MatchTuple>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
struct MatchState {
state: Option<GameState>,
start_s: Option<u32>,
start_ns: Option<u32>,
state: GameState,
start: f64,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
@ -261,6 +261,7 @@ struct MatchInfo {
struct Match {
state: Option<MatchState>,
info: Option<MatchInfo>,
score: Option<MatchScore>,
tuple: MatchTuple,
}
@ -620,6 +621,7 @@ struct StateChange {
field: FieldTuple,
}
#[derive(Serialize, Deserialize, Debug, Clone, Copy, Eq, PartialEq, Hash)]
struct FieldTuple {
set: i32,
id: i32,
@ -672,10 +674,7 @@ fn get_affected_match(notice: &tm::Notice) -> Option<MatchTuple> {
}
}
fn get_game_score(notice: &tm::Notice) -> Option<GameScore> {
match &notice.match_score {
None => None,
Some(scores) => {
fn get_game_score(scores: &tm::MatchScore) -> Option<MatchScore> {
if scores.alliances.len() != 2 {
return None;
}
@ -683,7 +682,7 @@ fn get_game_score(notice: &tm::Notice) -> Option<GameScore> {
let ref red_score = scores.alliances[0];
let ref blue_score = scores.alliances[1];
let mut out = GameScore{
let mut out = MatchScore{
autonomous_winner: None,
red_total: 0,
blue_total: 0,
@ -790,33 +789,32 @@ fn get_game_score(notice: &tm::Notice) -> Option<GameScore> {
}
}
}
return Some(out);
},
}
}
fn on_score_change(notice: tm::Notice, event: &mut Event, _connection: &mut TMConnection) -> Vec<MQTTMessage> {
match get_affected_match(&notice) {
None => Vec::new(),
Some(tuple) => {
match get_game_score(&notice) {
None => Vec::new(),
Some(score) => {
let serialized = serde_json::to_string(&score).unwrap();
let Some(tuple) = get_affected_match(&notice) else { return Vec::new() };
let Some(scores) = notice.match_score else { return Vec::new() };
let Some(score) = get_game_score(&scores) else { return Vec::new() };
let Some(division) = &mut event.divisions.get_mut(&tuple.division) else { return Vec::new() };
let Some(m) = &mut division.matches.iter_mut().find(|a| a.tuple == tuple) else { return Vec::new() };
m.score = Some(score.clone());
let serialized = serde_json::to_string_pretty(&score).unwrap();
let game_topic = format!("division/{}/{:?}/{}/score", tuple.division, tuple.round, tuple.match_num);
let mut out = Vec::new();
out.push(MQTTMessage{
topic: game_topic,
payload: serialized.clone(),
});
for (_, field_set) in &event.field_sets {
for field in &field_set.fields {
for (field_set_id, field_set) in &event.field_sets {
for (field_id, field) in &field_set.fields {
match field.last_known_match {
None => {},
Some(last_known_match) => {
if last_known_match == tuple {
let field_topic = format!("field/{}/score", field.id);
let field_topic = format!("field/{}/{}/score", field_set_id, field_id);
out.push(MQTTMessage{
topic: field_topic,
payload: serialized.clone(),
@ -827,10 +825,6 @@ fn on_score_change(notice: tm::Notice, event: &mut Event, _connection: &mut TMCo
}
}
return out;
},
}
},
}
}
fn on_match_list_update(_notice: tm::Notice, event: &mut Event, connection: &mut TMConnection) -> Vec<MQTTMessage> {
@ -843,7 +837,7 @@ fn on_match_list_update(_notice: tm::Notice, event: &mut Event, connection: &mut
for (division_id, division) in &event.divisions {
messages.push(MQTTMessage{
topic: format!("division/{}/schedule", division_id),
payload: serde_json::to_string(&division.matches).unwrap(),
payload: serde_json::to_string_pretty(&division.matches).unwrap(),
});
}
},
@ -851,18 +845,84 @@ fn on_match_list_update(_notice: tm::Notice, event: &mut Event, connection: &mut
return messages;
}
fn on_timer_start(notice: tm::Notice, _event: &mut Event, _connection: &mut TMConnection) -> Vec<MQTTMessage> {
println!("State: {:#?}", notice.field_time.unwrap());
fn on_field_assigned(notice: tm::Notice, event: &mut Event, connection: &mut TMConnection) -> Vec<MQTTMessage> {
let Some(field_info) = &notice.field else { return Vec::new() };
let Some(tuple) = get_affected_match(&notice) else { return Vec::new() };
let Some(field_set) = &mut event.field_sets.get_mut(&field_info.field_set_id()) else { return Vec::new() };
let Some(field) = &mut field_set.fields.get_mut(&field_info.id()) else { return Vec::new() };
let Some(division) = &event.divisions.get(&tuple.division) else { return Vec::new() };
let Some(m) = &division.matches.iter().find(|a| a.tuple == tuple) else { return Vec::new() };
field.last_known_match = Some(tuple);
let mut messages = Vec::new();
if let Some(score) = &m.score {
let serialized = serde_json::to_string_pretty(&score).unwrap();
let topic = format!("field/{}/{}/score", &field_info.field_set_id(), &field_info.id());
messages.push(MQTTMessage{
topic,
payload: serialized,
});
}
if let Some(state) = &m.state {
let serialized = serde_json::to_string_pretty(&state).unwrap();
let topic = format!("field/{}/{}/state", &field_info.field_set_id(), &field_info.id());
messages.push(MQTTMessage{
topic,
payload: serialized,
});
}
return messages;
}
fn on_timer_start(notice: tm::Notice, event: &mut Event, connection: &mut TMConnection) -> Vec<MQTTMessage> {
// 1) Find the state associated with the current block(driver or auton)
// 2) get the match tuple from the arena
// 3) add the mqtt messages for match & arena states
// 2) get the match tuple from the field
// 3) add the mqtt messages for match & field states
// 4) queue the state change to ${state}_done
// 5) add the cancel_send to the tmconnections map of timeout_sends
return Vec::new();
let Some(field_time) = notice.field_time else { return Vec::new() };
let Some(field_info) = &field_time.field else { return Vec::new() };
let Some(block_list) = &field_time.block_list else { return Vec::new() };
let Some(current_block_idx) = &field_time.current_block else { return Vec::new() };
let Some(current_block_start) = &field_time.current_block_start else { return Vec::new() };
let Some(current_block_end) = &field_time.current_block_end else { return Vec::new() };
let Some(field_set) = &event.field_sets.get(&field_info.field_set_id()) else { return Vec::new() };
let Some(field) = &field_set.fields.get(&field_info.id()) else { return Vec::new() };
let Some(tuple) = &field.last_known_match else { return Vec::new() };
let Some(division) = &mut event.divisions.get_mut(&tuple.division) else { return Vec::new() };
let Some(m) = &mut division.matches.iter_mut().find(|a| a.tuple == *tuple) else { return Vec::new() };
let current_block = &block_list.entries[*current_block_idx as usize];
let messages = Vec::new();
if current_block.r#type == Some(2) { //Auto
} else if current_block.r#type == Some(3) { //Driver
m.state = Some(MatchState{
state: GameState::Driver,
start: *current_block_start,
});
let field_tuple = FieldTuple{
set: field_info.field_set_id(),
id: field_info.id(),
};
let cancel_state = connection.queue_state_change(Duration::from_secs(current_block.seconds() as u64), StateChange{
next_state: MatchState{
state: GameState::DriverDone,
start: *current_block_end,
},
tuple: *tuple,
field: field_tuple.clone(),
});
connection.state_cancels.insert(field_tuple, cancel_state);
}
return messages;
}
enum Work {
Exit,
Notice(Box<tm::Notice>),
State(StateChange),
}
@ -875,6 +935,7 @@ fn main() {
callbacks.insert(tm::NoticeId::NoticeMatchScoreUpdated, on_score_change);
callbacks.insert(tm::NoticeId::NoticeMatchListUpdated, on_match_list_update);
callbacks.insert(tm::NoticeId::NoticeFieldTimerStarted, on_timer_start);
callbacks.insert(tm::NoticeId::NoticeFieldMatchAssigned, on_field_assigned);
let mut mqttoptions = MqttOptions::new("vex-bridge", "localhost", 1883);
mqttoptions.set_keep_alive(Duration::from_secs(5));
@ -917,12 +978,34 @@ fn main() {
let match_list_resp = tm_connection.request(1002, tm::BackendMessageData::default());
event.parse_match_list(match_list_resp.data.match_list.unwrap());
for (_, division) in &mut event.divisions {
for m in &mut division.matches {
let mut req = tm::BackendMessageData::default();
let mut filter = tm::MatchTuple::default();
filter.division = Some(m.tuple.division);
filter.r#match = Some(m.tuple.match_num);
filter.instance = Some(m.tuple.instance);
filter.session = Some(m.tuple.session);
filter.round = Some(m.tuple.round as i32);
req.match_tuple = Some(filter);
let resp = tm_connection.request(1000, req);
match resp.data.match_score {
None => {},
Some(scores) => {
let score = get_game_score(&scores);
m.score = score.clone();
client.publish(format!("division/{}/{:?}/{}", m.tuple.division, m.tuple.round, m.tuple.match_num), QoS::AtLeastOnce, true, serde_json::to_string_pretty(&score).unwrap()).expect("MQTT publish fail");
},
}
}
}
for (field_set_id, field_set) in &event.field_sets {
for field in &field_set.fields {
for (field_id, field) in &field_set.fields {
let mut field_req = tm::BackendMessageData::default();
let mut field_data = tm::OnFieldMatch::default();
let mut f = tm::Field::default();
f.id = Some(field.id);
f.id = Some(*field_id);
f.field_set_id = Some(*field_set_id);
field_data.field = Some(f);
field_req.on_field_match = Some(field_data);
@ -931,10 +1014,11 @@ fn main() {
}
}
println!("Event: {:#?}", event);
while running {
match tm_connection.work_queue.recv() {
Ok(work) => match work {
Work::Exit => running = false,
Work::Notice(notice) => {
let callback = callbacks.get(&notice.id());
match callback {
@ -947,7 +1031,7 @@ fn main() {
Some(callback) => {
let messages = callback(*notice, &mut event, &mut tm_connection);
for message in messages {
let result = client.publish(message.topic, QoS::AtMostOnce, true, message.payload);
let result = client.publish(message.topic, QoS::AtLeastOnce, true, message.payload);
match result {
Ok(_) => {},
Err(error) => log::error!("Publish error: {}", error),
@ -960,7 +1044,12 @@ fn main() {
match event.get_match(state_change.tuple) {
None => log::warn!("Received state change for unknown match {:#?}", state_change.tuple),
Some(m) => {
m.state = Some(state_change.next_state);
m.state = Some(state_change.next_state.clone());
let field_state_topic = format!("field/{}/{}/state", state_change.field.set, state_change.field.id);
let match_state_topic = format!("division/{}/{:?}/{}/score", state_change.tuple.division, state_change.tuple.round, state_change.tuple.match_num);
let payload = serde_json::to_vec_pretty(&state_change.next_state).unwrap();
client.publish(field_state_topic, QoS::AtLeastOnce, true, payload.clone()).expect("Failed MQTT publish");
client.publish(match_state_topic, QoS::AtLeastOnce, true, payload).expect("Failed MQTT publish");
},
}
},