git.rot13.org Git - zc/blob - zc-mqtt

   1 #!/usr/bin/perl
   2 use strict;
   3 use warnings;
   4
   5 # based on net-mqtt-sub example from Net::MQTT
   6 #
   7 # 2020-08-15 Dobrica Pavlinusic <dpavlin@rot13.org>
   8
   9 use strict;
  10 use Net::MQTT::Constants;
  11 use Net::MQTT::Message;
  12 use IO::Select;
  13 use IO::Socket::INET;
  14 use Time::HiRes;
  15 use Getopt::Long;
  16 use Pod::Usage;
  17 use POSIX qw(strftime);
  18
  19 $|=1; # flush STDOUT
  20
  21 my $help;
  22 my $man;
  23 my $verbose = 2;
  24 my $host = 'mqtt.zc-sensor.com';
  25 my $port = 1883;
  26 my $count;
  27 my $client_id;
  28 my $keep_alive_timer = 120;
  29 GetOptions('help|?' => \$help,
  30            'man' => \$man,
  31            'verbose+' => \$verbose,
  32            'host=s' => \$host,
  33            'port=i' => \$port,
  34            'count=i' => \$count,
  35            'one|1' => sub { $count = 1 },
  36            'client_id|client-id|C=s' => \$client_id,
  37            'keepalive=i' => \$keep_alive_timer) or pod2usage(2);
  38 pod2usage(1) if ($help);
  39 pod2usage(-exitstatus => 0, -verbose => 2) if $man;
  40 #pod2usage(2) unless (@ARGV); # need a topic
  41 push @ARGV, '#' unless (@ARGV);
  42
  43 open_again:
  44 print "Open $host:port keep_alive_timer: $keep_alive_timer\n";
  45 my $socket =
  46   IO::Socket::INET->new(PeerAddr => $host.':'.$port,
  47                         Timeout => $keep_alive_timer,
  48                        ) or die "Socket connect failed: $!\n";
  49
  50 my $buf = '';
  51 my $mid = 1;
  52 my $next_ping;
  53 my $got_ping_response = 1;
  54 my @connect = ( message_type => MQTT_CONNECT,
  55                 keep_alive_timer => $keep_alive_timer );
  56 push @connect, client_id => $client_id if (defined $client_id);
  57 send_message($socket, @connect);
  58 my $msg = read_message($socket, $buf) or die "No ConnAck\n";
  59 print 'Received: ', $msg->string, "\n" if ($verbose >= 2);
  60 send_message($socket, message_type => MQTT_SUBSCRIBE,
  61              message_id => $mid++,
  62              topics => [ map { [ $_ => MQTT_QOS_AT_MOST_ONCE ] } @ARGV ]);
  63 $msg = read_message($socket, $buf) or die "No SubAck\n";
  64 print 'Received: ', $msg->string, "\n" if ($verbose >= 2);
  65
  66 while (1) {
  67   $msg = read_message($socket, $buf);
  68   if ($msg) {
  69     print "\n",strftime("%Y-%m-%d %H:%M:%S ", localtime()) if $msg->message_type != MQTT_PINGRESP;
  70     if ($msg->message_type == MQTT_PUBLISH) {
  71       if ($verbose == 0) {
  72         print $msg->topic, " ", $msg->message, "\n";
  73       } else {
  74         print $msg->string, "\n";
  75       }
  76
  77 #      if ( $msg->topic =~ m{Inclinometer/ZCT330Ex_SWP_N_YK/869858031634109/up} ) {
  78       if ( substr($msg->message,2,1) eq "\x07" ) { # heartbeat
  79         my $raw = "\x5a\x0b\x03\x09\x00\x00\x04\xe8\x03\x00\x00\x21\x44\xaa\x4B\xF3";
  80         my $topic = $msg->topic;
  81         $topic =~ s/up$/down/;
  82         send_message($socket,
  83                 message_type => MQTT_PUBLISH,
  84                 retain => 0, #$retain,
  85                 topic => $topic,
  86                 message => $raw);
  87       }
  88
  89       if (defined $count && --$count == 0) {
  90         exit;
  91       }
  92     } elsif ($msg->message_type == MQTT_PINGRESP) {
  93       $got_ping_response = 1;
  94       print 'Received: ', $msg->string, "\n" if ($verbose >= 3);
  95     } else {
  96       print 'Received: ', $msg->string, "\n" if ($verbose >= 2);
  97     }
  98   }
  99   if (Time::HiRes::time > $next_ping) {
 100     die "Ping Response timeout.  Exiting\n" unless ($got_ping_response);
 101     send_message($socket, message_type => MQTT_PINGREQ);
 102   }
 103 }
 104
 105 sub send_message {
 106   my $socket = shift;
 107   my $msg = Net::MQTT::Message->new(@_);
 108   print 'Sending: ', $msg->string, "\n" if ($verbose >= 2 && $_[1] eq 'message_type' && $_[2] != MQTT_PINGREQ);
 109   $msg = $msg->bytes;
 110   syswrite $socket, $msg, length $msg;
 111   print dump_string($msg, 'Sent: '), "\n\n" if ($verbose >= 3);
 112   $next_ping = Time::HiRes::time + $keep_alive_timer;
 113 }
 114
 115 sub read_message {
 116   my $socket = shift;
 117   my $select = IO::Select->new($socket);
 118   my $timeout = $next_ping - Time::HiRes::time;
 119   do {
 120     my $mqtt = Net::MQTT::Message->new_from_bytes($_[0], 1);
 121     return $mqtt if (defined $mqtt);
 122     $select->can_read($timeout) || return;
 123     $timeout = $next_ping - Time::HiRes::time;
 124     my $bytes = sysread $socket, $_[0], 2048, length $_[0];
 125     unless ($bytes) {
 126       warn "Socket closed ", (defined $bytes ? 'gracefully' : 'error'), "\n";
 127       # FIXME reopen
 128       goto open_again;
 129     }
 130     print "Receive buffer: ", dump_string($_[0], '   '), "\n\n"
 131       if ($verbose >= 3);
 132   } while ($timeout > 0);
 133   return;
 134 }
 135
 136 __END__
 137
 138 =pod
 139
 140 =encoding UTF-8
 141
 142 =head1 NAME
 143
 144 net-mqtt-sub - Perl script for subscribing to an MQTT topic
 145
 146 =head1 VERSION
 147
 148 version 1.143260
 149
 150 =head1 SYNOPSIS
 151
 152   net-mqtt-sub [options] topic1 [topic2] [topic3] ...
 153
 154 =head1 DESCRIPTION
 155
 156 This script subscribes to one or more MQTT topics and prints any
 157 messages that it receives to stdout.
 158
 159 =head1 OPTIONS
 160
 161 =over
 162
 163 =item B<-help>
 164
 165 Print a brief help message.
 166
 167 =item B<-man>
 168
 169 Print the manual page.
 170
 171 =item B<-host>
 172
 173 The host running the MQTT service.  The default is C<127.0.0.1>.
 174
 175 =item B<-port>
 176
 177 The port of the running MQTT service.  The default is 1883.
 178
 179 =item B<-client-id>
 180
 181 The client id to use in the connect message.  The default is
 182 'NetMQTTpm' followed by the process id of the process.
 183
 184 =item B<-verbose>
 185
 186 Include more verbose output.  Without this option the script only
 187 outputs errors and received messages one per line in the form:
 188
 189   topic message
 190
 191 With one B<-verbose> options, publish messages are printed in a form
 192 of a summary of the header fields and the payload in hex dump and text
 193 form.
 194
 195 With two B<-verbose> options, summaries are printed for all messages
 196 sent and received.
 197
 198 With three B<-verbose> options, a hex dump of all data transmitted and
 199 received is printed.
 200
 201 =item B<-keepalive NNN>
 202
 203 The keep alive timer value.  Defaults to 120 seconds.  For simplicity,
 204 it is also currently used as the connection/subscription timeout.
 205
 206 =item B<-count NNN>
 207
 208 Read the specificed number of MQTT messages and then exit.  Default
 209 is 0 - read forever.
 210
 211 =item B<-one> or B<-1>
 212
 213 Short for B<-count 1>.  Read one message and exit.
 214
 215 =back
 216
 217 =head1 SEE ALSO
 218
 219 Net::MQTT::Message(3)
 220
 221 =head1 DISCLAIMER
 222
 223 This is B<not> official IBM code.  I work for IBM but I'm writing this
 224 in my spare time (with permission) for fun.
 225
 226 =head1 AUTHOR
 227
 228 Mark Hindess <soft-cpan@temporalanomaly.com>
 229
 230 =head1 COPYRIGHT AND LICENSE
 231
 232 This software is copyright (c) 2014 by Mark Hindess.
 233
 234 This is free software; you can redistribute it and/or modify it under
 235 the same terms as the Perl 5 programming language system itself.
 236
 237 =cut
 238
 239 # 6. Protocol Format
 240 # Valid data ID and parameter range supported by the product
 241 # DATA ID       ID Description  Data Type( Data Length）       R/W     Range   Default         Remark
 242
 243 __DATA__
 244 0x00    Seq #                   DWord(4)        R       /               0       The platform can carry the ID when the platform downstream reads and sets the device parameters, and the device returns the same data. Please refer to the example for use.Each seq # refer to one command and its response.
 245 0x01    PN                      DWord(4)        R       /                /      /
 246 0x02    Model                   Byte(1)         R       32              32      Inner number:32
 247 0x03    X axis angle            Float(4)        R        ‐90°~90°   /       X axis angle
 248 0x04    Y axis angle            Float(4)        R       ‐90°~90°    /       Y axis angle
 249 0x09    X axis relative angle   Float(4)        R       ‐90°~90°    0       Return the X angle value according to the set relative zero
 250 0x0A    Y axis relative angle   Float(4)        R       ‐90°~90°    0       Return the Y angle value according to the set relative zero
 251 0x0C    Sensor temperature      Word(2)         R       ‐32768~32767  /       signed,sensor temperature =Data/100, unit ℃
 252 0x0D    Power source voltage    Word(2)         R       0~65535         /       Voltage= Data/100, unit V
 253 0x11     Arming/disarming       Byte(1)         R/W     0~255           1       0 means disarming, non‐zero means arming
 254 0x12    Alarm delay time        Byte(1)         R/W     3~255           20       The unit is 0.1 second, which means that the product responds to the alarm only after the alarm has exceeded the alarm angle for a certain period of time.
 255
 256 0x13    Restore factory setting Byte(1)         R/W     0~255           0       0: Do nothing Non‐zero: Restore the non‐network‐related parameters of the sensor.
 257 0x14    Server IP&port          4*Byte(1)+Word(2)       R/W     /       CTIOT:117.60.157.137,5683 MQTT:0.0.0.0,0        The server address should be IP, not the domain name; using the domain name may cause the connection server to be unstable and cause data loss.
 258 0x17    Signal strength         Byte(1)         R       10~34           /       A larger value indicates a stronger signal
 259 0x18    Sensor operating mode   Byte(1)         R/W     0               0       The sensor can only work in absolute measurement mode
 260 0x19     Alarm axis             Byte(1)         R       0~      3       /       0: no alarm; 1: X‐axis alarm 2: Y axis alarm; 3: X/Y axis alarm at the same time
 261 0x1A    SIM card ID             QWord(8)        R       0~18446744073709551615  /       Take the first 19 digits, the last digit is discarded
 262 0x21     Heartbeat interval     DWord(4)        R/W     60~131071        86400  Interval at which the device periodically uploads data to the server
 263 0x22    IMEI number of the device       QWord(8)         R      0~18446744073709551615  /       Refers to the IMEI of the NB‐IOT network module in the product.
 264 0x23    Backup server IP&port   4*Byte(1)+Word(2)       R/W     /       CTIOT:117.60.157.137,5683 MQTT:0.0.0.0,0        /
 265 0x24    Backup server enable    Byte(1)         R/W     0~255   0       0 means off, non‐zero means on
 266 0x33    DNS IP address          4*Byte(1)       R/W     /       208.67.222.222  /
 267 0x34    Domain name and port    64*Byte(1)      R/W     /       mqtt.zc‐sensor.com,1883       Supports CTIOT and MQTT protocols. Priority IP in the case of IP (ID number 0x14); domain name and port should be separated with comma, length <=64
 268 0x35     MQTT‐ClientID                32*Byte(1)      R/W      /      IMEI number of the device       Length <=32, subject to MQTT related specifications
 269 0x36    MQTT‐Username                 32*Byte(1)      R/W     /       empty           Length <=32, subject to MQTT related specifications