Skip to content

Smc ftp

Upload PAD to SMC.

SmcFtpTrackObserver

Bases: TrackObserver

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

Source code in nowplaying/track/observers/smc_ftp.py 
 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
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
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
        """Options for SmcFtpTrackObserver."""

        @classmethod
        def args(
            cls: type[TTrackObserverOptions],
            args: configargparse.ArgParser,
        ) -> None:
            """Create args."""
            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: Self, hostname: str, username: str, password: str) -> None:
            """Create SmcFtpTrackObserver.Config."""
            self.hostname: str = hostname
            self.username: str = username
            self.password: str = password

    def __init__(self: Self, options: Options) -> None:
        """Create SmcFtpTrackObserver."""
        self._options = options

    def track_started(self: Self, track: Track) -> None:
        """Track started."""
        logger.info(
            "Updating DAB+ DLS for track artist=%s title=%s",
            track.artist,
            track.title,
        )

        if track.get_duration() < timedelta(seconds=5):
            logger.info(
                "Track is less than 5 seconds, not sending to SMC artist=%s title=%s",
                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 > _NOWPLAYING_DAB_MAXLEN:  # pragma: no cover
            logger.warning("SMC DLS to long %s", dls.getvalue().decode("latin1"))
            dls, dlplus = _dls_from_track(track, title=False)

        ftp = FTP_TLS()  # noqa: S321
        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(
            "SMC FTP hostname=%s dls=%s dlsplus=%s",
            self._options.hostname,
            dls.getvalue().decode("latin1"),
            dlplus.getvalue().decode("latin1"),
        )

    def track_finished(self: Self, _: Track) -> None:
        """Track finished."""
        return

Options

Bases: Options

Options for SmcFtpTrackObserver.

Source code in nowplaying/track/observers/smc_ftp.py 
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
class Options(TrackObserver.Options):  # pragma: no coverage
    """Options for SmcFtpTrackObserver."""

    @classmethod
    def args(
        cls: type[TTrackObserverOptions],
        args: configargparse.ArgParser,
    ) -> None:
        """Create args."""
        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: Self, hostname: str, username: str, password: str) -> None:
        """Create SmcFtpTrackObserver.Config."""
        self.hostname: str = hostname
        self.username: str = username
        self.password: str = password

__init__(hostname, username, password)

Create SmcFtpTrackObserver.Config.

Source code in nowplaying/track/observers/smc_ftp.py 
59
60
61
62
63
def __init__(self: Self, hostname: str, username: str, password: str) -> None:
    """Create SmcFtpTrackObserver.Config."""
    self.hostname: str = hostname
    self.username: str = username
    self.password: str = password

args(args) classmethod

Create args.

Source code in nowplaying/track/observers/smc_ftp.py 
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
@classmethod
def args(
    cls: type[TTrackObserverOptions],
    args: configargparse.ArgParser,
) -> None:
    """Create args."""
    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",
    )

__init__(options)

Create SmcFtpTrackObserver.

Source code in nowplaying/track/observers/smc_ftp.py 
65
66
67
def __init__(self: Self, options: Options) -> None:
    """Create SmcFtpTrackObserver."""
    self._options = options

track_finished(_)

Track finished.

Source code in nowplaying/track/observers/smc_ftp.py 
110
111
112
def track_finished(self: Self, _: Track) -> None:
    """Track finished."""
    return

track_started(track)

Track started.

Source code in nowplaying/track/observers/smc_ftp.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def track_started(self: Self, track: Track) -> None:
    """Track started."""
    logger.info(
        "Updating DAB+ DLS for track artist=%s title=%s",
        track.artist,
        track.title,
    )

    if track.get_duration() < timedelta(seconds=5):
        logger.info(
            "Track is less than 5 seconds, not sending to SMC artist=%s title=%s",
            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 > _NOWPLAYING_DAB_MAXLEN:  # pragma: no cover
        logger.warning("SMC DLS to long %s", dls.getvalue().decode("latin1"))
        dls, dlplus = _dls_from_track(track, title=False)

    ftp = FTP_TLS()  # noqa: S321
    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(
        "SMC FTP hostname=%s dls=%s dlsplus=%s",
        self._options.hostname,
        dls.getvalue().decode("latin1"),
        dlplus.getvalue().decode("latin1"),
    )