Skip to content

Smc ftp

SmcFtpTrackObserver

Bases: TrackObserver

Update track metadata for DLS and DL+ to the SMC FTP server.

Source code in nowplaying/track/observers/smc_ftp.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class SmcFtpTrackObserver(TrackObserver):
    """Update track metadata for DLS and DL+ to the SMC FTP server."""

    name = "SMC FTP"

    class Options(TrackObserver.Options):  # pragma: no coverage
        @classmethod
        def args(cls, args: configargparse.ArgParser):
            args.add_argument(
                "--dab-smc",
                help="Enable SMC FTP delivery",
                type=bool,
                default=False,
            )
            args.add_argument(
                "--dab-smc-ftp-hostname",
                help="Hostname of SMC FTP server",
                default=[],
            )
            args.add_argument(
                "--dab-smc-ftp-username",
                help="Username for SMC FTP server",
            )
            args.add_argument(
                "--dab-smc-ftp-password", help="Password for SMC FTP server"
            )

        def __init__(self, hostname: str, username: str, password: str) -> None:
            self.hostname: str = hostname
            self.username: str = username
            self.password: str = password

    def __init__(self, options: Options):
        self._options = options

    def track_started(self, track: Track):
        logger.info(f"Updating DAB+ DLS for track {track.artist=} {track.title=}")

        if track.get_duration() < timedelta(seconds=5):
            logger.info(
                "Track is less than 5 seconds, not sending to SMC"
                f"{track.artist=} {track.title=}"
            )
            return

        dls, dlplus = _dls_from_track(track)

        # check for too long meta and shorten to just artist
        if dls.getbuffer().nbytes > 128:  # pragma: no cover
            logger.warning(f"SMC DLS to long {dls.getvalue().decode('latin1')=}")
            dls, dlplus = _dls_from_track(track, title=False)

        ftp = FTP_TLS()
        ftp.connect(self._options.hostname)
        ftp.sendcmd(f"USER {self._options.username}")
        ftp.sendcmd(f"PASS {self._options.password}")

        ftp.storlines("STOR /dls/nowplaying.dls", dls)
        ftp.storlines("STOR /dlplus/nowplaying.dls", dlplus)

        ftp.quit()
        ftp.close()

        logger.info(
            f"SMC FTP {self._options.hostname=} "
            f"{dls.getvalue().decode('latin1')=} "
            f"{dlplus.getvalue().decode('latin1')=}"
        )

    def track_finished(self, track):
        return True