mirror/weewx
1
0
Fork 0
mirror of https://github.com/weewx/weewx.git synced 2026-05-19 07:15:18 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add configuration for station registration.

Browse Source 
Test suites.
This commit is contained in:
Tom Keffer
2022-12-16 15:59:37 -08:00
parent 26ca1ab1e2
commit 20213b2cae
4 changed files with 259 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
TODO.md
View File

@@ -4,8 +4,19 @@
Applications need to be converted into poetry scripts.
## Configuration related
Implement configuration for location.
Implement configuration for language.
Implement configuration for unit system.
Implement configuration for station registry.
## Obsolete
Function `prompt_for_info()`, and all other functions that manipulation `stn_info` go away.
Tests for `prompt_for_info()` go away.
Make sure distutil isn't used anywhere. E.g., distutils.copytree().
Make sure distutil isn't used anywhere. E.g., `distutils.copytree()`.

186
bin/weecfg/station_config.py
View File

@@ -16,17 +16,18 @@ import weecfg
import weeutil.config
import weeutil.weeutil
import weewx
from weeutil.weeutil import to_float
from weeutil.weeutil import to_float, to_bool
log = logging.getLogger(__name__)
def create_station(config_path, *args, **kwargs):
"""Create a brand new configuration file.
"""Create a brand-new configuration file.
Like config_station(), except it ensures that the config file does not already exists. It then
Like config_station(), except it ensures that the config file does not already exist. It then
retrieves the template config file from package resources and uses that.
"""
# Make sure there is not already a configuration file at the designated location.
if os.path.exists(config_path):
raise weewx.ViolatedPrecondition(f"Config file {config_path} already exists")
@@ -35,20 +36,100 @@ def create_station(config_path, *args, **kwargs):
with importlib.resources.open_text('wee_resources', 'weewx.conf', encoding='utf-8') as fd:
dist_config_dict = configobj.ConfigObj(fd, encoding='utf-8', file_error=True)
config_station(dist_config_dict, *args, **kwargs)
config_config(dist_config_dict, *args, **kwargs)
# Save the results. No backup.
weecfg.save(dist_config_dict, config_path)
def config_station(config_dict, driver=None,
latitude=None, longitude=None, altitude=None,
no_prompt=False):
def config_station(config_path, *args, **kwargs):
"Reconfigure an existing station"
config_dict = configobj.ConfigObj(config_path, encoding='utf-8', file_error=True)
config_config(config_dict, *args, **kwargs)
# Save the results with backup
weecfg.save_with_backup(config_dict, config_path)
def config_config(config_dict, driver=None,
latitude=None, longitude=None, altitude=None,
no_prompt=False):
"""Modify a configuration file."""
config_latlon(config_dict, latitude=latitude, longitude=longitude, no_prompt=no_prompt)
config_altitude(config_dict, altitude=altitude, no_prompt=no_prompt)
config_latlon(config_dict, latitude=latitude, longitude=longitude, no_prompt=no_prompt)
# config_registry(config_dict, register=register, no_prompt=no_prompt)
# config_units(config_dict, unit_system=units, no_prompt=noprompt)
# config_lang(config_dict, lang=lang, no_prompt=no_prompt)
config_driver(config_dict, driver=driver, no_prompt=no_prompt)
def config_altitude(config_dict, altitude=None, no_prompt=False):
"""Set a (possibly new) value and unit for altitude.
Args:
config_dict (configobj.ConfigObj): The configuration dictionary.
altitude (str): A string with value and unit, separated with a comma.
For example, "50, meter". Optional.
no_prompt(bool): Do not prompt the user for a value.
"""
if 'Station' not in config_dict:
return
# Start with assuming the existing value:
default_altitude = config_dict['Station'].get('altitude', ["0", 'foot'])
# Was a new value provided as an argument?
if altitude is not None:
# Yes. Extract and validate it.
value, unit = altitude.split(',')
# Fail hard if the value cannot be converted to a float
float(value)
# Fail hard if the unit is unknown:
unit = unit.strip().lower()
if unit not in ['foot', 'meter']:
raise ValueError(f"Unknown altitude unit {unit}")
# All is good. Use it.
final_altitude = [value, unit]
elif not no_prompt:
print("\nSpecify altitude, with units 'foot' or 'meter'. For example:")
print("35, foot")
print("12, meter")
msg = "altitude [%s]: " % weeutil.weeutil.list_as_string(default_altitude)
final_altitude = None
while final_altitude is None:
ans = input(msg).strip()
if ans:
value, unit = ans.split(',')
try:
# Test whether the first token can be converted into a
# number. If not, an exception will be raised.
float(value)
unit = unit.strip().lower()
if unit in ['foot', 'meter']:
final_altitude = [value.strip(), unit]
except (ValueError, TypeError):
pass
else:
# The user gave the null string. We're done
final_altitude = default_altitude
else:
# If we got here, there was no value in the args and we cannot prompt. Use the default.
final_altitude = default_altitude
config_dict['Station']['altitude'] = final_altitude
def config_latlon(config_dict, latitude=None, longitude=None, no_prompt=False):
"""Set a (possibly new) value for latitude and longitude"""
"""Set a (possibly new) value for latitude and longitude
Args:
config_dict (configobj.ConfigObj): The configuration dictionary.
latitude (float|None): The latitude. If specified, no prompting will happen.
longitude (float|None): The longitude. If specified no prompting will happen.
no_prompt(bool): Do not prompt the user for a value.
"""
if "Station" not in config_dict:
return
@@ -96,61 +177,51 @@ def config_latlon(config_dict, latitude=None, longitude=None, no_prompt=False):
config_dict['Station']['longitude'] = final_longitude
def config_altitude(config_dict, altitude=None, no_prompt=False):
"""Set a (possibly new) value and unit for altitude.
def config_registry(config_dict, register=None, station_url=None, no_prompt=False):
"""Configure whether to include the station in the weewx.com registry."""
Args:
config_dict (configobj.ConfigObj): The configuration dictionary.
altitude (str): A string with value and unit, separated with a comma.
For example, "50, meter". Optional.
no_prompt(bool): If altitude is not provided, and no_prompt is False, then the user will
be prompted to supply a value.
"""
if 'Station' not in config_dict:
return
# Start with assuming the existing value:
default_altitude = config_dict['Station'].get('altitude', ["0", 'foot'])
# Was a new value provided as an argument?
if altitude is not None:
# Yes. Extract and validate it.
value, unit = altitude.split(',')
# Fail hard if the value cannot be converted to a float
float(value)
# Fail hard if the unit is unknown:
unit = unit.strip().lower()
if unit not in ['foot', 'meter']:
raise ValueError(f"Unknown altitude unit {unit}")
# All is good. Use it.
final_altitude = [value, unit]
try:
default_register = to_bool(
config_dict['StdRESTful']['StationRegistry']['register_this_station'])
except KeyError:
default_register = False
default_station_url = config_dict['Station'].get('station_url')
if register is not None:
final_register = to_bool(register)
final_station_url = station_url or default_station_url
elif not no_prompt:
print("\nSpecify altitude, with units 'foot' or 'meter'. For example:")
print("35, foot")
print("12, meter")
msg = "altitude [%s]: " % weeutil.weeutil.list_as_string(default_altitude)
final_altitude = None
while final_altitude is None:
ans = input(msg).strip()
if ans:
value, unit = ans.split(',')
try:
# Test whether the first token can be converted into a
# number. If not, an exception will be raised.
float(value)
unit = unit.strip().lower()
if unit in ['foot', 'meter']:
final_altitude = [value.strip(), unit]
except (ValueError, TypeError):
pass
else:
# The user gave the null string. We're done
final_altitude = default_altitude
print("\nYou can register your station on weewx.com, where it will be included")
print("in a map. You will need a unique URL to identify your station (such as a")
print("website, or WeatherUnderground link).")
ans = weecfg.prompt_with_options("Include station in the station registry (y/n)?",
default_register,
['y', 'n'])
final_register = to_bool(ans)
if final_register:
while True:
url = weecfg.prompt_with_options("Unique URL:", default_station_url)
if url:
if url.startswith('http://www.example.com'):
print("Unique please!")
else:
final_station_url = url
break
else:
# If we got here, there was no value in the args and we cannot prompt. Use the default.
final_altitude = default_altitude
final_register = default_register
final_station_url = default_station_url
config_dict['Station']['altitude'] = final_altitude
if final_register and not final_station_url:
raise weewx.ViolatedPrecondition("Registering the station requires "
"option 'station_url'.")
config_dict['StdRESTful']['StationRegistry']['register_this_station'] = final_register
if final_station_url:
config_dict['Station']['station_url'] = final_station_url
def config_driver(config_dict, driver=None, no_prompt=False):
@@ -231,4 +302,3 @@ def config_driver(config_dict, driver=None, no_prompt=False):
if driver_editor:
# One final chance for the driver to modify other parts of the configuration
driver_editor.modify_config(config_dict)

123
bin/weecfg/tests/test_station_config.py
View File

@@ -17,6 +17,7 @@ import weecfg.station_config
import weecfg.update_config
import weeutil.config
import weeutil.weeutil
import weewx
CONFIG_DICT_STR = """
# WEEWX TEST CONFIGURATION FILE
@@ -61,10 +62,16 @@ version = 4.10.0a1
# If you have a website, you may specify an URL. This is required if you
# intend to register your station.
#station_url = http://www.example.com
[StdRESTful]
[[StationRegistry]]
register_this_station = false
"""
CONFIG_DICT = configobj.ConfigObj(io.StringIO(CONFIG_DICT_STR))
STATION_URL = 'http://weewx.com'
def suppress_stdout(func):
def wrapper(*args, **kwargs):
@@ -75,41 +82,6 @@ def suppress_stdout(func):
return wrapper
class LatLonConfigTest(unittest.TestCase):
def setUp(self):
self.config_dict = weeutil.config.deep_copy(CONFIG_DICT)
def test_default_config_latlon(self):
# Use the default as supplied by CONFIG_DICT
weecfg.station_config.config_latlon(self.config_dict, no_prompt=True)
self.assertEqual(float(self.config_dict['Station']['latitude']), 5.0)
self.assertEqual(float(self.config_dict['Station']['longitude']), 10.0)
# Delete the values in the configuration dictionary
del self.config_dict['Station']['latitude']
del self.config_dict['Station']['longitude']
# Now the defaults should be the hardwired defaults
weecfg.station_config.config_latlon(self.config_dict, no_prompt=True)
self.assertEqual(float(self.config_dict['Station']['latitude']), 0.0)
self.assertEqual(float(self.config_dict['Station']['longitude']), 0.0)
def test_arg_config_latlon(self):
weecfg.station_config.config_latlon(self.config_dict, latitude=-20, longitude=-40)
self.assertEqual(float(self.config_dict['Station']['latitude']), -20.0)
self.assertEqual(float(self.config_dict['Station']['longitude']), -40.0)
def test_badarg_config_latlon(self):
with self.assertRaises(ValueError):
weecfg.station_config.config_latlon(self.config_dict, latitude="-20f", longitude=-40)
@suppress_stdout
def test_prompt_config_latlong(self):
with patch('weecfg.input', side_effect=['-21', '-41']):
weecfg.station_config.config_latlon(self.config_dict)
self.assertEqual(float(self.config_dict['Station']['latitude']), -21.0)
self.assertEqual(float(self.config_dict['Station']['longitude']), -41.0)
class AltitudeConfigTest(unittest.TestCase):
def setUp(self):
@@ -157,6 +129,87 @@ class AltitudeConfigTest(unittest.TestCase):
self.assertEqual(self.config_dict['Station']['altitude'], ["110", "meter"])
class LatLonConfigTest(unittest.TestCase):
def setUp(self):
self.config_dict = weeutil.config.deep_copy(CONFIG_DICT)
def test_default_config_latlon(self):
# Use the default as supplied by CONFIG_DICT
weecfg.station_config.config_latlon(self.config_dict, no_prompt=True)
self.assertEqual(float(self.config_dict['Station']['latitude']), 5.0)
self.assertEqual(float(self.config_dict['Station']['longitude']), 10.0)
# Delete the values in the configuration dictionary
del self.config_dict['Station']['latitude']
del self.config_dict['Station']['longitude']
# Now the defaults should be the hardwired defaults
weecfg.station_config.config_latlon(self.config_dict, no_prompt=True)
self.assertEqual(float(self.config_dict['Station']['latitude']), 0.0)
self.assertEqual(float(self.config_dict['Station']['longitude']), 0.0)
def test_arg_config_latlon(self):
weecfg.station_config.config_latlon(self.config_dict, latitude=-20, longitude=-40)
self.assertEqual(float(self.config_dict['Station']['latitude']), -20.0)
self.assertEqual(float(self.config_dict['Station']['longitude']), -40.0)
def test_badarg_config_latlon(self):
with self.assertRaises(ValueError):
weecfg.station_config.config_latlon(self.config_dict, latitude="-20f", longitude=-40)
@suppress_stdout
def test_prompt_config_latlong(self):
with patch('weecfg.input', side_effect=['-21', '-41']):
weecfg.station_config.config_latlon(self.config_dict)
self.assertEqual(float(self.config_dict['Station']['latitude']), -21.0)
self.assertEqual(float(self.config_dict['Station']['longitude']), -41.0)
class RegistryConfigTest(unittest.TestCase):
def setUp(self):
self.config_dict = weeutil.config.deep_copy(CONFIG_DICT)
def test_default_register(self):
weecfg.station_config.config_registry(self.config_dict, no_prompt=True)
self.assertFalse(
self.config_dict['StdRESTful']['StationRegistry']['register_this_station'])
def test_args_register(self):
# Missing station_url:
with self.assertRaises(weewx.ViolatedPrecondition):
weecfg.station_config.config_registry(self.config_dict, register='True',
no_prompt=True)
# This time we supply a station_url. Should be OK.
weecfg.station_config.config_registry(self.config_dict, register='True',
station_url=STATION_URL, no_prompt=True)
self.assertTrue(self.config_dict['StdRESTful']['StationRegistry']['register_this_station'])
self.assertEqual(self.config_dict['Station']['station_url'], STATION_URL)
# Alternatively, the config file already had a station_url:
self.config_dict['Station']['station_url'] = STATION_URL
weecfg.station_config.config_registry(self.config_dict, register='True', no_prompt=True)
self.assertTrue(self.config_dict['StdRESTful']['StationRegistry']['register_this_station'])
self.assertEqual(self.config_dict['Station']['station_url'], STATION_URL)
@suppress_stdout
def test_prompt_register(self):
with patch('weecfg.input', side_effect=['y', STATION_URL]):
weecfg.station_config.config_registry(self.config_dict)
self.assertTrue(self.config_dict['StdRESTful']['StationRegistry']['register_this_station'])
self.assertEqual(self.config_dict['Station']['station_url'], STATION_URL)
# Try again, but without specifying an URL. Should ask twice.
with patch('weecfg.input', side_effect=['y', '', STATION_URL]):
weecfg.station_config.config_registry(self.config_dict)
self.assertTrue(self.config_dict['StdRESTful']['StationRegistry']['register_this_station'])
self.assertEqual(self.config_dict['Station']['station_url'], STATION_URL)
# Now with a bogus URL
with patch('weecfg.input', side_effect=['y', 'http://www.example.com', STATION_URL]):
weecfg.station_config.config_registry(self.config_dict)
self.assertTrue(self.config_dict['StdRESTful']['StationRegistry']['register_this_station'])
self.assertEqual(self.config_dict['Station']['station_url'], STATION_URL)
class DriverConfigTest(unittest.TestCase):
def setUp(self):

48
bin/weectl/station.py
View File

@@ -4,15 +4,19 @@
# See the file LICENSE.txt for your rights.
#
"""Entry point for the "station" subcommand."""
import sys
import weewx
from . import common_parser
import weecfg.station_config
station_create_usage = """weectl station create [--config=CONFIG-PATH]
[--html-root=HTML_ROOT] [--skin-root=SKIN_ROOT]
[--driver=DRIVER]
[--altitude=ALTITUDE,{foot|meter}]
[--latitude=LATITUDE] [--longitude=LONGITUDE]
[--altitude=ALTITUDE,{foot|meter}]"""
[--register={y,n} [--station-url=STATION_URL]]
[--html-root=HTML_ROOT] [--skin-root=SKIN_ROOT]
"""
station_reconfigure_usage = "weectl station reconfigure [--config=CONFIG-PATH] [--driver=DRIVER]"
station_upgrade_usage = "weectl station upgrade [--config=CONFIG-PATH]"
station_upgrade_skins_usage = "weectl station upgrade-skins [--config=CONFIG-PATH]"
@@ -37,20 +41,27 @@ def add_subparser(subparsers,
parents=[common_parser],
usage=station_create_usage,
help='Create a station config file')
create_station_parser.add_argument('--driver',
help="Driver to use. E.g., --driver=weewx.drivers.fousb")
create_station_parser.add_argument('--altitude', metavar="ALTITUDE,(foot|meter)",
help="The station altitude in either feet or meters."
" For example, '750,foot' or '320,meter'")
create_station_parser.add_argument('--latitude',
help="The station latitude in decimal degrees.")
create_station_parser.add_argument('--longitude',
help="The station longitude in decimal degrees.")
create_station_parser.add_argument('--register', choices=['y', 'n'],
help="Register this station in the weewx registry?")
create_station_parser.add_argument('--station-url',
help="Unique URL to be used if registering the station.")
create_station_parser.add_argument('--html_root',
default='public_html',
help='Set HTML_ROOT, relative to WEEWX_ROOT. '
'Default is "public_html".')
create_station_parser.add_argument('--skin_root', default='skins')
create_station_parser.add_argument('--driver', default='weewx.drivers.simulator')
create_station_parser.add_argument('--latitude',
help="The station latitude in decimal degrees.")
create_station_parser.add_argument('--longitude',
help="The station longitude in decimal degrees.")
create_station_parser.add_argument('--altitude', metavar="ALTITUDE,(foot|meter)",
help="The station altitude in either feet or meters."
" For example, '750,foot' or '320,meter'")
create_station_parser.set_defaults(func=weecfg.station_config.create_station)
create_station_parser.add_argument('--no-prompt', type=bool,
help="If true, suppress prompts")
create_station_parser.set_defaults(func=create_station)
# Action 'reconfigure'
reconfigure_station_parser = action_parser.add_parser('reconfigure',
@@ -73,9 +84,12 @@ def add_subparser(subparsers,
help='Upgrade the skins')
def create_station(namespace):
weecfg.station_config.create_station(config_path=namespace.config,
driver=namespace.driver,
latitude=namespace.latitude,
longitude=namespace.longitude,
altitude=namespace.altitude,
no_prompt=namespace.no_prompt)
try:
weecfg.station_config.create_station(config_path=namespace.config,
driver=namespace.driver,
latitude=namespace.latitude,
longitude=namespace.longitude,
altitude=namespace.altitude,
no_prompt=namespace.no_prompt)
except weewx.ViolatedPrecondition as e:
sys.exit(e)