如何在一个模拟代码中从一个CSV中全天重复查询

sdnqo3pr  于 2023-05-20  发布在  其他
关注(0)|答案(1)|浏览(114)

我正在尝试用python写一个模拟代码。此模拟代码依赖于大型csv文件的输入,并且模拟中的每一天都有一个单独的csv文件。我需要在每个模拟日进行大量的查询(查询是基于时间的,是csv文件中的列)。
我正在考虑使用pandas.read_csv将其作为 Dataframe 读入,并存储结果,然后从该 Dataframe 进行查询。一个编码要求是我不希望 Dataframe 存储在查询站点。
我认为最简单的方法是使用一个类,例如,

  1. import pandas as pd
  2. class DailyCSVLoader:
  3. def __init__(filepath):
  4. self.df = pd.read_csv(filepath)
  5. def query(time):
  6. # return the rows corresponding to time

使用方法:

  1. import datetime
  2. path = "/path/to/csv/file/filename.csv"
  3. time = datetime.datetime(year=2020, month=1, day=1, hour=12, minute=2, second=0)
  4. loader = DailyCSVLoader(path)
  5. loader.query()

然而,对于我的特定代码库,在类之外执行此操作可能会稍微容易一些,并且只需一个函数和一个保存 Dataframe 的静态变量,例如,

  1. import pandas as pd
  2. # because I don't want the calling site to store df, I decided to keep it as a static variable here
  3. def daily_csv_loader(filepath):
  4. daily_csv_loader.df = pd.read_csv(filepath)
  5. def query(time, df):
  6. # return rows from df corresponding to time

使用

  1. import datetime
  2. path = "/path/to/csv/file/filename.csv"
  3. time = datetime.datetime(year=2020, month=1, day=1, hour=12, minute=2, second=0)
  4. daily_csv_loader(filepath)
  5. query(time, daily_csv_loader.)

这里有没有其他的方法,最好是函数方法(我不想像前面提到的那样在这里使用OOP)?有没有一种函数式方法可以用一个函数来完成,也许是用嵌套的函数?

ffscu2ro

ffscu2ro1#

一个函数式的例子,使用嵌套函数,不会在调用站点暴露底层的csv/dataframe对象,最终通过单个函数调用:

  1. import csv
  2. import datetime as dt
  3. import io
  4. from typing import Any, Dict
  5. def load_queryable_csv(csv_str):
  6. # a quick & easy setup, could have used pandas instead and/or generalized parsing
  7. rows = csv.DictReader(io.StringIO(csv_str))
  8. parsed_rows = [
  9. {'a': row['a'], 'b': int(row['b']), 'time': dt.datetime.fromisoformat(row['time'])}
  10. for row in rows
  11. ]
  12. # "querying"
  13. def query_csv(**column_matchers):
  14. def matcher(row: Dict[str, Any]):
  15. return all(row[col] == val for col, val in column_matchers.items())
  16. return list(filter(matcher, parsed_rows))
  17. return query_csv
  18. query_csv = load_queryable_csv("""\
  19. a,b,time
  20. x,2,2020-01-01 12:02:00
  21. y,4,2020-01-01 12:02:01
  22. """)
  23. time = dt.datetime(year=2020, month=1, day=1, hour=12, minute=2, second=0)
  24. query_csv(time=time)
  25. # => [{'a': 'x', 'b': 2, 'time': datetime.datetime(2020, 1, 1, 12, 2)}]
  26. query_csv(a='y')
  27. # => [{'a': 'y', 'b': 4, 'time': datetime.datetime(2020, 1, 1, 12, 2, 1)}]
展开查看全部

相关问题