本文整理汇总了Python中opus_core.storage_factory.StorageFactory类的典型用法代码示例。如果您正苦于以下问题：Python StorageFactory类的具体用法？Python StorageFactory怎么用？Python StorageFactory使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了StorageFactory类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_my_inputs
def test_my_inputs(self):
storage = StorageFactory().get_storage("dict_storage")
persons_table_name = "persons"
storage.write_table(
table_name=persons_table_name,
table_data={
"person_id": array([1, 2, 3, 4, 5, 6]),
"household_id": array([1, 1, 2, 3, 3, 3]),
"member_id": array([1, 2, 1, 1, 2, 3]),
"work_nonhome_based": array([1, 0, 0, 1, 0, 1]),
"work_place_zone_id": array([71, -1, -1, 89, -1, -1]),
},
)
persons = PersonDataset(in_storage=storage, in_table_name=persons_table_name)
values = VariableTestToolbox().compute_variable(
self.variable_name,
data_dictionary={"household": {"household_id": array([1, 2, 3, 4])}, "person": persons},
dataset="household",
)
should_be = array([1, 0, 1, 0])
self.assert_(ma.allclose(values, should_be, rtol=1e-7), "Error in " + self.variable_name)
开发者ID:christianurich，项目名称:VIBe2UrbanSim，代码行数:27，代码来源:number_of_nonhome_based_workers_with_valid_work_place_zone_id.py
示例2: test_my_inputs
def test_my_inputs(self):
storage = StorageFactory().get_storage('dict_storage')
storage.write_table(
table_name='gridcells',
table_data={
'grid_id': array([1, 2, 3]),
'zone_id': array([1, 1, 3]),
}
)
storage.write_table(
table_name='zones',
table_data={
'zone_id': array([1, 2, 3]),
"trip_weighted_average_generalized_cost_hbw_to_work_am_drive_alone": array([4.1, 5.3, 6.2]),
}
)
dataset_pool = DatasetPool(package_order=['urbansim'],
storage=storage)
gridcell = dataset_pool.get_dataset('gridcell')
gridcell.compute_variables(self.variable_name,
dataset_pool=dataset_pool)
values = gridcell.get_attribute(self.variable_name)
should_be = array([4.1, 4.1, 6.2])
self.assert_(ma.allclose(values, should_be, rtol=1e-3),
msg="Error in " + self.variable_name)
开发者ID:，项目名称:，代码行数:30，代码来源:
示例3: test_my_inputs
def test_my_inputs(self):
storage = StorageFactory().get_storage('dict_storage')
persons_table_name = 'persons'
storage.write_table(
table_name=persons_table_name,
table_data={
'person_id':array([1, 2, 3, 4, 5]),
'household_id':array([1, 1, 3, 3, 3]),
'member_id':array([1,2,1,2,3]),
'home_zone_id': array([3, 1, 1, 2, 3]),
'work_place_zone_id':array([1, 3, 3, 1, 2])
},
)
persons = PersonDataset(in_storage=storage, in_table_name=persons_table_name)
values = VariableTestToolbox().compute_variable(self.variable_name,
data_dictionary = {
'person':persons,
'travel_data':{
'from_zone_id':array([3,3,1,1,1,2,2,3,2]),
'to_zone_id': array([1,3,1,3,2,1,3,2,2]),
'am_single_vehicle_to_work_travel_time':array([1.1, 2.2, 3.3, 4.4, 0.5, 0.7, 8.7, 7.8, 1.0])
}
},
dataset = 'person'
)
should_be = array([1.1, 4.4, 4.4, 0.7, 7.8])
self.assert_(ma.allclose(values, should_be, rtol=1e-2),
'Error in ' + self.variable_name)
开发者ID:christianurich，项目名称:VIBe2UrbanSim，代码行数:34，代码来源:travel_time_hbw_am_drive_alone_from_home_to_work.py
示例4: test
def test(self):
storage = StorageFactory().get_storage('dict_storage')
parcels_table_name = 'parcels'
storage.write_table(
table_name=parcels_table_name,
table_data={
'parcel_id':array([1,2,3,4])
},
)
parcels = ParcelDataset(in_storage=storage, in_table_name=parcels_table_name)
sid = number_of_surveyed_households.surveyed_households_starting_id
values = VariableTestToolbox().compute_variable(self.variable_name, \
data_dictionary = {
'parcel':parcels,
'household':{
'parcel_id':array([1, 2, 3, 4, 2, 2]),
'household_id':array([sid, sid+11, sid-1, sid-5, sid+7, sid-6]),
}
},
dataset = 'parcel'
)
should_be = array([1,1,0,0])
self.assert_(ma.allclose(values, should_be, rtol=1e-20),
'Error in ' + self.variable_name)
开发者ID:christianurich，项目名称:VIBe2UrbanSim，代码行数:30，代码来源:has_eg_DDD_surveyed_households.py
示例5: test_my_input
def test_my_input(self):
storage = StorageFactory().get_storage('dict_storage')
storage.write_table(
table_name='zones',
table_data={
'zone_id': array([1, 3]),
}
)
storage.write_table(
table_name='travel_data',
table_data={
"from_zone_id":array([1,3,3,1]),
"to_zone_id": array([1,1,3,3]),
"am_pk_period_drive_alone_vehicle_trips":array([1, 7, 3, 4]),
}
)
dataset_pool = DatasetPool(package_order=['urbansim'],
storage=storage)
zone = dataset_pool.get_dataset('zone')
zone.compute_variables(self.variable_name,
dataset_pool=dataset_pool)
values = zone.get_attribute(self.variable_name)
should_be = array([8, 7])
self.assert_(ma.allequal(values, should_be),
msg="Error in " + self.variable_name)
开发者ID:，项目名称:，代码行数:30，代码来源:
示例6: test_with_all_zero_denominator
def test_with_all_zero_denominator(self):
storage = StorageFactory().get_storage('dict_storage')
storage.write_table(
table_name='zones',
table_data={
"zone_id": array([1,2,3,4]),
}
)
storage.write_table(
table_name='travel_data',
table_data={
"from_zone_id":array([1,2,2,3,4]),
"to_zone_id":array([1,2,1,2,2]),
"am_single_vehicle_to_work_travel_time":array([1.1, 2.2, 3.3, 4.4, 5.5]),
"am_pk_period_drive_alone_vehicle_trips":array([0, 0.0, 0.0, 0.0, 0.0]),
}
)
dataset_pool = DatasetPool(package_order=['urbansim'],
storage=storage)
zone = dataset_pool.get_dataset('zone')
zone.compute_variables(self.variable_name,
dataset_pool=dataset_pool)
values = zone.get_attribute(self.variable_name)
should_be = array([0.0,
0.0,
0.0,
0.0])
self.assert_(ma.allclose(values, should_be, rtol=1e-7),
msg="Error in " + self.variable_name)
开发者ID:christianurich，项目名称:VIBe2UrbanSim，代码行数:34，代码来源:trip_weighted_average_time_hbw_from_home_am_drive_alone.py
示例7: test_my_inputs
def test_my_inputs(self):
storage = StorageFactory().get_storage('dict_storage')
building_types_table_name = 'building_types'
storage.write_table(
table_name=building_types_table_name,
table_data={
'building_type_id':array([1, 2, 3, 4]),
'generic_building_type_id': array([2,3,1,1])
}
)
buildings_table_name = 'buildings'
storage.write_table(
table_name=buildings_table_name,
table_data={
'building_id': array([1, 2, 3, 4, 5, 6]),
'building_type_id': array([2, 1, 2, 4, 3, 3])
}
)
dataset_pool = DatasetPool(package_order=['urbansim_parcel', 'urbansim'], storage=storage)
buildings = dataset_pool.get_dataset('building')
values = buildings.compute_variables(self.variable_name, dataset_pool=dataset_pool)
should_be = array([True, False, True, False, False, False])
self.assert_(ma.allequal(values, should_be),
'Error in ' + self.variable_name)
开发者ID:christianurich，项目名称:VIBe2UrbanSim，代码行数:29，代码来源:is_generic_building_type_DDD.py
示例8: test_scaling_households_model
def test_scaling_households_model(self):
storage = StorageFactory().get_storage('dict_storage')
hhs_table_name = 'households'
storage.write_table(
table_name=hhs_table_name,
table_data={
"household_id": arange(100)+1,
"building_id":array(10*[1]+60*[2]+30*[-1])
}
)
households = HouseholdDataset(in_storage=storage, in_table_name=hhs_table_name)
buildings_table_name = 'buildings'
storage.write_table(
table_name=buildings_table_name,
table_data={"building_id":arange(2)+1}
)
buildings = BuildingDataset(in_storage=storage, in_table_name=buildings_table_name)
# run model
model = ScalingAgentsModel(debuglevel=4)
model.run(buildings, households, agents_index = arange(70, 100))
# all households are placed
self.assertEqual((households['building_id']>0).all(), True)
# get results
buildings.compute_variables(["urbansim_parcel.building.number_of_households"],
resources = Resources({"household":households}))
result = buildings["number_of_households"]
self.assertEqual(result.sum(), 100)
res_incr = result - array([10, 60]) # building increments
# second building should get many more HHs than the first building (at least twice as much)
self.assertEqual(res_incr[1] > 2*res_incr[0], True)
开发者ID:psrc，项目名称:urbansim，代码行数:33，代码来源:scaling_agents_model.py
示例9: test_my_inputs
def test_my_inputs(self):
storage = StorageFactory().get_storage('dict_storage')
storage.write_table(
table_name='buildings',
table_data={
'building_id': array([1, 2, 3]),
}
)
storage.write_table(
table_name='jobs',
table_data={
'job_id': array([1, 2, 3, 4, 5, 6]),
'sector_id': array([1, 1, 3, 2, 3, 3]),
'building_id': array([1, 1, 1, 2, 3, 3])
}
)
dataset_pool = DatasetPool(package_order=['urbansim_parcel', 'urbansim'], storage=storage)
buildings = dataset_pool.get_dataset('building')
values = buildings.compute_variables(self.variable_name, dataset_pool=dataset_pool)
should_be = array([1/3., 0, 1])
self.assert_(ma.allequal(values, should_be),
'Error in ' + self.variable_name)
开发者ID:psrc，项目名称:urbansim，代码行数:26，代码来源:fraction_of_jobs_of_sector_DDD.py
示例10: test_my_inputs
def test_my_inputs(self):
storage = StorageFactory().get_storage("dict_storage")
storage.write_table(
table_name="gridcells",
table_data={
"grid_id": arange(50),
"residential_units": array(5 * [0] + 10 * [20] + 5 * [15] + 10 * [50] + 15 * [3] + 5 * [45]),
},
)
storage.write_table(table_name="households", table_data={"household_id": array([1, 2, 3])})
dataset_pool = DatasetPool(package_order=["urbansim"], storage=storage)
household_x_gridcell = dataset_pool.get_dataset(
"household_x_gridcell",
dataset_arguments={"index2": array([[13, 15, 23, 49], [5, 9, 17, 43], [17, 18, 40, 47]], dtype="int32")},
)
household_x_gridcell.compute_variables(self.variable_name)
values = household_x_gridcell.get_attribute(self.variable_name)
# The values are computed using formula from Ben-Akiva book (Chapter of correcting for sampling bias)
should_be = (
array(
[
[-11.3207881, -11.03310603, -12.23707884, -12.13171832],
[-15.01597613, -15.01597613, -14.72829406, -13.11885615],
[-14.18521949, -14.18521949, -12.57578158, -15.28383178],
]
)
+ 11.03310603
)
self.assert_(ma.allclose(values, should_be, rtol=1e-4), msg="Error in " + self.variable_name)
开发者ID:christianurich，项目名称:VIBe2UrbanSim，代码行数:33，代码来源:ln_sampling_probability_for_bias_correction_mnl_SSS.py
示例11: test_my_inputs
def test_my_inputs(self):
storage = StorageFactory().get_storage('dict_storage')
parcels_table_name = 'parcels'
storage.write_table(
table_name=parcels_table_name,
table_data={
'parcel_id':array([1,2,3,4,5]),
'grid_id':array([1, 1, 3, 2, 3]),
},
)
parcels = ParcelDataset(in_storage=storage, in_table_name=parcels_table_name)
values = VariableTestToolbox().compute_variable(self.variable_name, \
data_dictionary = {
'gridcell':{ \
'grid_id':array([1, 2, 3]),
'residential_units_within_walking_distance':array([100, 1000, 1500])
},
'parcel':parcels
},
dataset = 'parcel'
)
should_be = array([100.0, 100.0, 1500.0, 1000.0, 1500.0])
self.assert_(ma.allclose(values, should_be, rtol=1e-7),
'Error in ' + self.variable_name)
开发者ID:christianurich，项目名称:VIBe2UrbanSim，代码行数:30，代码来源:SSS_within_walking_distance.py
示例12: test_safely_divide_two_attributes
def test_safely_divide_two_attributes(self):
from opus_core.datasets.dataset_pool import DatasetPool
storage = StorageFactory().get_storage('dict_storage')
storage.write_table(
table_name='tests',
table_data={
'id': array([1,2,3,4]),
'numerator': array([1,2,3,0]),
'denominator': array([2.,0.,2.,0.]),
}
)
dataset_pool = DatasetPool(package_order=['opus_core'],
storage=storage)
test = dataset_pool.get_dataset('test')
variable = Variable()
variable.set_dataset(test)
result = variable.safely_divide_two_attributes('opus_core.test.numerator',
'opus_core.test.denominator')
self.assert_(ma.allclose(array([.5, 0, 1.5, 0]), result))
result = variable.safely_divide_two_attributes('opus_core.test.numerator',
'opus_core.test.denominator',
value_for_divide_by_zero=-1.0)
self.assert_(ma.allclose(array([.5, -1., 1.5, -1.]), result))
开发者ID:christianurich，项目名称:VIBe2UrbanSim，代码行数:27，代码来源:variable.py
示例13: test_scaling_households_model_with_weights
def test_scaling_households_model_with_weights(self):
storage = StorageFactory().get_storage('dict_storage')
hhs_table_name = 'households'
storage.write_table(
table_name=hhs_table_name,
table_data={
"household_id": arange(100)+1,
"building_id":array(10*[1]+50*[2]+10*[3]+30*[-1])
}
)
households = HouseholdDataset(in_storage=storage, in_table_name=hhs_table_name)
buildings_table_name = 'buildings'
storage.write_table(
table_name=buildings_table_name,
table_data={"building_id":arange(3)+1}
)
buildings = BuildingDataset(in_storage=storage, in_table_name=buildings_table_name)
# run model: Give the first building ten times as much weight as building 2. No weight for building 3.
model = ScalingAgentsModel(weights="10*(building.building_id == 1) + (building.building_id == 2)",
debuglevel=4)
model.run(buildings, households, agents_index = arange(70, 100))
# all households are placed
self.assertEqual((households['building_id']>0).all(), True)
# get results
buildings.compute_variables(["urbansim_parcel.building.number_of_households"],
resources = Resources({"household":households}))
result = buildings["number_of_households"]
self.assertEqual(result.sum(), 100)
res_incr = result - array([10, 50, 10]) # building increments
self.assertEqual(res_incr[2], 0) # third building should have no increment
# first building should get more HHs than the second building
self.assertEqual(res_incr[1] < res_incr[0], True)
开发者ID:psrc，项目名称:urbansim，代码行数:35，代码来源:scaling_agents_model.py
示例14: test_my_inputs
def test_my_inputs(self):
storage = StorageFactory().get_storage('dict_storage')
storage.write_table(
table_name='zones',
table_data={
'zone_id': array([1, 2]),
}
)
storage.write_table(
table_name='travel_data',
table_data={
"from_zone_id": array([1,1,2,2]),
'to_zone_id': array([1,2,1,2]),
"am_single_vehicle_to_work_travel_time":array([1.1, 2.2, 3.3, 4.4]),
"am_pk_period_drive_alone_vehicle_trips":array([1.0, 2.0, 3.0, 4.0]),
}
)
dataset_pool = DatasetPool(package_order=['urbansim'],
storage=storage)
zone = dataset_pool.get_dataset('zone')
zone.compute_variables(self.variable_name,
dataset_pool=dataset_pool)
values = zone.get_attribute(self.variable_name)
should_be = array([(1.1*1.0 +2.2*2.0)/(3.0),
(3.3*3.0 + 4.4*4.0)/(7.0)])
self.assert_(ma.allclose(values, should_be, rtol=1e-7),
msg="Error in " + self.variable_name)
开发者ID:christianurich，项目名称:VIBe2UrbanSim，代码行数:32，代码来源:trip_weighted_average_time_hbw_from_home_am_drive_alone.py
示例15: test_my_inputs
def test_my_inputs(self):
storage = StorageFactory().get_storage('dict_storage')
storage.write_table(
table_name='zones',
table_data={
"zone_id": array([1,2]),
"trip_mode_bike": array([3,1]),
"trip_mode_walk": array([5,6]),
"trip_mode_park_ride": array([3,2]),
"trip_mode_share_ride2": array([1,8]),
"trip_mode_drive_alone": array([2,9]),
"trip_mode_share_ride3": array([8,4]),
"trip_mode_transit": array([5,5]),
}
)
dataset_pool = DatasetPool(package_order=['urbansim'],
storage=storage)
zone = dataset_pool.get_dataset('zone')
zone.compute_variables(self.variable_name,
dataset_pool=dataset_pool)
values = zone.get_attribute(self.variable_name)
should_be = array([8.0/27.0, 7.0/35.0])
self.assert_(ma.allclose(values, should_be, rtol=1e-7),
msg="Error in " + self.variable_name)
开发者ID:christianurich，项目名称:VIBe2UrbanSim，代码行数:29，代码来源:mode_split_human_powered_over_all.py
示例16: get_values
def get_values(self, sector, threshold):
self.variable_name = "urbansim_parcel.zone.sector_%s_employment_within_%s_minutes_travel_time_hbw_am_transit_walk" % (sector, threshold)
storage = StorageFactory().get_storage('dict_storage')
storage.write_table(
table_name='zones',
table_data={
"zone_id":array([1,3]),
"number_of_jobs_of_sector_2":array([10, 1]),
"number_of_jobs_of_sector_3":array([7, 2]),
}
)
storage.write_table(
table_name='travel_data',
table_data={
"from_zone_id": array([3,3,1,1]),
"to_zone_id": array([1,3,1,3]),
"am_total_transit_time_walk": array([1.1, 2.2, 3.3, 4.4]),
}
)
dataset_pool = DatasetPool(package_order=['urbansim'],
storage=storage)
zone = dataset_pool.get_dataset('zone')
zone.compute_variables(self.variable_name,
dataset_pool=dataset_pool)
values = zone.get_attribute(self.variable_name)
return values
开发者ID:psrc，项目名称:urbansim，代码行数:29，代码来源:sector_DDD_employment_within_DDD_minutes_travel_time_hbw_am_transit_walk.py
示例17: test_full_tree
def test_full_tree(self):
storage = StorageFactory().get_storage('dict_storage')
storage.write_table(
table_name='parcels',
table_data={
'parcel_id':array([1,2,3,4]),
'is_in_city_seattle':array([1, 1, 0, 0])
},
)
storage.write_table(
table_name='households',
table_data={
'household_id':array([1,2,3,4,5]),
'income':array([1000, 300000, 50000, 0, 10550])
},
)
dataset_pool = DatasetPool(package_order=['psrc', 'urbansim'],
storage=storage)
household_x_parcel = dataset_pool.get_dataset('household_x_parcel')
household_x_parcel.compute_variables(self.variable_name,
dataset_pool=dataset_pool)
values = household_x_parcel.get_attribute(self.variable_name)
should_be = array([[1000*1, 1000*1, 1000*0, 1000*0],
[300000*1, 300000*1, 300000*0,300000*0 ],
[50000*1, 50000*1, 50000*0,50000*0],
[0, 0, 0, 0],
[10550*1, 10550*1, 10550*0, 10550*0]])
self.assert_(ma.allclose(values, should_be, rtol=1e-3),
msg="Error in " + self.variable_name)
开发者ID:christianurich，项目名称:VIBe2UrbanSim，代码行数:34，代码来源:income_and_is_in_city_SSS.py
示例18: test_create_indicator_multiple_years
def test_create_indicator_multiple_years(self):
indicator_path = os.path.join(self.temp_cache_path, "indicators")
self.assert_(not os.path.exists(indicator_path))
self.source_data.years = range(1980, 1984)
indicator = Indicator(dataset_name="test", attribute="opus_core.test.attribute")
maker = Maker(project_name="test", test=True)
maker.create(indicator=indicator, source_data=self.source_data)
for year in range(1980, 1984):
storage_location = os.path.join(self.source_data.get_indicator_directory(), "_stored_data", repr(year))
self.assert_(os.path.exists(os.path.join(storage_location, "test")))
store = StorageFactory().get_storage(type="flt_storage", storage_location=storage_location)
cols = store.get_column_names(table_name="test")
self.assertEqual(sorted(cols), sorted(["attribute", "id"]))
id_vals = [1, 2, 3, 4]
attribute_vals = [5, 6, 7, 8]
attribute_vals_1983 = [10, 12, 14, 16]
data = store.load_table(table_name="test", column_names=cols)
self.assertEqual(id_vals, list(data["id"]))
if year == 1983:
self.assertEqual(attribute_vals_1983, list(data["attribute"]))
else:
self.assertEqual(attribute_vals, list(data["attribute"]))
开发者ID:apdjustino，项目名称:DRCOG_Urbansim，代码行数:29，代码来源:maker.py
示例19: test_my_inputs
def test_my_inputs(self):
storage = StorageFactory().get_storage('dict_storage')
storage.write_table(
table_name='buildings',
table_data={
'building_id': arange(50),
'residential_units': array(5*[0]+10*[20]+5*[15]+10*[50]+15*[3]+5*[45]),
}
)
storage.write_table(
table_name='jobs',
table_data={
'job_id': array([1, 2, 3]),
}
)
dataset_pool = DatasetPool(package_order=['urbansim_parcel', 'urbansim'],
storage=storage)
job_x_building = dataset_pool.get_dataset('job_x_building',
dataset_arguments={"index2": array([[13, 15, 23, 49],
[5, 9, 17, 43],
[17, 18, 40, 47]], dtype="int32")})
job_x_building.compute_variables(self.variable_name)
values = job_x_building.get_attribute(self.variable_name)
# The values are computed using formula from Ben-Akiva book (Chapter of correcting for sampling bias)
should_be = array([[-11.3207881 , -11.03310603, -12.23707884, -12.13171832],
[-15.01597613, -15.01597613, -14.72829406, -13.11885615],
[-14.18521949, -14.18521949, -12.57578158, -15.28383178]]) + 11.03310603
self.assert_(ma.allclose(values, should_be, rtol=1e-4),
msg="Error in " + self.variable_name)
开发者ID:psrc，项目名称:urbansim，代码行数:33，代码来源:ln_sampling_probability_for_bias_correction_mnl_SSS.py
示例20: test_my_inputs
def test_my_inputs(self):
storage = StorageFactory().get_storage('dict_storage')
parcels_table_name = 'parcels'
storage.write_table(
table_name=parcels_table_name,
table_data={
'parcel_id':array([1,2,3,4,5]),
'residential_units':array([1, 1, 3, 2, 3]),
'number_of_households':array([2, 1, 3, 0, 5]),
'county':array(['033','061','035','033','033'])
},
)
parcels = ParcelDataset(in_storage=storage, in_table_name=parcels_table_name)
values = VariableTestToolbox().compute_variable(self.variable_name, \
data_dictionary = {
'parcel':parcels
},
dataset = 'parcel'
)
should_be = array([1, 0, 0, 0, 3]) #would be a 2-D array if it spanned more than one 'directory'
self.assert_(ma.allclose(values, should_be, rtol=1e-05),
'Error in ' + self.variable_name)
开发者ID:christianurich，项目名称:VIBe2UrbanSim，代码行数:28，代码来源:residential_units_when_has_eg_DDD_households_and_is_in_county_DDD.py
注：本文中的opus_core.storage_factory.StorageFactory类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论