mabi_cli/commands/
protocol.rs

1//! Protocol-specific commands.
2//!
3//! Provides subcommands for each supported protocol.
4
5use crate::context::CliContext;
6use crate::error::CliResult;
7use crate::output::{OutputFormat, PaginatedTable, StatusType, TableBuilder};
8use crate::runner::{Command, CommandOutput};
9use async_trait::async_trait;
10use mabi_core::prelude::*;
11use mabi_core::tags::Tags;
12use serde::Serialize;
13use std::net::SocketAddr;
14use std::sync::Arc;
15use std::time::Duration;
16use tokio::sync::Mutex;
17use tokio::task::JoinHandle;
18
19// Protocol-specific imports
20use mabi_bacnet::prelude::{
21    default_object_descriptors, BACnetServer, ObjectRegistry, ServerConfig as BacnetServerConfig,
22};
23use mabi_knx::{
24    DptId, GroupAddress, GroupObjectTable, IndividualAddress, KnxServer, KnxServerConfig,
25};
26use mabi_modbus::{tcp::ServerConfigV2, ModbusDevice, ModbusDeviceConfig, ModbusTcpServerV2};
27use mabi_opcua::{OpcUaServer, OpcUaServerConfig};
28
29/// Checks if a port is already in use and provides diagnostics.
30///
31/// This is an advisory check — it warns but does not block. If the port is held by
32/// a zombie/suspended process, it provides specific diagnostic instructions.
33async fn check_port_availability(addr: SocketAddr) {
34    use tokio::io::{AsyncReadExt, AsyncWriteExt};
35    use tokio::net::TcpStream;
36
37    // Try to connect to the port
38    let connect_result =
39        tokio::time::timeout(Duration::from_millis(500), TcpStream::connect(addr)).await;
40
41    match connect_result {
42        Ok(Ok(_first_stream)) => {
43            // Port is in use — something is listening. Try a Modbus probe.
44            drop(_first_stream);
45
46            let probe_result = tokio::time::timeout(Duration::from_secs(1), async {
47                let mut stream = TcpStream::connect(addr).await?;
48                // Send Modbus TCP ReadHoldingRegisters: txn=1, proto=0, len=6, unit=1, fc=3, addr=0, count=1
49                let request: [u8; 12] = [
50                    0x00, 0x01, 0x00, 0x00, 0x00, 0x06, 0x01, 0x03, 0x00, 0x00, 0x00, 0x01,
51                ];
52                stream.write_all(&request).await?;
53                let mut response = [0u8; 32];
54                let n = stream.read(&mut response).await?;
55                Ok::<_, std::io::Error>(n)
56            })
57            .await;
58
59            match probe_result {
60                Ok(Ok(n)) if n >= 7 => {
61                    tracing::warn!(
62                        port = addr.port(),
63                        "Port {} is already in use by a responding Modbus server. \
64                         The new server will fail to bind.",
65                        addr.port()
66                    );
67                }
68                _ => {
69                    // TCP connects but Modbus doesn't respond — possible zombie
70                    tracing::warn!(
71                        port = addr.port(),
72                        "Port {} is in use: TCP connects but no Modbus response. \
73                         This may be a suspended (zombie) process holding the port.\n  \
74                         Diagnostic: lsof -i :{} | grep LISTEN\n  \
75                         To kill:    kill $(lsof -ti :{} -sTCP:LISTEN)",
76                        addr.port(),
77                        addr.port(),
78                        addr.port()
79                    );
80                }
81            }
82        }
83        Ok(Err(_)) | Err(_) => {
84            // Port is available (connection refused or timeout) — good
85            tracing::debug!(port = addr.port(), "Port {} is available", addr.port());
86        }
87    }
88}
89
90/// Base trait for protocol-specific commands.
91#[async_trait]
92pub trait ProtocolCommand: Command {
93    /// Get the protocol type.
94    fn protocol(&self) -> Protocol;
95
96    /// Get the default port.
97    fn default_port(&self) -> u16;
98
99    /// Start the protocol server.
100    async fn start_server(&self, ctx: &mut CliContext) -> CliResult<()>;
101
102    /// Stop the protocol server.
103    async fn stop_server(&self, ctx: &mut CliContext) -> CliResult<()>;
104}
105
106// =============================================================================
107// Modbus Command
108// =============================================================================
109
110/// Modbus protocol command.
111pub struct ModbusCommand {
112    /// Binding address.
113    bind_addr: SocketAddr,
114    /// Number of devices to simulate.
115    devices: usize,
116    /// Points per device.
117    points_per_device: usize,
118    /// Use RTU mode instead of TCP.
119    rtu_mode: bool,
120    /// Serial port for RTU mode.
121    serial_port: Option<String>,
122    /// Device tags.
123    tags: Tags,
124    /// Server instance (for shutdown).
125    server: Arc<Mutex<Option<Arc<ModbusTcpServerV2>>>>,
126    /// Server task handle.
127    server_task: Arc<Mutex<Option<JoinHandle<()>>>>,
128}
129
130impl ModbusCommand {
131    pub fn new() -> Self {
132        Self {
133            bind_addr: "0.0.0.0:502".parse().unwrap(),
134            devices: 1,
135            points_per_device: 100,
136            rtu_mode: false,
137            serial_port: None,
138            tags: Tags::new(),
139            server: Arc::new(Mutex::new(None)),
140            server_task: Arc::new(Mutex::new(None)),
141        }
142    }
143
144    pub fn with_bind_addr(mut self, addr: SocketAddr) -> Self {
145        self.bind_addr = addr;
146        self
147    }
148
149    pub fn with_port(mut self, port: u16) -> Self {
150        self.bind_addr.set_port(port);
151        self
152    }
153
154    pub fn with_devices(mut self, devices: usize) -> Self {
155        self.devices = devices;
156        self
157    }
158
159    pub fn with_points(mut self, points: usize) -> Self {
160        self.points_per_device = points;
161        self
162    }
163
164    pub fn with_rtu_mode(mut self, serial_port: impl Into<String>) -> Self {
165        self.rtu_mode = true;
166        self.serial_port = Some(serial_port.into());
167        self
168    }
169
170    pub fn with_tags(mut self, tags: Tags) -> Self {
171        self.tags = tags;
172        self
173    }
174}
175
176impl Default for ModbusCommand {
177    fn default() -> Self {
178        Self::new()
179    }
180}
181
182#[async_trait]
183impl Command for ModbusCommand {
184    fn name(&self) -> &str {
185        "modbus"
186    }
187
188    fn description(&self) -> &str {
189        "Start a Modbus TCP/RTU simulator"
190    }
191
192    fn requires_engine(&self) -> bool {
193        true
194    }
195
196    fn supports_shutdown(&self) -> bool {
197        true
198    }
199
200    async fn execute(&self, ctx: &mut CliContext) -> CliResult<CommandOutput> {
201        let format = ctx.output().format();
202        let is_quiet = ctx.is_quiet();
203        let is_verbose = ctx.is_verbose();
204        let is_debug = ctx.is_debug();
205
206        if !is_quiet && matches!(format, OutputFormat::Table) {
207            let output = ctx.output();
208            if self.rtu_mode {
209                output.header("Modbus RTU Simulator");
210                output.kv("Serial Port", self.serial_port.as_deref().unwrap_or("N/A"));
211            } else {
212                output.header("Modbus TCP Simulator");
213                output.kv("Bind Address", self.bind_addr);
214            }
215            output.kv("Devices", self.devices);
216            output.kv("Points per Device", self.points_per_device);
217            output.kv("Total Points", self.devices * self.points_per_device);
218        }
219
220        // Verbose: show extra configuration details
221        if is_verbose {
222            ctx.vprintln(format!(
223                "  Protocol Mode: {}",
224                if self.rtu_mode { "RTU" } else { "TCP" }
225            ));
226            ctx.vprintln(format!(
227                "  Points Distribution: {} per register type",
228                self.points_per_device / 4
229            ));
230        }
231
232        // Debug: dump full configuration
233        if is_debug {
234            ctx.dprintln(format!("Bind address: {}", self.bind_addr));
235            ctx.dprintln(format!(
236                "RTU mode: {}, Serial: {:?}",
237                self.rtu_mode, self.serial_port
238            ));
239            ctx.dprintln(format!(
240                "Devices: {}, Points/device: {}",
241                self.devices, self.points_per_device
242            ));
243        }
244
245        self.start_server(ctx).await?;
246
247        let points_per_type = self.points_per_device / 4;
248
249        if !is_quiet {
250            match format {
251                OutputFormat::Table => {
252                    let colors_enabled = ctx.colors_enabled();
253                    let builder = TableBuilder::new(colors_enabled).header([
254                        "Unit ID",
255                        "Holding Regs",
256                        "Input Regs",
257                        "Coils",
258                        "Discrete",
259                        "Status",
260                    ]);
261
262                    let devices = self.devices;
263                    let pts = points_per_type.to_string();
264                    let table = PaginatedTable::default().render(builder, devices, 6, |i| {
265                        let unit_id = (i + 1).to_string();
266                        (
267                            vec![
268                                unit_id,
269                                pts.clone(),
270                                pts.clone(),
271                                pts.clone(),
272                                pts.clone(),
273                                "Online".into(),
274                            ],
275                            StatusType::Success,
276                        )
277                    });
278                    table.print();
279                }
280                _ => {
281                    #[derive(Serialize)]
282                    struct ModbusServerInfo {
283                        protocol: String,
284                        bind_address: String,
285                        devices: usize,
286                        points_per_device: usize,
287                        total_points: usize,
288                        rtu_mode: bool,
289                        serial_port: Option<String>,
290                        device_list: Vec<ModbusDeviceInfo>,
291                        status: String,
292                    }
293                    #[derive(Serialize)]
294                    struct ModbusDeviceInfo {
295                        unit_id: usize,
296                        holding_registers: usize,
297                        input_registers: usize,
298                        coils: usize,
299                        discrete_inputs: usize,
300                        status: String,
301                    }
302                    let device_list: Vec<ModbusDeviceInfo> = (0..self.devices)
303                        .map(|i| ModbusDeviceInfo {
304                            unit_id: i + 1,
305                            holding_registers: points_per_type,
306                            input_registers: points_per_type,
307                            coils: points_per_type,
308                            discrete_inputs: points_per_type,
309                            status: "Online".into(),
310                        })
311                        .collect();
312                    let info = ModbusServerInfo {
313                        protocol: if self.rtu_mode {
314                            "Modbus RTU".into()
315                        } else {
316                            "Modbus TCP".into()
317                        },
318                        bind_address: self.bind_addr.to_string(),
319                        devices: self.devices,
320                        points_per_device: self.points_per_device,
321                        total_points: self.devices * self.points_per_device,
322                        rtu_mode: self.rtu_mode,
323                        serial_port: self.serial_port.clone(),
324                        device_list,
325                        status: "Online".into(),
326                    };
327                    let _ = ctx.output().write(&info);
328                }
329            }
330        }
331
332        if !is_quiet {
333            ctx.output().info("Press Ctrl+C to stop");
334        }
335        ctx.shutdown_signal().notified().await;
336
337        self.stop_server(ctx).await?;
338        if !is_quiet {
339            ctx.output().success("Modbus simulator stopped");
340        }
341
342        Ok(CommandOutput::quiet_success())
343    }
344}
345
346#[async_trait]
347impl ProtocolCommand for ModbusCommand {
348    fn protocol(&self) -> Protocol {
349        if self.rtu_mode {
350            Protocol::ModbusRtu
351        } else {
352            Protocol::ModbusTcp
353        }
354    }
355
356    fn default_port(&self) -> u16 {
357        502
358    }
359
360    async fn start_server(&self, ctx: &mut CliContext) -> CliResult<()> {
361        let output = ctx.output();
362
363        // Advisory port availability check before starting
364        check_port_availability(self.bind_addr).await;
365
366        let spinner = output.spinner("Starting Modbus server...");
367
368        let config = ServerConfigV2 {
369            bind_address: self.bind_addr,
370            ..Default::default()
371        };
372
373        let server = Arc::new(ModbusTcpServerV2::new(config));
374
375        for i in 0..self.devices {
376            let unit_id = (i + 1) as u8;
377            let points = (self.points_per_device / 4) as u16;
378            let device_config = ModbusDeviceConfig {
379                unit_id,
380                name: format!("Device-{}", unit_id),
381                holding_registers: points,
382                input_registers: points,
383                coils: points,
384                discrete_inputs: points,
385                response_delay_ms: 0,
386                tags: self.tags.clone(),
387            };
388            let device = ModbusDevice::new(device_config);
389            server.add_device(device);
390        }
391
392        {
393            let mut server_guard = self.server.lock().await;
394            *server_guard = Some(server.clone());
395        }
396
397        let server_clone = server.clone();
398        let task = tokio::spawn(async move {
399            if let Err(e) = server_clone.run().await {
400                tracing::error!("Modbus server error: {}", e);
401            }
402        });
403
404        {
405            let mut task_guard = self.server_task.lock().await;
406            *task_guard = Some(task);
407        }
408
409        tokio::time::sleep(Duration::from_millis(100)).await;
410
411        // Check if server task exited early (likely a bind failure / port-in-use)
412        {
413            let task_guard = self.server_task.lock().await;
414            if let Some(task) = task_guard.as_ref() {
415                if task.is_finished() {
416                    spinner.finish_with_message("Failed to start server");
417                    return Err(crate::error::CliError::PortInUse {
418                        port: self.bind_addr.port(),
419                    });
420                }
421            }
422        }
423
424        spinner.finish_with_message(format!("Modbus server started on {}", self.bind_addr));
425        Ok(())
426    }
427
428    async fn stop_server(&self, _ctx: &mut CliContext) -> CliResult<()> {
429        if let Some(server) = self.server.lock().await.as_ref() {
430            server.shutdown();
431        }
432
433        if let Some(task) = self.server_task.lock().await.take() {
434            let _ = tokio::time::timeout(Duration::from_secs(5), task).await;
435        }
436
437        Ok(())
438    }
439}
440
441// =============================================================================
442// OPC UA Command
443// =============================================================================
444
445/// OPC UA protocol command.
446pub struct OpcuaCommand {
447    bind_addr: SocketAddr,
448    endpoint_path: String,
449    nodes: usize,
450    security_mode: String,
451    /// Device tags.
452    tags: Tags,
453    /// Server instance (for shutdown).
454    server: Arc<Mutex<Option<Arc<OpcUaServer>>>>,
455    /// Server task handle.
456    server_task: Arc<Mutex<Option<JoinHandle<()>>>>,
457}
458
459impl OpcuaCommand {
460    pub fn new() -> Self {
461        Self {
462            bind_addr: "0.0.0.0:4840".parse().unwrap(),
463            endpoint_path: "/".into(),
464            nodes: 1000,
465            security_mode: "None".into(),
466            tags: Tags::new(),
467            server: Arc::new(Mutex::new(None)),
468            server_task: Arc::new(Mutex::new(None)),
469        }
470    }
471
472    pub fn with_port(mut self, port: u16) -> Self {
473        self.bind_addr.set_port(port);
474        self
475    }
476
477    pub fn with_endpoint(mut self, path: impl Into<String>) -> Self {
478        self.endpoint_path = path.into();
479        self
480    }
481
482    pub fn with_nodes(mut self, nodes: usize) -> Self {
483        self.nodes = nodes;
484        self
485    }
486
487    pub fn with_security(mut self, mode: impl Into<String>) -> Self {
488        self.security_mode = mode.into();
489        self
490    }
491
492    pub fn with_tags(mut self, tags: Tags) -> Self {
493        self.tags = tags;
494        self
495    }
496}
497
498impl Default for OpcuaCommand {
499    fn default() -> Self {
500        Self::new()
501    }
502}
503
504#[async_trait]
505impl Command for OpcuaCommand {
506    fn name(&self) -> &str {
507        "opcua"
508    }
509
510    fn description(&self) -> &str {
511        "Start an OPC UA server simulator"
512    }
513
514    fn requires_engine(&self) -> bool {
515        true
516    }
517
518    fn supports_shutdown(&self) -> bool {
519        true
520    }
521
522    async fn execute(&self, ctx: &mut CliContext) -> CliResult<CommandOutput> {
523        let format = ctx.output().format();
524        let is_quiet = ctx.is_quiet();
525        let is_verbose = ctx.is_verbose();
526        let is_debug = ctx.is_debug();
527
528        if !is_quiet && matches!(format, OutputFormat::Table) {
529            let output = ctx.output();
530            output.header("OPC UA Simulator");
531            output.kv(
532                "Endpoint",
533                format!("opc.tcp://{}{}", self.bind_addr, self.endpoint_path),
534            );
535            output.kv("Nodes", self.nodes);
536            output.kv("Security Mode", &self.security_mode);
537        }
538
539        // Verbose: show extra details
540        if is_verbose {
541            ctx.vprintln(format!("  Bind Address: {}", self.bind_addr));
542            ctx.vprintln(format!("  Endpoint Path: {}", self.endpoint_path));
543            ctx.vprintln("  Max Subscriptions: 1000");
544            ctx.vprintln("  Max Monitored Items: 10000");
545        }
546
547        // Debug: dump full configuration
548        if is_debug {
549            ctx.dprintln(format!(
550                "Full endpoint URL: opc.tcp://{}{}",
551                self.bind_addr, self.endpoint_path
552            ));
553            ctx.dprintln(format!("Node count: {}", self.nodes));
554            ctx.dprintln(format!("Security mode: {}", self.security_mode));
555            ctx.dprintln(format!("Sample nodes created: {}", self.nodes.min(100)));
556        }
557
558        self.start_server(ctx).await?;
559
560        if !is_quiet {
561            match format {
562                OutputFormat::Table => {
563                    let colors_enabled = ctx.colors_enabled();
564                    let table = TableBuilder::new(colors_enabled)
565                        .header(["Namespace", "Nodes", "Subscriptions", "Status"])
566                        .status_row(["0", "Standard", "0", "Ready"], StatusType::Info)
567                        .status_row(
568                            ["1", &self.nodes.to_string(), "0", "Online"],
569                            StatusType::Success,
570                        );
571                    table.print();
572                }
573                _ => {
574                    #[derive(Serialize)]
575                    struct OpcuaServerInfo {
576                        protocol: String,
577                        endpoint: String,
578                        nodes: usize,
579                        security_mode: String,
580                        namespaces: Vec<NamespaceInfo>,
581                        status: String,
582                    }
583                    #[derive(Serialize)]
584                    struct NamespaceInfo {
585                        index: u32,
586                        nodes: String,
587                        subscriptions: u32,
588                        status: String,
589                    }
590                    let info = OpcuaServerInfo {
591                        protocol: "OPC UA".into(),
592                        endpoint: format!("opc.tcp://{}{}", self.bind_addr, self.endpoint_path),
593                        nodes: self.nodes,
594                        security_mode: self.security_mode.clone(),
595                        namespaces: vec![
596                            NamespaceInfo {
597                                index: 0,
598                                nodes: "Standard".into(),
599                                subscriptions: 0,
600                                status: "Ready".into(),
601                            },
602                            NamespaceInfo {
603                                index: 1,
604                                nodes: self.nodes.to_string(),
605                                subscriptions: 0,
606                                status: "Online".into(),
607                            },
608                        ],
609                        status: "Online".into(),
610                    };
611                    let _ = ctx.output().write(&info);
612                }
613            }
614        }
615
616        if !is_quiet {
617            ctx.output().info("Press Ctrl+C to stop");
618        }
619        ctx.shutdown_signal().notified().await;
620
621        self.stop_server(ctx).await?;
622        if !is_quiet {
623            ctx.output().success("OPC UA simulator stopped");
624        }
625
626        Ok(CommandOutput::quiet_success())
627    }
628}
629
630#[async_trait]
631impl ProtocolCommand for OpcuaCommand {
632    fn protocol(&self) -> Protocol {
633        Protocol::OpcUa
634    }
635
636    fn default_port(&self) -> u16 {
637        4840
638    }
639
640    async fn start_server(&self, ctx: &mut CliContext) -> CliResult<()> {
641        let output = ctx.output();
642        let spinner = output.spinner("Starting OPC UA server...");
643
644        let config = OpcUaServerConfig {
645            endpoint_url: format!("opc.tcp://{}{}", self.bind_addr, self.endpoint_path),
646            server_name: "Mabinogion OPC UA Simulator".to_string(),
647            max_subscriptions: 1000,
648            max_monitored_items: 10000,
649            ..Default::default()
650        };
651
652        let server = Arc::new(OpcUaServer::new(config).map_err(|e| {
653            crate::error::CliError::ExecutionFailed {
654                message: format!("Failed to create OPC UA server: {}", e),
655            }
656        })?);
657
658        // Add sample nodes with diverse types and mixed read-only / writable access.
659        // Even-indexed nodes are writable, odd-indexed are read-only.
660        let node_count = self.nodes.min(100);
661        for i in 0..node_count {
662            let node_id = format!("ns=2;i={}", 1000 + i);
663            let name = format!("Variable_{}", i);
664            let value = (i as f64) * 0.1;
665
666            if i % 2 == 0 {
667                let _ = server.add_writable_variable(node_id, name, value);
668            } else {
669                let _ = server.add_variable(node_id, name, value);
670            }
671        }
672
673        {
674            let mut server_guard = self.server.lock().await;
675            *server_guard = Some(server.clone());
676        }
677
678        let server_clone = server.clone();
679        let task = tokio::spawn(async move {
680            if let Err(e) = server_clone.start().await {
681                tracing::error!("OPC UA server error: {}", e);
682            }
683        });
684
685        {
686            let mut task_guard = self.server_task.lock().await;
687            *task_guard = Some(task);
688        }
689
690        tokio::time::sleep(Duration::from_millis(100)).await;
691
692        spinner.finish_with_message(format!("OPC UA server started on {}", self.bind_addr));
693        Ok(())
694    }
695
696    async fn stop_server(&self, _ctx: &mut CliContext) -> CliResult<()> {
697        if let Some(server) = self.server.lock().await.as_ref() {
698            let _ = server.stop().await;
699        }
700
701        if let Some(task) = self.server_task.lock().await.take() {
702            let _ = tokio::time::timeout(Duration::from_secs(5), task).await;
703        }
704
705        Ok(())
706    }
707}
708
709// =============================================================================
710// BACnet Command
711// =============================================================================
712
713/// BACnet protocol command.
714pub struct BacnetCommand {
715    bind_addr: SocketAddr,
716    device_instance: u32,
717    objects: usize,
718    bbmd_enabled: bool,
719    /// Device tags.
720    tags: Tags,
721    /// Server instance (for shutdown).
722    server: Arc<Mutex<Option<Arc<BACnetServer>>>>,
723    /// Server task handle.
724    server_task: Arc<Mutex<Option<JoinHandle<()>>>>,
725}
726
727impl BacnetCommand {
728    pub fn new() -> Self {
729        Self {
730            bind_addr: "0.0.0.0:47808".parse().unwrap(),
731            device_instance: 1234,
732            objects: 0,
733            bbmd_enabled: false,
734            tags: Tags::new(),
735            server: Arc::new(Mutex::new(None)),
736            server_task: Arc::new(Mutex::new(None)),
737        }
738    }
739
740    pub fn with_port(mut self, port: u16) -> Self {
741        self.bind_addr.set_port(port);
742        self
743    }
744
745    pub fn with_device_instance(mut self, instance: u32) -> Self {
746        self.device_instance = instance;
747        self
748    }
749
750    pub fn with_objects(mut self, objects: usize) -> Self {
751        self.objects = objects;
752        self
753    }
754
755    pub fn with_bbmd(mut self, enabled: bool) -> Self {
756        self.bbmd_enabled = enabled;
757        self
758    }
759
760    pub fn with_tags(mut self, tags: Tags) -> Self {
761        self.tags = tags;
762        self
763    }
764}
765
766impl Default for BacnetCommand {
767    fn default() -> Self {
768        Self::new()
769    }
770}
771
772#[async_trait]
773impl Command for BacnetCommand {
774    fn name(&self) -> &str {
775        "bacnet"
776    }
777
778    fn description(&self) -> &str {
779        "Start a BACnet/IP simulator"
780    }
781
782    fn requires_engine(&self) -> bool {
783        true
784    }
785
786    fn supports_shutdown(&self) -> bool {
787        true
788    }
789
790    async fn execute(&self, ctx: &mut CliContext) -> CliResult<CommandOutput> {
791        let format = ctx.output().format();
792        let is_quiet = ctx.is_quiet();
793        let is_verbose = ctx.is_verbose();
794        let is_debug = ctx.is_debug();
795
796        if !is_quiet && matches!(format, OutputFormat::Table) {
797            let output = ctx.output();
798            output.header("BACnet/IP Simulator");
799            output.kv("Bind Address", self.bind_addr);
800            output.kv("Device Instance", self.device_instance);
801            output.kv("Objects", self.objects);
802            output.kv(
803                "BBMD",
804                if self.bbmd_enabled {
805                    "Enabled"
806                } else {
807                    "Disabled"
808                },
809            );
810        }
811
812        // Verbose: show extra details
813        if is_verbose {
814            let per_type = if self.objects > 0 {
815                std::cmp::max(1, self.objects / 4)
816            } else {
817                0
818            };
819            if per_type == 0 {
820                ctx.vprintln("  Demo Objects: disabled (Device object only)");
821            } else {
822                ctx.vprintln(format!(
823                    "  Objects per Type: {} (AI: {}, AO: {}, BI: {}, BO: {})",
824                    per_type, per_type, per_type, per_type, per_type
825                ));
826            }
827            ctx.vprintln("  Device Name: Mabinogion BACnet Simulator");
828        }
829
830        // Debug: dump full configuration
831        if is_debug {
832            ctx.dprintln(format!("Bind address: {}", self.bind_addr));
833            ctx.dprintln(format!("Device instance: {}", self.device_instance));
834            ctx.dprintln(format!(
835                "Total objects: {}, BBMD: {}",
836                self.objects, self.bbmd_enabled
837            ));
838        }
839
840        self.start_server(ctx).await?;
841
842        let per_type = if self.objects > 0 {
843            std::cmp::max(1, self.objects / 4)
844        } else {
845            0
846        };
847
848        if !is_quiet {
849            match format {
850                OutputFormat::Table => {
851                    let colors_enabled = ctx.colors_enabled();
852                    let demo_status = if per_type == 0 {
853                        "Not created"
854                    } else {
855                        "Active"
856                    };
857                    let table = TableBuilder::new(colors_enabled)
858                        .header(["Object Type", "Count", "Status"])
859                        .status_row(["Device", "1", "Online"], StatusType::Success)
860                        .status_row(
861                            ["Analog Input", &per_type.to_string(), demo_status],
862                            StatusType::Success,
863                        )
864                        .status_row(
865                            ["Analog Output", &per_type.to_string(), demo_status],
866                            StatusType::Success,
867                        )
868                        .status_row(
869                            ["Binary Input", &per_type.to_string(), demo_status],
870                            StatusType::Success,
871                        )
872                        .status_row(
873                            ["Binary Output", &per_type.to_string(), demo_status],
874                            StatusType::Success,
875                        );
876                    table.print();
877                }
878                _ => {
879                    #[derive(Serialize)]
880                    struct BacnetServerInfo {
881                        protocol: String,
882                        bind_address: String,
883                        device_instance: u32,
884                        objects: usize,
885                        bbmd_enabled: bool,
886                        object_types: Vec<ObjectTypeInfo>,
887                        status: String,
888                    }
889                    #[derive(Serialize)]
890                    struct ObjectTypeInfo {
891                        object_type: String,
892                        count: usize,
893                        status: String,
894                    }
895                    let info = BacnetServerInfo {
896                        protocol: "BACnet/IP".into(),
897                        bind_address: self.bind_addr.to_string(),
898                        device_instance: self.device_instance,
899                        objects: self.objects,
900                        bbmd_enabled: self.bbmd_enabled,
901                        object_types: vec![
902                            ObjectTypeInfo {
903                                object_type: "Device".into(),
904                                count: 1,
905                                status: "Online".into(),
906                            },
907                            ObjectTypeInfo {
908                                object_type: "Analog Input".into(),
909                                count: per_type,
910                                status: if per_type == 0 {
911                                    "Not created".into()
912                                } else {
913                                    "Active".into()
914                                },
915                            },
916                            ObjectTypeInfo {
917                                object_type: "Analog Output".into(),
918                                count: per_type,
919                                status: if per_type == 0 {
920                                    "Not created".into()
921                                } else {
922                                    "Active".into()
923                                },
924                            },
925                            ObjectTypeInfo {
926                                object_type: "Binary Input".into(),
927                                count: per_type,
928                                status: if per_type == 0 {
929                                    "Not created".into()
930                                } else {
931                                    "Active".into()
932                                },
933                            },
934                            ObjectTypeInfo {
935                                object_type: "Binary Output".into(),
936                                count: per_type,
937                                status: if per_type == 0 {
938                                    "Not created".into()
939                                } else {
940                                    "Active".into()
941                                },
942                            },
943                        ],
944                        status: "Online".into(),
945                    };
946                    let _ = ctx.output().write(&info);
947                }
948            }
949        }
950
951        if !is_quiet {
952            ctx.output().info("Press Ctrl+C to stop");
953        }
954        ctx.shutdown_signal().notified().await;
955
956        self.stop_server(ctx).await?;
957        if !is_quiet {
958            ctx.output().success("BACnet simulator stopped");
959        }
960
961        Ok(CommandOutput::quiet_success())
962    }
963}
964
965#[async_trait]
966impl ProtocolCommand for BacnetCommand {
967    fn protocol(&self) -> Protocol {
968        Protocol::BacnetIp
969    }
970
971    fn default_port(&self) -> u16 {
972        47808
973    }
974
975    async fn start_server(&self, ctx: &mut CliContext) -> CliResult<()> {
976        let output = ctx.output();
977        let spinner = output.spinner("Starting BACnet server...");
978
979        let config = BacnetServerConfig::new(self.device_instance)
980            .with_bind_addr(self.bind_addr)
981            .with_device_name("Mabinogion BACnet Simulator");
982
983        let registry = ObjectRegistry::new();
984
985        if self.objects > 0 {
986            let descriptors = default_object_descriptors();
987            let objects_per_type = std::cmp::max(1, self.objects / descriptors.len());
988            registry.populate_standard_objects(&descriptors, objects_per_type);
989        }
990
991        let server = Arc::new(BACnetServer::new(config, registry));
992
993        {
994            let mut server_guard = self.server.lock().await;
995            *server_guard = Some(server.clone());
996        }
997
998        let server_clone = server.clone();
999        let task = tokio::spawn(async move {
1000            if let Err(e) = server_clone.run().await {
1001                tracing::error!("BACnet server error: {}", e);
1002            }
1003        });
1004
1005        {
1006            let mut task_guard = self.server_task.lock().await;
1007            *task_guard = Some(task);
1008        }
1009
1010        tokio::time::sleep(Duration::from_millis(100)).await;
1011
1012        spinner.finish_with_message(format!("BACnet server started on {}", self.bind_addr));
1013        Ok(())
1014    }
1015
1016    async fn stop_server(&self, _ctx: &mut CliContext) -> CliResult<()> {
1017        if let Some(server) = self.server.lock().await.as_ref() {
1018            server.shutdown();
1019        }
1020
1021        if let Some(task) = self.server_task.lock().await.take() {
1022            let _ = tokio::time::timeout(Duration::from_secs(5), task).await;
1023        }
1024
1025        Ok(())
1026    }
1027}
1028
1029// =============================================================================
1030// KNX Command
1031// =============================================================================
1032
1033/// KNX protocol command.
1034pub struct KnxCommand {
1035    bind_addr: SocketAddr,
1036    individual_address: String,
1037    group_objects: usize,
1038    /// Device tags.
1039    tags: Tags,
1040    /// Server instance (for shutdown).
1041    server: Arc<Mutex<Option<Arc<KnxServer>>>>,
1042    /// Server task handle.
1043    server_task: Arc<Mutex<Option<JoinHandle<()>>>>,
1044}
1045
1046impl KnxCommand {
1047    pub fn new() -> Self {
1048        Self {
1049            bind_addr: "0.0.0.0:3671".parse().unwrap(),
1050            individual_address: "1.1.1".into(),
1051            group_objects: 100,
1052            tags: Tags::new(),
1053            server: Arc::new(Mutex::new(None)),
1054            server_task: Arc::new(Mutex::new(None)),
1055        }
1056    }
1057
1058    pub fn with_port(mut self, port: u16) -> Self {
1059        self.bind_addr.set_port(port);
1060        self
1061    }
1062
1063    pub fn with_individual_address(mut self, addr: impl Into<String>) -> Self {
1064        self.individual_address = addr.into();
1065        self
1066    }
1067
1068    pub fn with_group_objects(mut self, count: usize) -> Self {
1069        self.group_objects = count;
1070        self
1071    }
1072
1073    pub fn with_tags(mut self, tags: Tags) -> Self {
1074        self.tags = tags;
1075        self
1076    }
1077}
1078
1079impl Default for KnxCommand {
1080    fn default() -> Self {
1081        Self::new()
1082    }
1083}
1084
1085#[async_trait]
1086impl Command for KnxCommand {
1087    fn name(&self) -> &str {
1088        "knx"
1089    }
1090
1091    fn description(&self) -> &str {
1092        "Start a KNXnet/IP simulator"
1093    }
1094
1095    fn requires_engine(&self) -> bool {
1096        true
1097    }
1098
1099    fn supports_shutdown(&self) -> bool {
1100        true
1101    }
1102
1103    async fn execute(&self, ctx: &mut CliContext) -> CliResult<CommandOutput> {
1104        let format = ctx.output().format();
1105        let is_quiet = ctx.is_quiet();
1106        let is_verbose = ctx.is_verbose();
1107        let is_debug = ctx.is_debug();
1108
1109        if !is_quiet && matches!(format, OutputFormat::Table) {
1110            let output = ctx.output();
1111            output.header("KNXnet/IP Simulator");
1112            output.kv("Bind Address", self.bind_addr);
1113            output.kv("Individual Address", &self.individual_address);
1114            output.kv("Group Objects", self.group_objects);
1115        }
1116
1117        // Verbose: show extra details
1118        if is_verbose {
1119            ctx.vprintln("  Max Connections: 10");
1120            ctx.vprintln("  Services: Core, Device Management, Tunneling");
1121        }
1122
1123        // Debug: dump full configuration
1124        if is_debug {
1125            ctx.dprintln(format!("Bind address: {}", self.bind_addr));
1126            ctx.dprintln(format!("Individual address: {}", self.individual_address));
1127            ctx.dprintln(format!("Group objects: {}", self.group_objects));
1128        }
1129
1130        self.start_server(ctx).await?;
1131
1132        if !is_quiet {
1133            match format {
1134                OutputFormat::Table => {
1135                    let colors_enabled = ctx.colors_enabled();
1136                    let table = TableBuilder::new(colors_enabled)
1137                        .header(["Service", "Status"])
1138                        .status_row(["Core", "Ready"], StatusType::Success)
1139                        .status_row(["Device Management", "Ready"], StatusType::Success)
1140                        .status_row(["Tunneling", "Ready"], StatusType::Success);
1141                    table.print();
1142                }
1143                _ => {
1144                    #[derive(Serialize)]
1145                    struct KnxServerInfo {
1146                        protocol: String,
1147                        bind_address: String,
1148                        individual_address: String,
1149                        group_objects: usize,
1150                        services: Vec<ServiceInfo>,
1151                        status: String,
1152                    }
1153                    #[derive(Serialize)]
1154                    struct ServiceInfo {
1155                        service: String,
1156                        status: String,
1157                    }
1158                    let info = KnxServerInfo {
1159                        protocol: "KNXnet/IP".into(),
1160                        bind_address: self.bind_addr.to_string(),
1161                        individual_address: self.individual_address.clone(),
1162                        group_objects: self.group_objects,
1163                        services: vec![
1164                            ServiceInfo {
1165                                service: "Core".into(),
1166                                status: "Ready".into(),
1167                            },
1168                            ServiceInfo {
1169                                service: "Device Management".into(),
1170                                status: "Ready".into(),
1171                            },
1172                            ServiceInfo {
1173                                service: "Tunneling".into(),
1174                                status: "Ready".into(),
1175                            },
1176                        ],
1177                        status: "Online".into(),
1178                    };
1179                    let _ = ctx.output().write(&info);
1180                }
1181            }
1182        }
1183
1184        if !is_quiet {
1185            ctx.output().info("Press Ctrl+C to stop");
1186        }
1187        ctx.shutdown_signal().notified().await;
1188
1189        self.stop_server(ctx).await?;
1190        if !is_quiet {
1191            ctx.output().success("KNX simulator stopped");
1192        }
1193
1194        Ok(CommandOutput::quiet_success())
1195    }
1196}
1197
1198#[async_trait]
1199impl ProtocolCommand for KnxCommand {
1200    fn protocol(&self) -> Protocol {
1201        Protocol::KnxIp
1202    }
1203
1204    fn default_port(&self) -> u16 {
1205        3671
1206    }
1207
1208    async fn start_server(&self, ctx: &mut CliContext) -> CliResult<()> {
1209        let output = ctx.output();
1210        let spinner = output.spinner("Starting KNX server...");
1211
1212        // Parse individual address
1213        let individual_address: IndividualAddress =
1214            self.individual_address.parse().map_err(|_| {
1215                crate::error::CliError::ExecutionFailed {
1216                    message: format!("Invalid individual address: {}", self.individual_address),
1217                }
1218            })?;
1219
1220        let config = KnxServerConfig {
1221            bind_addr: self.bind_addr,
1222            individual_address,
1223            max_connections: 256,
1224            ..Default::default()
1225        };
1226
1227        // Create group objects based on --groups parameter
1228        let group_table = Arc::new(GroupObjectTable::new());
1229        let dpt_types = [
1230            DptId::new(1, 1),   // Switch (bool)
1231            DptId::new(5, 1),   // Scaling (0-100%)
1232            DptId::new(9, 1),   // Temperature (float16)
1233            DptId::new(9, 4),   // Lux
1234            DptId::new(9, 7),   // Humidity
1235            DptId::new(12, 1),  // Counter (u32)
1236            DptId::new(13, 1),  // Counter signed (i32)
1237            DptId::new(14, 56), // Float (f32)
1238        ];
1239        let dpt_names = [
1240            "Switch",
1241            "Scaling",
1242            "Temperature",
1243            "Lux",
1244            "Humidity",
1245            "Counter",
1246            "SignedCounter",
1247            "Float",
1248        ];
1249
1250        for i in 0..self.group_objects {
1251            let main = ((i / 256) + 1) as u8;
1252            let middle = ((i / 8) % 8) as u8;
1253            let sub = (i % 256) as u8;
1254            let addr = GroupAddress::three_level(main, middle, sub);
1255            let dpt_idx = i % dpt_types.len();
1256            let name = format!("{}_{}", dpt_names[dpt_idx], i);
1257            if let Err(e) = group_table.create(addr, &name, &dpt_types[dpt_idx]) {
1258                tracing::warn!("Failed to create group object {}: {}", i, e);
1259            }
1260        }
1261
1262        let server = Arc::new(KnxServer::new(config).with_group_objects(group_table));
1263
1264        {
1265            let mut server_guard = self.server.lock().await;
1266            *server_guard = Some(server.clone());
1267        }
1268
1269        let server_clone = server.clone();
1270        let task = tokio::spawn(async move {
1271            if let Err(e) = server_clone.start().await {
1272                tracing::error!("KNX server error: {}", e);
1273            }
1274        });
1275
1276        {
1277            let mut task_guard = self.server_task.lock().await;
1278            *task_guard = Some(task);
1279        }
1280
1281        tokio::time::sleep(Duration::from_millis(100)).await;
1282
1283        spinner.finish_with_message(format!("KNX server started on {}", self.bind_addr));
1284        Ok(())
1285    }
1286
1287    async fn stop_server(&self, _ctx: &mut CliContext) -> CliResult<()> {
1288        // Take server out to call stop (KnxServer::stop has Send issues with parking_lot)
1289        let server_opt = self.server.lock().await.take();
1290        if let Some(server) = server_opt {
1291            // Use spawn_blocking to handle the non-Send future
1292            let _ = tokio::task::spawn_blocking(move || {
1293                let rt = tokio::runtime::Handle::current();
1294                rt.block_on(async {
1295                    let _ = server.stop().await;
1296                })
1297            })
1298            .await;
1299        }
1300
1301        if let Some(task) = self.server_task.lock().await.take() {
1302            let _ = tokio::time::timeout(Duration::from_secs(5), task).await;
1303        }
1304
1305        Ok(())
1306    }
1307}
mabi_cli/commands/protocol.rs

mabi_cli/commands/
protocol.rs
