masq/i3pystatus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add module to display sports scores

Browse Source 
This is a pluggable module which supports multiple backends. The first
backend displays MLB scores, with more planned (NHL to start).
This commit is contained in:
Erik Johnson 2016-04-04 00:06:09 -05:00
parent 8aeaab7fa5
commit afdcf32388
3 changed files with 988 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
docs/i3pystatus.rst
View File

@ -44,6 +44,15 @@ example configuration for the MaildirMail backend:
.. nothin'
.. _scorebackends:
Score Backends
--------------
.. autogen:: i3pystatus.scores SettingsBase
.. nothin'
.. _updatebackends:
Update Backends

661
i3pystatus/scores/__init__.py Normal file
View File

@ -0,0 +1,661 @@
import copy
import json
import operator
import pytz
import re
import threading
import time
from datetime import datetime, timedelta
from urllib.request import urlopen
from urllib.error import HTTPError, URLError
from i3pystatus import SettingsBase, Module, formatp
from i3pystatus.core.util import user_open, internet, require
class ScoresBackend(SettingsBase):
settings = ()
favorite_teams = []
all_games = True
date = None
games = {}
scroll_order = []
last_update = 0
def init(self):
# Merge the passed team colors with the global ones. A simple length
# check is sufficient here because i3pystatus.scores.Scores instance
# will already have checked to see if any invalid teams were specified
# in team_colors.
if len(self.team_colors) != len(self._default_colors):
self.logger.debug(
'Overriding %s team colors with: %s',
self.__class__.__name__,
self.team_colors
)
new_colors = copy.copy(self._default_colors)
new_colors.update(self.team_colors)
self.team_colors = new_colors
self.logger.debug('%s team colors: %s',
self.__class__.__name__, self.team_colors)
def api_request(self, url):
self.logger.debug('Making %s API request to %s',
self.__class__.__name__, url)
try:
with urlopen(url) as content:
try:
content_type = dict(content.getheaders())['Content-Type']
charset = re.search(r'charset=(.*)', content_type).group(1)
except AttributeError:
charset = 'utf-8'
response = json.loads(content.read().decode(charset))
self.logger.log(5, 'API response: %s', response)
return response
except HTTPError as exc:
self.logger.critical(
'Error %s (%s) making request to %s',
exc.code, exc.reason, exc.url,
)
return {}
except URLError as exc:
self.logger.critical('Error making request to %s: %s', url, exc)
return {}
def get_api_date(self):
'''
Figure out the date to use for API requests. Assumes yesterday's date
if between midnight and 10am Eastern time. Override this function in a
subclass to change how the API date is calculated.
'''
# NOTE: If you are writing your own function to get the date, make sure
# to include the first if block below to allow for the ``date``
# parameter to hard-code a date.
api_date = None
if self.date is not None and not isinstance(self.date, datetime):
try:
api_date = datetime.strptime(self.date, '%Y-%m-%d')
except (TypeError, ValueError):
self.logger.warning('Invalid date \'%s\'', self.date)
if api_date is None:
utc_time = pytz.utc.localize(datetime.utcnow())
eastern = pytz.timezone('US/Eastern')
api_date = eastern.normalize(utc_time.astimezone(eastern))
if api_date.hour < 10:
# The scores on NHL.com change at 10am Eastern, if it's before
# that time of day then we will use yesterday's date.
api_date -= timedelta(days=1)
self.date = api_date
@staticmethod
def add_ordinal(number):
try:
number = int(number)
except ValueError:
return number
if 4 <= number <= 20:
return '%d%s' % (number, 'th')
else:
ord_map = {1: 'st', 2: 'nd', 3: 'rd'}
return '%d%s' % (number, ord_map.get(number % 10, 'th'))
@staticmethod
def force_int(value):
try:
return int(value)
except (TypeError, ValueError):
return 0
def get_nested(self, data, expr, callback=None, default=None):
if callback is None:
def callback(x):
return x
try:
for key in expr.split(':'):
if key.isdigit() and isinstance(data, list):
key = int(key)
data = data[key]
except (KeyError, IndexError, TypeError):
self.logger.debug('No %s data found at %s, falling back to %s',
self.__class__.__name__, expr, repr(default))
return default
return callback(data)
def interpret_api_return(self, data, team_game_map):
favorite_games = []
# Cycle through the followed teams to ensure that games show up in the
# order of teams being followed.
for team in self.favorite_teams:
for id_ in team_game_map.get(team, []):
if id_ not in favorite_games:
favorite_games.append(id_)
# If all games are being tracked, add any games not from
# explicitly-followed teams.
if self.all_games:
additional_games = [x for x in data if x not in favorite_games]
else:
additional_games = []
# Process the API return data for each tracked game
self.games = {}
for game_id in favorite_games + additional_games:
self.games[game_id] = self.process_game(data[game_id])
# Favorite games come first
self.scroll_order = [self.games[x]['id'] for x in favorite_games]
# For any remaining games being tracked, sort each group by start time
# and add them to the list
for status in self.display_order:
time_map = {
x: self.games[x]['start_time'] for x in self.games
if x not in favorite_games and self.games[x]['status'] == status
}
sorted_games = sorted(time_map.items(), key=operator.itemgetter(1))
self.scroll_order.extend([x[0] for x in sorted_games])
# Reverse map so that we can know the scroll position for a given game
# by just its ID. This will help us to place the game in its new order
# when that order changes due to the game changing from one status to
# another.
self.scroll_order_revmap = {y: x for x, y in enumerate(self.scroll_order)}
class Scores(Module):
'''
This is a generic score checker, which must use at least one configured
:ref:`score backend <scorebackends>`.
Followed games can be scrolled through with the mouse/trackpad.
Left-clicking on the module will refresh the scores, while right-clicking
it will cycle through the configured backends. Double-clicking the module
with the left button will launch the league-specific (MLB Gameday / NHL
GameCenter / etc.) URL for the game. If there is not an active game,
double-clicking will launch the league-specific scoreboard URL containing
all games for the current day.
Double-clicking with the right button will reset the current backend to the
first game in the scroll list. This is useful for quickly switching back to
a followed team's game after looking at other game scores.
Scores for the previous day's games will be shown until 10am Eastern Time
(US), after which time the current day's games will be shown.
.. rubric:: Available formatters
Formatters are set in the backend instances, see the :ref:`scorebackends`
for more information.
This module supports the :ref:`formatp <formatp>` extended string format
syntax. This allows for values to be hidden when they evaluate as False
(e.g. when a formatter is blank (an empty string). The default values for
the format strings set in the :ref:`score backends <scorebackends>`
(``format_pregame``, ``format_in_progress``, etc.) make heavy use of
formatp, hiding many formatters when they are blank.
.. rubric:: Usage example
.. code-block:: python
from i3pystatus import Status
from i3pystatus.scores import mlb, nhl
status = Status()
status.register(
'scores',
hints={'markup': 'pango'},
colorize_teams=True,
favorite_icon='<span size="small" color="#F5FF00">★</span>',
backends=[
mlb.MLB(
teams=['CWS', 'SF'],
format_no_games='No games today :(',
inning_top='⬆',
inning_bottom='⬇',
),
nhl.NHL(teams=['CHI']),
nba.NBA(
teams=['GSW'],
all_games=False,
),
epl.EPL(),
],
)
status.run()
To enable colorized team name/city/abbbreviation, ``colorize_teams`` must
be set to ``True``. This also requires that i3bar is configured to use
Pango, and that the :ref:`hints <hints>` param is set for the module and
includes a ``markup`` key, as in the example above. To ensure that i3bar is
configured to use Pango, the `font param`__ in your i3 config file must
start with ``pango:``.
.. __: http://i3wm.org/docs/userguide.html#fonts
.. _scores-game-order:
If a ``teams`` param is not specified for the backend, then all games for
the current day will be tracked, and will be ordered by the start time of
the game. Otherwise, only games from explicitly-followed teams will be
tracked, and will be in the same order as listed. If ``ALL`` is part of the
list, then games from followed teams will be first in the scroll list,
followed by all remaining games in order of start time.
Therefore, in the above example, only White Sox and Giants games would be
tracked, while in the below example all games would be tracked, with
White Sox and Giants games appearing first in the scroll list and the
remaining games appearing after them, in order of start time.
.. code-block:: python
from i3pystatus import Status
from i3pystatus.scores import mlb
status = Status()
status.register(
'scores',
hints={'markup': 'pango'},
colorize_teams=True,
favorite_icon='<span size="small" color="#F5FF00">★</span>',
backends=[
mlb.MLB(
teams=['CWS', 'SF', 'ALL'],
team_colors={
'NYM': '#1D78CA',
},
),
],
)
status.run()
.. rubric:: Troubleshooting
If the module gets stuck during an update (i.e. the ``refresh_icon`` does
not go away), then the update thread probably encountered a traceback. This
traceback will (by default) be logged to ``~/.i3pystatus-<pid>`` where
``<pid>`` is the PID of the thread. However, it may be more convenient to
manually set the logfile to make the location of the log data reliable and
avoid clutter in your home directory. For example:
.. code-block:: python
import logging
from i3pystatus import Status
from i3pystatus.scores import mlb, nhl
status = Status(
logfile='/home/username/var/i3pystatus.log',
)
status.register(
'scores',
log_level=logging.DEBUG,
backends=[
mlb.MLB(
teams=['CWS', 'SF'],
log_level=logging.DEBUG,
),
nhl.NHL(
teams=['CHI'],
log_level=logging.DEBUG,
),
nba.NBA(
teams=['CHI'],
log_level=logging.DEBUG,
),
],
)
status.run()
.. note::
The ``log_level`` must be set separately in both the module and the
backend instances (as shown above), otherwise the backends will
still use the default log level.
'''
interval = 300
settings = (
('backends', 'List of backend instances'),
('favorite_icon', 'Value for the ``{away_favorite}`` and '
'``{home_favorite}`` formatter when the displayed game '
'is being played by a followed team'),
('color', 'Color to be used for non-colorized text (defaults to the '
'i3bar color)'),
('color_no_games', 'Color to use when no games are scheduled for the '
'currently-displayed backend (defaults to the '
'i3bar color)'),
('colorize_teams', 'Dislay team city, name, and abbreviation in the '
'team\'s color (as defined in the '
':ref:`backend <scorebackends>`\'s ``team_colors`` '
'attribute)'),
('scroll_arrow', 'Value used for the ``{scroll}`` formatter to '
'indicate that more than one game is being tracked '
'for the currently-displayed backend'),
('refresh_icon', 'Text to display (in addition to any text currently '
'shown by the module) when refreshing scores. '
'**NOTE:** Depending on how quickly the update is '
'performed, the icon may not be displayed.'),
)
backends = []
favorite_icon = '★'
color = None
color_no_games = None
colorize_teams = False
scroll_arrow = '⬍'
refresh_icon = '⟳'
output = {'full_text': ''}
game_map = {}
backend_id = 0
on_upscroll = ['scroll_game', 1]
on_downscroll = ['scroll_game', -1]
on_leftclick = ['check_scores', 'click event']
on_rightclick = ['cycle_backend', 1]
on_doubleleftclick = ['launch_web']
on_doublerightclick = ['reset_backend']
def init(self):
if not isinstance(self.backends, list):
self.backends = [self.backends]
if not self.backends:
raise ValueError('At least one backend is required')
# Initialize each backend's game index
for index in range(len(self.backends)):
self.game_map[index] = None
for backend in self.backends:
if hasattr(backend, '_valid_teams'):
for index in range(len(backend.favorite_teams)):
# Force team abbreviation to uppercase
team_uc = str(backend.favorite_teams[index]).upper()
# Check to make sure the team abbreviation is valid
if team_uc not in backend._valid_teams:
raise ValueError(
'Invalid %s team \'%s\'' % (
backend.__class__.__name__,
backend.favorite_teams[index]
)
)
backend.favorite_teams[index] = team_uc
for index in range(len(backend.display_order)):
order_lc = str(backend.display_order[index]).lower()
# Check to make sure the display order item is valid
if order_lc not in backend._valid_display_order:
raise ValueError(
'Invalid %s display_order \'%s\'' % (
backend.__class__.__name__,
backend.display_order[index]
)
)
backend.display_order[index] = order_lc
self.condition = threading.Condition()
self.thread = threading.Thread(target=self.update_thread, daemon=True)
self.thread.start()
def update_thread(self):
try:
self.check_scores(force='scheduled')
while True:
with self.condition:
self.condition.wait(self.interval)
self.check_scores(force='scheduled')
except:
msg = 'Exception in {thread} at {time}, module {name}'.format(
thread=threading.current_thread().name,
time=time.strftime('%c'),
name=self.__class__.__name__,
)
self.logger.error(msg, exc_info=True)
@property
def current_backend(self):
return self.backends[self.backend_id]
@property
def current_scroll_index(self):
return self.game_map[self.backend_id]
@property
def current_game_id(self):
try:
return self.current_backend.scroll_order[self.current_scroll_index]
except (AttributeError, TypeError):
return None
@property
def current_game(self):
try:
return self.current_backend.games[self.current_game_id]
except KeyError:
return None
def scroll_game(self, step=1):
cur_index = self.current_scroll_index
if cur_index is None:
self.logger.debug(
'Cannot scroll, no tracked {backend} games for '
'{date:%Y-%m-%d}'.format(
backend=self.current_backend.__class__.__name__,
date=self.current_backend.date,
)
)
else:
new_index = (cur_index + step) % len(self.current_backend.scroll_order)
if new_index != cur_index:
cur_id = self.current_game_id
# Don't reference self.current_scroll_index here, we're setting
# a new value for the data point for which
# self.current_scroll_index serves as a shorthand.
self.game_map[self.backend_id] = new_index
self.logger.debug(
'Scrolled from %s game %d (ID: %s) to %d (ID: %s)',
self.current_backend.__class__.__name__,
cur_index,
cur_id,
new_index,
self.current_backend.scroll_order[new_index],
)
self.refresh_display()
else:
self.logger.debug(
'Cannot scroll, only one tracked {backend} game '
'(ID: {id_}) for {date:%Y-%m-%d}'.format(
backend=self.current_backend.__class__.__name__,
id_=self.current_game_id,
date=self.current_backend.date,
)
)
def cycle_backend(self, step=1):
if len(self.backends) < 2:
self.logger.debug(
'Only one backend (%s) configured, backend cannot be changed',
self.current_backend.__class__.__name__,
)
return
old = self.backend_id
# Set the new backend
self.backend_id = (self.backend_id + step) % len(self.backends)
self.logger.debug(
'Changed scores backend from %s to %s',
self.backends[old].__class__.__name__,
self.current_backend.__class__.__name__,
)
# Display the score for the new backend. This gets rid of lag between
# when the mouse is clicked and when the new backend is shown, caused
# by any network latency encountered when updating scores.
self.refresh_display()
# Update scores (if necessary) and display them
self.check_scores()
def reset_backend(self):
if self.current_backend.games:
self.game_map[self.backend_id] = 0
self.logger.debug(
'Resetting to first game in %s scroll list (ID: %s)',
self.current_backend.__class__.__name__,
self.current_game_id,
)
self.refresh_display()
else:
self.logger.debug(
'No %s games, cannot reset to first game in scroll list',
self.current_backend.__class__.__name__,
)
def launch_web(self):
game = self.current_game
if game is None:
live_url = self.current_backend.scoreboard_url
else:
live_url = game['live_url']
self.logger.debug('Launching %s in browser', live_url)
user_open(live_url)
@require(internet)
def check_scores(self, force=False):
update_needed = False
if not self.current_backend.last_update:
update_needed = True
self.logger.debug(
'Performing initial %s score check',
self.current_backend.__class__.__name__,
)
elif force:
update_needed = True
self.logger.debug(
'%s score check triggered (%s)',
self.current_backend.__class__.__name__,
force
)
else:
update_diff = time.time() - self.current_backend.last_update
msg = ('Seconds since last %s update (%f) ' %
(self.current_backend.__class__.__name__, update_diff))
if update_diff >= self.interval:
update_needed = True
msg += ('meets or exceeds update interval (%d), update '
'triggered' % self.interval)
else:
msg += ('does not exceed update interval (%d), update '
'skipped' % self.interval)
self.logger.debug(msg)
if update_needed:
self.show_refresh_icon()
cur_id = self.current_game_id
cur_games = self.current_backend.games.keys()
self.current_backend.check_scores()
if cur_games == self.current_backend.games.keys():
# Set the index to the scroll position of the current game (it
# may have changed due to this game or other games changing
# status.
if cur_id is None:
self.logger.debug(
'No tracked {backend} games for {date:%Y-%m-%d}'.format(
backend=self.current_backend.__class__.__name__,
date=self.current_backend.date,
)
)
else:
cur_pos = self.game_map[self.backend_id]
new_pos = self.current_backend.scroll_order_revmap[cur_id]
if cur_pos != new_pos:
self.game_map[self.backend_id] = new_pos
self.logger.debug(
'Scroll position for current %s game (%s) updated '
'from %d to %d',
self.current_backend.__class__.__name__,
cur_id,
cur_pos,
new_pos,
)
else:
self.logger.debug(
'Scroll position (%d) for current %s game (ID: %s) '
'unchanged',
cur_pos,
self.current_backend.__class__.__name__,
cur_id,
)
else:
# Reset the index to 0 if there are any tracked games,
# otherwise set it to None to signify no tracked games for the
# backend.
if self.current_backend.games:
self.game_map[self.backend_id] = 0
self.logger.debug(
'Tracked %s games updated, setting scroll position to '
'0 (ID: %s)',
self.current_backend.__class__.__name__,
self.current_game_id
)
else:
self.game_map[self.backend_id] = None
self.logger.debug(
'No tracked {backend} games for {date:%Y-%m-%d}'.format(
backend=self.current_backend.__class__.__name__,
date=self.current_backend.date,
)
)
self.current_backend.last_update = time.time()
self.refresh_display()
def show_refresh_icon(self):
self.output['full_text'] = \
self.refresh_icon + self.output.get('full_text', '')
def refresh_display(self):
if self.current_scroll_index is None:
output = self.current_backend.format_no_games
color = self.color_no_games
else:
game = copy.copy(self.current_game)
fstr = str(getattr(
self.current_backend,
'format_%s' % game['status']
))
for team in ('home', 'away'):
abbrev_key = '%s_abbrev' % team
# Set favorite icon, if applicable
game['%s_favorite' % team] = self.favorite_icon \
if game[abbrev_key] in self.current_backend.favorite_teams \
else ''
if self.colorize_teams:
# Wrap in Pango markup
color = self.current_backend.team_colors.get(
game.get(abbrev_key)
)
if color is not None:
for item in ('abbrev', 'city', 'name', 'name_short'):
key = '%s_%s' % (team, item)
if key in game:
val = '<span color="%s">%s</span>' % (color, game[key])
game[key] = val
game['scroll'] = self.scroll_arrow \
if len(self.current_backend.games) > 1 \
else ''
output = formatp(fstr, **game).strip()
self.output = {'full_text': output, 'color': self.color}
def run(self):
pass

318
i3pystatus/scores/mlb.py Normal file
View File

@ -0,0 +1,318 @@
from i3pystatus.core.util import internet, require
from i3pystatus.scores import ScoresBackend
import copy
import json
import pytz
import re
import time
from datetime import datetime
from urllib.request import urlopen
LIVE_URL = 'http://mlb.mlb.com/mlb/gameday/index.jsp?gid=%s'
SCOREBOARD_URL = 'http://m.mlb.com/scoreboard'
API_URL = 'http://gd2.mlb.com/components/game/mlb/year_%04d/month_%02d/day_%02d/miniscoreboard.json'
class MLB(ScoresBackend):
'''
Backend to retrieve MLB scores. For usage examples, see :py:mod:`here
<.scores>`.
.. rubric:: Available formatters
* `{home_name}` — Name of home team
* `{home_city}` — Name of home team's city
* `{home_abbrev}` — 2 or 3-letter abbreviation for home team's city
* `{home_score}` — Home team's current score
* `{home_wins}` — Home team's number of wins
* `{home_losses}` — Home team's number of losses
* `{home_favorite}` — Displays the value for the :py:mod:`.scores` module's
``favorite`` attribute, if the home team is one of the teams being
followed. Otherwise, this formatter will be blank.
* `{away_name}` — Name of away team
* `{away_city}` — Name of away team's city
* `{away_abbrev}` — 2 or 3-letter abbreviation for away team's city
* `{away_score}` — Away team's current score
* `{away_wins}` — Away team's number of wins
* `{away_losses}` — Away team's number of losses
* `{away_favorite}` — Displays the value for the :py:mod:`.scores` module's
``favorite`` attribute, if the away team is one of the teams being
followed. Otherwise, this formatter will be blank.
* `{top_bottom}` — Displays the value of either ``inning_top`` or
``inning_bottom`` based on whether the game is in the top or bottom of an
inning.
* `{inning}` — Current inning
* `{outs}` — Number of outs in current inning
* `{venue}` — Name of ballpark where game is being played
* `{start_time}` — Start time of game in system's localtime (supports
strftime formatting, e.g. `{start_time:%I:%M %p}`)
* `{delay}` — Reason for delay, if game is currently delayed. Otherwise,
this formatter will be blank.
* `{postponed}` — Reason for postponement, if game has been postponed.
Otherwise, this formatter will be blank.
* `{extra_innings}` — When a game lasts longer than 9 innings, this
formatter will show that number of innings. Otherwise, it will blank.
.. rubric:: Team abbreviations
* **ARI** — Arizona Diamondbacks
* **ATL** — Atlanta Braves
* **BAL** — Baltimore Orioles
* **BOS** — Boston Red Sox
* **CHC** — Chicago Cubs
* **CIN** — Cincinnati Reds
* **CLE** — Cleveland Indians
* **COL** — Colorado Rockies
* **CWS** — Chicago White Sox
* **DET** — Detroit Tigers
* **HOU** — Houston Astros
* **KC** — Kansas City Royals
* **LAA** — Los Angeles Angels of Anaheim
* **LAD** — Los Angeles Dodgers
* **MIA** — Miami Marlins
* **MIL** — Milwaukee Brewers
* **MIN** — Minnesota Twins
* **NYY** — New York Yankees
* **NYM** — New York Mets
* **OAK** — Oakland Athletics
* **PHI** — Philadelphia Phillies
* **PIT** — Pittsburgh Pirates
* **SD** — San Diego Padres
* **SEA** — Seattle Mariners
* **SF** — San Francisco Giants
* **STL** — St. Louis Cardinals
* **TB** — Tampa Bay Rays
* **TEX** — Texas Rangers
* **TOR** — Toronto Blue Jays
* **WSH** — Washington Nationals
'''
interval = 300
settings = (
('favorite_teams', 'List of abbreviations of favorite teams. Games '
'for these teams will appear first in the scroll '
'list. A detailed description of how games are '
'ordered can be found '
':ref:`here <scores-game-order>`.'),
('all_games', 'If set to ``True``, all games will be present in '
'the scroll list. If set to ``False``, then only '
'games from **favorite_teams** will be present in '
'the scroll list.'),
('display_order', 'When **all_games** is set to ``True``, this '
'option will dictate the order in which games from '
'teams not in **favorite_teams** are displayed'),
('format_no_games', 'Format used when no tracked games are scheduled '
'for the current day (does not support formatter '
'placeholders)'),
('format_pregame', 'Format used when the game has not yet started'),
('format_in_progress', 'Format used when the game is in progress'),
('format_final', 'Format used when the game is complete'),
('format_postponed', 'Format used when the game has been postponed'),
('inning_top', 'Value for the ``{top_bottom}`` formatter when game '
'is in the top half of an inning'),
('inning_bottom', 'Value for the ``{top_bottom}`` formatter when game '
'is in the bottom half of an inning'),
('team_colors', 'Dictionary mapping team abbreviations to hex color '
'codes. If overridden, the passed values will be '
'merged with the defaults, so it is not necessary to '
'define all teams if specifying this value.'),
('date', 'Date for which to display game scores, in **YYYY-MM-DD** '
'format. If unspecified, the current day\'s games will be '
'displayed starting at 10am Eastern time, with last '
'evening\'s scores being shown before then. This option '
'exists primarily for troubleshooting purposes.'),
('live_url', 'Alternate URL string to launch MLB Gameday. This value '
'should not need to be changed'),
('scoreboard_url', 'Link to the MLB.com scoreboard page. Like '
'**live_url**, this value should not need to be '
'changed.'),
('api_url', 'Alternate URL string from which to retrieve score data. '
'Like **live_url*** this value should not need to be '
'changed.'),
)
required = ()
_default_colors = {
'ARI': '#A71930',
'ATL': '#CE1141',
'BAL': '#DF4601',
'BOS': '#BD3039',
'CHC': '#004EC1',
'CIN': '#C6011F',
'CLE': '#E31937',
'COL': '#5E5EB6',
'CWS': '#DADADA',
'DET': '#FF6600',
'HOU': '#EB6E1F',
'KC': '#0046DD',
'LAA': '#BA0021',
'LAD': '#005A9C',
'MIA': '#F14634',
'MIL': '#0747CC',
'MIN': '#D31145',
'NYY': '#0747CC',
'NYM': '#FF5910',
'OAK': '#006659',
'PHI': '#E81828',
'PIT': '#FFCC01',
'SD': '#285F9A',
'SEA': '#2E8B90',
'SF': '#FD5A1E',
'STL': '#B53B30',
'TB': '#8FBCE6',
'TEX': '#C0111F',
'TOR': '#0046DD',
'WSH': '#C70003',
}
_valid_teams = [x for x in _default_colors]
_valid_display_order = ['in_progress', 'final', 'postponed', 'pregame']
display_order = _valid_display_order
format_no_games = 'MLB: No games'
format_pregame = '[{scroll} ]MLB: [{away_favorite} ]{away_abbrev} ({away_wins}-{away_losses}) at [{home_favorite} ]{home_abbrev} ({home_wins}-{home_losses}) {start_time:%H:%M %Z}[ ({delay} Delay)]'
format_in_progress = '[{scroll} ]MLB: [{away_favorite} ]{away_abbrev} {away_score}, [{home_favorite} ]{home_abbrev} {home_score} ({top_bottom} {inning}, {outs} Out)[ ({delay} Delay)]'
format_final = '[{scroll} ]MLB: [{away_favorite} ]{away_abbrev} {away_score} ({away_wins}-{away_losses}) at [{home_favorite} ]{home_abbrev} {home_score} ({home_wins}-{home_losses}) (Final[/{extra_innings}])'
format_postponed = '[{scroll} ]MLB: [{away_favorite} ]{away_abbrev} ({away_wins}-{away_losses}) at [{home_favorite} ]{home_abbrev} ({home_wins}-{home_losses}) (PPD: {postponed})'
inning_top = 'Top'
inning_bottom = 'Bot'
team_colors = _default_colors
live_url = LIVE_URL
scoreboard_url = SCOREBOARD_URL
api_url = API_URL
@require(internet)
def check_scores(self):
self.get_api_date()
url = self.api_url % (self.date.year, self.date.month, self.date.day)
game_list = self.get_nested(self.api_request(url),
'data:games:game',
default=[])
# Convert list of games to dictionary for easy reference later on
data = {}
team_game_map = {}
for game in game_list:
try:
id_ = game['id']
except KeyError:
continue
try:
for team in (game['home_name_abbrev'], game['away_name_abbrev']):
team = team.upper()
if team in self.favorite_teams:
team_game_map.setdefault(team, []).append(id_)
except KeyError:
continue
data[id_] = game
self.interpret_api_return(data, team_game_map)
def process_game(self, game):
ret = {}
def _update(ret_key, game_key=None, callback=None, default='?'):
ret[ret_key] = self.get_nested(game,
game_key or ret_key,
callback=callback,
default=default)
self.logger.debug('Processing %s game data: %s',
self.__class__.__name__, game)
for key in ('id', 'venue'):
_update(key)
for key in ('inning', 'outs'):
_update(key, callback=self.force_int, default=0)
ret['live_url'] = self.live_url % game['gameday_link']
for team in ('home', 'away'):
_update('%s_wins' % team, '%s_win' % team,
callback=self.force_int)
_update('%s_losses' % team, '%s_loss' % team,
callback=self.force_int)
_update('%s_score' % team, '%s_team_runs' % team,
callback=self.force_int, default=0)
_update('%s_abbrev' % team, '%s_name_abbrev' % team)
for item in ('city', 'name'):
_update('%s_%s' % (team, item), '%s_team_%s' % (team, item))
try:
ret['status'] = game.get('status').lower().replace(' ', '_')
except AttributeError:
# During warmup ret['status'] may be a dictionary. Treat these as
# pregame
ret['status'] = 'pregame'
for key in ('delay', 'postponed'):
ret[key] = ''
if ret['status'] == 'delayed_start':
ret['status'] = 'pregame'
ret['delay'] = game.get('reason', 'Unknown')
elif ret['status'] == 'postponed':
ret['postponed'] = game.get('reason', 'Unknown Reason')
elif ret['status'] == 'game_over':
ret['status'] = 'final'
elif ret['status'] not in ('in_progress', 'final'):
ret['status'] = 'pregame'
try:
inning = game.get('inning', '0')
ret['extra_innings'] = inning \
if ret['status'] == 'final' and int(inning) > 9 \
else ''
except ValueError:
ret['extra_innings'] = ''
top_bottom = game.get('top_inning')
ret['top_bottom'] = self.inning_top if top_bottom == 'Y' \
else self.inning_bottom if top_bottom == 'N' \
else ''
time_zones = {
'PT': 'US/Pacific',
'MT': 'US/Mountain',
'CT': 'US/Central',
'ET': 'US/Eastern',
}
game_tz = pytz.timezone(
time_zones.get(
game.get('time_zone', 'ET'),
'US/Eastern'
)
)
game_time_str = ' '.join((
game.get('time_date', ''),
game.get('ampm', '')
))
try:
game_time = datetime.strptime(game_time_str, '%Y/%m/%d %I:%M %p')
except ValueError as exc:
# Log when the date retrieved from the API return doesn't match the
# expected format (to help troubleshoot API changes), and set an
# actual datetime so format strings work as expected. The times
# will all be wrong, but the logging here will help us make the
# necessary changes to adapt to any API changes.
self.logger.error(
'Error encountered determining %s game time for game %s:',
self.__class__.__name__,
game['id'],
exc_info=True
)
game_time = datetime.datetime(1970, 1, 1)
ret['start_time'] = game_tz.localize(game_time).astimezone()
self.logger.debug('Returned %s formatter data: %s',
self.__class__.__name__, ret)
return ret