Farnsworth contact dedupe

src/dzcb/farnsworth.py

@@ -50,8 +50,14 @@ def Contact_to_dict(c):
 )
 
 
-def GroupList_to_dict(g):
-    return {"Name": g.name, "Contact": [tg.name for tg in g.contacts]}
+def GroupList_to_dict(g, contacts_by_id):
+    return {
+        "Name": g.name,
+        "Contact": [
+            contacts_by_id.get(tg.dmrid, tg).name
+            for tg in g.contacts
+        ],
+    }
 
 
 def ScanList_to_dict(s):
@@ -168,13 +174,17 @@ def AnalogChannel_to_dict(c, codeplug):
 )
 
 
-def DigitalChannel_to_dict(c, codeplug):
+def DigitalChannel_to_dict(c, codeplug, contacts_by_id):
     d = DefaultChannel.copy()
+    talkgroup_name = "Parrot 1"
+    if c.talkgroup:
+        # get the dedupe'd contact's name for the given ID
+        talkgroup_name = str(contacts_by_id.get(c.talkgroup.dmrid, c.talkgroup).name)
     d.update(
         {
             "ChannelMode": "Digital",
             "RepeaterSlot": str(c.talkgroup.timeslot) if c.talkgroup else 1,
-            "ContactName": str(c.talkgroup.name) if c.talkgroup else "Parrot 1",
+            "ContactName": talkgroup_name,
             "GroupList": str(c.grouplist_name(codeplug)) if c.grouplist else None,
             "ScanList": dzcb.munge.zone_name(c.scanlist_name(codeplug), NAME_MAX),
         }
@@ -197,12 +207,12 @@ def DigitalChannel_to_dict(c, codeplug):
 }
 
 
-def Channel_to_dict(c, codeplug):
+def Channel_to_dict(c, codeplug, contacts_by_id):
     d = None
     if isinstance(c, AnalogChannel):
         d = AnalogChannel_to_dict(c, codeplug)
     elif isinstance(c, DigitalChannel):
-        d = DigitalChannel_to_dict(c, codeplug)
+        d = DigitalChannel_to_dict(c, codeplug, contacts_by_id)
     if d is None:
         raise ValueError("Unknown type: {}".format(c))
     return {k: Channel_value_replacements.get(v, str(v)) for k, v in d.items()}
@@ -233,14 +243,15 @@ def Codeplug_to_json(cp, based_on=None):
         ranges.append((basic_info["LowFrequencyB"], basic_info["HighFrequencyB"]))
     if ranges:
         cp = cp.filter(ranges=ranges)
+    contacts_by_id = {
+        c.dmrid: c
+        for c in uniquify_contacts(cp.contacts, ignore_timeslot=True)
+    }
     cp_dict.update(
         dict(
-            Contacts=[
-                Contact_to_dict(c)
-                for c in uniquify_contacts(cp.contacts, ignore_timeslot=True)
-            ],
-            Channels=[Channel_to_dict(c, cp) for c in cp.channels],
-            GroupLists=[GroupList_to_dict(c) for c in cp.grouplists],
+            Contacts=[Contact_to_dict(c) for c in contacts_by_id.values()],
+            Channels=[Channel_to_dict(c, cp, contacts_by_id) for c in cp.channels],
+            GroupLists=[GroupList_to_dict(c, contacts_by_id) for c in cp.grouplists],
             ScanLists=[ScanList_to_dict(c) for c in cp.scanlists],
             Zones=[Zone_to_dict(c) for c in cp.zones],
         )

tests/digital-repeaters-same-tg-different-ts/Digital-Repeaters__A.csv

@@ -0,0 +1,2 @@
+Zone Name,Comment,Power,RX Freq,TX Freq,Color Code,PNW Rgnl
+DPHS Static;DHS,Static Tgs,Low,430.4375,439.4375,1,2

tests/digital-repeaters-same-tg-different-ts/Digital-Repeaters__F.csv

@@ -0,0 +1,3 @@
+Zone Name,Comment,Power,RX Freq,TX Freq,Color Code,PNW Rgnl 2
+MM/Bridget 430;MMB,,Low,430.4125,430.4125,1,1
+Moses Lk/Frechmn;MOF,,High,440.925,445.925,1,2

tests/digital-repeaters-same-tg-different-ts/Talkgroups__A.csv

@@ -0,0 +1 @@
+PNW Rgnl,31771

tests/digital-repeaters-same-tg-different-ts/Talkgroups__F.csv

@@ -0,0 +1 @@
+PNW Rgnl 2,31771

tests/test_k7abd.py

@@ -1,9 +1,10 @@
+import json
 import os
 from pathlib import Path
 
 import pytest
 
-from dzcb import k7abd
+from dzcb import farnsworth, k7abd
 from dzcb.model import Timeslot, ContactType
 
 
@@ -155,3 +156,48 @@ def test_digital_repeaters_private_contacts():
         ("Parrot", 9998, Timeslot.ONE, ContactType.GROUP),
         ("Private Parrot", 9998, Timeslot.ONE, ContactType.PRIVATE),
     ]
+
+
+def test_digital_repeaters_same_tg_different_ts():
+    """
+    A talkgroup may be carried on either timeslot.
+    """
+
+    cp = codeplug_from_relative_dir("digital-repeaters-same-tg-different-ts")
+    assert len(cp.zones) == 2
+    assert len(cp.contacts) == 2
+    assert len(cp.channels) == 3
+
+    expanded_cp = cp.expand_static_talkgroups()
+    assert len(expanded_cp.zones) == 3
+    assert len(expanded_cp.contacts) == 2
+    assert len(expanded_cp.channels) == 3
+
+    expect_channels = [
+        ("PNW Rgnl 2 DHS", "PNW Rgnl", Timeslot.TWO),
+        ("PNW Rgnl 2 1 MMB", "PNW Rgnl 2", Timeslot.ONE),
+        ("PNW Rgnl 2 MOF", "PNW Rgnl 2", Timeslot.TWO),
+    ]
+
+    print("EXPECT CHANNELS:\n{}".format("\n".join(str(ch) for ch in expect_channels)))
+    print(
+        "ACTUAL CHANNELS:\n{}".format("\n".join(str(ch) for ch in expanded_cp.channels))
+    )
+
+    for ch, exp_ch in zip(expanded_cp.channels, expect_channels):
+        assert ch.name == exp_ch[0]
+        assert ch.talkgroup.name == exp_ch[1]
+        assert ch.talkgroup.timeslot == exp_ch[2]
+
+    # farnsworth output collapses talkgroups by ID, so only 1 FW contact should be generated
+    # for the 2 contacts in the codeplug
+    fw_cp = json.loads(farnsworth.Codeplug_to_json(expanded_cp))
+    assert len(fw_cp["Contacts"]) == 1
+    tg_name = fw_cp["Contacts"][0]["Name"]
+    assert len(fw_cp["Channels"]) == 3
+    for channel in fw_cp["Channels"]:
+        assert channel["ContactName"] == tg_name
+
+    assert len(fw_cp["GroupLists"]) == 3
+    for grouplist in fw_cp["GroupLists"]:
+        assert grouplist["Contact"] == [tg_name]