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

sdnqo3pr 于 2023-05-20 发布在其他

关注(0)|答案(1)|浏览(106)

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

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

使用方法：

import datetime

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

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

import pandas as pd

# because I don't want the calling site to store df, I decided to keep it as a static variable here
def daily_csv_loader(filepath):
  daily_csv_loader.df = pd.read_csv(filepath)

def query(time, df):
  # return rows from df corresponding to time

使用

import datetime

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

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

csv

来源：https://stackoverflow.com/questions/76275068/how-to-query-from-a-csv-repeatedly-throughout-the-day-in-a-simulation-code

1条答案

按热度按时间

ffscu2ro1#

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

import csv
import datetime as dt
import io
from typing import Any, Dict

def load_queryable_csv(csv_str):
    # a quick & easy setup, could have used pandas instead and/or generalized parsing
    rows = csv.DictReader(io.StringIO(csv_str))
    parsed_rows = [
        {'a': row['a'], 'b': int(row['b']), 'time': dt.datetime.fromisoformat(row['time'])}
        for row in rows
    ]

    # "querying"
    def query_csv(**column_matchers):
        def matcher(row: Dict[str, Any]):
            return all(row[col] == val for col, val in column_matchers.items())

        return list(filter(matcher, parsed_rows))

    return query_csv

query_csv = load_queryable_csv("""\
a,b,time
x,2,2020-01-01 12:02:00
y,4,2020-01-01 12:02:01
""")

time = dt.datetime(year=2020, month=1, day=1, hour=12, minute=2, second=0)

query_csv(time=time)
# => [{'a': 'x', 'b': 2, 'time': datetime.datetime(2020, 1, 1, 12, 2)}]

query_csv(a='y')
# => [{'a': 'y', 'b': 4, 'time': datetime.datetime(2020, 1, 1, 12, 2, 1)}]

赞(0）回复(0）举报 2023-05-20

我来回答

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

1条答案

相关问题

热门标签

最新问答