Merge pull request #38 from GlobalFishingWatch/update-tagblock-cli

cli command for update_tagblock

ais_tools/__init__.py

@@ -2,7 +2,7 @@
 Tools for managing AIS messages
 """
 
-__version__ = 'v0.1.5.dev2'
+__version__ = 'v0.1.5.dev3'
 __author__ = 'Paul Woods'
 __email__ = 'paul@globalfishingwatch.org'
 __source__ = 'https://github.com/GlobalFishingWatch/ais-tools'

ais_tools/cli.py

@@ -72,6 +72,39 @@ def add_tagblock(input, output, station):
         output.write('\n')
 
 
+@cli.command(
+    short_help="Update existing tagblock with specified field values.  Create a new tagblock if none is present",
+    help="Utility for updating a stream of raw AIVDM sentences with tagblocks, modifying the tagblock to"
+         "add or overwrite selected fields "
+         "\n\n"
+         "INPUT should be a text stream with one NMEA message per line, and defaults to stdin.  Use '-' to explicitly "
+         "use stdin"
+         "\n\n"
+         "OUTPUT is a text stream of the input NMEA with modified tagblock"
+         "\n\n"
+         "For example:"
+         "\n\n"
+         "$ echo '\\c:1577762601537,s:99\\!AIVDM,1,1,,A,15NTES0P00J>tC4@@FOhMgvD0D0M,0*49' | \\"
+         "  ais_tools update-tagblock -s my-station"
+         "\n\n"
+         "outputs something like"
+         "\n\n"
+         "\\c:1577762601537,s:my-station*5D\\!AIVDM,1,1,,A,15NTES0P00J>tC4@@FOhMgvD0D0M,0*49"
+)
+@click.argument('input', type=click.File('r'), default='-')
+@click.argument('output', type=click.File('w'), default='-')
+@click.option('-s', '--station',
+              help="identifier for the receiving station")
+@click.option('-t', '--text',
+              help="tagblock text field")
+def update_tagblock(input, output, station, text):
+    fields = {'tagblock_station': station, 'tagblock_text': text}
+    fields = {k: v for k, v in fields.items() if v is not None}
+    for nmea in input:
+        output.write(tagblock.update_tagblock(nmea, **fields))
+        output.write('\n')
+
+
 @cli.command(
     short_help="Decode AIS from NMEA to JSON",
     help="Decode AIS from NMEA to JSON"

tests/test_cli.py

@@ -4,6 +4,7 @@
 import re
 
 from ais_tools.cli import add_tagblock
+from ais_tools.cli import update_tagblock
 from ais_tools.cli import decode
 from ais_tools.cli import encode
 from ais_tools.cli import join_multipart
@@ -36,6 +37,20 @@ def test_add_tagblock():
     assert tagblock['tagblock_station'] == 'test'
 
 
+def test_update_tagblock():
+
+    runner = CliRunner()
+    input = '!AIVDM,1,1,,A,B>cSnNP00FVur7UaC7WQ3wS1jCJJ,0*73'
+    args = '--station=test'
+    result = runner.invoke(update_tagblock, input=input, args=args)
+    assert not result.exception
+
+    tagblock_str, nmea = split_tagblock(result.output.strip())
+    tagblock = decode_tagblock(tagblock_str)
+    assert nmea == input
+    assert tagblock['tagblock_station'] == 'test'
+
+
 def test_decode():
     runner = CliRunner()
     input = '\\c:1599239526500,s:ais-tools,T:2020-09-04 18.12.06*5D\\!AIVDM,1,1,,A,B>cSnNP00FVur7UaC7WQ3wS1jCJJ,0*73'

tests/test_tagblock.py

@@ -118,13 +118,13 @@ def test_decode_tagblock_invalid(tagblock_str):
 
 
 @pytest.mark.parametrize("tagblock_str,new_fields,expected", [
-    ('!AIVDM', {'q':123}, '\\q:123*7B\\!AIVDM'),
-    ('\\!AIVDM', {'q':123}, '\\q:123*7B\\!AIVDM'),
-    ('\\s:00*00\\!AIVDM', {'tagblock_station':99}, '\\s:99*49\\!AIVDM'),
-    ('\\s:00*00\\!AIVDM\\s:00*00\\!AIVDM', {'tagblock_station':99}, '\\s:99*49\\!AIVDM\\s:00*00\\!AIVDM'),
+    ('!AIVDM', {'q': 123}, '\\q:123*7B\\!AIVDM'),
+    ('\\!AIVDM', {'q': 123}, '\\q:123*7B\\!AIVDM'),
+    ('\\s:00*00\\!AIVDM', {'tagblock_station': 99}, '\\s:99*49\\!AIVDM'),
+    ('\\s:00*00\\!AIVDM\\s:00*00\\!AIVDM', {'tagblock_station': 99}, '\\s:99*49\\!AIVDM\\s:00*00\\!AIVDM'),
     ('\\c:123456789*68\\!AIVDM', {}, '\\c:123456789*68\\!AIVDM'),
-    ('\\c:123456789*68\\!AIVDM', {'tagblock_station':99}, '\\c:123456789,s:99*0D\\!AIVDM'),
-    ('\\c:123456789*68\\!AIVDM', {'q':123}, '\\c:123456789,q:123*3F\\!AIVDM'),
+    ('\\c:123456789*68\\!AIVDM', {'tagblock_station': 99}, '\\c:123456789,s:99*0D\\!AIVDM'),
+    ('\\c:123456789*68\\!AIVDM', {'q': 123}, '\\c:123456789,q:123*3F\\!AIVDM'),
 ])
 def test_update_tagblock(tagblock_str, new_fields, expected):
-    assert expected == tagblock.update_tagblock(tagblock_str, **new_fields)
+    assert expected == tagblock.update_tagblock(tagblock_str, **new_fields)