pamoja_modbus/
server.rs

1//! A Modbus server: the device's side of the conversation.
2
3use alloc::collections::BTreeMap;
4use alloc::vec::Vec;
5
6use crate::adu::Adu;
7use crate::error::ModbusError;
8use crate::function::Exception;
9use crate::pdu::Pdu;
10use crate::request::Request;
11
12/// The unit address every server acts on and none answers.
13pub const BROADCAST: u8 = 0;
14
15/// The lowest and highest address a server may have; 248 to 255 are reserved.
16pub const UNITS: core::ops::RangeInclusive<u8> = 1..=247;
17
18/// A Modbus device: a unit address and the four tables it serves, coils, discrete inputs,
19/// holding registers, and input registers, each holding only the addresses it was given.
20///
21/// [`answer`](Server::answer) takes one RTU frame and returns the frame the device sends
22/// back, following the Modbus over Serial Line specification: a frame that fails its CRC,
23/// is for another unit, or is a broadcast gets no answer, and a broadcast write is still
24/// carried out. [`serve`](Server::serve) is the same at the PDU level, for a transport with
25/// its own framing. A request the device cannot serve is answered with the exception the
26/// application protocol's state diagrams give: an unknown function, a quantity or value out
27/// of range, then an address the device does not have.
28///
29/// # Examples
30///
31/// ```
32/// use pamoja_modbus::{Adu, Pdu, Server};
33///
34/// // A meter at unit 17 holding voltage, current, and a fault word from register 107.
35/// let mut meter = Server::new(17)?.with_holding_registers(107, &[2301, 418, 0]);
36///
37/// let request = Pdu::read_holding_registers(107, 3).to_adu(17);
38/// let reply = meter.answer(request.as_bytes()).expect("the meter answers its own unit");
39/// let registers: Vec<u16> = reply.response().registers()?.collect();
40/// assert_eq!(registers, [2301, 418, 0]);
41///
42/// // Register 110 is not in its table.
43/// let past = Pdu::read_holding_registers(108, 3).to_adu(17);
44/// let refused = meter.answer(past.as_bytes()).expect("an exception is an answer");
45/// assert_eq!(refused.exception(), Some(pamoja_modbus::Exception::IllegalDataAddress));
46/// # Ok::<(), pamoja_modbus::ModbusError>(())
47/// ```
48#[derive(Clone, Debug, PartialEq, Eq)]
49pub struct Server {
50    unit: u8,
51    coils: BTreeMap<u16, bool>,
52    discrete_inputs: BTreeMap<u16, bool>,
53    holding_registers: BTreeMap<u16, u16>,
54    input_registers: BTreeMap<u16, u16>,
55    served: usize,
56}
57
58fn fill<T: Copy>(table: &mut BTreeMap<u16, T>, start: u16, values: &[T]) {
59    for (address, &value) in (start..=u16::MAX).zip(values) {
60        table.insert(address, value);
61    }
62}
63
64fn read<T: Copy>(table: &BTreeMap<u16, T>, start: u16, count: usize) -> Vec<T> {
65    (start..=u16::MAX)
66        .take(count)
67        .filter_map(|address| table.get(&address).copied())
68        .collect()
69}
70
71fn holds<T>(table: &BTreeMap<u16, T>, start: u16, count: usize) -> bool {
72    let last = usize::from(start) + count - 1;
73    last <= usize::from(u16::MAX) && (start..=last as u16).all(|a| table.contains_key(&a))
74}
75
76impl Server {
77    /// Makes a device at a unit address, with every table empty.
78    ///
79    /// # Arguments
80    ///
81    /// * `unit` - its address, 1 to 247.
82    ///
83    /// # Returns
84    ///
85    /// The device.
86    ///
87    /// # Errors
88    ///
89    /// [`ModbusError::UnitOutOfRange`] for 0, the broadcast address, or 248 to 255, which the
90    /// specification reserves.
91    pub fn new(unit: u8) -> Result<Server, ModbusError> {
92        if !UNITS.contains(&unit) {
93            return Err(ModbusError::UnitOutOfRange { unit });
94        }
95        Ok(Server {
96            unit,
97            coils: BTreeMap::new(),
98            discrete_inputs: BTreeMap::new(),
99            holding_registers: BTreeMap::new(),
100            input_registers: BTreeMap::new(),
101            served: 0,
102        })
103    }
104
105    /// Gives the device coils at consecutive addresses.
106    ///
107    /// # Arguments
108    ///
109    /// * `start` - the first coil's address.
110    /// * `values` - their states, in address order.
111    ///
112    /// # Returns
113    ///
114    /// The device.
115    #[must_use]
116    pub fn with_coils(mut self, start: u16, values: &[bool]) -> Server {
117        fill(&mut self.coils, start, values);
118        self
119    }
120
121    /// Gives the device discrete inputs at consecutive addresses.
122    ///
123    /// # Arguments
124    ///
125    /// * `start` - the first input's address.
126    /// * `values` - their states, in address order.
127    ///
128    /// # Returns
129    ///
130    /// The device.
131    #[must_use]
132    pub fn with_discrete_inputs(mut self, start: u16, values: &[bool]) -> Server {
133        fill(&mut self.discrete_inputs, start, values);
134        self
135    }
136
137    /// Gives the device holding registers at consecutive addresses.
138    ///
139    /// # Arguments
140    ///
141    /// * `start` - the first register's address.
142    /// * `values` - their values, in address order.
143    ///
144    /// # Returns
145    ///
146    /// The device.
147    #[must_use]
148    pub fn with_holding_registers(mut self, start: u16, values: &[u16]) -> Server {
149        fill(&mut self.holding_registers, start, values);
150        self
151    }
152
153    /// Gives the device input registers at consecutive addresses.
154    ///
155    /// # Arguments
156    ///
157    /// * `start` - the first register's address.
158    /// * `values` - their values, in address order.
159    ///
160    /// # Returns
161    ///
162    /// The device.
163    #[must_use]
164    pub fn with_input_registers(mut self, start: u16, values: &[u16]) -> Server {
165        fill(&mut self.input_registers, start, values);
166        self
167    }
168
169    /// Sets holding registers from an address on, adding any the device did not have, as its
170    /// own measurements change.
171    ///
172    /// # Arguments
173    ///
174    /// * `start` - the first register's address.
175    /// * `values` - their values, in address order.
176    pub fn set_holding_registers(&mut self, start: u16, values: &[u16]) {
177        fill(&mut self.holding_registers, start, values);
178    }
179
180    /// Sets input registers from an address on, adding any the device did not have.
181    ///
182    /// # Arguments
183    ///
184    /// * `start` - the first register's address.
185    /// * `values` - their values, in address order.
186    pub fn set_input_registers(&mut self, start: u16, values: &[u16]) {
187        fill(&mut self.input_registers, start, values);
188    }
189
190    /// Sets coils from an address on, adding any the device did not have.
191    ///
192    /// # Arguments
193    ///
194    /// * `start` - the first coil's address.
195    /// * `values` - their states, in address order.
196    pub fn set_coils(&mut self, start: u16, values: &[bool]) {
197        fill(&mut self.coils, start, values);
198    }
199
200    /// Sets discrete inputs from an address on, adding any the device did not have.
201    ///
202    /// # Arguments
203    ///
204    /// * `start` - the first input's address.
205    /// * `values` - their states, in address order.
206    pub fn set_discrete_inputs(&mut self, start: u16, values: &[bool]) {
207        fill(&mut self.discrete_inputs, start, values);
208    }
209
210    /// Returns a coil's state, or `None` when the device has no coil there.
211    pub fn coil(&self, address: u16) -> Option<bool> {
212        self.coils.get(&address).copied()
213    }
214
215    /// Returns a discrete input's state, or `None` when the device has no input there.
216    pub fn discrete_input(&self, address: u16) -> Option<bool> {
217        self.discrete_inputs.get(&address).copied()
218    }
219
220    /// Returns a holding register's value, or `None` when the device has no register there.
221    pub fn holding_register(&self, address: u16) -> Option<u16> {
222        self.holding_registers.get(&address).copied()
223    }
224
225    /// Returns an input register's value, or `None` when the device has no register there.
226    pub fn input_register(&self, address: u16) -> Option<u16> {
227        self.input_registers.get(&address).copied()
228    }
229
230    /// Returns the device's unit address.
231    pub fn unit(&self) -> u8 {
232        self.unit
233    }
234
235    /// Returns how many requests the device has carried out, broadcasts included and
236    /// refusals not.
237    pub fn served(&self) -> usize {
238        self.served
239    }
240
241    /// Serves one request PDU.
242    ///
243    /// # Arguments
244    ///
245    /// * `pdu` - the request: a function code followed by its data.
246    ///
247    /// # Returns
248    ///
249    /// The reply PDU: the normal response, or the exception the request earned.
250    pub fn serve(&mut self, pdu: &[u8]) -> Pdu {
251        let function = pdu.first().copied().unwrap_or(0);
252        let request = match Request::parse(pdu) {
253            Ok(request) => request,
254            Err(exception) => return Pdu::exception(function, exception),
255        };
256        let (start, count) = request.span();
257        let present = match request {
258            Request::ReadCoils { .. }
259            | Request::WriteSingleCoil { .. }
260            | Request::WriteMultipleCoils { .. } => holds(&self.coils, start, count),
261            Request::ReadDiscreteInputs { .. } => holds(&self.discrete_inputs, start, count),
262            Request::ReadHoldingRegisters { .. }
263            | Request::WriteSingleRegister { .. }
264            | Request::WriteMultipleRegisters { .. } => {
265                holds(&self.holding_registers, start, count)
266            }
267            Request::ReadInputRegisters { .. } => holds(&self.input_registers, start, count),
268        };
269        if !present {
270            return Pdu::exception(function, Exception::IllegalDataAddress);
271        }
272        self.served += 1;
273        let quantity = count as u16;
274        let reply = match request {
275            Request::ReadCoils { .. } => Pdu::read_coils_reply(&read(&self.coils, start, count)),
276            Request::ReadDiscreteInputs { .. } => {
277                Pdu::read_discrete_inputs_reply(&read(&self.discrete_inputs, start, count))
278            }
279            Request::ReadHoldingRegisters { .. } => {
280                Pdu::read_holding_registers_reply(&read(&self.holding_registers, start, count))
281            }
282            Request::ReadInputRegisters { .. } => {
283                Pdu::read_input_registers_reply(&read(&self.input_registers, start, count))
284            }
285            Request::WriteSingleCoil { address, on } => {
286                self.coils.insert(address, on);
287                Ok(Pdu::write_single_coil(address, on))
288            }
289            Request::WriteSingleRegister { address, value } => {
290                self.holding_registers.insert(address, value);
291                Ok(Pdu::write_single_register(address, value))
292            }
293            Request::WriteMultipleCoils { values, .. } => {
294                for (address, on) in (start..=u16::MAX).zip(values) {
295                    self.coils.insert(address, on);
296                }
297                Ok(Pdu::write_multiple_coils_reply(start, quantity))
298            }
299            Request::WriteMultipleRegisters { values, .. } => {
300                for (address, value) in (start..=u16::MAX).zip(values) {
301                    self.holding_registers.insert(address, value);
302                }
303                Ok(Pdu::write_multiple_registers_reply(start, quantity))
304            }
305        };
306        reply.unwrap_or_else(|_| Pdu::exception(function, Exception::ServerDeviceFailure))
307    }
308
309    /// Answers one RTU frame, as a device on the line does.
310    ///
311    /// # Arguments
312    ///
313    /// * `frame` - the frame as it came off the line, CRC included.
314    ///
315    /// # Returns
316    ///
317    /// The frame to send back, or `None` when the device stays silent: the frame failed its
318    /// CRC, is for another unit, or is a broadcast, whose write the device still carries out.
319    pub fn answer(&mut self, frame: &[u8]) -> Option<Adu> {
320        let adu = Adu::parse(frame).ok()?;
321        if adu.address() == BROADCAST {
322            if Request::parse(adu.pdu()).is_ok_and(|request| !request.reads()) {
323                self.serve(adu.pdu());
324            }
325            return None;
326        }
327        if adu.address() != self.unit {
328            return None;
329        }
330        Some(self.serve(adu.pdu()).to_adu(self.unit))
331    }
332}
333
334/// A server on the far end of a simulated port answers each write as one frame.
335#[cfg(feature = "port")]
336impl pamoja_hal::port::Peer for Server {
337    fn receive(&mut self, bytes: &[u8]) -> Vec<u8> {
338        self.answer(bytes)
339            .map(|adu| adu.as_bytes().to_vec())
340            .unwrap_or_default()
341    }
342}
343
344/// Several servers on one simulated line, as devices share an RS485 pair: every frame reaches
345/// all of them, the one it is addressed to answers, and each carries out a broadcast write.
346///
347/// A line is a [`Peer`](pamoja_hal::port::Peer), so a
348/// [`SerialPort::simulated`](pamoja_hal::port::SerialPort::simulated) port with a line on
349/// the far end is a bus of devices a [`Client`](crate::Client) polls. Each server is kept
350/// behind a shared handle, so a program looks at what a write did to it afterward.
351///
352/// # Examples
353///
354/// ```
355/// use pamoja_hal::port::{Parity, SerialPort, Settings};
356/// use pamoja_modbus::{Client, Line, Server};
357///
358/// let mut line = Line::new();
359/// let meter = line.attach(Server::new(17)?.with_input_registers(0, &[2301]));
360/// let pump = line.attach(Server::new(18)?.with_coils(0, &[false]));
361///
362/// let settings = Settings::new(19_200).with_parity(Parity::Even);
363/// let mut client = Client::new(SerialPort::simulated(settings, line));
364/// assert_eq!(client.read_input_registers(17, 0, 1)?, [2301]);
365/// client.write_single_coil(18, 0, true)?;
366///
367/// assert_eq!(pump.lock().unwrap().coil(0), Some(true));
368/// assert_eq!(meter.lock().unwrap().served(), 1);
369/// # Ok::<(), Box<dyn std::error::Error>>(())
370/// ```
371#[cfg(feature = "port")]
372#[derive(Clone, Debug, Default)]
373pub struct Line {
374    servers: Vec<std::sync::Arc<std::sync::Mutex<Server>>>,
375}
376
377#[cfg(feature = "port")]
378impl Line {
379    /// Makes a line with no devices on it.
380    pub fn new() -> Line {
381        Line::default()
382    }
383
384    /// Puts a device on the line.
385    ///
386    /// # Arguments
387    ///
388    /// * `server` - the device.
389    ///
390    /// # Returns
391    ///
392    /// A shared handle to it, for looking at it or changing its tables later.
393    pub fn attach(&mut self, server: Server) -> std::sync::Arc<std::sync::Mutex<Server>> {
394        let shared = std::sync::Arc::new(std::sync::Mutex::new(server));
395        self.servers.push(std::sync::Arc::clone(&shared));
396        shared
397    }
398
399    /// Puts a device the program already shares on the line.
400    ///
401    /// # Arguments
402    ///
403    /// * `server` - the shared device.
404    pub fn attach_shared(&mut self, server: std::sync::Arc<std::sync::Mutex<Server>>) {
405        self.servers.push(server);
406    }
407
408    /// Returns the unit addresses on the line, in the order they were attached.
409    pub fn units(&self) -> Vec<u8> {
410        self.servers
411            .iter()
412            .map(|server| lock(server).unit())
413            .collect()
414    }
415}
416
417#[cfg(feature = "port")]
418fn lock(server: &std::sync::Mutex<Server>) -> std::sync::MutexGuard<'_, Server> {
419    server
420        .lock()
421        .unwrap_or_else(std::sync::PoisonError::into_inner)
422}
423
424#[cfg(feature = "port")]
425impl pamoja_hal::port::Peer for Line {
426    fn receive(&mut self, bytes: &[u8]) -> Vec<u8> {
427        let mut replies = Vec::new();
428        for server in &self.servers {
429            if let Some(reply) = lock(server).answer(bytes) {
430                replies.extend_from_slice(reply.as_bytes());
431            }
432        }
433        replies
434    }
435}
436
437#[cfg(test)]
438mod tests {
439    use super::*;
440    use crate::Response;
441
442    fn reply(server: &mut Server, request: Pdu) -> Pdu {
443        server.serve(request.as_bytes())
444    }
445
446    #[test]
447    fn it_answers_each_specification_example_with_the_frame_the_specification_shows() {
448        // 6.1: coils 20 to 38, at addresses 19 to 37, answered CD 6B 05.
449        let coils = [
450            true, false, true, true, false, false, true, true, true, true, false, true, false,
451            true, true, false, true, false, true,
452        ];
453        let mut device = Server::new(1)
454            .unwrap()
455            .with_coils(0x0013, &coils)
456            .with_holding_registers(0x006B, &[0x022B, 0x0000, 0x0064])
457            .with_input_registers(0x0008, &[0x000A])
458            .with_holding_registers(0x0001, &[0, 0]);
459        assert_eq!(
460            reply(&mut device, Pdu::read_coils(0x0013, 0x0013)).as_bytes(),
461            &[0x01, 0x03, 0xCD, 0x6B, 0x05]
462        );
463        // 6.2: discrete inputs 197 to 218, at addresses 196 to 217, answered AC DB 35.
464        let inputs = [
465            false, false, true, true, false, true, false, true, true, true, false, true, true,
466            false, true, true, true, false, true, false, true, true,
467        ];
468        let mut device = device.with_discrete_inputs(0x00C4, &inputs);
469        assert_eq!(
470            reply(&mut device, Pdu::read_discrete_inputs(0x00C4, 0x0016)).as_bytes(),
471            &[0x02, 0x03, 0xAC, 0xDB, 0x35]
472        );
473        // 6.3 and 6.4.
474        assert_eq!(
475            reply(&mut device, Pdu::read_holding_registers(0x006B, 3)).as_bytes(),
476            &[0x03, 0x06, 0x02, 0x2B, 0x00, 0x00, 0x00, 0x64]
477        );
478        assert_eq!(
479            reply(&mut device, Pdu::read_input_registers(0x0008, 1)).as_bytes(),
480            &[0x04, 0x02, 0x00, 0x0A]
481        );
482        // 6.5: coil 173 on, answered with an echo.
483        let on = Pdu::write_single_coil(0x00AC, true);
484        let mut device = device.with_coils(0x00AC, &[false]);
485        assert_eq!(reply(&mut device, on).as_bytes(), on.as_bytes());
486        assert_eq!(device.coil(0x00AC), Some(true));
487        // 6.11: coils 20 to 29 written from CD 01, answered with where and how many.
488        let ten = [
489            true, false, true, true, false, false, true, true, true, false,
490        ];
491        let many = Pdu::write_multiple_coils(0x0013, &ten).unwrap();
492        assert_eq!(
493            reply(&mut device, many).as_bytes(),
494            &[0x0F, 0x00, 0x13, 0x00, 0x0A]
495        );
496        // 6.6 and 6.12.
497        let single = Pdu::write_single_register(0x0001, 0x0003);
498        assert_eq!(reply(&mut device, single).as_bytes(), single.as_bytes());
499        let two = Pdu::write_multiple_registers(0x0001, &[0x000A, 0x0102]).unwrap();
500        assert_eq!(
501            reply(&mut device, two).as_bytes(),
502            &[0x10, 0x00, 0x01, 0x00, 0x02]
503        );
504        assert_eq!(device.holding_register(0x0002), Some(0x0102));
505        assert_eq!(device.coil(0x0013 + 9), Some(false));
506        assert_eq!(device.served(), 8);
507    }
508
509    #[test]
510    fn it_refuses_in_the_order_the_state_diagrams_check() {
511        let mut device = Server::new(1).unwrap().with_holding_registers(0, &[0; 100]);
512        // Section 7: an output that does not exist answers 81 02.
513        assert_eq!(
514            reply(&mut device, Pdu::read_coils(0x04A1, 1)).as_bytes(),
515            &[0x81, 0x02]
516        );
517        // Exception code 02's definition: of 100 registers, 96 and 4 more is fine, and 96
518        // and 5 more is not.
519        assert!(
520            Response::new(reply(&mut device, Pdu::read_holding_registers(96, 4)).as_bytes())
521                .exception()
522                .is_none()
523        );
524        assert_eq!(
525            reply(&mut device, Pdu::read_holding_registers(96, 5)).as_bytes(),
526            &[0x83, 0x02]
527        );
528        // A quantity out of range is refused as a value before any address is looked at.
529        assert_eq!(
530            reply(&mut device, Pdu::read_holding_registers(5_000, 126)).as_bytes(),
531            &[0x83, 0x03]
532        );
533        // A function the device does not serve.
534        let diagnostics = Pdu::raw(0x08, &[0x00, 0x00, 0xA5, 0x37]).unwrap();
535        assert_eq!(reply(&mut device, diagnostics).as_bytes(), &[0x88, 0x01]);
536        // A span that would run past address 0xFFFF.
537        assert_eq!(
538            reply(&mut device, Pdu::read_holding_registers(0xFFFF, 2)).as_bytes(),
539            &[0x83, 0x02]
540        );
541        assert_eq!(device.served(), 1, "only the one good read was carried out");
542    }
543
544    #[test]
545    fn a_frame_gets_an_answer_only_when_it_should() {
546        let mut device = Server::new(17)
547            .unwrap()
548            .with_holding_registers(107, &[2301, 418, 0]);
549        let request = Pdu::read_holding_registers(107, 1).to_adu(17);
550        let answer = device.answer(request.as_bytes()).unwrap();
551        assert_eq!(answer.address(), 17);
552
553        // Another unit's frame, and a frame that fails its CRC, get silence.
554        let other = Pdu::read_holding_registers(107, 1).to_adu(18);
555        assert!(device.answer(other.as_bytes()).is_none());
556        let mut mangled = request.as_bytes().to_vec();
557        mangled[3] ^= 0x01;
558        assert!(device.answer(&mangled).is_none());
559
560        // A broadcast write is carried out and not answered; a broadcast read is neither.
561        let write = Pdu::write_single_register(108, 500).to_adu(BROADCAST);
562        assert!(device.answer(write.as_bytes()).is_none());
563        assert_eq!(device.holding_register(108), Some(500));
564        let read_all = Pdu::read_holding_registers(107, 1).to_adu(BROADCAST);
565        assert!(device.answer(read_all.as_bytes()).is_none());
566        assert_eq!(device.served(), 2);
567    }
568
569    #[test]
570    fn a_unit_outside_one_to_247_is_refused() {
571        for unit in [0, 248, 255] {
572            assert_eq!(Server::new(unit), Err(ModbusError::UnitOutOfRange { unit }));
573        }
574        assert!(Server::new(247).is_ok());
575    }
576
577    #[test]
578    fn many_coils_and_discrete_inputs_round_trip() {
579        let ten = [
580            true, false, true, true, false, false, true, true, true, false,
581        ];
582        let mut device = Server::new(1)
583            .unwrap()
584            .with_coils(0x0013, &[false; 10])
585            .with_discrete_inputs(0x00C4, &[true, false, true]);
586        let write = Pdu::write_multiple_coils(0x0013, &ten).unwrap();
587        assert_eq!(
588            reply(&mut device, write).as_bytes(),
589            &[0x0F, 0x00, 0x13, 0x00, 0x0A]
590        );
591        let back = reply(&mut device, Pdu::read_coils(0x0013, 10));
592        assert!(Response::new(back.as_bytes()).coils(10).unwrap().eq(ten));
593        let inputs = reply(&mut device, Pdu::read_discrete_inputs(0x00C4, 3));
594        assert_eq!(inputs.as_bytes(), &[0x02, 0x01, 0x05]);
595    }
596}
pamoja_modbus/server.rs

pamoja_modbus/
server.rs
